govoplan-files/tests/test_share_lifecycle.py

from __future__ import annotations

import unittest
from datetime import timedelta
from unittest.mock import patch

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from govoplan_access.backend.db.models import Account, Group, User
from govoplan_core.core.change_sequence import ChangeSequenceEntry
from govoplan_core.db.base import Base
from govoplan_files.backend.db.models import FileAsset, FileShare
from govoplan_files.backend.storage.common import FileStorageError, utcnow
from govoplan_files.backend.storage.files import (
    get_asset_for_user,
    list_file_shares,
    revoke_file_share,
    share_file,
)


TENANT_ID = "tenant-1"
OWNER_ID = "owner-1"
RECIPIENT_ID = "recipient-1"
GROUP_ID = "group-1"


class FileShareLifecycleTests(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__,
                FileAsset.__table__,
                FileShare.__table__,
            ],
        )
        self.session = sessionmaker(bind=self.engine, future=True)()
        self.asset = FileAsset(
            id="file-1",
            tenant_id=TENANT_ID,
            owner_type="user",
            owner_user_id=OWNER_ID,
            display_path="shared.pdf",
            filename="shared.pdf",
        )
        self.session.add(self.asset)
        self.session.commit()

    def tearDown(self) -> None:
        self.session.close()
        self.engine.dispose()

    @patch(
        "govoplan_files.backend.storage.files.user_group_ids",
        return_value=[],
    )
    def test_expired_share_stops_access_immediately(self, _groups) -> None:
        self.session.add(
            FileShare(
                id="expired-share",
                tenant_id=TENANT_ID,
                file_asset_id=self.asset.id,
                target_type="user",
                target_id=RECIPIENT_ID,
                permission="read",
                expires_at=utcnow() - timedelta(seconds=1),
            )
        )
        self.session.commit()

        with self.assertRaisesRegex(FileStorageError, "No access"):
            get_asset_for_user(
                self.session,
                tenant_id=TENANT_ID,
                user_id=RECIPIENT_ID,
                asset_id=self.asset.id,
            )
        self.assertEqual(
            [],
            list_file_shares(
                self.session,
                tenant_id=TENANT_ID,
                asset_id=self.asset.id,
            ),
        )

    @patch(
        "govoplan_files.backend.storage.files.user_group_ids",
        return_value=[GROUP_ID],
    )
    def test_independent_active_grant_survives_other_expiry(self, _groups) -> None:
        self.session.add_all(
            [
                FileShare(
                    id="expired-user-share",
                    tenant_id=TENANT_ID,
                    file_asset_id=self.asset.id,
                    target_type="user",
                    target_id=RECIPIENT_ID,
                    permission="read",
                    expires_at=utcnow() - timedelta(seconds=1),
                ),
                FileShare(
                    id="active-group-share",
                    tenant_id=TENANT_ID,
                    file_asset_id=self.asset.id,
                    target_type="group",
                    target_id=GROUP_ID,
                    permission="read",
                    expires_at=utcnow() + timedelta(hours=1),
                ),
            ]
        )
        self.session.commit()

        result = get_asset_for_user(
            self.session,
            tenant_id=TENANT_ID,
            user_id=RECIPIENT_ID,
            asset_id=self.asset.id,
        )

        self.assertEqual(self.asset.id, result.id)
        self.assertEqual(
            ["active-group-share"],
            [
                share.id
                for share in list_file_shares(
                    self.session,
                    tenant_id=TENANT_ID,
                    asset_id=self.asset.id,
                )
            ],
        )

    @patch(
        "govoplan_files.backend.storage.files.ensure_share_target_exists",
        return_value=None,
    )
    def test_revoke_is_idempotent_and_future_expiry_is_persisted(
        self, _target_exists
    ) -> None:
        expiry = utcnow() + timedelta(days=1)
        share = share_file(
            self.session,
            tenant_id=TENANT_ID,
            asset=self.asset,
            target_type="user",
            target_id=RECIPIENT_ID,
            permission="read",
            user_id=OWNER_ID,
            expires_at=expiry,
        )
        self.session.commit()

        revoked, first_changed = revoke_file_share(
            self.session,
            tenant_id=TENANT_ID,
            asset_id=self.asset.id,
            share_id=share.id,
            user_id=OWNER_ID,
        )
        self.session.commit()
        repeated, second_changed = revoke_file_share(
            self.session,
            tenant_id=TENANT_ID,
            asset_id=self.asset.id,
            share_id=share.id,
            user_id=OWNER_ID,
        )

        self.assertTrue(first_changed)
        self.assertFalse(second_changed)
        self.assertEqual(OWNER_ID, revoked.revoked_by_user_id)
        self.assertEqual(revoked.revoked_at, repeated.revoked_at)
        self.assertEqual([], list_file_shares(
            self.session,
            tenant_id=TENANT_ID,
            asset_id=self.asset.id,
        ))
        self.assertEqual(
            [share.id],
            [
                item.id
                for item in list_file_shares(
                    self.session,
                    tenant_id=TENANT_ID,
                    asset_id=self.asset.id,
                    include_inactive=True,
                )
            ],
        )

    @patch(
        "govoplan_files.backend.storage.files.ensure_share_target_exists",
        return_value=None,
    )
    def test_past_expiry_is_rejected(self, _target_exists) -> None:
        with self.assertRaisesRegex(FileStorageError, "future"):
            share_file(
                self.session,
                tenant_id=TENANT_ID,
                asset=self.asset,
                target_type="user",
                target_id=RECIPIENT_ID,
                permission="read",
                user_id=OWNER_ID,
                expires_at=utcnow() - timedelta(seconds=1),
            )


if __name__ == "__main__":
    unittest.main()