from __future__ import annotations import unittest from sqlalchemy import create_engine, text from sqlalchemy.orm import sessionmaker from govoplan_access.backend.db.models import Account, Group, User from govoplan_core.core.access import PrincipalRef from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.db.base import Base from govoplan_files.backend.capabilities import FilesAccessService, virtual_folder_resource_id from govoplan_files.backend.db.models import FileAsset, FileFolder, FileShare from govoplan_files.backend.storage.files import count_assets_for_user, list_assets_for_user TENANT_ID = "tenant-1" USER_ID = "user-1" OTHER_USER_ID = "user-2" GROUP_ID = "group-1" class FilesAccessProviderTests(unittest.TestCase): def test_file_access_provider_explains_owner_share_admin_and_missing_resources(self) -> None: session = _session() self.addCleanup(_close_session, session) _seed_access_subjects(session) owned = FileAsset(id="file-owned", tenant_id=TENANT_ID, owner_type="user", owner_user_id=USER_ID, display_path="owned.pdf", filename="owned.pdf") shared = FileAsset(id="file-shared", tenant_id=TENANT_ID, owner_type="user", owner_user_id=OTHER_USER_ID, display_path="shared.pdf", filename="shared.pdf") session.add_all([ owned, shared, FileShare(id="share-group", tenant_id=TENANT_ID, file_asset_id=shared.id, target_type="group", target_id=GROUP_ID, permission="read"), ]) session.commit() service = FilesAccessService() owned_items = service.explain_resource_provenance(session, _principal(), resource_type="file", resource_id=owned.id, action="files:file:read") shared_items = service.explain_resource_provenance(session, _principal(group_ids={GROUP_ID}), resource_type="file", resource_id=shared.id, action="files:file:read") admin_items = service.explain_resource_provenance(session, _principal(scopes={"files:file:admin"}), resource_type="file", resource_id=shared.id, action="files:file:read") missing_items = service.explain_resource_provenance(session, _principal(), resource_type="file", resource_id="missing-file", action="files:file:read") self.assertTrue(any(item.kind == "resource" and item.source == "files.file" and item.id == owned.id for item in owned_items)) self.assertTrue(any(item.kind == "owner" and item.id == USER_ID for item in owned_items)) self.assertTrue(any(item.kind == "share" and item.id == "share-group" for item in shared_items)) self.assertTrue(any(item.kind == "policy" and item.id == "files:file:admin" for item in admin_items)) self.assertEqual("files.not_found", missing_items[0].source) self.assertIs(missing_items[0].details["found"], False) def test_file_access_provider_explains_persisted_and_virtual_folders(self) -> None: session = _session() self.addCleanup(_close_session, session) _seed_access_subjects(session) persisted = FileFolder(id="folder-persisted", tenant_id=TENANT_ID, owner_type="group", owner_group_id=GROUP_ID, path="records") virtual_child = FileAsset( id="file-in-virtual-folder", tenant_id=TENANT_ID, owner_type="group", owner_group_id=GROUP_ID, display_path="inferred/sub/file.pdf", filename="file.pdf", ) session.add_all([persisted, virtual_child]) session.commit() service = FilesAccessService() principal = _principal(group_ids={GROUP_ID}) persisted_items = service.explain_resource_provenance(session, principal, resource_type="folder", resource_id=persisted.id, action="files:file:read") virtual_id = virtual_folder_resource_id(tenant_id=TENANT_ID, owner_type="group", owner_id=GROUP_ID, path="inferred/sub") virtual_items = service.explain_resource_provenance(session, principal, resource_type="folder", resource_id=virtual_id, action="files:file:read") missing_virtual_id = virtual_folder_resource_id(tenant_id=TENANT_ID, owner_type="group", owner_id=GROUP_ID, path="inferred/missing") missing_items = service.explain_resource_provenance(session, principal, resource_type="folder", resource_id=missing_virtual_id, action="files:file:read") self.assertTrue(any(item.kind == "resource" and item.source == "files.folder" and item.id == persisted.id for item in persisted_items)) self.assertTrue(any(item.kind == "owner" and item.id == GROUP_ID for item in persisted_items)) self.assertTrue(any(item.kind == "resource" and item.source == "files.virtual_folder" and item.id == virtual_id for item in virtual_items)) self.assertTrue(any(item.kind == "owner" and item.id == GROUP_ID for item in virtual_items)) self.assertEqual("files.not_found", missing_items[0].source) def test_file_property_filters_cover_campaign_and_audit_usage(self) -> None: session = _session() self.addCleanup(_close_session, session) _seed_access_subjects(session) assets = [ FileAsset( id=f"file-{index}", tenant_id=TENANT_ID, owner_type="user", owner_user_id=USER_ID, display_path=f"{index}.pdf", filename=f"{index}.pdf", ) for index in range(1, 4) ] session.add_all([ *assets, FileShare( id="share-campaign", tenant_id=TENANT_ID, file_asset_id=assets[0].id, target_type="campaign", target_id="campaign-1", permission="read", ), ]) session.commit() session.execute( text( "INSERT INTO campaign_attachment_uses " "(id, tenant_id, file_asset_id, use_stage) " "VALUES (:id, :tenant_id, :file_asset_id, :use_stage)" ), { "id": "attachment-use-1", "tenant_id": TENANT_ID, "file_asset_id": assets[1].id, "use_stage": "sent", }, ) session.commit() linked = list_assets_for_user( session, tenant_id=TENANT_ID, user_id=USER_ID, owner_type="user", owner_id=USER_ID, campaign_usage="linked", ) unlinked = list_assets_for_user( session, tenant_id=TENANT_ID, user_id=USER_ID, owner_type="user", owner_id=USER_ID, campaign_usage="unlinked", ) audit_relevant = list_assets_for_user( session, tenant_id=TENANT_ID, user_id=USER_ID, owner_type="user", owner_id=USER_ID, audit_relevant=True, ) self.assertEqual([asset.id for asset in linked], ["file-1", "file-2"]) self.assertEqual([asset.id for asset in unlinked], ["file-3"]) self.assertEqual([asset.id for asset in audit_relevant], ["file-2"]) self.assertEqual( count_assets_for_user( session, tenant_id=TENANT_ID, user_id=USER_ID, owner_type="user", owner_id=USER_ID, campaign_usage="unlinked", ), 1, ) def _session(): engine = create_engine("sqlite:///:memory:", future=True) Base.metadata.create_all( bind=engine, tables=[ Account.__table__, User.__table__, Group.__table__, ChangeSequenceEntry.__table__, FileAsset.__table__, FileFolder.__table__, FileShare.__table__, ], ) with engine.begin() as connection: connection.execute( text( "CREATE TABLE campaign_attachment_uses (" "id VARCHAR(36) PRIMARY KEY, " "tenant_id VARCHAR(36) NOT NULL, " "file_asset_id VARCHAR(36) NOT NULL, " "use_stage VARCHAR(20) NOT NULL" ")" ) ) return sessionmaker(bind=engine, future=True)() def _close_session(session) -> None: engine = session.get_bind() session.close() engine.dispose() def _seed_access_subjects(session) -> None: session.add_all([ Account(id="account-1", email="one@example.test", normalized_email="one@example.test"), Account(id="account-2", email="two@example.test", normalized_email="two@example.test"), User(id=USER_ID, tenant_id=TENANT_ID, account_id="account-1", email="one@example.test"), User(id=OTHER_USER_ID, tenant_id=TENANT_ID, account_id="account-2", email="two@example.test"), Group(id=GROUP_ID, tenant_id=TENANT_ID, slug="group", name="Group"), ]) session.commit() def _principal(*, scopes: set[str] | None = None, group_ids: set[str] | None = None) -> PrincipalRef: return PrincipalRef( account_id="account-1", membership_id=USER_ID, tenant_id=TENANT_ID, scopes=frozenset(scopes or {"files:file:read"}), group_ids=frozenset(group_ids or set()), )