bambuddy/backend/tests/integration/test_read_permission_backfill_migration.py

"""Migration tests for maziggy/bambuddy-security #2 — read permission OWN/ALL backfill.

Pre-fix, ARCHIVES_READ / LIBRARY_READ / QUEUE_READ were flat "read all" flags.
Post-fix they split into OWN/ALL. The migration in seed_default_groups must:

  1. Rename legacy `archives:read` etc to `archives:read_all` on Administrators
     and to `archives:read_own` on every other role (fail-closed default).
  2. Backfill `_own` AND `_all` variants for the Administrators group on upgrade
     so an upgraded install matches a fresh install's permission set.
  3. Backfill `_own` variants for Operators and Viewers so they keep read access
     even if their stored row didn't carry the legacy flag.

These regressions are the failure shape Maziggy hit on a live upgrade — the
admin role ended up missing queue:read_own AND queue:read after migration.
"""

import pytest
from httpx import AsyncClient
from sqlalchemy import select

from backend.app.core import database as _database_module
from backend.app.core.database import seed_default_groups
from backend.app.models.group import Group

_READ_FLAGS = frozenset(
    {
        "archives:read",
        "archives:read_own",
        "archives:read_all",
        "library:read",
        "library:read_own",
        "library:read_all",
        "queue:read",
        "queue:read_own",
        "queue:read_all",
    }
)


async def _strip_and_set(group_name: str, extra: list[str] | None = None) -> None:
    """Strip every read flag from ``group_name`` then add ``extra`` flags.

    Simulates a pre-migration state where the group either had only the
    legacy flat permission (set ``extra=['archives:read']``) or no read
    permission at all (set ``extra=None``).
    """
    async with _database_module.async_session() as session:
        grp = (await session.execute(select(Group).where(Group.name == group_name))).scalar_one_or_none()
        assert grp is not None, f"group {group_name} not pre-seeded"
        stripped = [p for p in (grp.permissions or []) if p not in _READ_FLAGS]
        stripped.extend(extra or [])
        grp.permissions = stripped
        await session.commit()


async def _get_perms(group_name: str) -> set[str]:
    async with _database_module.async_session() as session:
        grp = (await session.execute(select(Group).where(Group.name == group_name))).scalar_one_or_none()
        assert grp is not None
        return set(grp.permissions or [])


# Note: ``async_client`` is depended upon (even though unused) so pytest-asyncio
# uses the same event loop the conftest fixture uses for async_session(). Without
# it, calling ``async_session()`` twice in one test trips an asyncpg
# "got Future attached to a different loop" RuntimeError.


