diff --git a/src/grimoire/mcp/tools/repos.py b/src/grimoire/mcp/tools/repos.py index db87598f..ff6255f0 100644 --- a/src/grimoire/mcp/tools/repos.py +++ b/src/grimoire/mcp/tools/repos.py @@ -13,7 +13,9 @@ from grimoire.services import projects as projects_svc from grimoire.services import repo_bindings as repo_bindings_svc -async def bind_repo(repo_url: str, project_id: int, ref: str = "") -> dict: +async def bind_repo( + repo_url: str, project_id: int, ref: str = "", moved_from: str = "", +) -> dict: """Bind a git repository to a Grimoire project for session-start context. After this, any session started in that repo auto-loads the project's @@ -27,21 +29,34 @@ async def bind_repo(repo_url: str, project_id: int, ref: str = "") -> dict: the push, not the merge. Re-binding with ref="" keeps the standing ref; pass ref="-" to clear it back to the default branch. + A repo renamed on its forge has a new remote, and its ledger is stored + under the old one. Pass the old remote as `moved_from` and the binding and + every shape judgment move to the new name together. Binding the new name + without it starts an empty ledger beside the old one. + Args: repo_url: the repo's git remote (e.g. the output of `git remote get-url origin` — ssh or https form, both work). project_id: the Grimoire project this repo represents. ref: branch the ledger follows ("" = leave as is / default branch on a new binding; "-" = clear to the default branch). + moved_from: the repo's remote before a rename on the forge. The + old name must be bound to this project, and the new one must be + neither bound nor holding a ledger of this project's. """ uid = current_user_id() project = await projects_svc.get_project(uid, project_id) if project is None: raise ValueError(f"project {project_id} not found") ref_arg = None if not ref else ("" if ref.strip() == "-" else ref) + moved = None + if moved_from: + _, moved = await repo_bindings_svc.move_binding( + uid, moved_from, repo_url, project_id, + ) binding = await repo_bindings_svc.set_binding(uid, repo_url, project_id, ref_arg) follows = binding.ref or "the default branch" - return { + out = { "repo_key": binding.repo_key, "project_id": binding.project_id, "project_title": project.title, @@ -51,6 +66,12 @@ async def bind_repo(repo_url: str, project_id: int, ref: str = "") -> dict: f"the ledger follows {follows}." ), } + if moved is not None: + old_key = repo_bindings_svc.normalize_repo_key(moved_from) + out["moved_from"] = old_key + out["shapes_moved"] = moved + out["message"] += f" Moved from `{old_key}` with {moved} shape(s)." + return out async def list_repo_bindings() -> dict: diff --git a/src/grimoire/services/repo_bindings.py b/src/grimoire/services/repo_bindings.py index 63a62ffe..90bb10f8 100644 --- a/src/grimoire/services/repo_bindings.py +++ b/src/grimoire/services/repo_bindings.py @@ -9,9 +9,10 @@ from __future__ import annotations import re -from sqlalchemy import select +from sqlalchemy import select, update from grimoire.models import async_session +from grimoire.models.code_shape import CodeShape from grimoire.models.repo_binding import RepoBinding @@ -100,6 +101,58 @@ async def set_binding( return binding +async def move_binding( + user_id: int, old_repo: str, new_repo: str, project_id: int, +) -> tuple[RepoBinding, int]: + """Carry a renamed repo's binding, and its project's shape ledger, from the + old remote to the new one. Returns the binding and how many shapes moved. + + Every ledger row is keyed by the repo it was found in, so a repo renamed on + its forge would otherwise start an empty ledger under the new name and + leave every judgment behind under the old one. Both keys change in one + transaction. It refuses rather than merge: the old key must be bound to + this project by the caller, and the new key must have neither a binding of + the caller's nor ledger rows of this project's. + """ + old_key = normalize_repo_key(old_repo) + new_key = normalize_repo_key(new_repo) + if not old_key or not new_key: + raise ValueError("both remotes must be parseable repo URLs") + if old_key == new_key: + raise ValueError(f"`{old_key}` is the same repo under both names") + async with async_session() as session: + mine = select(RepoBinding).where(RepoBinding.user_id == user_id) + binding = ( + await session.execute(mine.where(RepoBinding.repo_key == old_key)) + ).scalar_one_or_none() + if binding is None or binding.project_id != project_id: + raise ValueError(f"`{old_key}` is not bound to project {project_id}") + if ( + await session.execute(mine.where(RepoBinding.repo_key == new_key)) + ).scalar_one_or_none() is not None: + raise ValueError( + f"`{new_key}` is already bound; unbind it first, then move" + ) + shapes_here = select(CodeShape.id).where(CodeShape.project_id == project_id) + if ( + await session.execute( + shapes_here.where(CodeShape.repo_key == new_key).limit(1) + ) + ).first() is not None: + raise ValueError( + f"project {project_id} already has a ledger under `{new_key}`" + ) + moved = await session.execute( + update(CodeShape) + .where(CodeShape.project_id == project_id, CodeShape.repo_key == old_key) + .values(repo_key=new_key) + ) + binding.repo_key = new_key + await session.commit() + await session.refresh(binding) + return binding, moved.rowcount or 0 + + async def list_bindings(user_id: int) -> list[RepoBinding]: async with async_session() as session: rows = await session.execute( diff --git a/tests/test_integration_repo_move.py b/tests/test_integration_repo_move.py new file mode 100644 index 00000000..0d9b1f88 --- /dev/null +++ b/tests/test_integration_repo_move.py @@ -0,0 +1,102 @@ +"""Real-Postgres integration tests for moving a renamed repo's binding. + +A repo renamed on its forge has a new remote. The binding and every ledger row +are keyed by the remote, so the move has to carry both, in one transaction, and +refuse anything that would merge two ledgers or move another project's repo. +""" +import pytest +import pytest_asyncio +from sqlalchemy import select + +from grimoire.models import async_session +from grimoire.models.code_shape import CodeShape +from grimoire.models.project import Project +from grimoire.services import repo_bindings as svc +from grimoire.services.shape_ledger import classify_shapes, sync_repo_shapes +from tests.helpers import ensure_user + +pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")] + +OLD = "https://git.example.com/alice/OldName.git" +NEW = "git@git.example.com:alice/NewName.git" +OLD_KEY = "git.example.com/alice/oldname" +NEW_KEY = "git.example.com/alice/newname" +SHAPES = [("src/app.py", "sym", "make_app"), ("src/util.py", "sym", "helper")] + + +@pytest_asyncio.fixture +async def bound(): + """A project bound under the old name, its two-shape ledger, one judged.""" + async with async_session() as s: + owner = await ensure_user(s, "repo_move_owner") + project = Project(user_id=owner.id, title="Renamed repo") + other = Project(user_id=owner.id, title="Some other project") + s.add_all([project, other]) + await s.flush() + ids = {"owner": owner.id, "pid": project.id, "other": other.id} + await s.commit() + + await svc.set_binding(ids["owner"], OLD, ids["pid"], "dev") + await sync_repo_shapes(ids["pid"], OLD_KEY, SHAPES, seen_marker="dev") + await classify_shapes(ids["owner"], ids["pid"], [ + {"path": "src/util.py", "symbol": "helper", "status": "exempt", + "reason": "one caller"}, + ]) + yield ids + await svc.delete_binding(ids["owner"], OLD) + await svc.delete_binding(ids["owner"], NEW) + + +async def _ledger(pid: int) -> dict[str, tuple[str, str]]: + async with async_session() as s: + rows = await s.execute(select(CodeShape).where(CodeShape.project_id == pid)) + return {r.symbol: (r.repo_key, r.status) for r in rows.scalars()} + + +@pytest.mark.integration +async def test_the_binding_and_every_judgment_move_to_the_new_name(bound): + binding, moved = await svc.move_binding(bound["owner"], OLD, NEW, bound["pid"]) + + assert moved == 2 + assert binding.repo_key == NEW_KEY + assert binding.project_id == bound["pid"] + assert binding.ref == "dev" + assert await svc.resolve_project(bound["owner"], NEW) == bound["pid"] + assert await svc.resolve_project(bound["owner"], OLD) is None + assert await _ledger(bound["pid"]) == { + "make_app": (NEW_KEY, "unclassified"), + "helper": (NEW_KEY, "exempt"), + } + + +@pytest.mark.integration +async def test_it_refuses_a_repo_bound_to_another_project(bound): + with pytest.raises(ValueError, match="not bound to project"): + await svc.move_binding(bound["owner"], OLD, NEW, bound["other"]) + assert {k for k, _ in (await _ledger(bound["pid"])).values()} == {OLD_KEY} + + +@pytest.mark.integration +async def test_it_refuses_when_the_new_name_is_already_bound(bound): + await svc.set_binding(bound["owner"], NEW, bound["pid"]) + with pytest.raises(ValueError, match="already bound"): + await svc.move_binding(bound["owner"], OLD, NEW, bound["pid"]) + assert {k for k, _ in (await _ledger(bound["pid"])).values()} == {OLD_KEY} + + +@pytest.mark.integration +async def test_it_refuses_to_merge_into_a_ledger_the_new_name_already_has(bound): + await sync_repo_shapes( + bound["pid"], NEW_KEY, [("src/new.py", "sym", "fresh")], seen_marker="dev", + ) + with pytest.raises(ValueError, match="already has a ledger"): + await svc.move_binding(bound["owner"], OLD, NEW, bound["pid"]) + assert await svc.resolve_project(bound["owner"], OLD) == bound["pid"] + + +@pytest.mark.integration +async def test_the_same_repo_under_both_names_is_refused(bound): + with pytest.raises(ValueError, match="same repo"): + await svc.move_binding( + bound["owner"], OLD, "git@git.example.com:alice/oldname", bound["pid"], + ) diff --git a/tests/test_pattern_coverage.py b/tests/test_pattern_coverage.py index 6d4630be..f42ec226 100644 --- a/tests/test_pattern_coverage.py +++ b/tests/test_pattern_coverage.py @@ -376,6 +376,13 @@ def test_bind_repo_tool_takes_a_ref(): assert "ref" in tool.parameters.get("properties", {}) +def test_bind_repo_tool_can_move_a_renamed_repo(): + """A repo renamed on its forge carries its ledger to the new name.""" + from grimoire.mcp.server import build_mcp_server + tool = build_mcp_server()._tool_manager.get_tool("bind_repo") + assert "moved_from" in tool.parameters.get("properties", {}) + + def test_coverage_routes_are_registered(): from grimoire.app import create_app