tests/test_tenant_foundation.py

from __future__ import annotations

from pathlib import Path
from uuid import uuid4

import pytest

from app.container import build_container
from app.core.config import Settings
from app.security.assets import CanonicalAssetNotFoundError
from app.security.schemas import APIKeyCreate
from app.social.models import SocialAccount


def foundation_settings(tmp_path: Path) -> Settings:
    return Settings(
        _env_file=None,
        database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}",
        social_database_url=f"sqlite+aiosqlite:///{tmp_path / 'social.db'}",
        social_auto_migrate=True,
        social_worker_enabled=False,
        auth_enabled=True,
        social_oauth_encryption_key="test-only-encryption-material",
        temp_dir=tmp_path / "temp",
        output_dir=tmp_path / "outputs",
        cleanup_interval_seconds=3600,
        whisper_model="tiny",
    )


async def _create_key(container, name: str, **kwargs: str):
    return await container.api_keys.create(
        APIKeyCreate(
            name=name,
            environment="test",
            role=None,
            scopes=["operations:execute", "operations:read"],
        ),
        created_by="tests",
        **kwargs,
    )


@pytest.mark.asyncio
async def test_api_key_is_resolved_to_persisted_membership_and_rotation_preserves_it(
    tmp_path: Path,
) -> None:
    container = build_container(foundation_settings(tmp_path))
    await container.security_database.initialize()
    try:
        first, first_secret = await _create_key(container, "First")
        first_context = await container.api_keys.authenticate(first_secret)
        assert first_context.workspace_id and first_context.user_id
        assert first_context.workspace_id != first.id
        assert first_context.user_id != first.id

        sibling, sibling_secret = await _create_key(
            container,
            "Sibling",
            workspace_id=first_context.workspace_id,
            user_id=first_context.user_id,
        )
        sibling_context = await container.api_keys.authenticate(sibling_secret)
        assert sibling_context.workspace_id == first_context.workspace_id
        assert sibling_context.user_id == first_context.user_id

        isolated, isolated_secret = await _create_key(container, "Isolated")
        isolated_context = await container.api_keys.authenticate(isolated_secret)
        assert isolated_context.workspace_id != first_context.workspace_id

        rotated, rotated_secret = await container.api_keys.rotate(
            first.id, 0, created_by="tests"
        )
        rotated_context = await container.api_keys.authenticate(rotated_secret)
        assert rotated_context.workspace_id == first_context.workspace_id
        assert rotated_context.user_id == first_context.user_id
    finally:
        await container.security_database.close()


@pytest.mark.asyncio
async def test_canonical_asset_cannot_be_claimed_by_another_workspace(tmp_path: Path) -> None:
    container = build_container(foundation_settings(tmp_path))
    await container.security_database.initialize()
    try:
        _, secret_a = await _create_key(container, "A")
        _, secret_b = await _create_key(container, "B")
        context_a = await container.api_keys.authenticate(secret_a)
        context_b = await container.api_keys.authenticate(secret_b)
        request_id = str(uuid4())
        output = container.settings.output_dir / request_id
        output.mkdir(parents=True)
        path = output / "asset.mp4"
        path.write_bytes(b"owned output")

        asset = await container.assets.register_output(
            workspace_id=str(context_a.workspace_id),
            user_id=context_a.user_id,
            request_id=request_id,
            path=path,
            mime_type="video/mp4",
        )
        assert asset.workspace_id == context_a.workspace_id
        owned = await container.assets.get_owned(
            workspace_id=str(context_a.workspace_id),
            request_id=request_id,
            filename=path.name,
        )
        assert owned.id == asset.id
        with pytest.raises(CanonicalAssetNotFoundError):
            await container.assets.get_owned(
                workspace_id=str(context_b.workspace_id),
                request_id=request_id,
                filename=path.name,
            )
        path.write_bytes(b"tampered")
        with pytest.raises(CanonicalAssetNotFoundError):
            await container.assets.verify_file(asset, path)
    finally:
        await container.security_database.close()


@pytest.mark.asyncio
async def test_legacy_social_rows_are_adopted_without_reusing_api_key_tenant_id(
    tmp_path: Path,
) -> None:
    container = build_container(foundation_settings(tmp_path))
    await container.security_database.initialize()
    await container.social.initialize()
    try:
        key, secret = await _create_key(container, "Legacy")
        context = await container.api_keys.authenticate(secret)
        legacy_account = await container.social.accounts.repository.create(
            SocialAccount(
                workspace_id=key.id,
                provider="youtube",
                account_type="channel",
                external_account_id="legacy-channel",
                status="connected",
            )
        )
        await container.social.adopt_legacy_workspaces(
            await container.tenants.list_principals()
        )
        adopted = await container.social.accounts.repository.get(
            str(context.workspace_id), legacy_account.id
        )
        assert adopted.workspace_id == context.workspace_id
    finally:
        await container.social.close()
        await container.security_database.close()