"""Exercise the public opt-in CLI with isolated pools and no provider/credential access.""" import json import os from pathlib import Path import subprocess import tempfile import tomllib import unittest REPO = Path(__file__).resolve().parents[1] KEY = "max_concurrent_threads_per_session" class CodexSettingsTests(unittest.TestCase): def setUp(self): self.work = tempfile.TemporaryDirectory() self.addCleanup(self.work.cleanup) self.home = Path(self.work.name) self.pool = self.home / "pool" self.pool.mkdir() (self.home / ".codex").mkdir() self.global_config = self.home / ".codex/config.toml" self.global_config.write_text('model = "global"\n[features]\nmulti_agent_v2 = true\n') self.manifest = {"version": 1, "server": "none", "accounts": [ {"id": "acct-01", "email": "fixture@example.test", "home": "mac"}]} (self.pool / "accounts.json").write_text(json.dumps(self.manifest)) self.account = self.pool / "acct-01/config.toml" self.account.parent.mkdir() self.account.write_text('model = "account"\n[agents]\nmax_threads = 3\n') self.env = {**os.environ, "HOME": str(self.home), "CODEX_ACCOUNTS_ROOT": str(self.pool), "CLAUDE_MULTIACC_PYTHON": "python3"} def command(self, *args): return subprocess.run([str(REPO / "bin/codex-accounts"), *args], env=self.env, capture_output=True, text=True, check=False) def test_configure_opt_in_preserves_other_settings_and_retired_dirs(self): retired = self.pool / "acct-99/config.toml" retired.parent.mkdir() retired.write_text("retired = true\n") original_global = self.global_config.read_text() for _ in range(2): result = self.command("configure", "--max-subagents", "8", "--json") self.assertEqual(result.returncode, 0, result.stderr) report = json.loads(result.stdout) self.assertTrue(report["verified"]) self.assertEqual(report["accounts_total"], 1) self.assertEqual(report["accounts_configured"], 1) self.assertNotIn("acct-01", result.stdout) self.assertNotIn("fixture@example", result.stdout) self.assertEqual(tomllib.loads(self.account.read_text())["model"], "account") self.assertEqual(self.global_config.read_text(), original_global) self.assertEqual(retired.read_text(), "retired = true\n") def test_global_opt_in_and_read_only_drift_detection(self): result = self.command("configure", "--max-subagents", "8", "--include-global", "--json") self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(json.loads(result.stdout)["global_configured"]) self.account.write_text('[agents]\nmax_concurrent_threads_per_session = 20\n') before = self.account.read_text() result = self.command("configure", "--max-subagents", "8", "--include-global", "--check", "--json") self.assertEqual(result.returncode, 1) self.assertEqual(json.loads(result.stdout)["mismatch_count"], 1) self.assertFalse(json.loads(result.stdout)["verified"]) self.assertEqual(self.account.read_text(), before) def test_future_seeding_and_local_sync_apply_policy(self): result = self.command("configure", "--max-subagents", "8") self.assertEqual(result.returncode, 0, result.stderr) self.manifest["accounts"].append({"id": "acct-02", "email": "new@example.test", "home": "mac"}) (self.pool / "accounts.json").write_text(json.dumps(self.manifest)) self.account.write_text('model = "changed"\n') result = self.command("sync", "--no-server") self.assertEqual(result.returncode, 0, result.stderr) for path in (self.account, self.pool / "acct-02/config.toml"): self.assertEqual(tomllib.loads(path.read_text())["agents"][KEY], 8) self.assertEqual(tomllib.loads(self.account.read_text())["model"], "changed") def test_invalid_config_is_rejected_before_any_configuration_is_changed(self): before = self.account.read_text() self.global_config.write_text('[profiles.quick.features.multi_agent_v2]\n' 'max_concurrent_threads_per_session = 20\n') result = self.command("configure", "--max-subagents", "8", "--include-global", "--json") self.assertNotEqual(result.returncode, 0) self.assertIn("overrides", result.stderr) self.assertEqual(self.account.read_text(), before) self.assertFalse((self.pool / "codex-settings-policy.json").exists()) def test_seven_thread_policy_requires_v2_setting_and_repairs_legacy_only_configs(self): result = self.command("configure", "--max-subagents", "8", "--include-global", "--json") self.assertEqual(result.returncode, 0, result.stderr) result = self.command("configure", "--max-subagents", "7", "--include-global", "--json") self.assertEqual(result.returncode, 0, result.stderr) for path in (self.account, self.global_config): config = tomllib.loads(path.read_text()) self.assertEqual(config["features"]["multi_agent_v2"][KEY], 7) self.assertEqual(config["agents"][KEY], 7) self.account.write_text('[agents]\nmax_concurrent_threads_per_session = 7\n') result = self.command("configure", "--include-global", "--check", "--json") self.assertEqual(result.returncode, 1, result.stderr) self.assertEqual(json.loads(result.stdout)["mismatch_count"], 1) result = self.command("configure", "--max-subagents", "7", "--include-global", "--json") self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(json.loads(result.stdout)["verified"]) def test_symlinked_global_config_remains_linked(self): self.account.unlink() self.account.symlink_to(self.global_config) result = self.command("configure", "--max-subagents", "8", "--include-global", "--json") self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(self.account.is_symlink()) self.assertTrue(json.loads(result.stdout)["verified"]) @unittest.skipUnless(Path("/usr/bin/python3").is_file(), "System Python required") def test_configuration_via_system_python_uses_installed_modern_runtime(self): self.env["CLAUDE_MULTIACC_PYTHON"] = "/usr/bin/python3" result = self.command("configure", "--max-subagents", "8", "--include-global", "--json") self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(json.loads(result.stdout)["verified"]) def test_sync_verifies_opted_in_policy_on_remote_target(self): result = self.command("configure", "--max-subagents", "8") self.assertEqual(result.returncode, 0, result.stderr) tools = self.home / "tools" tools.mkdir() for name, body in {"uname": "echo Darwin", "rsync": "exit 0", "ssh": 'printf "%s\\n" "$*" >> "$HOME/ssh-calls"\nexit 0'}.items(): path = tools / name path.write_text("#!/bin/sh\n" + body + "\n") path.chmod(0o755) self.env.update({"PATH": str(tools) + os.pathsep + self.env["PATH"], "CODEX_MULTIACC_SYNC_TARGET": "operator@runner", "CODEX_MULTIACC_SYNC_ROOT": "/remote/pool", "CODEX_MULTIACC_SYNC_REPO": "/remote/plugin"}) result = self.command("sync") self.assertEqual(result.returncode, 0, result.stderr) calls = (self.home / "ssh-calls").read_text() self.assertIn("configure --max-subagents '8' --json", calls) self.assertIn("CODEX_ACCOUNTS_ROOT='/remote/pool'", calls) (tools / "ssh").write_text('#!/bin/sh\ncase "$*" in *configure*) exit 1 ;; esac\nexit 0\n') result = self.command("sync") self.assertNotEqual(result.returncode, 0) self.assertIn("settings policy was not verified", result.stderr) if __name__ == "__main__": unittest.main()