govoplan-files/tests/test_tabular_content_capability.py

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()