FabledScribe/tests/test_integration_forge_keyring.py

"""Real-Postgres integration tests for the forge keyring (#2778).

Runs only in the CI integration lane (real Postgres, schema built by
`alembic upgrade head`, which includes migration 0078's forge_connections
table and the projects.forge_connection_id pin). This exercises what the
unit tests cannot: the own-rows ACL on connection CRUD, host-keyed
resolution against stored rows, the pin's only-that-connection semantics,
the owner-only pin eligibility, and the admin-only scoping of the env
fallback — every one of which is a cross-user isolation property, and
isolation properties are exactly what mocks cannot prove.
"""
from unittest.mock import patch

import pytest
import pytest_asyncio

from scribe.config import Config
from scribe.models import async_session
from scribe.models.project import Project
from scribe.services.forge import get_forges
from scribe.services.forge_connections import (
    create_connection,
    delete_connection,
    list_connections,
    set_project_pin,
    update_connection,
)
from tests.helpers import ensure_user

pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]

GITEA = "https://git.example.com"
GITHUB = "https://github.com"


@pytest_asyncio.fixture
async def seeded():
    """An owner with a project, plus an unrelated user and an admin."""
    async with async_session() as s:
        owner = await ensure_user(s, "keyring_owner")
        other = await ensure_user(s, "keyring_other")
        admin = await ensure_user(s, "keyring_admin", role="admin")
        project = Project(user_id=owner.id, title="Keyring project")
        s.add(project)
        await s.flush()
        ids = {
            "owner": owner.id, "other": other.id, "admin": admin.id,
            "project": project.id,
        }
        await s.commit()
    # Start each test from a clean keyring — the fixture users persist
    # across tests in one lane run.
    for uid in (ids["owner"], ids["other"], ids["admin"]):
        for row in await list_connections(uid):
            await delete_connection(uid, row.id)
    return ids


@pytest.mark.integration
async def test_crud_is_own_rows_only(seeded):
    owner, other = seeded["owner"], seeded["other"]
    row = await create_connection(owner, kind="gitea", base_url=GITEA, token="tok")
    assert row.host == "git.example.com"

    # One row per (user, host) IS the resolution model.
    with pytest.raises(ValueError):
        await create_connection(owner, kind="gitea", base_url=GITEA + "/", token="t2")
    # ...but the same host under ANOTHER user is that user's own business.
    theirs = await create_connection(other, kind="gitea", base_url=GITEA, token="t3")

    # Another user can neither see, edit, nor delete it.
    assert [r.id for r in await list_connections(other)] == [theirs.id]
    assert await update_connection(other, row.id, token="stolen") is None
    assert await delete_connection(other, row.id) is False

    updated = await update_connection(owner, row.id, base_url=GITEA + ":3000")
    assert updated is not None and updated.base_url.endswith(":3000")
    assert await delete_connection(owner, row.id) is True
    assert await delete_connection(other, theirs.id) is True


@pytest.mark.integration
async def test_resolution_runs_on_the_owners_keyring_only(seeded):
    owner, other = seeded["owner"], seeded["other"]
    await create_connection(owner, kind="gitea", base_url=GITEA, token="tok")

    mine = await get_forges(owner)
    assert mine.configured
    hit = mine.resolve(f"{GITEA}/alice/widget.git")
    assert hit is not None and hit[1] == "alice/widget"

    # The other user's reads never ride the owner's token.
    assert not (await get_forges(other)).configured


@pytest.mark.integration
async def test_pin_means_only_that_connection(seeded):
    owner, project = seeded["owner"], seeded["project"]
    gitea = await create_connection(owner, kind="gitea", base_url=GITEA, token="t1")
    await create_connection(owner, kind="github", base_url=GITHUB, token="t2")

    # Unpinned: both hosts resolve from the keyring.
    selector = await get_forges(owner, project)
    assert selector.resolve(f"{GITEA}/a/b") is not None
    assert selector.resolve(f"{GITHUB}/a/b") is not None

    assert await set_project_pin(owner, project, gitea.id) is True
    pinned = await get_forges(owner, project)
    assert pinned.resolve(f"{GITEA}/a/b") is not None
    # The pin is exclusive: a host the pinned connection can't serve reads as
    # unserved, exactly like the documented repo-not-on-this-forge state.
    assert pinned.resolve(f"{GITHUB}/a/b") is None

    assert await set_project_pin(owner, project, None) is True
    assert (await get_forges(owner, project)).resolve(f"{GITHUB}/a/b") is not None


@pytest.mark.integration
async def test_pin_only_accepts_the_owners_connections(seeded):
    owner, other, project = seeded["owner"], seeded["other"], seeded["project"]
    theirs = await create_connection(other, kind="gitea", base_url=GITEA, token="t")

    # A collaborator's token can never be attached to someone else's project —
    # the confused-deputy channel this feature exists to close.
    assert await set_project_pin(owner, project, theirs.id) is False
    # And only the owner's projects accept a pin at all.
    assert await set_project_pin(other, project, theirs.id) is False


@pytest.mark.integration
async def test_stale_pin_after_ownership_transfer_is_ignored_not_honored(seeded):
    owner, other, project = seeded["owner"], seeded["other"], seeded["project"]
    gitea = await create_connection(owner, kind="gitea", base_url=GITEA, token="t")
    assert await set_project_pin(owner, project, gitea.id) is True

    async with async_session() as s:
        proj = await s.get(Project, project)
        proj.user_id = other
        await s.commit()
    try:
        # The pin now names a connection the (new) owner does not hold: it
        # must fall back to the new owner's keyring — empty — never keep
        # reading with the previous owner's token.
        assert not (await get_forges(other, project)).configured
    finally:
        async with async_session() as s:
            proj = await s.get(Project, project)
            proj.user_id = owner
            proj.forge_connection_id = None
            await s.commit()


@pytest.mark.integration
async def test_env_config_serves_admins_only_and_rows_beat_it(seeded):
    owner, admin = seeded["owner"], seeded["admin"]
    with patch.object(Config, "FORGE_KIND", "gitea"), \
            patch.object(Config, "FORGE_BASE_URL", GITEA), \
            patch.object(Config, "FORGE_TOKEN", "env-tok"):
        # The env entry is the OPERATOR's token: admin projects only.
        assert (await get_forges(admin)).configured
        assert not (await get_forges(owner)).configured

        # A stored row for the same host wins — the UI writes rows, and a UI
        # edit that silently lost to an env var would look like a broken form.
        row = await create_connection(
            admin, kind="gitea", base_url=GITEA, token="row-tok"
        )
        selector = await get_forges(admin)
        assert len(selector.adapters) == 1
        assert selector.adapters[0]._token == "row-tok"
        await delete_connection(admin, row.id)