from __future__ import annotations import json import os import stat import sys from dataclasses import replace from datetime import UTC, datetime from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import pytest from account_registry import Account, AccountRegistry, AccountRegistryKind, AuthorityMode from authority_migration import ( AuthorityMigrationCoordinator, AuthorityProvision, MigrationAcceptance, MigrationCleanupError, MigrationError, ) from health_client import AccountSnapshot, HealthStatus from provider_services import LifecycleEvent class FakeAdmin: def __init__(self, grant: Path, *, generation: int = 1) -> None: self.grant = grant self.generation = generation self.calls: list[tuple[str, str, str | None]] = [] def login_and_provision(self, account: Account, authority_name: str) -> AuthorityProvision: self.calls.append(("login", account.tool, account.slug)) return AuthorityProvision( authority_name=authority_name, route_id=f"route-{account.tool}-fixture", proxy_grant_ref=self.grant, generation=self.generation, ) def status_generation(self, provider: str, account_id: str) -> int: self.calls.append(("status", provider, account_id)) return self.generation def disable_route(self, route_id: str) -> None: self.calls.append(("disable-route", route_id, None)) def remove_route_grants(self, route_id: str) -> None: self.calls.append(("remove-route-grants", route_id, None)) def remove_route(self, route_id: str) -> None: self.calls.append(("remove-route", route_id, None)) def remove_account(self, provider: str, account_id: str) -> None: self.calls.append(("remove-account", provider, account_id)) class NativeLifecycle: def __init__(self, registry: AccountRegistry, *, succeed: bool = True) -> None: self.registry = registry self.succeed = succeed self.seen_quiesced = False def add(self, alias: str, *, login_hint: str | None = None, slug: str | None = None): del alias, login_hint, slug raise AssertionError("add must not be called by rollback") def reauthenticate(self, account: Account): current = next(item for item in self.registry.list() if item.slug == account.slug) self.seen_quiesced = bool(current.authority_binding and current.authority_binding.quiesced) assert account.authority_binding is None if self.succeed: yield LifecycleEvent(kind="success", account=account) else: yield LifecycleEvent(kind="failure", account=account, message="synthetic native failure") def submit_code(self, session, code: str) -> None: del session, code class NativeHealth: def __init__(self, registry: AccountRegistry, status: HealthStatus = HealthStatus.OK) -> None: self.registry = registry self.status = status self.seen_quiesced = False def fetch(self, account: Account, timeout_secs: float = 10.0) -> AccountSnapshot: del timeout_secs current = next(item for item in self.registry.list() if item.slug == account.slug) self.seen_quiesced = bool(current.authority_binding and current.authority_binding.quiesced) assert account.authority_binding is None return AccountSnapshot(self.status, None, None, detail="synthetic native readiness") def make_registry(tmp_path: Path, provider: str) -> tuple[AccountRegistry, Account]: kind = AccountRegistryKind(provider) registry = AccountRegistry( base_dir=tmp_path / "tray", kind=kind, legacy_codex_home=tmp_path / "native-codex", legacy_claude_home=tmp_path / "native-claude", active_claude_json=tmp_path / "native-claude.json", ) slug = f"fixture-{provider}" registry.add_dir(slug, f"Fixture {provider.title()}") account = next(item for item in registry.list() if item.slug == slug) return registry, account def complete_acceptance(provider: str) -> MigrationAcceptance: common = dict( cli=True, resume=True, streaming=True, quota_health=True, repair=True, restart=True, exact_route=True, no_provider_credentials=True, ) if provider == "codex": return MigrationAcceptance(**common, responses=True, realtime=True, app_server=True) return MigrationAcceptance(**common, http_sse=True) @pytest.mark.parametrize("provider", ["codex", "claude"]) def test_dark_stage_and_promotion_are_fixture_only_and_evidence_gated( tmp_path: Path, provider: str ) -> None: registry, account = make_registry(tmp_path, provider) grant = tmp_path / "authority-grant" grant.write_text("synthetic-proxy-key", encoding="utf-8") grant.chmod(0o600) admin = FakeAdmin(grant) now = datetime(2026, 8, 22, 3, 0, tzinfo=UTC) coordinator = AuthorityMigrationCoordinator(registry, admin, clock=lambda: now) staged = coordinator.stage_dark(account, authority_name="fixture-authority") current = registry.list()[0] assert current.authority_binding is not None assert current.authority_binding.mode == AuthorityMode.SUBROUTER_DARK assert current.authority_binding.quiesced is False assert staged.state == "dark" assert staged.provider_credential_copied is False assert staged.native_fallback_used is False receipt_path = coordinator._receipt_path(current) assert receipt_path.stat().st_mode & 0o777 == 0o600 receipt_text = receipt_path.read_text(encoding="utf-8") assert account.slug not in receipt_text assert current.authority_binding.route_id not in receipt_text assert str(grant) not in receipt_text assert "synthetic-proxy-key" not in receipt_text with pytest.raises(MigrationError, match="acceptance incomplete"): coordinator.promote(current, MigrationAcceptance(cli=True)) assert registry.list()[0].authority_binding.mode == AuthorityMode.SUBROUTER_DARK admin.generation = 2 promoted = coordinator.promote(current, complete_acceptance(provider)) bound = registry.list()[0].authority_binding assert bound is not None and bound.mode == AuthorityMode.SUBROUTER assert promoted.state == "authority" assert promoted.generation == 2 assert promoted.flow_checks is not None assert not complete_acceptance(provider).missing(provider) assert ("status", provider, account.slug) in admin.calls @pytest.mark.parametrize("provider", ["codex", "claude"]) def test_rollback_quiesces_then_fresh_reauths_before_native_switch_and_cleanup( tmp_path: Path, provider: str ) -> None: registry, account = make_registry(tmp_path, provider) grant = tmp_path / "authority-grant" grant.write_text("synthetic-proxy-key", encoding="utf-8") grant.chmod(0o600) admin = FakeAdmin(grant) coordinator = AuthorityMigrationCoordinator(registry, admin) coordinator.stage_dark(account, authority_name="fixture-authority") coordinator.promote(registry.list()[0], complete_acceptance(provider)) lifecycle = NativeLifecycle(registry) health = NativeHealth(registry) receipt = coordinator.rollback( registry.list()[0], native_lifecycle=lifecycle, native_health=health ) assert lifecycle.seen_quiesced is True assert health.seen_quiesced is True assert registry.list()[0].authority_binding is None assert receipt.state == "native" assert receipt.cleanup_complete is True assert receipt.provider_credential_copied is False assert receipt.native_fallback_used is False assert not grant.exists() names = [call[0] for call in admin.calls] assert names.index("disable-route") < names.index("remove-route-grants") assert names.index("remove-route-grants") < names.index("remove-route") assert names.index("remove-route") < names.index("remove-account") @pytest.mark.parametrize("failure", ["reauth", "health"]) def test_failed_rollback_stays_authority_bound_and_quiesced( tmp_path: Path, failure: str ) -> None: registry, account = make_registry(tmp_path, "codex") grant = tmp_path / "authority-grant" grant.write_text("synthetic-proxy-key", encoding="utf-8") grant.chmod(0o600) admin = FakeAdmin(grant) coordinator = AuthorityMigrationCoordinator(registry, admin) coordinator.stage_dark(account, authority_name="fixture-authority") coordinator.promote(registry.list()[0], complete_acceptance("codex")) lifecycle = NativeLifecycle(registry, succeed=failure != "reauth") health = NativeHealth( registry, status=HealthStatus.BROKEN if failure == "health" else HealthStatus.OK, ) with pytest.raises(MigrationError): coordinator.rollback( registry.list()[0], native_lifecycle=lifecycle, native_health=health ) binding = registry.list()[0].authority_binding assert binding is not None assert binding.mode == AuthorityMode.SUBROUTER assert binding.quiesced is True assert grant.exists() assert not any(call[0] == "remove-account" for call in admin.calls) def test_cleanup_refuses_symlink_grant_after_native_commit(tmp_path: Path) -> None: registry, account = make_registry(tmp_path, "claude") real = tmp_path / "real-grant" real.write_text("synthetic-proxy-key", encoding="utf-8") real.chmod(0o600) link = tmp_path / "grant-link" link.symlink_to(real) admin = FakeAdmin(link) coordinator = AuthorityMigrationCoordinator(registry, admin) coordinator.stage_dark(account, authority_name="fixture-authority") coordinator.promote(registry.list()[0], complete_acceptance("claude")) with pytest.raises(MigrationCleanupError): coordinator.rollback( registry.list()[0], native_lifecycle=NativeLifecycle(registry), native_health=NativeHealth(registry), ) assert registry.list()[0].authority_binding is None assert link.is_symlink() receipt = json.loads(coordinator._receipt_path(account).read_text(encoding="utf-8")) assert receipt["state"] == "native-cleanup-required" assert receipt["cleanup_complete"] is False def test_concrete_admin_provisions_material_without_privilege_bridge(tmp_path: Path) -> None: from authority_migration import SubrouterMigrationAdmin registry, account = make_registry(tmp_path, "codex") calls: list[tuple[str, dict[str, str], bool]] = [] def requester(operation, params, *, interactive=False): calls.append((operation, dict(params), interactive)) if operation == "provision": return { "route_id": "routefixture1234567890", "proxy_key": "synthetic-bearer-never-output", "generation": 3, } if operation == "status": return {"generation": 3} return {"status": "ok"} material_root = registry.base_dir / "authority-material" provider_root = material_root / "codex" material_root.mkdir(mode=0o775) provider_root.mkdir(mode=0o775) material_root.chmod(0o775) provider_root.chmod(0o775) admin = SubrouterMigrationAdmin(registry, requester=requester) provision = admin.login_and_provision(account, "fixture-authority") assert stat.S_IMODE(material_root.stat().st_mode) == 0o700 assert stat.S_IMODE(provider_root.stat().st_mode) == 0o700 assert stat.S_IMODE(provision.proxy_grant_ref.parent.stat().st_mode) == 0o700 assert provision.generation == 3 assert provision.route_id == "routefixture1234567890" assert provision.proxy_grant_ref.name == "proxy.key" assert provision.proxy_grant_ref.stat().st_mode & 0o777 == 0o600 assert calls == [ ( "provision", {"provider": "codex", "account_id": "fixture-codex"}, True, ) ] source = Path(__import__("authority_migration").__file__).read_text(encoding="utf-8") concrete = source[source.index("class SubrouterMigrationAdmin") : source.index("class MigrationAcceptance")] assert "deck-sudo" not in concrete assert "runuser" not in concrete assert "migration-provision" not in concrete def test_concrete_admin_rejects_malformed_generation_and_compensates(tmp_path: Path) -> None: from authority_migration import SubrouterMigrationAdmin registry, account = make_registry(tmp_path, "claude") calls: list[tuple[str, dict[str, str], bool]] = [] def requester(operation, params, *, interactive=False): calls.append((operation, dict(params), interactive)) if operation == "provision": return { "route_id": "routefixture1234567890", "proxy_key": "synthetic", "generation": "secret", } return {"status": "ok"} admin = SubrouterMigrationAdmin(registry, requester=requester) with pytest.raises(MigrationError, match="response"): admin.login_and_provision(account, "fixture-authority") assert [item[0] for item in calls] == [ "provision", "remove-route-grants", "remove-route", "remove-account", ] material_root = registry.base_dir / "authority-material" / "claude" assert not any(material_root.glob("*/route.id")) assert not any(material_root.glob("*/proxy.key"))