#!/bin/sh """:" ACCOUNT_REGISTRY_SHIM_ACTIVE=1 exec python3 "$0" "$@" ":""" from __future__ import annotations import importlib.util import base64 import errno import json import os from pathlib import Path import subprocess import stat import sys import threading import pytest sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import account_registry as account_registry_module from account_lock import AccountLockedError from account_registry import ( Account, AccountRef, AccountRegistry, AccountRegistryKind, AuthorityBinding, AuthorityMode, StagedAccount, ) from claude_credentials import pin_account_uuid SHARED_FIXTURE_NAMES = ("config.toml", "skills", "commands") def make_auth_payload(email: str, plan: str, account_id: str) -> str: payload = { "email": email, "https://api.openai.com/auth": { "chatgpt_plan_type": plan, "chatgpt_account_id": account_id, }, } encoded = base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")).decode("ascii") encoded = encoded.rstrip("=") return f"header.{encoded}.signature" def make_auth_payload_with_claims( email: str, account_id: str, auth_claims: dict[str, object], ) -> str: payload = { "email": email, "https://api.openai.com/auth": { **auth_claims, "chatgpt_account_id": account_id, }, } encoded = base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")).decode("ascii") encoded = encoded.rstrip("=") return f"header.{encoded}.signature" def write_auth(path: Path, email: str, plan: str, account_id: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps({"tokens": {"id_token": make_auth_payload(email, plan, account_id)}}), encoding="utf-8", ) @pytest.fixture def registry(tmp_path: Path) -> AccountRegistry: legacy = tmp_path / "legacy" base = tmp_path / "tray" legacy.mkdir() for name in SHARED_FIXTURE_NAMES: target = legacy / name if "." in name: target.write_text(name, encoding="utf-8") else: target.mkdir() return AccountRegistry(base_dir=base, legacy_codex_home=legacy) def test_migrate_legacy_is_additive_idempotent_and_sets_default(registry: AccountRegistry) -> None: legacy = registry.legacy_codex_home write_auth(legacy / "auth.roy.json", "roy@example.com", "plus", "acct-roy") write_auth(legacy / "auth.rafa.json", "rafa@example.com", "pro", "acct-rafa") write_auth(legacy / "auth.json", "rafa@example.com", "pro", "acct-rafa") original_bytes = (legacy / "auth.roy.json").read_bytes() registry.migrate_legacy() registry.migrate_legacy() accounts = registry.list() assert [account.slug for account in accounts] == ["rafa", "roy"] assert registry.default_slug() == "rafa" assert (legacy / "auth.roy.json").read_bytes() == original_bytes rafa_home = registry.accounts_dir / "rafa" / "CODEX_HOME" assert (rafa_home / "auth.json").exists() for name in SHARED_FIXTURE_NAMES: link = rafa_home / name assert link.is_symlink() assert os.readlink(link) == str(legacy / name) def test_removing_legacy_account_prevents_future_legacy_migration( registry: AccountRegistry, ) -> None: write_auth( registry.legacy_codex_home / "auth.avi.json", "avi@example.com", "plus", "acct-avi", ) registry.migrate_legacy() registry.remove("avi") fresh_registry = AccountRegistry( base_dir=registry.base_dir, legacy_codex_home=registry.legacy_codex_home, ) fresh_registry.migrate_legacy() assert fresh_registry.list() == [] def test_explicitly_readding_a_deleted_legacy_slug_clears_its_tombstone( registry: AccountRegistry, ) -> None: write_auth( registry.legacy_codex_home / "auth.avi.json", "avi@example.com", "plus", "acct-avi", ) registry.migrate_legacy() registry.remove("avi") registry.add_dir("avi", "Avi") state = json.loads(registry.registry_path.read_text(encoding="utf-8")) assert state["deleted_legacy_slugs"] == [] def test_committing_a_deleted_legacy_slug_clears_its_tombstone( registry: AccountRegistry, ) -> None: write_auth( registry.legacy_codex_home / "auth.avi.json", "avi@example.com", "plus", "acct-avi", ) registry.migrate_legacy() registry.remove("avi") staged = registry.stage_add("Avi") write_auth(staged.account_home / "auth.json", "avi@example.com", "plus", "acct-avi") staged.commit() state = json.loads(registry.registry_path.read_text(encoding="utf-8")) assert state["deleted_legacy_slugs"] == [] def test_sync_shared_links_shares_every_noncredential_entry(registry: AccountRegistry) -> None: codex_home = registry.accounts_dir / "roy" / "CODEX_HOME" codex_home.mkdir(parents=True) (codex_home / "auth.json").write_text("account-auth", encoding="utf-8") (codex_home / "auth.json.bak").write_text("account-backup", encoding="utf-8") (codex_home / "state_5.sqlite").write_text("account-db", encoding="utf-8") (codex_home / "history.jsonl").write_text( '{"session_id":"account","ts":2,"text":"account"}\n', encoding="utf-8", ) agents = codex_home / "agents" agents.mkdir() (agents / "reviewer.md").write_text("reviewer", encoding="utf-8") (codex_home / "future-state.json").write_text("future", encoding="utf-8") (registry.legacy_codex_home / "history.jsonl").write_text( '{"session_id":"shared","ts":1,"text":"shared"}\n', encoding="utf-8", ) (registry.legacy_codex_home / "hooks.json").write_text("hooks", encoding="utf-8") shared_log = registry.legacy_codex_home / "log" / "codex-login.log" shared_log.parent.mkdir() shared_log.write_text("shared-log\n", encoding="utf-8") account_log = codex_home / "log" / "codex-login.log" account_log.parent.mkdir() account_log.write_text("account-log\n", encoding="utf-8") shared_cache = registry.legacy_codex_home / ".tmp" / "plugins" / ".git" / "FETCH_HEAD" shared_cache.parent.mkdir(parents=True) shared_cache.write_text("shared-cache", encoding="utf-8") account_cache = codex_home / ".tmp" / "plugins" / ".git" / "FETCH_HEAD" account_cache.parent.mkdir(parents=True) account_cache.write_text("account-cache", encoding="utf-8") registry.sync_shared_links(codex_home) for name in ( *SHARED_FIXTURE_NAMES, ".tmp", "agents", "future-state.json", "history.jsonl", "hooks.json", "log", ): shared_entry = codex_home / name assert shared_entry.is_symlink() assert shared_entry.resolve() == (registry.legacy_codex_home / name).resolve() assert (codex_home / "agents" / "reviewer.md").read_text(encoding="utf-8") == "reviewer" assert (registry.legacy_codex_home / "history.jsonl").read_text( encoding="utf-8" ).splitlines() == [ '{"session_id":"shared","ts":1,"text":"shared"}', '{"session_id":"account","ts":2,"text":"account"}', ] assert (codex_home / "auth.json").read_text(encoding="utf-8") == "account-auth" assert not (codex_home / "auth.json").is_symlink() assert (codex_home / "auth.json.bak").read_text(encoding="utf-8") == "account-backup" assert not (codex_home / "auth.json.bak").is_symlink() assert (codex_home / "state_5.sqlite").read_text(encoding="utf-8") == "account-db" assert not (codex_home / "state_5.sqlite").is_symlink() assert shared_log.read_text(encoding="utf-8") == "shared-log\naccount-log\n" assert shared_cache.read_text(encoding="utf-8") == "shared-cache" def test_sync_all_shared_links_merges_existing_conversation_state( registry: AccountRegistry, ) -> None: rafa_home = registry.accounts_dir / "rafa" / "CODEX_HOME" roy_home = registry.accounts_dir / "roy" / "CODEX_HOME" shared_session = Path("2026/07/02/rollout-shared.jsonl") rafa_only_session = Path("2026/07/02/rollout-rafa.jsonl") roy_only_archive = Path("2026/07/01/rollout-roy.jsonl") for codex_home in (rafa_home, roy_home): path = codex_home / "sessions" / shared_session path.parent.mkdir(parents=True) path.write_text("shared\n", encoding="utf-8") rafa_only = rafa_home / "sessions" / rafa_only_session rafa_only.write_text("rafa\n", encoding="utf-8") roy_archive = roy_home / "archived_sessions" / roy_only_archive roy_archive.parent.mkdir(parents=True) roy_archive.write_text("roy\n", encoding="utf-8") registry.sync_all_shared_links() registry.sync_all_shared_links() for codex_home in (rafa_home, roy_home): assert (codex_home / "sessions").is_symlink() assert (codex_home / "archived_sessions").is_symlink() assert (codex_home / "sessions" / shared_session).read_text(encoding="utf-8") == ( "shared\n" ) assert (codex_home / "sessions" / rafa_only_session).read_text( encoding="utf-8" ) == "rafa\n" assert (codex_home / "archived_sessions" / roy_only_archive).read_text( encoding="utf-8" ) == "roy\n" def test_sync_all_shared_links_preserves_conflicting_session_files( registry: AccountRegistry, ) -> None: relative_path = Path("2026/07/02/rollout-conflict.jsonl") rafa_session = registry.accounts_dir / "rafa" / "CODEX_HOME" / "sessions" / relative_path roy_session = registry.accounts_dir / "roy" / "CODEX_HOME" / "sessions" / relative_path rafa_session.parent.mkdir(parents=True) roy_session.parent.mkdir(parents=True) rafa_session.write_text("rafa\n", encoding="utf-8") roy_session.write_text("roy\n", encoding="utf-8") conflicts = registry.sync_all_shared_links() assert roy_session in conflicts canonical = registry.legacy_codex_home / "sessions" / relative_path assert canonical.read_text(encoding="utf-8") == "rafa\n" assert roy_session.read_text(encoding="utf-8") == "roy\n" def test_list_keeps_undecodable_auth_entries( registry: AccountRegistry, capsys: pytest.CaptureFixture[str] ) -> None: codex_home = registry.add_dir("broken", "Broken") (codex_home / "auth.json").write_text("{not-json", encoding="utf-8") accounts = registry.list() captured = capsys.readouterr() assert accounts == [ Account( ref=AccountRef("codex", "broken"), alias="Broken", account_home=codex_home, email=None, plan=None, account_id=None, ) ] assert accounts[0].tray_key == "codex:broken" assert "failed to decode auth.json for account 'broken'" in captured.err def test_list_ignores_profile_account_type_and_uses_plan_fallback( registry: AccountRegistry, ) -> None: codex_home = registry.add_dir("rafa", "Rafa") (codex_home / "auth.json").write_text( json.dumps( { "tokens": { "id_token": make_auth_payload_with_claims( "rafa@example.com", "acct-rafa", { "chatgpt_plan_type": "profile", "plan": "plus", }, ) } } ), encoding="utf-8", ) assert registry.list() == [ Account( ref=AccountRef("codex", "rafa"), alias="Rafa", account_home=codex_home, email="rafa@example.com", plan="plus", account_id="acct-rafa", ) ] def test_list_ignores_unrelated_auth_fields(registry: AccountRegistry) -> None: codex_home = registry.add_dir("rafa", "Rafa") codex_home.joinpath("auth.json").write_text( json.dumps( { "tokens": {"id_token": make_auth_payload("rafa@example.com", "plus", "acct-rafa")}, "rate_limits": { "primary_reset_at": "2026-07-01T12:00:00Z", "secondary_reset_at": "2026-07-07T00:00:00Z", }, } ), encoding="utf-8", ) assert registry.list() == [ Account( slug="rafa", alias="Rafa", codex_home=codex_home, email="rafa@example.com", plan="plus", account_id="acct-rafa", ) ] def test_set_default_writes_pointer_and_symlinks_courtesy_auth( registry: AccountRegistry, ) -> None: codex_home = registry.add_dir("rafa", "Rafa") other_home = registry.add_dir("roy", "Roy") write_auth(codex_home / "auth.json", "rafa@example.com", "plus", "acct-rafa") write_auth(other_home / "auth.json", "roy@example.com", "plus", "acct-roy") account = registry.list()[0] registry.set_default(account) assert registry.default_path.read_text(encoding="utf-8") == "rafa\n" courtesy = registry.legacy_codex_home / "auth.json" assert courtesy.is_symlink() assert courtesy.resolve() == (codex_home / "auth.json").resolve() assert (other_home / "auth.json").exists() (codex_home / "auth.json").write_text('{"rotated": true}', encoding="utf-8") assert courtesy.read_text(encoding="utf-8") == '{"rotated": true}' registry.set_default(registry.list()[1]) assert courtesy.is_symlink() assert courtesy.resolve() == (other_home / "auth.json").resolve() def test_set_default_touches_ratelimit_wake_file(registry: AccountRegistry) -> None: codex_home = registry.add_dir("rafa", "Rafa") write_auth(codex_home / "auth.json", "rafa@example.com", "plus", "acct-rafa") wake = registry.base_dir / "ratelimit-wake" assert not wake.exists() registry.set_default(registry.list()[0]) assert wake.exists() def test_locking_default_disconnects_live_auth_and_unlock_restores_it( registry: AccountRegistry, ) -> None: codex_home = registry.add_dir("rafa", "Rafa") write_auth(codex_home / "auth.json", "rafa@example.com", "plus", "acct-rafa") account = registry.list()[0] registry.set_default(account) live_auth = registry.legacy_codex_home / "auth.json" assert live_auth.is_symlink() assert registry.set_locked("rafa", True) is True assert registry.is_locked("rafa") is True assert not live_auth.exists() assert not live_auth.is_symlink() with pytest.raises(AccountLockedError): registry.set_default(account) assert registry.set_locked("rafa", False) is True assert registry.is_locked("rafa") is False assert live_auth.is_symlink() assert live_auth.resolve() == (codex_home / "auth.json").resolve() def test_failed_unlock_relocks_and_removes_partially_activated_credentials( registry: AccountRegistry, monkeypatch: pytest.MonkeyPatch, ) -> None: codex_home = registry.add_dir("rafa", "Rafa") write_auth(codex_home / "auth.json", "rafa@example.com", "plus", "acct-rafa") registry.set_default(registry.list()[0]) registry.set_locked("rafa", True) live_auth = registry.legacy_codex_home / "auth.json" def fail_after_activation(slug: str | None) -> None: assert slug == "rafa" registry._atomic_symlink(codex_home / "auth.json", live_auth) raise RuntimeError("activation failed") monkeypatch.setattr(registry, "_sync_active_links", fail_after_activation) with pytest.raises(RuntimeError, match="activation failed"): registry.set_locked("rafa", False) assert registry.is_locked("rafa") is True assert not live_auth.exists() assert not live_auth.is_symlink() def test_locking_default_moves_vendor_replaced_live_auth_back_to_account( registry: AccountRegistry, ) -> None: codex_home = registry.add_dir("rafa", "Rafa") write_auth(codex_home / "auth.json", "rafa@example.com", "plus", "acct-old") registry.set_default(registry.list()[0]) live_auth = registry.legacy_codex_home / "auth.json" live_auth.unlink() live_auth.write_text('{"rotated":true}', encoding="utf-8") registry.set_locked("rafa", True) assert not live_auth.exists() assert (codex_home / "auth.json").read_text(encoding="utf-8") == '{"rotated":true}' assert stat.S_IMODE((codex_home / "auth.json").stat().st_mode) == 0o600 def test_locked_credential_cleanup_falls_back_across_filesystems( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: active = tmp_path / "live" / "auth.json" account_copy = tmp_path / "account" / "auth.json" active.parent.mkdir() account_copy.parent.mkdir() active.write_text('{"rotated":true}', encoding="utf-8") account_copy.write_text('{"old":true}', encoding="utf-8") real_replace = os.replace def replace_with_cross_device_once(source: str | os.PathLike[str], target: str | os.PathLike[str]) -> None: if Path(source) == active and Path(target) == account_copy: raise OSError(errno.EXDEV, "cross-device link") real_replace(source, target) monkeypatch.setattr(account_registry_module.os, "replace", replace_with_cross_device_once) AccountRegistry._deactivate_live_path(active, account_copy) assert not active.exists() assert account_copy.read_text(encoding="utf-8") == '{"rotated":true}' assert stat.S_IMODE(account_copy.stat().st_mode) == 0o600 assert list(account_copy.parent.glob(".auth.json.lock-move-*")) == [] def test_rename_and_remove_keep_account_lock_state_consistent( registry: AccountRegistry, ) -> None: codex_home = registry.add_dir("work", "Work") write_auth(codex_home / "auth.json", "work@example.com", "plus", "acct-work") registry.set_locked("work", True) renamed = registry.rename("work", "Night Account") assert renamed == "night-account" assert registry.is_locked("work") is False assert registry.is_locked("night-account") is True registry.remove("night-account") assert registry.locked_slugs() == frozenset() def test_stage_add_commit_registers_account_without_exposing_partial_state( registry: AccountRegistry, monkeypatch: pytest.MonkeyPatch ) -> None: staged = registry.stage_add("Rafa Work") assert isinstance(staged, StagedAccount) assert staged.slug == "rafa-work" assert staged.account_home.name == "CODEX_HOME" write_auth(staged.account_home / "auth.json", "rafa@example.com", "plus", "acct-rafa") original_write_registry = registry._write_registry def verifying_write_registry(payload: dict[str, list[dict[str, str]]]) -> None: assert registry.list() == [] original_write_registry(payload) monkeypatch.setattr(registry, "_write_registry", verifying_write_registry) account = staged.commit() assert account == Account( ref=AccountRef("codex", "rafa-work"), alias="Rafa Work", account_home=registry.accounts_dir / "rafa-work" / "CODEX_HOME", email="rafa@example.com", plan="plus", account_id="acct-rafa", ) assert account.account_home.exists() assert not staged._staged_dir.exists() assert registry.list() == [account] with pytest.raises(RuntimeError, match="staged account already closed"): staged.commit() with pytest.raises(RuntimeError, match="staged account already closed"): staged.rollback() def test_stage_add_uses_explicit_slug_when_provided(registry: AccountRegistry) -> None: staged = registry.stage_add("Display Name", slug="my-work") assert staged.slug == "my-work" assert staged.alias == "Display Name" def test_stage_add_rollback_discards_directory_and_cannot_be_reused( registry: AccountRegistry, ) -> None: staged = registry.stage_add("Disposable") write_auth(staged.account_home / "auth.json", "temp@example.com", "plus", "acct-temp") staged.rollback() assert registry.list() == [] assert not staged.account_home.exists() with pytest.raises(RuntimeError, match="staged account already closed"): staged.rollback() with pytest.raises(RuntimeError, match="staged account already closed"): staged.commit() def test_stage_add_commit_rejects_concurrent_final_destination(tmp_path: Path) -> None: legacy = tmp_path / "legacy" base = tmp_path / "tray" legacy.mkdir() for name in SHARED_FIXTURE_NAMES: target = legacy / name if "." in name: target.write_text(name, encoding="utf-8") else: target.mkdir() first_registry = AccountRegistry(base_dir=base, legacy_codex_home=legacy) second_registry = AccountRegistry(base_dir=base, legacy_codex_home=legacy) first_staged = first_registry.stage_add("Roy") second_staged = second_registry.stage_add("Roy") write_auth(first_staged.account_home / "auth.json", "roy@example.com", "plus", "acct-one") write_auth(second_staged.account_home / "auth.json", "roy@example.com", "pro", "acct-two") barrier = threading.Barrier(2) results: list[tuple[str, object]] = [] def commit(staged: StagedAccount) -> None: try: barrier.wait(timeout=5) results.append(("ok", staged.commit())) except BaseException as exc: # noqa: BLE001 results.append(("err", exc)) first_thread = threading.Thread(target=commit, args=(first_staged,)) second_thread = threading.Thread(target=commit, args=(second_staged,)) first_thread.start() second_thread.start() first_thread.join() second_thread.join() successes = [value for kind, value in results if kind == "ok"] failures = [value for kind, value in results if kind == "err"] assert len(successes) == 1 assert len(failures) == 1 assert isinstance(failures[0], FileExistsError) assert first_registry.list() == [successes[0]] def test_claude_registry_uses_separate_storage_default_and_credentials_symlink(tmp_path: Path) -> None: from account_registry import AccountRegistryKind legacy = tmp_path / "claude" base = tmp_path / "tray" legacy.mkdir() registry = AccountRegistry( base_dir=base, legacy_codex_home=tmp_path / "codex", kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy, ) claude_home = registry.add_dir("rafa", "Rafa") (claude_home / ".credentials.json").write_text('{"accessToken":"first"}', encoding="utf-8") account = registry.list()[0] registry.set_default(account) assert registry.accounts_dir == base / "claude-accounts" assert registry.registry_path == base / "claude_accounts.json" assert registry.default_path == base / "claude_default_slug" assert claude_home == base / "claude-accounts" / "rafa" / "CLAUDE_HOME" assert account.ref == AccountRef("claude", "rafa") assert account.account_home == claude_home assert account.tray_key == "claude:rafa" courtesy = legacy / ".credentials.json" assert courtesy.is_symlink() assert courtesy.resolve() == (claude_home / ".credentials.json").resolve() (claude_home / ".credentials.json").write_text('{"accessToken":"rotated"}', encoding="utf-8") assert courtesy.read_text(encoding="utf-8") == '{"accessToken":"rotated"}' class _FakeTokenIdentity: def __init__(self, owner_uuid: str | None = "shared-uuid") -> None: self._owner_uuid = owner_uuid def account_uuid_for(self, _credentials_path: Path) -> str | None: return self._owner_uuid def _claude_registry(tmp_path: Path) -> AccountRegistry: legacy = tmp_path / "claude" legacy.mkdir() return AccountRegistry( base_dir=tmp_path / "tray", legacy_codex_home=tmp_path / "codex", kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy, ) def test_preserve_live_claude_grant_writes_only_to_the_pinned_matching_account( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: registry = _claude_registry(tmp_path) zync_home = registry.add_dir("zync", "Zync") other_home = registry.add_dir("other", "Other") pin_account_uuid(zync_home, "shared-uuid") pin_account_uuid(other_home, "some-other-uuid") live = registry.legacy_claude_home / ".credentials.json" live.write_text('{"rotated":true}', encoding="utf-8") monkeypatch.setattr( account_registry_module, "ClaudeTokenIdentity", lambda: _FakeTokenIdentity("shared-uuid"), raising=False, ) registry._preserve_live_claude_grant() assert (zync_home / ".credentials.json").read_text(encoding="utf-8") == '{"rotated":true}' assert stat.S_IMODE((zync_home / ".credentials.json").stat().st_mode) == 0o600 assert not (other_home / ".credentials.json").exists() def test_preserve_live_claude_grant_skips_an_account_with_a_poisoned_but_unpinned_claude_json( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: registry = _claude_registry(tmp_path) poisoned_home = registry.add_dir("poisoned", "Poisoned") (poisoned_home / ".claude.json").write_text( json.dumps({"oauthAccount": {"accountUuid": "shared-uuid"}}), encoding="utf-8", ) live = registry.legacy_claude_home / ".credentials.json" live.write_text('{"rotated":true}', encoding="utf-8") monkeypatch.setattr( account_registry_module, "ClaudeTokenIdentity", lambda: _FakeTokenIdentity("shared-uuid"), raising=False, ) registry._preserve_live_claude_grant() assert not (poisoned_home / ".credentials.json").exists() def test_claude_registry_resolves_default_home_paths_at_construction( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setattr(account_registry_module.Path, "home", lambda: tmp_path) registry = account_registry_module.AccountRegistry( kind=account_registry_module.AccountRegistryKind.CLAUDE ) assert registry.base_dir == tmp_path / ".local/state/overdeck/systray/runtime" assert registry.legacy_claude_home == tmp_path / ".claude" assert registry.active_claude_json == tmp_path / ".claude.json" def test_claude_registry_derives_metadata_path_from_custom_claude_home( tmp_path: Path, ) -> None: claude_home = tmp_path / ".claude" registry = account_registry_module.AccountRegistry( base_dir=tmp_path / "tray", kind=account_registry_module.AccountRegistryKind.CLAUDE, legacy_claude_home=claude_home, ) assert registry.active_claude_json == tmp_path / ".claude.json" def test_claude_migrate_legacy_imports_current_credentials_and_sets_default(tmp_path: Path) -> None: from account_registry import AccountRegistryKind legacy_claude_home = tmp_path / "claude" legacy_codex_home = tmp_path / "codex" base = tmp_path / "tray" legacy_claude_home.mkdir() legacy_codex_home.mkdir() (legacy_claude_home / ".credentials.json").write_text( '{"accessToken":"legacy"}', encoding="utf-8", ) active_claude_json = tmp_path / ".claude.json" active_claude_json.write_text( '{"oauthAccount":{"emailAddress":"legacy@example.com","organizationRateLimitTier":"max"}}', encoding="utf-8", ) registry = AccountRegistry( base_dir=base, legacy_codex_home=legacy_codex_home, kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy_claude_home, active_claude_json=active_claude_json, ) registry.migrate_legacy() registry.migrate_legacy() accounts = registry.list() assert [account.slug for account in accounts] == ["legacy"] assert registry.default_slug() == "legacy" claude_home = base / "claude-accounts" / "legacy" / "CLAUDE_HOME" assert (claude_home / ".credentials.json").read_text(encoding="utf-8") == '{"accessToken":"legacy"}' assert (claude_home / "claude.json").read_text(encoding="utf-8") == ( '{"oauthAccount":{"emailAddress":"legacy@example.com","organizationRateLimitTier":"max"}}' ) def test_claude_migrate_legacy_does_not_import_phantom_when_registry_exists( tmp_path: Path, ) -> None: from account_registry import AccountRegistryKind legacy_claude_home = tmp_path / "claude" legacy_codex_home = tmp_path / "codex" active_claude_json = tmp_path / ".claude.json" base = tmp_path / "tray" legacy_claude_home.mkdir() legacy_codex_home.mkdir() (legacy_claude_home / ".credentials.json").write_text( '{"accessToken":"live-other"}', encoding="utf-8", ) active_claude_json.write_text( '{"oauthAccount":{"emailAddress":"other@example.com"}}', encoding="utf-8", ) registry = AccountRegistry( base_dir=base, legacy_codex_home=legacy_codex_home, kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy_claude_home, active_claude_json=active_claude_json, ) claude_home = registry.add_dir("avi", "Avi") (claude_home / ".credentials.json").write_text( '{"accessToken":"avi"}', encoding="utf-8", ) registry.migrate_legacy() assert [account.slug for account in registry.list()] == ["avi"] assert registry.default_slug() == "avi" assert not (base / "claude-accounts" / "other").exists() def test_claude_list_hides_registry_entries_with_empty_credentials(tmp_path: Path) -> None: from account_registry import AccountRegistryKind legacy_claude_home = tmp_path / "claude" legacy_codex_home = tmp_path / "codex" base = tmp_path / "tray" legacy_claude_home.mkdir() legacy_codex_home.mkdir() registry = AccountRegistry( base_dir=base, legacy_codex_home=legacy_codex_home, kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy_claude_home, active_claude_json=tmp_path / ".claude.json", ) live_home = registry.add_dir("live", "Live") phantom_home = registry.add_dir("phantom", "Phantom") (live_home / ".credentials.json").write_text( '{"claudeAiOauth":{"accessToken":"live-token","refreshToken":""}}', encoding="utf-8", ) (phantom_home / ".credentials.json").write_text( '{"claudeAiOauth":{"accessToken":"","refreshToken":""}}', encoding="utf-8", ) accounts = registry.list() assert [account.slug for account in accounts] == ["live"] def test_claude_list_keeps_identified_account_with_cleared_credentials( tmp_path: Path, ) -> None: from account_registry import AccountRegistryKind legacy_claude_home = tmp_path / "claude" legacy_codex_home = tmp_path / "codex" legacy_claude_home.mkdir() legacy_codex_home.mkdir() registry = AccountRegistry( base_dir=tmp_path / "tray", legacy_codex_home=legacy_codex_home, kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy_claude_home, active_claude_json=tmp_path / ".claude.json", ) account_home = registry.add_dir("logged-out", "Logged out") (account_home / ".credentials.json").write_text( '{"claudeAiOauth":{"accessToken":"","refreshToken":""}}', encoding="utf-8", ) (account_home / "account_identity.json").write_text( '{"email":"user@example.com","plan":"max"}', encoding="utf-8", ) accounts = registry.list() assert [(account.slug, account.email) for account in accounts] == [ ("logged-out", "user@example.com") ] def test_claude_migrate_legacy_skips_live_symlink_into_managed_storage( tmp_path: Path, ) -> None: from account_registry import AccountRegistryKind legacy_claude_home = tmp_path / "claude" legacy_codex_home = tmp_path / "codex" base = tmp_path / "tray" managed_home = base / "claude-accounts" / "avi" / "CLAUDE_HOME" legacy_claude_home.mkdir() legacy_codex_home.mkdir() managed_home.mkdir(parents=True) (managed_home / ".credentials.json").write_text('{"accessToken":"avi"}', encoding="utf-8") (legacy_claude_home / ".credentials.json").symlink_to(managed_home / ".credentials.json") registry = AccountRegistry( base_dir=base, legacy_codex_home=legacy_codex_home, kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy_claude_home, active_claude_json=tmp_path / ".claude.json", ) registry.migrate_legacy() assert registry.list() == [] assert not registry.default_path.exists() def test_claude_set_default_links_dot_claude_json_metadata(tmp_path: Path) -> None: from account_registry import AccountRegistryKind legacy_claude_home = tmp_path / "claude" legacy_codex_home = tmp_path / "codex" active_claude_json = tmp_path / ".claude.json" base = tmp_path / "tray" legacy_claude_home.mkdir() legacy_codex_home.mkdir() registry = AccountRegistry( base_dir=base, legacy_codex_home=legacy_codex_home, kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy_claude_home, active_claude_json=active_claude_json, ) claude_home = registry.add_dir("avi", "Avi") (claude_home / ".credentials.json").write_text('{"accessToken":"avi"}', encoding="utf-8") (claude_home / ".claude.json").write_text( '{"oauthAccount":{"emailAddress":"avi@example.com"}}', encoding="utf-8", ) registry.set_default(registry.list()[0]) assert active_claude_json.is_symlink() assert active_claude_json.resolve() == (claude_home / ".claude.json").resolve() assert json.loads(active_claude_json.read_text(encoding="utf-8")) == { "oauthAccount": {"emailAddress": "avi@example.com"}, "hasCompletedOnboarding": True, } def test_claude_set_default_ignores_generated_dot_config_stub(tmp_path: Path) -> None: from account_registry import AccountRegistryKind legacy_claude_home = tmp_path / "claude" legacy_codex_home = tmp_path / "codex" active_claude_json = tmp_path / ".claude.json" base = tmp_path / "tray" legacy_claude_home.mkdir() legacy_codex_home.mkdir() registry = AccountRegistry( base_dir=base, legacy_codex_home=legacy_codex_home, kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy_claude_home, active_claude_json=active_claude_json, ) claude_home = registry.add_dir("avi", "Avi") (claude_home / ".credentials.json").write_text('{"accessToken":"avi"}', encoding="utf-8") (claude_home / ".claude.json").write_text( '{"firstStartTime":"now","migrationVersion":13}', encoding="utf-8", ) (claude_home / "claude.json").write_text( '{"oauthAccount":{"emailAddress":"avi@example.com"},"projects":{"/repo":{}}}', encoding="utf-8", ) registry.set_default(registry.list()[0]) assert active_claude_json.is_symlink() assert active_claude_json.resolve() == (claude_home / "claude.json").resolve() assert json.loads(active_claude_json.read_text(encoding="utf-8")) == { "oauthAccount": {"emailAddress": "avi@example.com"}, "projects": {"/repo": {}}, "hasCompletedOnboarding": True, } def test_claude_list_prefers_account_identity_over_legacy_claude_json(tmp_path: Path) -> None: from account_registry import AccountRegistryKind legacy_claude_home = tmp_path / "claude" legacy_codex_home = tmp_path / "codex" base = tmp_path / "tray" legacy_claude_home.mkdir() legacy_codex_home.mkdir() registry = AccountRegistry( base_dir=base, legacy_codex_home=legacy_codex_home, kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy_claude_home, ) claude_home = registry.add_dir("rafa", "Rafa") (claude_home / "account_identity.json").write_text( json.dumps( { "email": "preferred@example.com", "org_id": "org-preferred", "subscription_type": "max", } ), encoding="utf-8", ) (claude_home / "claude.json").write_text( json.dumps( { "oauthAccount": { "emailAddress": "legacy@example.com", "accountUuid": "legacy-account", "organizationRateLimitTier": "legacy-plan", } } ), encoding="utf-8", ) assert registry.list() == [ Account( ref=AccountRef("claude", "rafa"), alias="Rafa", account_home=claude_home, email="preferred@example.com", plan="max", account_id="org-preferred", ) ] def test_rename_moves_the_account_directory_to_the_new_slug( registry: AccountRegistry, ) -> None: codex_home = registry.add_dir("roy", "Roy") write_auth(codex_home / "auth.json", "roy@example.com", "plus", "acct-roy") registry.set_default(registry.list()[0]) assert registry.rename("roy", "Roy Work") == "roy-work" account = registry.list()[0] assert (account.slug, account.alias) == ("roy-work", "Roy Work") assert (registry.accounts_dir / "roy-work" / "CODEX_HOME" / "auth.json").is_file() assert not (registry.accounts_dir / "roy").exists() assert registry.default_slug() == "roy-work" def test_rename_keeps_the_slug_when_the_label_still_derives_it( registry: AccountRegistry, ) -> None: registry.add_dir("roy", "Roy") assert registry.rename("roy", "ROY") == "roy" assert (registry.accounts_dir / "roy").is_dir() def test_rename_rekeys_health_cache_and_routing_rules(registry: AccountRegistry) -> None: registry.add_dir("roy", "Roy") registry.add_dir("avi", "Avi") registry.health_cache_path.write_text( json.dumps({"roy": {"status": "ok"}, "avi": {"status": "ok"}}), encoding="utf-8" ) assert registry.routing_rules_path is not None registry.routing_rules_path.write_text( json.dumps( { "projects": {"zync": {"account": "roy"}, "other": {"account": "avi"}}, "default": "roy", "fallback_chain": ["avi", "roy"], } ), encoding="utf-8", ) registry.rename("roy", "Roy Work") health = json.loads(registry.health_cache_path.read_text(encoding="utf-8")) assert set(health) == {"roy-work", "avi"} rules = json.loads(registry.routing_rules_path.read_text(encoding="utf-8")) assert rules["projects"]["zync"]["account"] == "roy-work" assert rules["projects"]["other"]["account"] == "avi" assert rules["default"] == "roy-work" assert rules["fallback_chain"] == ["avi", "roy-work"] def test_rename_rekeys_account_caps(registry: AccountRegistry) -> None: registry.add_dir("roy", "Roy") registry.add_dir("avi", "Avi") assert registry.routing_rules_path is not None registry.routing_rules_path.write_text( json.dumps( { "projects": {}, "default": "roy", "account_caps": {"roy": {"7d": 40}, "avi": {"5h": 20}}, } ), encoding="utf-8", ) registry.rename("roy", "Roy Work") rules = json.loads(registry.routing_rules_path.read_text(encoding="utf-8")) assert rules["account_caps"] == {"roy-work": {"7d": 40}, "avi": {"5h": 20}} def test_remove_drops_account_caps(registry: AccountRegistry) -> None: registry.add_dir("roy", "Roy") avi_home = registry.add_dir("avi", "Avi") write_auth(avi_home / "auth.json", "avi@example.com", "plus", "acct-avi") assert registry.routing_rules_path is not None registry.routing_rules_path.write_text( json.dumps( { "projects": {}, "default": "avi", "account_caps": {"roy": {"7d": 40}, "avi": {"5h": 20}}, } ), encoding="utf-8", ) registry.remove("roy") rules = json.loads(registry.routing_rules_path.read_text(encoding="utf-8")) assert rules["account_caps"] == {"avi": {"5h": 20}} def test_rename_rejects_a_slug_already_taken_on_disk(registry: AccountRegistry) -> None: registry.add_dir("roy", "Roy") (registry.accounts_dir / "roy-work").mkdir() with pytest.raises(FileExistsError, match="account directory already in use: roy-work"): registry.rename("roy", "Roy Work") assert registry.list()[0].slug == "roy" assert (registry.accounts_dir / "roy").is_dir() def test_account_names_are_unique_across_slugs_aliases_and_reserved_names( registry: AccountRegistry, ) -> None: registry.add_dir("roy", "Personal") with pytest.raises(FileExistsError, match="account name already in use: ROY"): registry.add_dir("work", "ROY") with pytest.raises(FileExistsError, match="account name already in use: personal"): registry.stage_add("personal") with pytest.raises(ValueError, match="reserved account name: Dynamic"): registry.rename("roy", "Dynamic") def test_rename_rejects_another_account_slug_but_allows_own_slug( registry: AccountRegistry, ) -> None: registry.add_dir("roy", "Personal") registry.add_dir("work", "Office") registry.rename("roy", "ROY") with pytest.raises(FileExistsError, match="account name already in use: work"): registry.rename("roy", "work") def test_resolve_profile_token_prefers_exact_slug_then_casefolded_alias( registry: AccountRegistry, ) -> None: for slug in ("roy", "work"): (registry.accounts_dir / slug / "CODEX_HOME").mkdir(parents=True) registry._write_registry( { "accounts": [ {"slug": "roy", "alias": "Personal"}, {"slug": "work", "alias": "Roy"}, ] } ) assert registry.resolve_profile_token("roy") == "roy" assert registry.resolve_profile_token(" personal ") == "roy" assert registry.resolve_profile_token("ROY") == "work" assert registry.resolve_profile_token("dynamic") is None assert registry.resolve_profile_token("missing") is None def test_resolve_follows_the_slug_a_rename_produced(registry: AccountRegistry) -> None: registry.add_dir("avi", "Avi") assert registry.resolve_profile_token("Avi") == "avi" registry.rename("avi", "Roy Home") assert registry.resolve_profile_token("Avi") is None assert registry.resolve_profile_token("avi") is None assert registry.resolve_profile_token("Roy Home") == "roy-home" assert registry.resolve_profile_token("roy-home") == "roy-home" def test_remove_reassigns_default_and_clears_active_links_on_final_removal( registry: AccountRegistry, ) -> None: rafa_home = registry.add_dir("rafa", "Rafa") roy_home = registry.add_dir("roy", "Roy") write_auth(rafa_home / "auth.json", "rafa@example.com", "plus", "acct-rafa") write_auth(roy_home / "auth.json", "roy@example.com", "pro", "acct-roy") registry.set_default(registry.list()[0]) registry.remove("rafa") courtesy = registry.legacy_codex_home / "auth.json" assert [account.slug for account in registry.list()] == ["roy"] assert registry.default_slug() == "roy" assert registry.default_path.read_text(encoding="utf-8") == "roy\n" assert courtesy.is_symlink() assert courtesy.resolve() == (roy_home / "auth.json").resolve() registry.remove("roy") assert registry.list() == [] assert registry.default_slug() is None assert not registry.default_path.exists() assert not courtesy.exists() def test_remove_restores_quarantined_account_when_state_write_fails( registry: AccountRegistry, monkeypatch: pytest.MonkeyPatch ) -> None: rafa_home = registry.add_dir("rafa", "Rafa") roy_home = registry.add_dir("roy", "Roy") write_auth(rafa_home / "auth.json", "rafa@example.com", "plus", "acct-rafa") write_auth(roy_home / "auth.json", "roy@example.com", "pro", "acct-roy") registry.set_default(registry.list()[0]) original_atomic_write_text = account_registry_module.AccountRegistry._atomic_write_text failed = False def fail_registry_write(path: Path, content: str) -> None: nonlocal failed original_atomic_write_text(path, content) if path == registry.registry_path and not failed: failed = True raise OSError("boom") monkeypatch.setattr( account_registry_module.AccountRegistry, "_atomic_write_text", staticmethod(fail_registry_write), ) with pytest.raises(OSError, match="boom"): registry.remove("rafa") courtesy = registry.legacy_codex_home / "auth.json" assert [account.slug for account in registry.list()] == ["rafa", "roy"] assert registry.default_slug() == "rafa" assert courtesy.is_symlink() assert courtesy.resolve() == (rafa_home / "auth.json").resolve() assert (registry.accounts_dir / "rafa" / "CODEX_HOME" / "auth.json").exists() def test_add_dir_raises_when_slug_already_exists(registry: AccountRegistry) -> None: registry.add_dir("roy", "Roy") with pytest.raises(FileExistsError): registry.add_dir("roy", "Roy Again") def test_account_registry_delegates_shared_codex_state( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: calls: dict[str, object] = { "init": None, "sync_shared_links": [], "sync_all_shared_links": 0, } class FakeSharedCodexState: def __init__(self, legacy_codex_home: Path, accounts_dir: Path) -> None: calls["init"] = (legacy_codex_home, accounts_dir) def sync_shared_links(self, account_home: Path) -> None: calls["sync_shared_links"].append(account_home) def sync_all_shared_links(self) -> None: calls["sync_all_shared_links"] += 1 monkeypatch.setattr(account_registry_module, "SharedCodexState", FakeSharedCodexState) legacy = tmp_path / "legacy" base = tmp_path / "tray" legacy.mkdir() registry = account_registry_module.AccountRegistry(base_dir=base, legacy_codex_home=legacy) account_home = registry.accounts_dir / "roy" / "CODEX_HOME" account_home.mkdir(parents=True) registry.sync_shared_links(account_home) registry.sync_all_shared_links() assert calls["init"] == (legacy, registry.accounts_dir) assert calls["sync_shared_links"] == [account_home] assert calls["sync_all_shared_links"] == 1 def test_shell_shim_delegates_to_account_registry_suite( monkeypatch: pytest.MonkeyPatch, ) -> None: shim_path = Path(__file__) spec = importlib.util.spec_from_file_location("test_account_registry_shim", shim_path) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) pytest_call: dict[str, object] = {} def fake_pytest_main(args: list[str]) -> int: pytest_call["args"] = args return 0 monkeypatch.setattr(module.pytest, "main", fake_pytest_main) assert module.main() == 0 assert pytest_call["args"] == ["tests/test_account_registry.py", "-v"] def test_shell_shim_runs_when_invoked_via_shell(tmp_path: Path) -> None: if os.environ.get("ACCOUNT_REGISTRY_SHIM_ACTIVE") == "1": pytest.skip("shell shim subprocess test must not recurse under shim execution") repo_root = Path(__file__).resolve().parent.parent shim_path = repo_root / "tests" / "test_account_registry.py" # Shim execs bare `python3`: pin it to this suite's interpreter (has pytest). env = os.environ.copy() env["PATH"] = f"{Path(sys.executable).parent}{os.pathsep}{env.get('PATH', '')}" env["TMPDIR"] = str(tmp_path) result = subprocess.run( ["sh", str(shim_path)], cwd=repo_root, capture_output=True, text=True, check=False, env=env, ) assert result.returncode == 0, result.stderr def test_claude_commit_seeds_first_run_flags_without_leaking_identity( tmp_path: Path, ) -> None: from account_registry import AccountRegistryKind legacy_claude_home = tmp_path / "claude" legacy_codex_home = tmp_path / "codex" active_claude_json = tmp_path / ".claude.json" base = tmp_path / "tray" legacy_claude_home.mkdir() legacy_codex_home.mkdir() active_claude_json.write_text( json.dumps( { "oauthAccount": {"emailAddress": "active@example.com"}, "userID": "active-user-id", "machineID": "active-machine-id", "hasCompletedOnboarding": True, "lastOnboardingVersion": "2.1.211", "officialMarketplaceAutoInstallAttempted": True, "officialMarketplaceAutoInstalled": True, "migrationVersion": 13, } ), encoding="utf-8", ) registry = AccountRegistry( base_dir=base, legacy_codex_home=legacy_codex_home, kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy_claude_home, active_claude_json=active_claude_json, ) staged = registry.stage_add("Rafa") (staged.account_home / ".credentials.json").write_text( '{"accessToken":"rafa"}', encoding="utf-8" ) (staged.account_home / ".claude.json").write_text( json.dumps( { "oauthAccount": {"emailAddress": "rafa@example.com"}, "firstStartTime": "2026-07-16T00:00:00Z", } ), encoding="utf-8", ) staged.commit() seeded = json.loads( (base / "claude-accounts" / "rafa" / "CLAUDE_HOME" / ".claude.json").read_text( encoding="utf-8" ) ) assert seeded["hasCompletedOnboarding"] is True assert seeded["officialMarketplaceAutoInstallAttempted"] is True assert seeded["migrationVersion"] == 13 assert seeded["oauthAccount"] == {"emailAddress": "rafa@example.com"} assert seeded["firstStartTime"] == "2026-07-16T00:00:00Z" assert "userID" not in seeded assert "machineID" not in seeded def _claude_registry(tmp_path: Path) -> AccountRegistry: legacy_claude_home = tmp_path / "claude" legacy_codex_home = tmp_path / "codex" legacy_claude_home.mkdir(exist_ok=True) legacy_codex_home.mkdir(exist_ok=True) return AccountRegistry( base_dir=tmp_path / "tray", legacy_codex_home=legacy_codex_home, kind=AccountRegistryKind.CLAUDE, legacy_claude_home=legacy_claude_home, active_claude_json=tmp_path / ".claude.json", ) @pytest.mark.parametrize( ("metadata_name", "metadata"), [ (".claude.json", {"oauthAccount": {"emailAddress": "a@example.com"}}), ("claude.json", {"firstStartTime": "2026-07-16T00:00:00Z"}), ], ) def test_seed_first_run_flags_targets_resolved_metadata( tmp_path: Path, metadata_name: str, metadata: dict[str, object], ) -> None: registry = _claude_registry(tmp_path) registry.active_claude_json.write_text( json.dumps({"officialMarketplaceAutoInstalled": True}), encoding="utf-8", ) account_home = tmp_path / "account" account_home.mkdir() target = account_home / metadata_name target.write_text(json.dumps(metadata), encoding="utf-8") registry._seed_first_run_flags(account_home) assert json.loads(target.read_text(encoding="utf-8"))[ "hasCompletedOnboarding" ] is True other_name = "claude.json" if metadata_name == ".claude.json" else ".claude.json" assert not (account_home / other_name).exists() def test_seed_first_run_flags_forces_false_to_true_with_empty_source( tmp_path: Path, ) -> None: registry = _claude_registry(tmp_path) registry.active_claude_json.write_text("{}", encoding="utf-8") account_home = tmp_path / "account" account_home.mkdir() target = account_home / "claude.json" target.write_text( json.dumps({"hasCompletedOnboarding": False, "identity": "preserved"}), encoding="utf-8", ) registry._seed_first_run_flags(account_home) assert json.loads(target.read_text(encoding="utf-8")) == { "hasCompletedOnboarding": True, "identity": "preserved", } def test_seed_first_run_flags_is_idempotent( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: registry = _claude_registry(tmp_path) registry.active_claude_json.write_text("{}", encoding="utf-8") account_home = tmp_path / "account" account_home.mkdir() target = account_home / "claude.json" target.write_text( json.dumps({"hasCompletedOnboarding": True}), encoding="utf-8", ) writes: list[tuple[Path, str]] = [] monkeypatch.setattr( registry, "_atomic_write_text", lambda path, text: writes.append((path, text)), ) registry._seed_first_run_flags(account_home) assert writes == [] @pytest.mark.parametrize("malformed", ["not-json", "[]"]) def test_seed_first_run_flags_does_not_overwrite_malformed_target( tmp_path: Path, malformed: str, ) -> None: registry = _claude_registry(tmp_path) registry.active_claude_json.write_text("{}", encoding="utf-8") account_home = tmp_path / "account" account_home.mkdir() target = account_home / "claude.json" target.write_text(malformed, encoding="utf-8") registry._seed_first_run_flags(account_home) assert target.read_text(encoding="utf-8") == malformed def test_set_default_seeds_preexisting_stale_claude_account(tmp_path: Path) -> None: registry = _claude_registry(tmp_path) registry.active_claude_json.write_text("{}", encoding="utf-8") staged = registry.stage_add("stale") (staged.account_home / ".credentials.json").write_text("{}", encoding="utf-8") target = staged.account_home / "claude.json" target.write_text("{}", encoding="utf-8") account = staged.commit() target = account.account_home / "claude.json" target.write_text( json.dumps({"hasCompletedOnboarding": False}), encoding="utf-8", ) registry.set_default(account) assert json.loads(target.read_text(encoding="utf-8"))[ "hasCompletedOnboarding" ] is True def test_set_default_backs_up_settings_even_when_seed_skipped(tmp_path: Path) -> None: # Malformed target metadata makes _seed_first_run_flags a no-op — exactly when Claude # will hit first-run and clobber the shared settings.json — so activation must still # back it up unconditionally, not only when seeding succeeded. registry = _claude_registry(tmp_path) registry.active_claude_json.write_text("{}", encoding="utf-8") staged = registry.stage_add("malformed") (staged.account_home / ".credentials.json").write_text("{}", encoding="utf-8") (staged.account_home / "claude.json").write_text("{}", encoding="utf-8") account = staged.commit() (account.account_home / "claude.json").write_text("not-json", encoding="utf-8") settings = registry.legacy_claude_home / "settings.json" settings.write_bytes(b'{"hooks": ["x"]}\n') registry.set_default(account) backups = list((registry.legacy_claude_home / "backups").glob("settings.json.*")) assert len(backups) == 1 assert backups[0].read_bytes() == b'{"hooks": ["x"]}\n' def test_backup_claude_settings_preserves_bytes_and_source(tmp_path: Path) -> None: registry = _claude_registry(tmp_path) settings = registry.legacy_claude_home / "settings.json" original = b'{"hooks": ["\xff"]}\n\x00' settings.write_bytes(original) registry._backup_claude_settings() backups = list((registry.legacy_claude_home / "backups").glob("settings.json.*")) assert len(backups) == 1 assert backups[0].read_bytes() == original assert settings.read_bytes() == original def test_backup_claude_settings_missing_file_is_noop(tmp_path: Path) -> None: registry = _claude_registry(tmp_path) registry._backup_claude_settings() assert not (registry.legacy_claude_home / "backups").exists() def test_backup_claude_settings_retains_newest_five(tmp_path: Path) -> None: registry = _claude_registry(tmp_path) settings = registry.legacy_claude_home / "settings.json" settings.write_bytes(b"current") backups_dir = registry.legacy_claude_home / "backups" backups_dir.mkdir() for index in range(6): (backups_dir / f"settings.json.20260718T00000{index}.000000Z").write_bytes( str(index).encode() ) registry._backup_claude_settings() backups = sorted(backups_dir.glob("settings.json.*")) assert len(backups) == 5 assert not (backups_dir / "settings.json.20260718T000000.000000Z").exists() assert settings.read_bytes() == b"current" def test_authority_binding_round_trips_and_native_remains_absent( registry: AccountRegistry, tmp_path: Path ) -> None: account_home = registry.add_dir("dark", "Dark") write_auth(account_home / "auth.json", "dark@example.com", "plus", "acct-dark") binding = AuthorityBinding( mode=AuthorityMode.SUBROUTER_DARK, authority_name="workstation", route_id="route-fixture", provider="codex", proxy_grant_ref=(tmp_path / "grant.key").resolve(), ) bound = registry.set_authority_binding("dark", binding) assert bound.authority_binding == binding assert registry.list()[0].authority_binding == binding persisted = json.loads(registry.registry_path.read_text(encoding="utf-8")) assert persisted["accounts"][0]["authority_binding"] == binding.as_dict() assert "token" not in json.dumps(persisted).lower() native = registry.set_authority_binding("dark", None) assert native.authority_binding is None assert "authority_binding" not in json.loads( registry.registry_path.read_text(encoding="utf-8") )["accounts"][0] def test_authority_binding_survives_rekey_and_removal_does_not_touch_grant( registry: AccountRegistry, tmp_path: Path ) -> None: account_home = registry.add_dir("dark", "Dark") write_auth(account_home / "auth.json", "dark@example.com", "plus", "acct-dark") grant = tmp_path / "grant.key" grant.write_text("synthetic", encoding="utf-8") binding = AuthorityBinding( mode=AuthorityMode.SUBROUTER_DARK, authority_name="workstation", route_id="route-fixture", provider="codex", proxy_grant_ref=grant.resolve(), ) registry.set_authority_binding("dark", binding) renamed = registry.rename("dark", "Renamed") assert registry.list()[0].slug == renamed assert registry.list()[0].authority_binding == binding registry.remove(renamed) assert grant.read_text(encoding="utf-8") == "synthetic" @pytest.mark.parametrize("route_id", [".", "..", "../admin", "route/name", "route%2fname"]) def test_authority_binding_rejects_ambiguous_route_ids( route_id: str, tmp_path: Path ) -> None: with pytest.raises(ValueError, match="route identifier"): AuthorityBinding( mode=AuthorityMode.SUBROUTER_DARK, authority_name="workstation", route_id=route_id, provider="codex", proxy_grant_ref=(tmp_path / "grant.key").resolve(), ) def test_registry_rejects_inline_native_binding( registry: AccountRegistry, tmp_path: Path ) -> None: registry.add_dir("native", "Native") binding = AuthorityBinding( mode=AuthorityMode.NATIVE, authority_name="workstation", route_id="route-fixture", provider="codex", proxy_grant_ref=(tmp_path / "grant.key").resolve(), ) with pytest.raises(ValueError, match="must omit"): registry.set_authority_binding("native", binding) def test_account_construction_rejects_native_or_cross_provider_binding( tmp_path: Path, ) -> None: base = { "authority_name": "workstation", "route_id": "route-fixture", "proxy_grant_ref": (tmp_path / "grant.key").resolve(), } for binding in ( AuthorityBinding(mode=AuthorityMode.NATIVE, provider="codex", **base), AuthorityBinding(mode=AuthorityMode.SUBROUTER_DARK, provider="claude", **base), ): with pytest.raises(ValueError): Account( ref=AccountRef("codex", "fixture"), alias="Fixture", account_home=tmp_path / "CODEX_HOME", email=None, plan=None, account_id=None, authority_binding=binding, ) def test_authority_binding_rejects_inline_secrets_and_provider_mismatch( registry: AccountRegistry, tmp_path: Path ) -> None: registry.add_dir("dark", "Dark") payload = { "mode": "subrouter-dark", "authority_name": "workstation", "route_id": "route-fixture", "provider": "claude", "proxy_grant_ref": str((tmp_path / "grant.key").resolve()), } registry.registry_path.write_text( json.dumps({"accounts": [{"slug": "dark", "alias": "Dark", "authority_binding": payload}]}), encoding="utf-8", ) with pytest.raises(ValueError, match="provider"): registry.list() payload["provider"] = "codex" payload["refresh_token"] = "synthetic" registry.registry_path.write_text( json.dumps({"accounts": [{"slug": "dark", "alias": "Dark", "authority_binding": payload}]}), encoding="utf-8", ) with pytest.raises(ValueError, match="unknown or missing"): registry.list() def main() -> int: return pytest.main(["tests/test_account_registry.py", "-v"]) if __name__ == "__main__": raise SystemExit(main()) def test_authority_binding_quiesced_is_additive_and_legacy_payload_defaults_false( registry: AccountRegistry, tmp_path: Path ) -> None: registry.add_dir("fixture-quiesce", "Fixture Quiesce") legacy = { "mode": "subrouter", "authority_name": "workstation", "route_id": "route-fixture", "provider": "codex", "proxy_grant_ref": str((tmp_path / "grant.key").resolve()), } registry.registry_path.write_text( json.dumps( { "accounts": [ { "slug": "fixture-quiesce", "alias": "Fixture Quiesce", "authority_binding": legacy, } ], "deleted_legacy_slugs": [], } ), encoding="utf-8", ) account = registry.list()[0] assert account.authority_binding is not None assert account.authority_binding.quiesced is False quiesced = AuthorityBinding( mode=AuthorityMode.SUBROUTER, authority_name="workstation", route_id="route-fixture", provider="codex", proxy_grant_ref=(tmp_path / "grant.key").resolve(), quiesced=True, ) registry.set_authority_binding("fixture-quiesce", quiesced) persisted = json.loads(registry.registry_path.read_text(encoding="utf-8")) assert persisted["accounts"][0]["authority_binding"]["quiesced"] is True bad = dict(persisted["accounts"][0]["authority_binding"]) bad["quiesced"] = "yes" persisted["accounts"][0]["authority_binding"] = bad registry.registry_path.write_text(json.dumps(persisted), encoding="utf-8") with pytest.raises(ValueError, match="quiesced"): registry.list() def test_authority_binding_for_is_metadata_only_and_never_reads_native_auth( registry: AccountRegistry, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: registry.add_dir("fixture-binding-only", "Fixture Binding Only") grant = (tmp_path / "grant.key").resolve() binding = AuthorityBinding( mode=AuthorityMode.SUBROUTER, authority_name="workstation", route_id="routefixture1234567890", provider="codex", proxy_grant_ref=grant, ) registry.set_authority_binding("fixture-binding-only", binding) monkeypatch.setattr( registry, "_read_account_metadata", lambda *_args, **_kwargs: pytest.fail("binding lookup read native account metadata"), ) assert registry.authority_binding_for("fixture-binding-only") == binding with pytest.raises(KeyError): registry.authority_binding_for("absent")