"""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)