tegwick
9f0a104b56
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a049a4-ee9f-78e1-9d66-2cb0f9bea3e3
931 lines
35 KiB
Python
931 lines
35 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import delete, func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from api.config import settings
|
|
from api.main import app
|
|
from api.models.managed_repo import ManagedRepo
|
|
from api.models.repository_rename import RepositoryRenameOperation, RepositorySlug
|
|
from api.models.agent_message import AgentMessage
|
|
from api.models.decision import Decision
|
|
from api.models.sbom_entry import Ecosystem, SBOMEntry
|
|
from api.models.sbom_snapshot import SBOMSnapshot
|
|
from api.models.token_event import TokenEvent
|
|
from api.models.workplan import Workplan
|
|
from api.routers.repository_renames import repository_rename_events
|
|
from api.services.forge_repository import (
|
|
ForgeRepositorySnapshot,
|
|
get_forge_repository_gateway,
|
|
)
|
|
from tests.conftest import (
|
|
create_test_domain,
|
|
create_test_repo,
|
|
create_test_workplan,
|
|
)
|
|
|
|
|
|
class FakeForge:
|
|
def __init__(self, *, name: str = "flex-auth", repository_id: int = 417, head: str = "a" * 40):
|
|
self.name = name
|
|
self.repository_id = repository_id
|
|
self.head = head
|
|
self.rename_calls = 0
|
|
self.extra: dict[str, int] = {}
|
|
self.unreadable = False
|
|
|
|
def snapshot(self, name: str, repository_id: int | None = None) -> ForgeRepositorySnapshot:
|
|
return ForgeRepositorySnapshot(
|
|
repository_id=repository_id or self.repository_id,
|
|
owner="coulomb",
|
|
name=name,
|
|
full_name=f"coulomb/{name}",
|
|
default_branch="main",
|
|
head_commit=self.head,
|
|
clone_url=f"https://forge.example/coulomb/{name}.git",
|
|
html_url=f"https://forge.example/coulomb/{name}",
|
|
projection_readable=True,
|
|
projection_source_present=True,
|
|
projection_entry_count=1,
|
|
)
|
|
|
|
async def inspect(self, *, instance: str, owner: str, name: str):
|
|
if self.unreadable:
|
|
from api.services.forge_repository import ForgeRepositoryUnreadable
|
|
|
|
raise ForgeRepositoryUnreadable("simulated unreadable forge")
|
|
if name == self.name:
|
|
return self.snapshot(name)
|
|
if name in self.extra:
|
|
return self.snapshot(name, self.extra[name])
|
|
return None
|
|
|
|
async def rename(self, *, instance: str, owner: str, old_name: str, new_name: str):
|
|
assert old_name == self.name
|
|
assert new_name not in self.extra
|
|
self.rename_calls += 1
|
|
self.name = new_name
|
|
return self.snapshot(new_name)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def rename_setup(client, monkeypatch):
|
|
monkeypatch.setattr(settings, "repository_rename_preflight_secret", "test-only-preflight-secret")
|
|
domain = await create_test_domain(client)
|
|
repo = await create_test_repo(
|
|
client,
|
|
domain_slug=domain["slug"],
|
|
slug="flex-auth",
|
|
remote_url="https://forge.example/coulomb/flex-auth.git",
|
|
local_path="/srv/flex-auth",
|
|
host_paths={"workstation": "/home/operator/flex-auth"},
|
|
)
|
|
forge = FakeForge()
|
|
app.dependency_overrides[get_forge_repository_gateway] = lambda: forge
|
|
verified = await client.post(
|
|
f"/repos/{repo['id']}/forge-identity/verify",
|
|
json={
|
|
"provider": "forgejo",
|
|
"forge_instance": "https://forge.example",
|
|
"forge_owner": "coulomb",
|
|
"forge_repository_id": forge.repository_id,
|
|
"verified_by": "pytest",
|
|
},
|
|
)
|
|
assert verified.status_code == 200, verified.text
|
|
return repo, forge
|
|
|
|
|
|
async def _preflight(client, repo_id: str, new_slug: str = "access-engine"):
|
|
response = await client.post(
|
|
f"/repos/{repo_id}/rename/preflight", json={"new_slug": new_slug}
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
async def _operation(client, repo_id: str, preflight: dict):
|
|
confirmation = f"rename:{repo_id}:flex-auth:access-engine"
|
|
response = await client.post(
|
|
f"/repos/{repo_id}/rename/operations",
|
|
json={
|
|
"new_slug": "access-engine",
|
|
"preflight_token": preflight["preflight_token"],
|
|
"confirmation": confirmation,
|
|
"actor": "pytest",
|
|
},
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
return response.json(), confirmation
|
|
|
|
|
|
async def _phase(client, repo_id: str, operation_id: str, phase: str, expected: str, confirmation: str, **extra):
|
|
response = await client.post(
|
|
f"/repos/{repo_id}/rename/operations/{operation_id}/phases/{phase}",
|
|
json={
|
|
"expected_phase": expected,
|
|
"confirmation": confirmation,
|
|
**extra,
|
|
},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dry_run_has_no_persistent_changes(client, test_engine, rename_setup):
|
|
repo, _forge = rename_setup
|
|
factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with factory() as session:
|
|
before_operations = await session.scalar(select(func.count()).select_from(RepositoryRenameOperation))
|
|
before_slugs = await session.scalar(select(func.count()).select_from(RepositorySlug))
|
|
before_repo = await session.get(ManagedRepo, uuid.UUID(repo["id"]))
|
|
before = (before_operations, before_slugs, before_repo.slug, before_repo.updated_at)
|
|
|
|
report = await _preflight(client, repo["id"])
|
|
assert report["safe_to_apply"] is True
|
|
assert report["preflight_token"]
|
|
assert report["current"]["statehub"]["repo_id"] == repo["id"]
|
|
assert report["current"]["forge"]["repository_id"] == 417
|
|
assert report["retained_history"]
|
|
|
|
async with factory() as session:
|
|
after_operations = await session.scalar(select(func.count()).select_from(RepositoryRenameOperation))
|
|
after_slugs = await session.scalar(select(func.count()).select_from(RepositorySlug))
|
|
after_repo = await session.get(ManagedRepo, uuid.UUID(repo["id"]))
|
|
after = (after_operations, after_slugs, after_repo.slug, after_repo.updated_at)
|
|
assert after == before
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_operation_id_is_idempotent_and_globally_discoverable(
|
|
client, rename_setup
|
|
):
|
|
repo, _forge = rename_setup
|
|
preflight = await _preflight(client, repo["id"])
|
|
operation_id = str(uuid.uuid4())
|
|
confirmation = f"rename:{repo['id']}:flex-auth:access-engine"
|
|
payload = {
|
|
"operation_id": operation_id,
|
|
"new_slug": "access-engine",
|
|
"preflight_token": preflight["preflight_token"],
|
|
"confirmation": confirmation,
|
|
"actor": "helixforge-test",
|
|
}
|
|
|
|
created = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations", json=payload
|
|
)
|
|
replay = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations",
|
|
json={**payload, "preflight_token": "expired-after-creation"},
|
|
)
|
|
assert created.status_code == replay.status_code == 201
|
|
assert created.json()["id"] == replay.json()["id"] == operation_id
|
|
assert created.json()["no_op"] is False
|
|
assert replay.json()["no_op"] is True
|
|
|
|
discovered = await client.get(
|
|
f"/repository-renames/operations/{operation_id}"
|
|
)
|
|
assert discovered.status_code == 200
|
|
assert discovered.json()["repo_id"] == repo["id"]
|
|
assert discovered.json()["phase"] == "preflighted"
|
|
|
|
collision = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations",
|
|
json={**payload, "new_slug": "another-name"},
|
|
)
|
|
assert collision.status_code == 412
|
|
assert collision.json()["detail"]["code"] == "repository_rename_precondition_failed"
|
|
|
|
actor_collision = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations",
|
|
json={**payload, "actor": "different-actor"},
|
|
)
|
|
assert actor_collision.status_code == 412
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_interrupt_resume_every_phase_and_preserve_uuid(client, rename_setup):
|
|
repo, forge = rename_setup
|
|
workplan = await create_test_workplan(
|
|
client, repo_id=repo["id"], slug="FLEX-WP-0001", status="active"
|
|
)
|
|
task = await client.post(
|
|
"/tasks/",
|
|
json={
|
|
"workplan_id": workplan["id"],
|
|
"title": "Security migration",
|
|
"status": "todo",
|
|
"priority": "high",
|
|
},
|
|
)
|
|
assert task.status_code == 201, task.text
|
|
progress = await client.post(
|
|
"/progress/",
|
|
json={"workplan_id": workplan["id"], "task_id": task.json()["id"], "summary": "baseline", "event_type": "note"},
|
|
)
|
|
assert progress.status_code == 201, progress.text
|
|
|
|
preflight = await _preflight(client, repo["id"])
|
|
operation, confirmation = await _operation(client, repo["id"], preflight)
|
|
operation_id = operation["id"]
|
|
discover = await client.get(
|
|
f"/repos/{repo['id']}/rename/operations", params={"active_only": True}
|
|
)
|
|
assert discover.status_code == 200
|
|
assert [item["id"] for item in discover.json()] == [operation_id]
|
|
|
|
# Simulate the hardest interruption: Forge committed the rename but the
|
|
# State Hub request died before recording forge-renamed.
|
|
await forge.rename(
|
|
instance="https://forge.example",
|
|
owner="coulomb",
|
|
old_name="flex-auth",
|
|
new_name="access-engine",
|
|
)
|
|
phase = await _phase(
|
|
client, repo["id"], operation_id, "forge-renamed", "preflighted", confirmation
|
|
)
|
|
assert phase["phase"] == "forge-renamed"
|
|
assert phase["evidence"]["phases"]["forge-renamed"]["resumed"] is True
|
|
assert forge.rename_calls == 1
|
|
|
|
sequence = [
|
|
("statehub-rebound", "forge-renamed", {}),
|
|
(
|
|
"source-synced",
|
|
"statehub-rebound",
|
|
{
|
|
"evidence": {
|
|
"fresh_clone": True,
|
|
"forge_repository_id": 417,
|
|
"head_commit": "a" * 40,
|
|
}
|
|
},
|
|
),
|
|
("consumers-verified", "source-synced", {"checks": {"routes": True, "security-stack": True}}),
|
|
("completed", "consumers-verified", {}),
|
|
]
|
|
for requested, expected, extra in sequence:
|
|
read = await client.get(f"/repos/{repo['id']}/rename/operations/{operation_id}")
|
|
assert read.status_code == 200
|
|
assert read.json()["phase"] == expected
|
|
phase = await _phase(
|
|
client, repo["id"], operation_id, requested, expected, confirmation, **extra
|
|
)
|
|
assert phase["phase"] == requested
|
|
if requested == "source-synced":
|
|
appended = await client.post(
|
|
"/progress/",
|
|
json={
|
|
"workplan_id": workplan["id"],
|
|
"task_id": task.json()["id"],
|
|
"summary": "rename phase completed",
|
|
"event_type": "note",
|
|
},
|
|
)
|
|
assert appended.status_code == 201, appended.text
|
|
replay = await _phase(
|
|
client, repo["id"], operation_id, requested, expected, confirmation, **extra
|
|
)
|
|
assert replay["id"] == operation_id
|
|
assert replay["no_op"] is True
|
|
|
|
verification = await client.get(
|
|
f"/repos/{repo['id']}/rename/operations/{operation_id}/verify"
|
|
)
|
|
assert verification.status_code == 200, verification.text
|
|
assert verification.json()["ok"] is True
|
|
discover = await client.get(
|
|
f"/repos/{repo['id']}/rename/operations", params={"active_only": True}
|
|
)
|
|
assert discover.json() == []
|
|
current = await client.get(f"/repos/by-id/{repo['id']}")
|
|
assert current.status_code == 200, current.text
|
|
assert current.json()["id"] == repo["id"]
|
|
assert current.json()["slug"] == "access-engine"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stale_head_wrong_id_target_conflict_and_queued_writes_fail_closed(client, rename_setup):
|
|
repo, forge = rename_setup
|
|
|
|
queued = await client.post(
|
|
f"/repos/{repo['id']}/rename/preflight",
|
|
json={
|
|
"new_slug": "access-engine",
|
|
"queued_edge_writes": [{"id": "edge-1", "source_host": "laptop"}],
|
|
},
|
|
)
|
|
assert queued.status_code == 200
|
|
assert queued.json()["safe_to_apply"] is False
|
|
assert queued.json()["preflight_token"] is None
|
|
|
|
forge.extra["access-engine"] = 999
|
|
conflict = await _preflight(client, repo["id"])
|
|
assert conflict["safe_to_apply"] is False
|
|
assert {item["code"] for item in conflict["blockers"]} >= {"forge_target_claimed"}
|
|
del forge.extra["access-engine"]
|
|
|
|
forge.repository_id = 999
|
|
wrong_id = await _preflight(client, repo["id"])
|
|
assert wrong_id["safe_to_apply"] is False
|
|
assert {item["code"] for item in wrong_id["blockers"]} >= {"wrong_forge_repository_id"}
|
|
forge.repository_id = 417
|
|
|
|
preflight = await _preflight(client, repo["id"])
|
|
forge.head = "b" * 40
|
|
create = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations",
|
|
json={
|
|
"new_slug": "access-engine",
|
|
"preflight_token": preflight["preflight_token"],
|
|
"confirmation": f"rename:{repo['id']}:flex-auth:access-engine",
|
|
"actor": "pytest",
|
|
},
|
|
)
|
|
assert create.status_code == 412, create.text
|
|
assert "stale" in create.json()["detail"]["message"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expired_token_bad_confirmation_and_unreadable_forge_fail_closed(
|
|
client, rename_setup, monkeypatch
|
|
):
|
|
repo, forge = rename_setup
|
|
monkeypatch.setattr(settings, "repository_rename_preflight_ttl_seconds", -1)
|
|
expired = await _preflight(client, repo["id"])
|
|
response = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations",
|
|
json={
|
|
"new_slug": "access-engine",
|
|
"preflight_token": expired["preflight_token"],
|
|
"confirmation": f"rename:{repo['id']}:flex-auth:access-engine",
|
|
"actor": "pytest",
|
|
},
|
|
)
|
|
assert response.status_code == 412
|
|
assert "expired" in response.json()["detail"]["message"].lower()
|
|
|
|
monkeypatch.setattr(settings, "repository_rename_preflight_ttl_seconds", 900)
|
|
preflight = await _preflight(client, repo["id"])
|
|
response = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations",
|
|
json={
|
|
"new_slug": "access-engine",
|
|
"preflight_token": preflight["preflight_token"],
|
|
"confirmation": "yes",
|
|
"actor": "pytest",
|
|
},
|
|
)
|
|
assert response.status_code == 412
|
|
forge.unreadable = True
|
|
unreadable = await _preflight(client, repo["id"])
|
|
assert unreadable["safe_to_apply"] is False
|
|
assert {item["code"] for item in unreadable["blockers"]} >= {"forge_unreadable"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rollback_restores_old_canonical_and_remains_auditable(client, rename_setup):
|
|
repo, forge = rename_setup
|
|
preflight = await _preflight(client, repo["id"])
|
|
operation, confirmation = await _operation(client, repo["id"], preflight)
|
|
operation_id = operation["id"]
|
|
for requested, expected, extra in [
|
|
("forge-renamed", "preflighted", {}),
|
|
("statehub-rebound", "forge-renamed", {}),
|
|
(
|
|
"source-synced",
|
|
"statehub-rebound",
|
|
{
|
|
"evidence": {
|
|
"fresh_clone": True,
|
|
"forge_repository_id": 417,
|
|
"head_commit": "a" * 40,
|
|
}
|
|
},
|
|
),
|
|
]:
|
|
await _phase(client, repo["id"], operation_id, requested, expected, confirmation, **extra)
|
|
|
|
rollback_confirmation = f"rollback:{operation_id}"
|
|
report = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation_id}/rollback-preflight",
|
|
json={"expected_phase": "source-synced", "confirmation": rollback_confirmation},
|
|
)
|
|
assert report.status_code == 200, report.text
|
|
assert report.json()["safe_to_rollback"] is True
|
|
assert report.json()["irreversible"]
|
|
|
|
rollback = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation_id}/rollback",
|
|
json={"expected_phase": "rollback-preflight", "confirmation": rollback_confirmation},
|
|
)
|
|
assert rollback.status_code == 200, rollback.text
|
|
assert rollback.json()["phase"] == "rolled-back"
|
|
assert forge.name == "flex-auth"
|
|
current = await client.get(f"/repos/by-id/{repo['id']}")
|
|
assert current.json()["id"] == repo["id"]
|
|
assert current.json()["slug"] == "flex-auth"
|
|
history = rollback.json()["evidence"]["phases"]
|
|
assert set(history) >= {"preflighted", "forge-renamed", "statehub-rebound", "source-synced", "rollback-preflight", "rolled-back"}
|
|
replay = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation_id}/rollback",
|
|
json={"expected_phase": "rollback-preflight", "confirmation": rollback_confirmation},
|
|
)
|
|
assert replay.status_code == 200
|
|
assert replay.json()["no_op"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rollback_recovers_unrecorded_forge_rename(client, rename_setup):
|
|
repo, forge = rename_setup
|
|
preflight = await _preflight(client, repo["id"])
|
|
operation, _confirmation = await _operation(client, repo["id"], preflight)
|
|
await forge.rename(
|
|
instance="https://forge.example",
|
|
owner="coulomb",
|
|
old_name="flex-auth",
|
|
new_name="access-engine",
|
|
)
|
|
rollback_confirmation = f"rollback:{operation['id']}"
|
|
report = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation['id']}/rollback-preflight",
|
|
json={"expected_phase": "preflighted", "confirmation": rollback_confirmation},
|
|
)
|
|
assert report.status_code == 200, report.text
|
|
assert report.json()["rollback_from_phase"] == "forge-renamed-unrecorded"
|
|
rollback = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation['id']}/rollback",
|
|
json={"expected_phase": "rollback-preflight", "confirmation": rollback_confirmation},
|
|
)
|
|
assert rollback.status_code == 200, rollback.text
|
|
assert rollback.json()["phase"] == "rolled-back"
|
|
assert forge.name == "flex-auth"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_continuity_evidence_covers_relationships_totals_routes_and_dispatch(
|
|
client, test_engine, rename_setup
|
|
):
|
|
repo, _forge = rename_setup
|
|
workplan = await create_test_workplan(
|
|
client, repo_id=repo["id"], slug="FLEX-WP-0042", status="active"
|
|
)
|
|
task = await client.post(
|
|
"/tasks/",
|
|
json={
|
|
"workplan_id": workplan["id"],
|
|
"title": "Preserve security-stack lineage",
|
|
"status": "todo",
|
|
"priority": "high",
|
|
},
|
|
)
|
|
assert task.status_code == 201, task.text
|
|
now = datetime.now(timezone.utc)
|
|
factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with factory() as session:
|
|
bound = await session.get(Workplan, uuid.UUID(workplan["id"]))
|
|
bound.backing_filename = "FLEX-WP-0042-lineage.md"
|
|
bound.backing_relative_path = "workplans/FLEX-WP-0042-lineage.md"
|
|
decision = Decision(
|
|
workplan_id=bound.id,
|
|
title="Retain immutable repository identity",
|
|
)
|
|
token = TokenEvent(
|
|
repo_id=uuid.UUID(repo["id"]),
|
|
workplan_id=bound.id,
|
|
task_id=uuid.UUID(task.json()["id"]),
|
|
tokens_in=120,
|
|
tokens_out=30,
|
|
cached_input_tokens=10,
|
|
reasoning_output_tokens=5,
|
|
raw_total_tokens=165,
|
|
)
|
|
snapshot = SBOMSnapshot(
|
|
repo_id=uuid.UUID(repo["id"]),
|
|
snapshot_at=now,
|
|
source="pytest",
|
|
entry_count=1,
|
|
created_at=now,
|
|
)
|
|
message = AgentMessage(
|
|
from_agent="security-review",
|
|
to_agent="flex-auth",
|
|
subject="Historical route",
|
|
body="Keep the recorded coordinate",
|
|
)
|
|
session.add_all([decision, token, snapshot, message])
|
|
await session.flush()
|
|
session.add(
|
|
SBOMEntry(
|
|
repo_id=uuid.UUID(repo["id"]),
|
|
snapshot_id=snapshot.id,
|
|
package_name="policy-engine",
|
|
package_version="1.0",
|
|
ecosystem=Ecosystem.python,
|
|
snapshot_at=now,
|
|
created_at=now,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
preflight = await _preflight(client, repo["id"])
|
|
baseline = preflight["baselines"]
|
|
assert set(baseline) >= {
|
|
"repository",
|
|
"workplans",
|
|
"tasks",
|
|
"progress_events",
|
|
"decisions",
|
|
"token_events",
|
|
"sbom_snapshots",
|
|
"active_dispatch",
|
|
"aliases",
|
|
"messages",
|
|
"workplan_bindings",
|
|
"continuity_checksum",
|
|
}
|
|
assert baseline["token_events"]["totals"] == {
|
|
"tokens_in": 120,
|
|
"tokens_out": 30,
|
|
"cached_input_tokens": 10,
|
|
"reasoning_output_tokens": 5,
|
|
"raw_total_tokens": 165,
|
|
}
|
|
assert baseline["workplan_bindings"]["relationships"][0]["repo_id"] == repo["id"]
|
|
|
|
operation, confirmation = await _operation(client, repo["id"], preflight)
|
|
for requested, expected, extra in [
|
|
("forge-renamed", "preflighted", {}),
|
|
("statehub-rebound", "forge-renamed", {}),
|
|
(
|
|
"source-synced",
|
|
"statehub-rebound",
|
|
{
|
|
"evidence": {
|
|
"fresh_clone": True,
|
|
"forge_repository_id": 417,
|
|
"head_commit": "a" * 40,
|
|
}
|
|
},
|
|
),
|
|
("consumers-verified", "source-synced", {"checks": {"dispatch": True}}),
|
|
("completed", "consumers-verified", {}),
|
|
]:
|
|
await _phase(
|
|
client,
|
|
repo["id"],
|
|
operation["id"],
|
|
requested,
|
|
expected,
|
|
confirmation,
|
|
**extra,
|
|
)
|
|
|
|
verification = await client.get(
|
|
f"/repos/{repo['id']}/rename/operations/{operation['id']}/verify"
|
|
)
|
|
assert verification.status_code == 200, verification.text
|
|
evidence = verification.json()
|
|
assert evidence["ok"] is True
|
|
for name, expected_checksum in evidence["relationship_checksums"]["baseline"].items():
|
|
assert evidence["relationship_checksums"]["preserved"][name] == expected_checksum
|
|
assert (await client.get("/repos/flex-auth")).json()["slug_status"] == "alias"
|
|
dispatch = await client.get("/repos/flex-auth/dispatch")
|
|
assert dispatch.status_code == 200
|
|
assert dispatch.json()["canonical_slug"] == "access-engine"
|
|
assert dispatch.json()["active_workplans"][0]["id"] == workplan["id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_equal_counts_do_not_hide_detached_work_and_telemetry(
|
|
client, test_engine, rename_setup
|
|
):
|
|
repo, _forge = rename_setup
|
|
workplan = await create_test_workplan(
|
|
client, repo_id=repo["id"], slug="FLEX-WP-0050", status="active"
|
|
)
|
|
factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with factory() as session:
|
|
original_token = TokenEvent(
|
|
repo_id=uuid.UUID(repo["id"]), tokens_in=11, tokens_out=7
|
|
)
|
|
session.add(original_token)
|
|
await session.commit()
|
|
preflight = await _preflight(client, repo["id"])
|
|
operation, _confirmation = await _operation(client, repo["id"], preflight)
|
|
|
|
other_domain = await create_test_domain(client, slug="other-domain")
|
|
other_repo = await create_test_repo(
|
|
client, domain_slug=other_domain["slug"], slug="other-repo"
|
|
)
|
|
async with factory() as session:
|
|
detached = await session.get(Workplan, uuid.UUID(workplan["id"]))
|
|
detached.repo_id = uuid.UUID(other_repo["id"])
|
|
token = await session.get(TokenEvent, original_token.id)
|
|
token.repo_id = uuid.UUID(other_repo["id"])
|
|
session.add(
|
|
Workplan(
|
|
repo_id=uuid.UUID(repo["id"]),
|
|
slug="FLEX-WP-0050-RECREATED",
|
|
title="Count-preserving replacement",
|
|
status="active",
|
|
)
|
|
)
|
|
session.add(
|
|
TokenEvent(repo_id=uuid.UUID(repo["id"]), tokens_in=11, tokens_out=7)
|
|
)
|
|
await session.commit()
|
|
|
|
verification = await client.get(
|
|
f"/repos/{repo['id']}/rename/operations/{operation['id']}/verify"
|
|
)
|
|
assert verification.status_code == 200, verification.text
|
|
result = verification.json()
|
|
assert result["ok"] is False
|
|
relationship = next(
|
|
check for check in result["checks"] if check["name"] == "relationship_continuity"
|
|
)
|
|
assert relationship["ok"] is False
|
|
assert set(relationship["actual"]["missing"]) >= {"workplans", "token_events"}
|
|
counts = next(
|
|
check for check in result["checks"] if check["name"] == "record_counts_non_decreasing"
|
|
)
|
|
assert counts["actual"]["workplans"] == counts["expected"]["workplans"]
|
|
assert counts["actual"]["token_events"] == counts["expected"]["token_events"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redirected_forge_and_wrong_fresh_clone_identity_fail_closed_then_resume(
|
|
client, rename_setup
|
|
):
|
|
repo, forge = rename_setup
|
|
original_inspect = forge.inspect
|
|
|
|
async def redirected(*, instance: str, owner: str, name: str):
|
|
if name == "flex-auth":
|
|
return forge.snapshot("access-engine")
|
|
if name == "access-engine":
|
|
return None
|
|
return await original_inspect(instance=instance, owner=owner, name=name)
|
|
|
|
forge.inspect = redirected
|
|
redirected_report = await _preflight(client, repo["id"])
|
|
assert redirected_report["safe_to_apply"] is False
|
|
assert {item["code"] for item in redirected_report["blockers"]} >= {"forge_redirected"}
|
|
forge.inspect = original_inspect
|
|
|
|
preflight = await _preflight(client, repo["id"])
|
|
operation, confirmation = await _operation(client, repo["id"], preflight)
|
|
await _phase(client, repo["id"], operation["id"], "forge-renamed", "preflighted", confirmation)
|
|
await _phase(client, repo["id"], operation["id"], "statehub-rebound", "forge-renamed", confirmation)
|
|
wrong = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation['id']}/phases/source-synced",
|
|
json={
|
|
"expected_phase": "statehub-rebound",
|
|
"confirmation": confirmation,
|
|
"evidence": {
|
|
"fresh_clone": True,
|
|
"forge_repository_id": 999,
|
|
"head_commit": "a" * 40,
|
|
"password": "must-never-reach-an-event",
|
|
},
|
|
},
|
|
)
|
|
assert wrong.status_code == 412
|
|
status = (await client.get(f"/repository-renames/operations/{operation['id']}")).json()
|
|
assert status["phase"] == "statehub-rebound"
|
|
assert status["evidence"]["telemetry"]["failures"] == 1
|
|
resumed = await _phase(
|
|
client,
|
|
repo["id"],
|
|
operation["id"],
|
|
"source-synced",
|
|
"statehub-rebound",
|
|
confirmation,
|
|
evidence={
|
|
"fresh_clone": True,
|
|
"forge_repository_id": 417,
|
|
"head_commit": "a" * 40,
|
|
},
|
|
)
|
|
replay = await _phase(
|
|
client,
|
|
repo["id"],
|
|
operation["id"],
|
|
"source-synced",
|
|
"statehub-rebound",
|
|
confirmation,
|
|
evidence={
|
|
"fresh_clone": True,
|
|
"forge_repository_id": 417,
|
|
"head_commit": "a" * 40,
|
|
},
|
|
)
|
|
assert resumed["phase"] == "source-synced"
|
|
assert replay["no_op"] is True
|
|
assert replay["evidence"]["telemetry"]["retries"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rollback_recovers_when_old_statehub_slug_temporarily_unavailable(
|
|
client, test_engine, rename_setup
|
|
):
|
|
repo, forge = rename_setup
|
|
preflight = await _preflight(client, repo["id"])
|
|
operation, confirmation = await _operation(client, repo["id"], preflight)
|
|
for requested, expected, extra in [
|
|
("forge-renamed", "preflighted", {}),
|
|
("statehub-rebound", "forge-renamed", {}),
|
|
(
|
|
"source-synced",
|
|
"statehub-rebound",
|
|
{
|
|
"evidence": {
|
|
"fresh_clone": True,
|
|
"forge_repository_id": 417,
|
|
"head_commit": "a" * 40,
|
|
}
|
|
},
|
|
),
|
|
]:
|
|
await _phase(client, repo["id"], operation["id"], requested, expected, confirmation, **extra)
|
|
rollback_confirmation = f"rollback:{operation['id']}"
|
|
report = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation['id']}/rollback-preflight",
|
|
json={"expected_phase": "source-synced", "confirmation": rollback_confirmation},
|
|
)
|
|
assert report.status_code == 200, report.text
|
|
|
|
factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with factory() as session:
|
|
await session.execute(delete(RepositorySlug).where(RepositorySlug.slug == "flex-auth"))
|
|
await session.commit()
|
|
failed = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation['id']}/rollback",
|
|
json={"expected_phase": "rollback-preflight", "confirmation": rollback_confirmation},
|
|
)
|
|
assert failed.status_code == 412
|
|
assert forge.name == "flex-auth"
|
|
async with factory() as session:
|
|
session.add(
|
|
RepositorySlug(
|
|
repo_id=uuid.UUID(repo["id"]),
|
|
slug="flex-auth",
|
|
kind="alias",
|
|
protected=True,
|
|
source_operation_id=uuid.UUID(operation["id"]),
|
|
)
|
|
)
|
|
await session.commit()
|
|
resumed = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation['id']}/rollback",
|
|
json={"expected_phase": "rollback-preflight", "confirmation": rollback_confirmation},
|
|
)
|
|
assert resumed.status_code == 200, resumed.text
|
|
assert resumed.json()["phase"] == "rolled-back"
|
|
assert resumed.json()["evidence"]["telemetry"]["failures"] == 1
|
|
assert resumed.json()["evidence"]["telemetry"]["rollback_outcome"] == "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_events_expose_metrics_without_operator_evidence_or_secrets(
|
|
client, test_engine, rename_setup
|
|
):
|
|
repo, _forge = rename_setup
|
|
preflight = await _preflight(client, repo["id"])
|
|
operation, confirmation = await _operation(client, repo["id"], preflight)
|
|
await _phase(client, repo["id"], operation["id"], "forge-renamed", "preflighted", confirmation)
|
|
replay = await _phase(client, repo["id"], operation["id"], "forge-renamed", "preflighted", confirmation)
|
|
assert replay["evidence"]["telemetry"]["retries"] == 1
|
|
|
|
factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with factory() as session:
|
|
persisted = await session.get(RepositoryRenameOperation, uuid.UUID(operation["id"]))
|
|
events = repository_rename_events(persisted, outcome="succeeded", no_op=True)
|
|
assert [subject for subject, _event in events] == ["org.statehub.repo.rename.phase"]
|
|
attributes = events[0][1].attributes
|
|
assert attributes["repo_id"] == repo["id"]
|
|
assert attributes["phase"] == "forge-renamed"
|
|
assert attributes["phase_duration_ms"] >= 0
|
|
assert attributes["retries"] == 1
|
|
assert set(attributes) >= {
|
|
"failures",
|
|
"rollback_attempts",
|
|
"verification_outcome",
|
|
"evidence_ref",
|
|
}
|
|
serialized = events[0][1].model_dump_json().lower()
|
|
assert "preflight_token" not in serialized
|
|
assert "operator_evidence" not in serialized
|
|
assert "password" not in serialized
|
|
persisted.phase = "completed"
|
|
completion = repository_rename_events(persisted, outcome="succeeded")
|
|
assert [subject for subject, _event in completion] == [
|
|
"org.statehub.repo.rename.phase",
|
|
"org.statehub.repo.renamed",
|
|
]
|
|
persisted.phase = "forge-renamed"
|
|
persisted.error_code = "repository_rename_precondition_failed"
|
|
persisted.error_details = {"authorization": "Bearer must-not-publish"}
|
|
failure = repository_rename_events(persisted, outcome="failed")
|
|
failure_json = failure[0][1].model_dump_json()
|
|
assert failure[0][0] == "org.statehub.repo.rename.failed"
|
|
assert "must-not-publish" not in failure_json
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_every_forward_phase_failure_is_retry_safe(client, rename_setup):
|
|
repo, forge = rename_setup
|
|
preflight = await _preflight(client, repo["id"])
|
|
operation, confirmation = await _operation(client, repo["id"], preflight)
|
|
operation_id = operation["id"]
|
|
|
|
async def fail_unreadable(requested: str, expected: str) -> None:
|
|
forge.unreadable = True
|
|
response = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation_id}/phases/{requested}",
|
|
json={"expected_phase": expected, "confirmation": confirmation},
|
|
)
|
|
forge.unreadable = False
|
|
assert response.status_code == 412
|
|
status = await client.get(f"/repository-renames/operations/{operation_id}")
|
|
assert status.json()["phase"] == expected
|
|
|
|
await fail_unreadable("forge-renamed", "preflighted")
|
|
await _phase(client, repo["id"], operation_id, "forge-renamed", "preflighted", confirmation)
|
|
await fail_unreadable("statehub-rebound", "forge-renamed")
|
|
await _phase(client, repo["id"], operation_id, "statehub-rebound", "forge-renamed", confirmation)
|
|
|
|
wrong_clone = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation_id}/phases/source-synced",
|
|
json={
|
|
"expected_phase": "statehub-rebound",
|
|
"confirmation": confirmation,
|
|
"evidence": {
|
|
"fresh_clone": True,
|
|
"forge_repository_id": 999,
|
|
"head_commit": "a" * 40,
|
|
},
|
|
},
|
|
)
|
|
assert wrong_clone.status_code == 412
|
|
await _phase(
|
|
client,
|
|
repo["id"],
|
|
operation_id,
|
|
"source-synced",
|
|
"statehub-rebound",
|
|
confirmation,
|
|
evidence={
|
|
"fresh_clone": True,
|
|
"forge_repository_id": 417,
|
|
"head_commit": "a" * 40,
|
|
},
|
|
)
|
|
|
|
bad_consumers = await client.post(
|
|
f"/repos/{repo['id']}/rename/operations/{operation_id}/phases/consumers-verified",
|
|
json={
|
|
"expected_phase": "source-synced",
|
|
"confirmation": confirmation,
|
|
"checks": {"security-stack": False},
|
|
},
|
|
)
|
|
assert bad_consumers.status_code == 412
|
|
await _phase(
|
|
client,
|
|
repo["id"],
|
|
operation_id,
|
|
"consumers-verified",
|
|
"source-synced",
|
|
confirmation,
|
|
checks={"security-stack": True},
|
|
)
|
|
|
|
await fail_unreadable("completed", "consumers-verified")
|
|
completed = await _phase(
|
|
client,
|
|
repo["id"],
|
|
operation_id,
|
|
"completed",
|
|
"consumers-verified",
|
|
confirmation,
|
|
)
|
|
assert completed["evidence"]["telemetry"]["failures"] == 5
|
|
assert completed["evidence"]["telemetry"]["verification_outcome"] == "passed"
|
|
assert set(completed["evidence"]["telemetry"]["failures_by_phase"]) == {
|
|
"forge-renamed",
|
|
"statehub-rebound",
|
|
"source-synced",
|
|
"consumers-verified",
|
|
"completed",
|
|
}
|