coulomb/repo-manager
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(identifiers): verify batch projections

Browse source 
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
This commit is contained in:
tegwick 2026-08-22 23:50:34 +02:00
parent e6cc18bf18
commit 055c6971ab
11 changed files with 2964 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

88
tests/test_identifiers.py
View file

@ -3,6 +3,7 @@ from __future__ import annotations
import subprocess
from pathlib import Path
import httpx
import pytest
from repo_manager.identifiers import (
@ -213,10 +214,24 @@ def _push_fixture_to_upstream(repo: Path, remote: Path) -> None:
subprocess.run(["git", "push", "-u", "origin", "HEAD:main"], cwd=repo, check=True)
def test_migration_batch_plan_pins_clean_synchronized_repo(tmp_path: Path) -> None:
def test_migration_batch_plan_pins_clean_synchronized_repo(
tmp_path: Path, monkeypatch
) -> None:
repo = tmp_path / "one"
path = repo / "workplans" / "one.md"
_workplan(path, "ONE-WP-0001", "active")
path.write_text(
path.read_text(encoding="utf-8")
.replace(
"status: active\n---",
'status: active\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---',
)
.replace(
"status: todo\n```",
'status: todo\nstate_hub_task_id: "22222222-2222-4222-8222-222222222222"\n```',
),
encoding="utf-8",
)
subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
subprocess.run(["git", "add", "."], cwd=repo, check=True, capture_output=True)
subprocess.run(
@ -228,15 +243,84 @@ def test_migration_batch_plan_pins_clean_synchronized_repo(tmp_path: Path) -> No
_push_fixture_to_upstream(repo, tmp_path / "remote.git")
plan = plan_identifier_migration(tmp_path, "helixforge")
batch = plan_identifier_migration_batch(plan, repo_slugs=["one"])
current_uuids = {
mapping["current_uuid"] for mapping in plan["repositories"][0]["mappings"]
}
def projection_get(self, url):
status = 200 if any(str(url).endswith(str(value)) for value in current_uuids) else 404
return httpx.Response(status, request=httpx.Request("GET", url))
monkeypatch.setattr(httpx.Client, "get", projection_get)
batch = plan_identifier_migration_batch(
plan,
repo_slugs=["one"],
projection_api_bases=["http://hub-one.test", "http://hub-two.test"],
)
assert batch["ok"] is True
assert batch["ready_for_approval"] is True
assert batch["apply_authorized"] is False
assert batch["approval_required"] is True
assert batch["projection_api_bases"] == [
"http://hub-one.test",
"http://hub-two.test",
]
assert batch["repositories"][0]["projection_preflight"]["ok"] is True
assert batch["repositories"][0]["git_preflight"]["upstream"] == "origin/main"
assert len(batch["batch_sha256"]) == 64
def projection_drift(self, url):
return httpx.Response(404, request=httpx.Request("GET", url))
monkeypatch.setattr(httpx.Client, "get", projection_drift)
verification = verify_identifier_migration_batch(batch, plan=plan)
assert verification["ok"] is False
assert verification["ready_for_decision"] is False
def test_migration_batch_plan_rejects_projection_gap(tmp_path: Path, monkeypatch) -> None:
repo = tmp_path / "one"
path = repo / "workplans" / "one.md"
_workplan(path, "ONE-WP-0001", "active")
path.write_text(
path.read_text(encoding="utf-8")
.replace(
"status: active\n---",
'status: active\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---',
)
.replace(
"status: todo\n```",
'status: todo\nstate_hub_task_id: "22222222-2222-4222-8222-222222222222"\n```',
),
encoding="utf-8",
)
subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
subprocess.run(["git", "add", "."], cwd=repo, check=True, capture_output=True)
subprocess.run(
["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-m", "seed"],
cwd=repo,
check=True,
capture_output=True,
)
_push_fixture_to_upstream(repo, tmp_path / "remote.git")
plan = plan_identifier_migration(tmp_path, "helixforge")
def projection_get(self, url):
return httpx.Response(404, request=httpx.Request("GET", url))
monkeypatch.setattr(httpx.Client, "get", projection_get)
batch = plan_identifier_migration_batch(
plan,
repo_slugs=["one"],
projection_api_bases=["http://hub.test"],
)
assert batch["ok"] is False
assert batch["ready_for_approval"] is False
assert batch["repositories"][0]["projection_preflight"]["ok"] is False
assert any("current UUID=200" in error["reason"] for error in batch["errors"])
def test_migration_batch_plan_rejects_dirty_or_duplicate_scope(tmp_path: Path) -> None:
repo = tmp_path / "one"