"""Konfigurationsdatei: strikt, fail-closed, sichere Defaults.""" import json import tempfile import unittest from pathlib import Path from aria_router.config import ConfigurationError, load_gateway_config, parse_gateway_config from aria_router.core import PrivacyTier from aria_router.gateway import Gateway from aria_router.runner import MockRunner, SubprocessRunner from tests.helpers import OWNER_ID, config_dict, write_key EXAMPLE = Path(__file__).resolve().parents[1] / "config" / "gateway.example.json" class ConfigFileTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.root = Path(self._tmp.name) write_key(self.root / "keys" / "clearance.key") def tearDown(self): self._tmp.cleanup() def parse(self, overrides=None): return parse_gateway_config(config_dict(self.root, overrides)) def assertRejected(self, overrides, fragment=None): with self.assertRaises(ConfigurationError) as ctx: self.parse(overrides) if fragment: self.assertIn(fragment, str(ctx.exception)) def test_example_config_loads_with_safe_defaults(self): config = load_gateway_config(EXAMPLE) self.assertEqual(config.allowed_modes, frozenset({"shadow"})) self.assertFalse(config.codex.real_execution) # Owner-Entscheidung 24.10.2026: Codex erhält dieselben Datenklassen wie Claude. self.assertTrue(config.privacy.codex_parity_with_claude) self.assertEqual(config.privacy.matrix.tiers_for("codex"), config.privacy.matrix.tiers_for("claude")) self.assertIn(PrivacyTier.CONFIDENTIAL, config.privacy.matrix.tiers_for("claude")) self.assertEqual(config.routing.manual_fallback, "block") def test_real_codex_example_configs_load(self): # Vorlagen für den Produktivwechsel (Phase G): echte Codex-Ausführung im Käfig from aria_router.codex_adapter import validate_command_template for profile in ("text", "tools"): config = load_gateway_config(EXAMPLE.with_name(f"gateway.codex-{profile}.example.json")) codex = config.codex self.assertTrue(codex.real_execution) self.assertEqual(codex.profile, profile) self.assertFalse(codex.codex_auth_writable) self.assertEqual(codex.resource_limits.address_space_bytes is None, profile == "tools") self.assertIn("-p", codex.isolation_wrapper) self.assertIn("MemoryMax=2G", codex.isolation_wrapper) validate_command_template(codex.command_template, codex.isolation_wrapper, allow_bypass_sandbox=profile == "tools", require_cage=profile == "tools") self.assertTrue(config.privacy.codex_parity_with_claude) def test_unknown_keys_rejected(self): self.assertRejected({"surprise": 1}, "unknown key") self.assertRejected({"privacy": {"matrx": {}}}, "unknown key") self.assertRejected({"codex": {"allow_danger": True}}, "unknown key") def test_real_execution_requires_isolation_wrapper(self): self.assertRejected({"codex": {"real_execution": True, "resource_limits": {"cpu_seconds": 60}}}, "isolation_wrapper") def test_real_execution_requires_resource_limits(self): self.assertRejected({"codex": {"real_execution": True, "isolation_wrapper": ["/usr/bin/systemd-run"]}}, "resource_limits") def test_real_execution_requires_workspace(self): data = config_dict(self.root, {"codex": {"real_execution": True, "isolation_wrapper": ["/usr/bin/x"], "resource_limits": {"cpu_seconds": 60}}}) del data["codex"]["workspace_root"] with self.assertRaises(ConfigurationError): parse_gateway_config(data) def test_status_quo_switch(self): on = self.parse({"privacy": {"claude_status_quo_confidential": True}}).privacy.matrix off = self.parse({"privacy": {"claude_status_quo_confidential": False}}).privacy.matrix self.assertTrue(on.permits("claude", PrivacyTier.CONFIDENTIAL)) self.assertFalse(on.permits("codex", PrivacyTier.CONFIDENTIAL)) self.assertFalse(off.permits("claude", PrivacyTier.CONFIDENTIAL)) def test_codex_parity_copies_claude_tiers(self): data = config_dict(self.root, {"privacy": {"claude_status_quo_confidential": True, "codex_parity_with_claude": True}}) del data["privacy"]["matrix"]["codex"] matrix = parse_gateway_config(data).privacy.matrix for tier in PrivacyTier: self.assertEqual(matrix.permits("codex", tier), matrix.permits("claude", tier), tier) self.assertTrue(matrix.permits("codex", PrivacyTier.CONFIDENTIAL)) self.assertTrue(matrix.permits("codex", PrivacyTier.UNKNOWN)) def test_codex_parity_follows_claude_without_status_quo(self): data = config_dict(self.root, {"privacy": {"claude_status_quo_confidential": False, "codex_parity_with_claude": True}}) del data["privacy"]["matrix"]["codex"] matrix = parse_gateway_config(data).privacy.matrix self.assertFalse(matrix.permits("codex", PrivacyTier.CONFIDENTIAL)) self.assertTrue(matrix.permits("codex", PrivacyTier.UNKNOWN)) def test_codex_parity_rejects_separate_codex_matrix(self): data = config_dict(self.root, {"privacy": {"codex_parity_with_claude": True}}) self.assertIn("codex", data["privacy"]["matrix"]) with self.assertRaises(ConfigurationError): parse_gateway_config(data) def test_codex_parity_off_by_default(self): matrix = self.parse({}).privacy.matrix self.assertFalse(matrix.permits("codex", PrivacyTier.UNKNOWN)) def test_budget_required_for_enabled_backend(self): data = config_dict(self.root) del data["budget"]["codex"] with self.assertRaises(ConfigurationError): parse_gateway_config(data) def test_auth_required(self): data = config_dict(self.root) data["auth"] = {} with self.assertRaises(ConfigurationError) as ctx: parse_gateway_config(data) self.assertIn("auth", str(ctx.exception)) def test_invalid_sender_ids_rejected(self): self.assertRejected({"auth": {"allowed_sender_ids": ["kais"]}}) def test_types_strict(self): self.assertRejected({"backends": {"codex": {"quality": True}}}) self.assertRejected({"backends": {"codex": {"quality": "90"}}}) self.assertRejected({"clearance": {"ttl_seconds": 601}}) self.assertRejected({"allowed_modes": ["live"]}) self.assertRejected({"state_dir": "relative/path"}) def test_load_rejects_invalid_files(self): broken = self.root / "broken.json" broken.write_text("{nope", encoding="utf-8") for path in (broken, self.root / "missing.json"): with self.subTest(path=path), self.assertRaises(ConfigurationError): load_gateway_config(path) def test_default_runner_is_mock(self): gateway = Gateway(self.parse()) self.assertIsInstance(gateway.adapters["codex"].runner, MockRunner) def test_real_execution_selects_subprocess_runner_without_running(self): config = self.parse({"codex": {"real_execution": True, "isolation_wrapper": ["/usr/bin/env"], "resource_limits": {"cpu_seconds": 60}}}) gateway = Gateway(config) runner = gateway.adapters["codex"].runner self.assertIsInstance(runner, SubprocessRunner) self.assertEqual(runner.resource_limits.cpu_seconds, 60) self.assertEqual(gateway.adapters["codex"].argv[0], "/usr/bin/env") if __name__ == "__main__": unittest.main()