CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 14s
CI & Build / TypeScript typecheck (push) Successful in 1m13s
CI & Build / integration (push) Successful in 1m15s
CI & Build / Python tests (push) Successful in 2m8s
CI & Build / Build & push image (push) Successful in 28s
A binding and every row of its project's code-shape ledger are keyed by the repo's remote. A rename on the forge changes the remote, so binding the new name started an empty ledger beside the old one and left every judgment behind. bind_repo now takes moved_from, the old remote. Its binding and the project's ledger rows move to the new key in one transaction, and the reply says how many shapes moved. The move is refused when: - the old name isn't bound to that project; - the new name is already bound; - the project already has a ledger under the new name. Each refusal leaves both names exactly as they were. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
103 lines
4.1 KiB
Python
103 lines
4.1 KiB
Python
"""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"],
|
|
)
|