feat(files): expose governed tabular content

tests/test_manifest_documentation.py

@@ -21,6 +21,7 @@ STATIC_TOPIC_IDS = {
     "files.records.exact-version-source",
     "files.forms-runtime.managed-evidence",
     "files.postbox.exact-version-references",
+    "files.tabular-content",
     "files.assurance.process-and-release-readiness",
 }
 RUNTIME_TOPIC_IDS = {

tests/test_tabular_content_capability.py

@@ -0,0 +1,175 @@
+from __future__ import annotations
+
+import unittest
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.files import (
+    ManagedTabularFileAccessError,
+    ManagedTabularFileValidationError,
+)
+from govoplan_files.backend.capabilities import FilesManagedTabularFileService
+from govoplan_files.backend.storage.common import FileStorageError
+
+
+def principal(
+    tenant_id: str = "tenant-1",
+    *,
+    scopes: tuple[str, ...] = (
+        "files:file:read",
+        "files:file:download",
+        "files:file:admin",
+    ),
+) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id=tenant_id,
+            scopes=frozenset(scopes),
+        ),
+        account=object(),
+        user=SimpleNamespace(id="user-1"),
+    )
+
+
+class FilesManagedTabularContentTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        self.session = sessionmaker(bind=self.engine)()
+        self.blob = SimpleNamespace(
+            id="blob-1",
+            tenant_id="tenant-1",
+            storage_backend="local",
+            storage_key="tenants/tenant-1/files/blob-1",
+            checksum_sha256="a" * 64,
+            size_bytes=8,
+            content_type="text/csv",
+            integrity_status="verified",
+        )
+        self.asset = SimpleNamespace(
+            id="asset-1",
+            tenant_id="tenant-1",
+            owner_type="user",
+            owner_user_id="user-1",
+            current_version_id="version-1",
+            display_path="Imports/source.csv",
+            filename="source.csv",
+            description=None,
+            updated_at=datetime(2026, 8, 21, tzinfo=UTC),
+        )
+        self.version = SimpleNamespace(
+            id="version-1",
+            tenant_id="tenant-1",
+            file_asset_id="asset-1",
+            blob_id="blob-1",
+            version_number=1,
+            filename_at_upload="source.csv",
+            display_path_at_upload="Imports/source.csv",
+            content_type="text/csv",
+            size_bytes=8,
+            checksum_sha256="a" * 64,
+            created_by_user_id="user-1",
+            created_at=datetime(2026, 8, 21, tzinfo=UTC),
+        )
+        self.provider = FilesManagedTabularFileService()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_lists_and_reads_only_exact_authorized_tabular_versions(self) -> None:
+        with (
+            patch(
+                "govoplan_files.backend.capabilities.list_recent_assets_for_user",
+                return_value=[self.asset],
+            ),
+            patch(
+                "govoplan_files.backend.capabilities.current_version_and_blob",
+                return_value=(self.version, self.blob),
+            ),
+            patch(
+                "govoplan_files.backend.capabilities.get_asset_for_user",
+                return_value=self.asset,
+            ),
+            patch.object(
+                self.session,
+                "get",
+                side_effect=lambda model, object_id: {
+                    "version-1": self.version,
+                    "blob-1": self.blob,
+                }.get(object_id),
+            ),
+            patch(
+                "govoplan_files.backend.capabilities.read_asset_version_bytes",
+                return_value=(b"id\n1\n", self.version, self.blob),
+            ) as read_bytes,
+            patch("govoplan_files.backend.capabilities.audit_from_principal"),
+        ):
+            listed = self.provider.list_tabular_files(self.session, principal())
+            result = self.provider.read_tabular_file(
+                self.session,
+                principal(),
+                file_asset_id="asset-1",
+                file_version_id="version-1",
+                max_bytes=100,
+            )
+
+        self.assertEqual(("version-1",), tuple(item.file_version_id for item in listed))
+        self.assertEqual(b"id\n1\n", result.payload)
+        self.assertTrue(result.file.current_version)
+        read_bytes.assert_called_once()
+
+    def test_tenant_scope_permissions_and_pre_read_size_limit_fail_closed(self) -> None:
+        with patch(
+            "govoplan_files.backend.capabilities.get_asset_for_user",
+            side_effect=FileStorageError("File not found"),
+        ):
+            self.assertIsNone(
+                self.provider.get_tabular_file(
+                    self.session,
+                    principal("tenant-2"),
+                    file_asset_id="asset-1",
+                )
+            )
+        with self.assertRaises(ManagedTabularFileAccessError):
+            self.provider.list_tabular_files(
+                self.session,
+                principal(scopes=()),
+            )
+        with (
+            patch(
+                "govoplan_files.backend.capabilities.get_asset_for_user",
+                return_value=self.asset,
+            ),
+            patch.object(
+                self.session,
+                "get",
+                side_effect=lambda model, object_id: {
+                    "version-1": self.version,
+                    "blob-1": self.blob,
+                }.get(object_id),
+            ),
+            patch(
+                "govoplan_files.backend.capabilities.read_asset_version_bytes"
+            ) as read_bytes,
+            self.assertRaises(ManagedTabularFileValidationError),
+        ):
+            self.provider.read_tabular_file(
+                self.session,
+                principal(),
+                file_asset_id="asset-1",
+                file_version_id="version-1",
+                max_bytes=4,
+            )
+        read_bytes.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()