from __future__ import annotations from datetime import UTC, datetime, timedelta from io import BytesIO from types import SimpleNamespace import unittest from unittest.mock import patch from fastapi import UploadFile from sqlalchemy import create_engine from sqlalchemy.orm import Session from starlette.datastructures import Headers from govoplan_access.backend.db.models import Account, User from govoplan_core.core.access import ( CAPABILITY_ACCESS_DIRECTORY, AccessSubjectRef, UserRef, ) from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.core.form_evidence import ( FormEvidenceContractError, FormEvidenceGrantRequest, FormEvidenceInspectionRequest, ) from govoplan_core.core.institutional import EvidenceReference, InstitutionalReference from govoplan_core.db.base import Base from govoplan_files.backend.db.models import ( FileAsset, FileBlob, FileFormEvidenceGrant, FileVersion, ) from govoplan_files.backend.form_evidence import FilesFormEvidenceProvider from govoplan_files.backend.routes.form_evidence import upload_form_evidence class _Directory: def __init__(self, user: UserRef) -> None: self._user = user def get_account(self, account_id: str): del account_id return None def get_user(self, user_id: str): return self._user if user_id == self._user.id else None def get_users(self, user_ids): return {user_id: self._user for user_id in user_ids if user_id == self._user.id} def users_for_tenant(self, tenant_id: str): return (self._user,) if tenant_id == self._user.tenant_id else () def get_group(self, group_id: str): del group_id return None def get_groups(self, group_ids): del group_ids return {} def groups_for_tenant(self, tenant_id: str): del tenant_id return () def groups_for_user(self, user_id: str, *, tenant_id: str): del user_id, tenant_id return () def display_label(self, subject: AccessSubjectRef): del subject return self._user.display_name class _Registry: def __init__(self, directory: _Directory) -> None: self._directory = directory def has_capability(self, name: str) -> bool: return name == CAPABILITY_ACCESS_DIRECTORY def require_capability(self, name: str): if not self.has_capability(name): raise KeyError(name) return self._directory class FilesFormEvidenceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all( self.engine, tables=[ Account.__table__, User.__table__, FileBlob.__table__, FileAsset.__table__, FileVersion.__table__, FileFormEvidenceGrant.__table__, ChangeSequenceEntry.__table__, ], ) self.session = Session(self.engine) user_ref = UserRef( id="user-1", account_id="account-1", tenant_id="tenant-1", email="user@example.test", display_name="Evidence Custodian", ) self.provider = FilesFormEvidenceProvider( _Registry(_Directory(user_ref)), SimpleNamespace(file_upload_max_bytes=2_000_000), ) self.principal = SimpleNamespace(tenant_id="tenant-1") self.definition_ref = InstitutionalReference( kind="form", owner_module="forms", object_id="form-1", tenant_id="tenant-1", version="3", ) def tearDown(self) -> None: self.session.close() self.engine.dispose() def request(self, **overrides: object) -> FormEvidenceGrantRequest: values: dict[str, object] = { "tenant_id": "tenant-1", "instance_id": "instance-1", "definition_ref": self.definition_ref, "evidence_kind": "document", "purpose": "supporting document", "idempotency_key": "grant-1", "expires_at": datetime.now(UTC) + timedelta(hours=1), "custodian_ref": "user:user-1", "max_size_bytes": 1_000_000, "allowed_content_types": ("application/pdf",), } values.update(overrides) return FormEvidenceGrantRequest(**values) # type: ignore[arg-type] def test_grant_persists_only_token_hash_and_replays_without_secret(self) -> None: request = self.request() issued = self.provider.create_upload_grant( self.session, self.principal, request=request, ) row = self.session.query(FileFormEvidenceGrant).one() self.assertIsNotNone(issued.upload_token) self.assertNotEqual(issued.upload_token, row.token_sha256) self.assertLessEqual( issued.expires_at, datetime.now(UTC) + timedelta(minutes=15, seconds=1), ) replay = self.provider.create_upload_grant( self.session, self.principal, request=request, ) self.assertTrue(replay.replayed) self.assertIsNone(replay.upload_token) self.assertEqual(issued.grant_id, replay.grant_id) def test_grant_capacity_counts_outstanding_uploads_but_allows_replay(self) -> None: request = self.request(metadata={"remaining_attachments": 1}) self.provider.create_upload_grant(self.session, self.principal, request=request) replay = self.provider.create_upload_grant( self.session, self.principal, request=request, ) self.assertTrue(replay.replayed) with self.assertRaisesRegex( FormEvidenceContractError, "maximum number of active attachment uploads", ): self.provider.create_upload_grant( self.session, self.principal, request=self.request( idempotency_key="grant-2", metadata={"remaining_attachments": 1}, ), ) def test_idempotency_key_cannot_be_reused_for_another_request(self) -> None: self.provider.create_upload_grant( self.session, self.principal, request=self.request(), ) with self.assertRaisesRegex( FormEvidenceContractError, "idempotency conflict", ): self.provider.create_upload_grant( self.session, self.principal, request=self.request(purpose="another purpose"), ) def test_inspection_accepts_only_exact_verified_managed_version(self) -> None: issued = self.provider.create_upload_grant( self.session, self.principal, request=self.request(), ) grant = self.session.get(FileFormEvidenceGrant, issued.grant_id) assert grant is not None blob = FileBlob( id="blob-1", tenant_id="tenant-1", storage_backend="local", storage_key="tenant-1/blob-1", checksum_sha256="a" * 64, size_bytes=100, integrity_status="verified", ) asset = FileAsset( id="asset-1", tenant_id="tenant-1", owner_type="user", owner_user_id="user-1", current_version_id="version-1", display_path="Form submissions/instance-1/evidence.pdf", filename="evidence.pdf", ) version = FileVersion( id="version-1", tenant_id="tenant-1", file_asset_id="asset-1", blob_id="blob-1", version_number=1, filename_at_upload="evidence.pdf", display_path_at_upload=asset.display_path, content_type="application/pdf", size_bytes=100, checksum_sha256="a" * 64, ) self.session.add_all((blob, asset, version)) grant.status = "uploaded" grant.file_asset_id = asset.id grant.file_version_id = version.id self.session.flush() reference = EvidenceReference( kind="document", owner_module="files", evidence_id=asset.id, tenant_id="tenant-1", version=version.id, checksum="a" * 64, ) accepted = self.provider.inspect_evidence( self.session, self.principal, request=FormEvidenceInspectionRequest( tenant_id="tenant-1", instance_id="instance-1", definition_ref=self.definition_ref, evidence=reference, purpose="final submission", final=True, ), ) self.assertTrue(accepted.accepted) rejected = self.provider.inspect_evidence( self.session, self.principal, request=FormEvidenceInspectionRequest( tenant_id="tenant-1", instance_id="another-instance", definition_ref=self.definition_ref, evidence=reference, purpose="final submission", final=True, ), ) self.assertEqual("rejected", rejected.state) def test_public_upload_consumes_grant_and_returns_exact_evidence(self) -> None: issued = self.provider.create_upload_grant( self.session, self.principal, request=self.request(), ) assert issued.upload_token is not None stored = SimpleNamespace( asset=SimpleNamespace(id="asset-uploaded"), version=SimpleNamespace( id="version-uploaded", checksum_sha256="c" * 64, size_bytes=8, ), ) upload = UploadFile( filename="evidence.pdf", file=BytesIO(b"evidence"), headers=Headers({"content-type": "application/pdf"}), ) with ( patch( "govoplan_files.backend.routes.form_evidence.create_file_asset", return_value=stored, ) as create, patch("govoplan_files.backend.routes.form_evidence.emit_platform_event"), ): response = upload_form_evidence( file=upload, x_form_evidence_token=issued.upload_token, session=self.session, ) grant = self.session.get(FileFormEvidenceGrant, issued.grant_id) assert grant is not None self.assertEqual("uploaded", grant.status) self.assertEqual("asset-uploaded", grant.file_asset_id) self.assertEqual("version-uploaded", grant.file_version_id) self.assertEqual("asset-uploaded", response["evidence"]["evidence_id"]) self.assertEqual(b"evidence", create.call_args.kwargs["data"]) self.assertEqual("user-1", create.call_args.kwargs["owner_id"]) if __name__ == "__main__": unittest.main()