govoplan-files/tests/test_access_provider.py

from __future__ import annotations

import unittest
from datetime import datetime, timedelta, timezone

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",
            ),
            FileShare(
                id="share-campaign-expired",
                tenant_id=TENANT_ID,
                file_asset_id=assets[2].id,
                target_type="campaign",
                target_id="campaign-1",
                permission="read",
                expires_at=datetime.now(timezone.utc) - timedelta(seconds=1),
            ),
        ])
        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()),
    )