from __future__ import annotations import unittest from datetime import datetime, timedelta, timezone from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_access.backend.db.models import Account, Group, User from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.db.base import Base from govoplan_core.privacy.dsar_workflow import ( DataSubjectRequest, create_data_subject_request, execute_data_subject_erasure, plan_data_subject_erasure, search_data_subject_request, ) from govoplan_files.backend.db.models import ( FileAsset, FileBlob, FileConnectorCredential, FileConnectorPolicy, FileConnectorProfile, FileConnectorSpace, FileFolder, FileFormEvidenceGrant, FileIntegrityFinding, FileIntegrityScan, FileShare, FileVersion, ) from govoplan_files.backend.dsar_provider import ( FILES_DSAR_CAPABILITY, FilesDsarProvider, ) from govoplan_files.backend.manifest import manifest class _Registry: def __init__(self, provider: FilesDsarProvider, *, files_active: bool = True) -> None: self.provider = provider self.files_active = files_active def capability_names(self): return (FILES_DSAR_CAPABILITY,) def capability_owner(self, name): self._assert_capability(name) return "files" def tenant_entitlement_resolver(self): files_active = self.files_active class _Resolver: @staticmethod def resolve(session, tenant_id): del session, tenant_id return type( "State", (), {"effective_modules": ("files",) if files_active else ()}, )() return _Resolver() def require_tenant_capability(self, name, session, **kwargs): del session, kwargs self._assert_capability(name) return self.provider def manifests(self): return (type("Manifest", (), {"id": "files"})(),) @staticmethod def _assert_capability(name: str) -> None: if name != FILES_DSAR_CAPABILITY: raise KeyError(name) class FilesDsarProviderTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:", future=True) Base.metadata.create_all( bind=self.engine, tables=[ Account.__table__, User.__table__, Group.__table__, ChangeSequenceEntry.__table__, DataSubjectRequest.__table__, FileBlob.__table__, FileAsset.__table__, FileVersion.__table__, FileFolder.__table__, FileFormEvidenceGrant.__table__, FileShare.__table__, FileConnectorCredential.__table__, FileConnectorPolicy.__table__, FileConnectorProfile.__table__, FileConnectorSpace.__table__, FileIntegrityScan.__table__, FileIntegrityFinding.__table__, ], ) self.session = sessionmaker(bind=self.engine, future=True)() self.account = Account( id="account-1", email="subject@example.test", normalized_email="subject@example.test", display_name="Subject", password_hash="not-exported", ) self.user = User( id="membership-1", tenant_id="tenant-1", account_id=self.account.id, email="subject@example.test", display_name="Subject", ) self.blob = FileBlob( id="blob-1", tenant_id="tenant-1", storage_backend="local", storage_key="private/storage/key-do-not-export", checksum_sha256="a" * 64, size_bytes=12, ref_count=1, ) self.asset = FileAsset( id="asset-1", tenant_id="tenant-1", owner_type="user", owner_user_id=self.user.id, created_by_user_id=self.user.id, current_version_id="version-1", display_path="subjects/private.txt", filename="private.txt", description="Subject-provided document", retained_until=datetime.now(timezone.utc) + timedelta(days=30), lifecycle_reason="Pending proceeding", metadata_={"password": "metadata-secret-do-not-export"}, ) self.version = FileVersion( id="version-1", tenant_id="tenant-1", file_asset_id=self.asset.id, blob_id=self.blob.id, version_number=1, filename_at_upload=self.asset.filename, display_path_at_upload=self.asset.display_path, content_type="text/plain", size_bytes=12, checksum_sha256="a" * 64, created_by_user_id=self.user.id, ) self.folder = FileFolder( id="folder-1", tenant_id="tenant-1", owner_type="user", owner_user_id=self.user.id, path="subjects", created_by_user_id=self.user.id, ) self.share = FileShare( id="share-1", tenant_id="tenant-1", file_asset_id=self.asset.id, target_type="user", target_id=self.user.id, permission="read", created_by_user_id=self.user.id, ) self.form_evidence = FileFormEvidenceGrant( id="evidence-1", tenant_id="tenant-1", form_instance_id="form-instance-1", form_definition_id="application", form_definition_revision="7", token_sha256="b" * 64, idempotency_key="evidence-key-1", request_sha256="c" * 64, custodian_user_id=self.user.id, evidence_kind="attachment", purpose="Submitted application evidence", status="uploaded", expires_at=datetime.now(timezone.utc) + timedelta(days=1), max_size_bytes=1024, allowed_content_types=["text/plain"], file_asset_id=self.asset.id, file_version_id=self.version.id, metadata_={}, ) self.credential = FileConnectorCredential( id="credential-1", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", label="Subject-created credential", provider="s3", credential_mode="database", username="subject-user", password_encrypted="encrypted-password-do-not-export", token_encrypted="encrypted-token-do-not-export", password_env="PASSWORD_ENV_DO_NOT_EXPORT", secret_ref="vault://do-not-export", created_by_user_id=self.user.id, updated_by_user_id=self.user.id, ) self.integrity_scan = FileIntegrityScan( id="scan-1", tenant_id="tenant-1", storage_backend="local", storage_prefix="private-prefix-do-not-export", created_by_user_id=self.user.id, ) tenant_two_asset = FileAsset( id="asset-tenant-2", tenant_id="tenant-2", owner_type="user", owner_user_id=self.user.id, display_path="other-tenant.txt", filename="other-tenant.txt", ) self.session.add_all( [ self.account, self.user, self.blob, self.asset, self.version, self.folder, self.share, self.form_evidence, self.credential, self.integrity_scan, tenant_two_asset, ] ) self.session.commit() self.provider = FilesDsarProvider() self.subject = DsarSubjectRef(membership_id=self.user.id) def tearDown(self) -> None: self.session.close() self.engine.dispose() def test_manifest_publishes_protocol_conforming_provider(self) -> None: provided_names = {item.name for item in manifest.provides_interfaces} self.assertIn(FILES_DSAR_CAPABILITY, provided_names) provider = manifest.capability_factories[FILES_DSAR_CAPABILITY](None) self.assertIsInstance(provider, DsarProvider) def test_search_is_tenant_scoped_and_excludes_bytes_and_secrets(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self.subject, ) resource_types = {record.resource_type for record in records} self.assertTrue( { "file_asset", "file_version", "file_folder", "file_share", "file_form_evidence", "connector_credential", "file_integrity_scan", }.issubset(resource_types) ) serialized = repr([record.to_dict() for record in records]) self.assertNotIn("asset-tenant-2", serialized) self.assertNotIn("private/storage/key-do-not-export", serialized) self.assertNotIn("metadata-secret-do-not-export", serialized) self.assertNotIn("encrypted-password-do-not-export", serialized) self.assertNotIn("encrypted-token-do-not-export", serialized) self.assertNotIn("subject-user", serialized) self.assertNotIn("PASSWORD_ENV_DO_NOT_EXPORT", serialized) self.assertNotIn("vault://do-not-export", serialized) self.assertNotIn("private-prefix-do-not-export", serialized) def test_conflicting_direct_subject_references_fail_closed(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=DsarSubjectRef( membership_id=self.user.id, external_references={"files.user": "another-membership"}, ), ) self.assertEqual((), records) def test_plan_classifies_retention_and_manual_review(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self.subject, ) actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=self.subject, records=records, ) kinds = {action.kind for action in actions} self.assertTrue({"retain", "manual_review", "revoke", "detach"}.issubset(kinds)) asset_retention = next( action for action in actions if action.action_id == "files:retain:file_asset:asset-1" ) self.assertIn("retained until", asset_retention.rationale) self.assertTrue( any(action.action_id == "files:revoke:file_share:share-1" for action in actions) ) self.assertFalse( any(action.kind == "delete" and action.executable for action in actions) ) def test_execution_is_revalidated_and_idempotent(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self.subject, ) actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=self.subject, records=records, ) executable = tuple(action for action in actions if action.executable) first = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self.subject, actions=executable, request_id="dsar-1", ) self.assertEqual({"executed"}, {result.status for result in first}) self.assertIsNotNone(self.share.revoked_at) self.assertIsNone(self.asset.created_by_user_id) self.assertIsNone(self.version.created_by_user_id) self.assertIsNone(self.folder.created_by_user_id) self.assertIsNone(self.credential.created_by_user_id) self.assertIsNone(self.credential.updated_by_user_id) repeated = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self.subject, actions=executable, request_id="dsar-1", ) self.assertEqual({"unchanged"}, {result.status for result in repeated}) def test_execution_blocks_when_reference_changed_after_planning(self) -> None: records = self.provider.search_subject( self.session, tenant_id="tenant-1", subject=self.subject, ) actions = self.provider.plan_erasure( self.session, tenant_id="tenant-1", subject=self.subject, records=records, ) detach = next( action for action in actions if action.action_id == "files:detach:file_version:created_by_user_id:version-1" ) self.version.created_by_user_id = "replacement-user" self.session.flush() result = self.provider.execute_erasure( self.session, tenant_id="tenant-1", subject=self.subject, actions=(detach,), request_id="dsar-2", ) self.assertEqual("blocked", result[0].status) self.assertEqual("replacement-user", self.version.created_by_user_id) def test_core_workflow_discovers_active_provider_and_skips_it_when_disabled( self, ) -> None: request = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-FILES-1", request_kind="access_and_erasure", subject=self.subject, purpose="Respond to an authorized privacy request.", legal_basis="Article 15 and 17 GDPR", due_at=None, requested_by_account_id="privacy-officer", ) self.session.commit() registry = _Registry(self.provider) search_data_subject_request( self.session, registry=registry, row=request, expected_revision=1, ) self.assertEqual( "searched", request.status, request.search_result["provider_runs"] ) self.assertEqual(["files"], request.coverage["covered_modules"]) self.assertEqual([], request.coverage["modules_without_provider"]) plan_data_subject_erasure( self.session, registry=registry, row=request, expected_revision=2, ) executable_ids = [ action["action_id"] for action in request.erasure_plan["actions"] if action["executable"] ] execute_data_subject_erasure( self.session, registry=registry, row=request, expected_revision=3, action_ids=executable_ids, ) self.assertEqual("completed", request.status) disabled = create_data_subject_request( self.session, tenant_id="tenant-1", reference="DSAR-FILES-DISABLED", request_kind="access", subject=self.subject, purpose="Verify disabled-module coverage.", legal_basis="Article 15 GDPR", due_at=None, requested_by_account_id="privacy-officer", ) search_data_subject_request( self.session, registry=_Registry(self.provider, files_active=False), row=disabled, expected_revision=1, ) self.assertEqual(0, disabled.search_result["record_count"]) self.assertEqual( [FILES_DSAR_CAPABILITY], disabled.coverage["inactive_provider_capabilities"], ) if __name__ == "__main__": unittest.main()