class TestReadPermissionMigration:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_legacy_archives_read_renamed_to_all_for_administrators(self, async_client: AsyncClient):
        """Existing Administrators group with legacy `archives:read` → gets
        `archives:read_all` after seed_default_groups runs, and gets the
        `_own` companion backfilled too."""
        await seed_default_groups()
        await _strip_and_set("Administrators", extra=["archives:read"])

        await seed_default_groups()

        perms = await _get_perms("Administrators")
        # Rename happened: legacy renamed to _all
        assert "archives:read_all" in perms
        # Backfill also added _own so fresh install and upgraded install match
        assert "archives:read_own" in perms

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_administrators_backfill_adds_all_six_read_flags(self, async_client: AsyncClient):
        """Even with NO legacy flags present, Administrators ends up with both
        OWN and ALL variants for archives / library / queue after the backfill
        pass. This is the case Maziggy hit — admin missing `queue:read_own`
        after upgrade."""
        await seed_default_groups()
        await _strip_and_set("Administrators")

        await seed_default_groups()

        perms = await _get_perms("Administrators")
        for needed in (
            "archives:read_own",
            "archives:read_all",
            "library:read_own",
            "library:read_all",
            "queue:read_own",
            "queue:read_all",
        ):
            assert needed in perms, f"{needed} must be backfilled for Administrators"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_operators_backfill_adds_own_read_flags(self, async_client: AsyncClient):
        """Operators with no read flags get the _OWN variants backfilled
        (fail-closed — no _ALL)."""
        await seed_default_groups()
        await _strip_and_set("Operators")

        await seed_default_groups()

        perms = await _get_perms("Operators")
        assert "archives:read_own" in perms
        assert "library:read_own" in perms
        assert "queue:read_own" in perms
        assert "archives:read_all" not in perms
        assert "library:read_all" not in perms
        assert "queue:read_all" not in perms

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_operators_legacy_archives_read_renamed_to_own(self, async_client: AsyncClient):
        """Pre-PR Operators with legacy `archives:read` get the _OWN rename
        (fail-closed — close the IDOR, the operator can re-request _ALL via
        admin if cross-user visibility is genuinely needed)."""
        await seed_default_groups()
        await _strip_and_set("Operators", extra=["archives:read"])

        await seed_default_groups()

        perms = await _get_perms("Operators")
        assert "archives:read_own" in perms
        assert "archives:read_all" not in perms

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_administrators_legacy_archives_read_retained(self, async_client: AsyncClient):
        """Admin keeps the LEGACY `archives:read` flag — the frontend gates
        download / preview UI on it (ArchivesPage / FileManagerPage), and
        removing it on rename was leaving admin with no visible download
        buttons after upgrade. The new API gates use the _ALL variant which
        the backfill also ensures is present."""
        await seed_default_groups()
        await _strip_and_set("Administrators", extra=["archives:read"])

        await seed_default_groups()

        perms = await _get_perms("Administrators")
        # Both the legacy flag (for the UI) and the _all variant (for the API)
        # must coexist on admin.
        assert "archives:read" in perms
        assert "archives:read_all" in perms
        assert "archives:read_own" in perms

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_administrators_backfill_adds_legacy_read_flags(self, async_client: AsyncClient):
        """Admin with NO read flags at all (hand-edited or stripped role) ends
        up with the legacy `archives:read` / `queue:read` / `library:read`
        backfilled — so the UI gates work — alongside the OWN/ALL split."""
        await seed_default_groups()
        await _strip_and_set("Administrators")

        await seed_default_groups()

        perms = await _get_perms("Administrators")
        for needed in (
            "archives:read",
            "library:read",
            "queue:read",
            "archives:read_own",
            "archives:read_all",
            "library:read_own",
            "library:read_all",
            "queue:read_own",
            "queue:read_all",
        ):
            assert needed in perms, f"{needed} must be backfilled for Administrators"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_administrators_orca_cloud_auth_backfilled(self, async_client: AsyncClient):
        """Admin without `orca_cloud:auth` (older custom edit) gets it
        backfilled — matches the fresh-install default."""
        await seed_default_groups()
        async with _database_module.async_session() as session:
            grp = (await session.execute(select(Group).where(Group.name == "Administrators"))).scalar_one()
            grp.permissions = [p for p in (grp.permissions or []) if p != "orca_cloud:auth"]
            await session.commit()

        await seed_default_groups()

        perms = await _get_perms("Administrators")
        assert "orca_cloud:auth" in perms

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_operators_orca_cloud_auth_backfilled(self, async_client: AsyncClient):
        """Operators on upgraded installs get `orca_cloud:auth` backfilled
        (the new default — needed for the Slice modal's Orca Cloud preset
        picker)."""
        await seed_default_groups()
        async with _database_module.async_session() as session:
            grp = (await session.execute(select(Group).where(Group.name == "Operators"))).scalar_one()
            grp.permissions = [p for p in (grp.permissions or []) if p != "orca_cloud:auth"]
            await session.commit()

        await seed_default_groups()

        perms = await _get_perms("Operators")
        assert "orca_cloud:auth" in perms

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_administrators_printer_sensor_history_read_backfilled(self, async_client: AsyncClient):
        """Admin without `printer_sensor_history:read` (older custom edit or
        a DB seeded before that permission existed) gets it backfilled —
        regression for the gap maziggy hit on a live install where the
        per-permission admin backfills missed it."""
        await seed_default_groups()
        async with _database_module.async_session() as session:
            grp = (await session.execute(select(Group).where(Group.name == "Administrators"))).scalar_one()
            grp.permissions = [p for p in (grp.permissions or []) if p != "printer_sensor_history:read"]
            await session.commit()

        await seed_default_groups()

        perms = await _get_perms("Administrators")
        assert "printer_sensor_history:read" in perms

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_administrators_sync_covers_every_current_permission(self, async_client: AsyncClient):
        """Generic invariant: ALL_PERMISSIONS sync ensures every Permission
        enum value is present on the Administrators group, no matter what
        was stripped pre-backfill. Catches every future "new permission
        missing on upgrade" regression without needing a one-off test."""
        from backend.app.core.permissions import ALL_PERMISSIONS

        await seed_default_groups()
        # Wipe the admin group's permission list entirely and force the sync
        # to put everything back.
        async with _database_module.async_session() as session:
            grp = (await session.execute(select(Group).where(Group.name == "Administrators"))).scalar_one()
            grp.permissions = []
            await session.commit()

        await seed_default_groups()

        perms = await _get_perms("Administrators")
        missing = [p for p in ALL_PERMISSIONS if p not in perms]
        assert not missing, f"Administrators missing permissions after backfill: {missing}"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_administrators_sync_is_additive_only(self, async_client: AsyncClient):
        """The sync block must never remove a permission an operator added by
        hand — only add missing entries from ALL_PERMISSIONS."""
        await seed_default_groups()
        async with _database_module.async_session() as session:
            grp = (await session.execute(select(Group).where(Group.name == "Administrators"))).scalar_one()
            grp.permissions = [*(grp.permissions or []), "custom:plugin_permission"]
            await session.commit()

        await seed_default_groups()

        perms = await _get_perms("Administrators")
        assert "custom:plugin_permission" in perms

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_viewers_do_not_get_orca_cloud_auth(self, async_client: AsyncClient):
        """Viewers stay read-only — orca_cloud:auth is not added by the
        backfill (matches the fresh-install Viewers bootstrap, which
        intentionally excludes cloud-auth permissions)."""
        await seed_default_groups()
        async with _database_module.async_session() as session:
            grp = (await session.execute(select(Group).where(Group.name == "Viewers"))).scalar_one()
            grp.permissions = [p for p in (grp.permissions or []) if p != "orca_cloud:auth"]
            await session.commit()

        await seed_default_groups()

        perms = await _get_perms("Viewers")
        assert "orca_cloud:auth" not in perms