"""Gemeinsame Test-Hilfen. Schlüssel werden pro Test frisch per os.urandom erzeugt.""" from __future__ import annotations import copy import json import os import sys import tempfile import threading import unittest from pathlib import Path from aria_router.adapter_contract import AdapterJob, ExecutionPolicy from aria_router.authorization import ClearanceIssuer, ClearanceKeyring, ClearanceVerifier from aria_router.classify import PrivacyClassifier from aria_router.config import parse_gateway_config from aria_router.context import Attachment, ContextEnvelope, MemorySnippet from aria_router.core import PrivacyMatrix, PrivacyTier from aria_router.gateway import Gateway, GatewayRequest from aria_router.runner import RunResult, RunSpec, SubprocessRunner from aria_router.state import NonceStore OWNER_ID = "100200300" # synthetische Telegram-ID OTHER_ID = "999888777" # synthetische, nicht freigegebene ID GROUP_ID = "-1001234567" FAKE_CODEX = str(Path(__file__).resolve().parent / "fixtures" / "fake_codex.py") PYTHON = sys.executable def fake_codex_template(mode: str = "success", *extra: str) -> list[str]: return [PYTHON, "-I", FAKE_CODEX, mode, *extra] class RealBinaryGuardError(AssertionError): pass class GuardedSubprocessRunner(SubprocessRunner): """Echter SubprocessRunner, der in Tests NUR den Test-Interpreter startet. Sicherung nach dem Vorfall vom 24.10.: Ein Test darf nie /usr/bin/codex (oder ein anderes Programm) ausführen. Verstöße werfen, statt zu laufen. """ def run(self, spec: RunSpec) -> RunResult: if not spec.argv or os.path.realpath(spec.argv[0]) != os.path.realpath(PYTHON): raise RealBinaryGuardError(f"test tried to execute {spec.argv[:1]}") return super().run(spec) class FakeClock: def __init__(self, start: float = 1_800_000_000.0): self.now = start self._lock = threading.Lock() def __call__(self) -> float: with self._lock: return self.now def advance(self, seconds: float) -> None: with self._lock: self.now += seconds def write_key(path: Path, size: int = 32, mode: int = 0o600) -> Path: path.parent.mkdir(parents=True, exist_ok=True) fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, "wb") as handle: handle.write(os.urandom(size)) # generierter Testschlüssel, kein echtes Secret os.chmod(path, mode) return path def deep_merge(base: dict, overrides: dict) -> dict: result = copy.deepcopy(base) for key, value in overrides.items(): if isinstance(value, dict) and isinstance(result.get(key), dict): result[key] = deep_merge(result[key], value) else: result[key] = copy.deepcopy(value) return result def config_dict(root: Path, overrides: dict | None = None) -> dict: base = { "version": 1, "allowed_modes": ["shadow", "execute"], "state_dir": str(root / "state"), "clearance": {"key_file": str(root / "keys" / "clearance.key"), "ttl_seconds": 120}, "auth": {"allowed_sender_ids": [OWNER_ID]}, "privacy": { "matrix": {"codex": ["public"], "claude": ["public", "internal", "unknown"]}, "claude_status_quo_confidential": True, "public_marker_enabled": True, }, "routing": {"min_quality": 0, "manual_fallback": "block", "default_task_class": "analysis"}, "backends": { "codex": {"enabled": True, "capabilities": ["text", "code"], "tasks": ["analysis", "coding", "review", "writing", "planning"], "quality": 90, "max_context_bytes": 131072}, "claude": {"enabled": True, "capabilities": ["text", "code", "vision", "files"], "tasks": ["analysis", "coding", "review", "writing", "planning", "vision"], "quality": 90, "max_context_bytes": 1048576}, }, "budget": {"codex": {"daily_jobs": 100}, "claude": {"daily_jobs": 100}}, "health": {"failure_threshold": 3, "open_seconds": 300}, "limits": {"max_text_bytes": 65536, "max_reply_chars": 12000, "max_attachments": 1}, "codex": { "real_execution": False, # Sicherung: In Tests ist das Default-Template immer das Fake-Skript. "command_template": fake_codex_template("success"), "timeout_seconds": 30, "workspace_root": str(root / "work"), "env": {"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"}, }, } return deep_merge(base, overrides or {}) def make_config(root: Path, overrides: dict | None = None): return parse_gateway_config(config_dict(root, overrides)) class RecordingSink: """Mock des Telegram-Plugins: führt Weiterleitungsanweisungen 'aus'.""" def __init__(self, fail: bool = False): self.delivered: list[dict] = [] self.fail = fail def deliver(self, instruction: dict) -> None: if self.fail: raise ConnectionError("simulated plugin failure") self.delivered.append(instruction) class GatewayTestCase(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.root = Path(self._tmp.name) write_key(self.root / "keys" / "clearance.key") (self.root / "work").mkdir() self.clock = FakeClock() self.sink = RecordingSink() def tearDown(self): self._tmp.cleanup() def gateway(self, config: dict | None = None, **kwargs) -> Gateway: kwargs.setdefault("forward_sink", self.sink) kwargs.setdefault("clock", self.clock) return Gateway(make_config(self.root, config), **kwargs) def request(self, message_id: int, text: str, *, sender_id: str = OWNER_ID, chat_id: str | None = None, chat_type: str = "private", attachments=(), username: str | None = "kais_test", ts: str = "2026-10-24T18:00:00.000Z") -> GatewayRequest: return GatewayRequest(sender_id=sender_id, chat_id=chat_id if chat_id is not None else sender_id, chat_type=chat_type, message_id=message_id, text=text, sender_username=username, ts=ts, attachments=tuple(attachments)) def log_records(self, gateway: Gateway) -> list[dict]: return gateway.decision_log.read_all() def decision_records(self, gateway: Gateway) -> list[dict]: return [r for r in gateway.decision_log.read_all() if r.get("phase") == "decision"] def write_access_file(self, data: dict) -> Path: path = self.root / "access.json" path.write_text(json.dumps(data), encoding="utf-8") return path DEFAULT_MATRIX = PrivacyMatrix.from_mapping({ "codex": {PrivacyTier.PUBLIC}, "claude": {PrivacyTier.PUBLIC, PrivacyTier.INTERNAL, PrivacyTier.UNKNOWN, PrivacyTier.CONFIDENTIAL}, }) class AdapterHarness: """Issuer + Verifier + Nonce-Speicher ohne Gateway (für Adapter-/Clearance-Tests).""" def __init__(self, root: Path, *, matrix: PrivacyMatrix = DEFAULT_MATRIX, clock=None, key_name="clearance.key"): self.root = root self.clock = clock or FakeClock() self.key_file = write_key(root / "keys" / key_name) self.keyring = ClearanceKeyring(self.key_file) self.matrix = matrix self.classifier = PrivacyClassifier() self.nonces = NonceStore(root / "state" / f"nonces-{key_name}.json") self.issuer = ClearanceIssuer(self.keyring, matrix, clock=self.clock) self.verifier = ClearanceVerifier(self.keyring, matrix, self.nonces, self.classifier, clock=self.clock) def job(self, prompt: str = "#public hallo", *, memory=(), attachments=(), job_id="job-1", sender_id=OWNER_ID, chat_id=OWNER_ID, message_id=7, username="kais_test", ts="2026-10-24T18:00:00.000Z", timeout: int = 30) -> AdapterJob: return AdapterJob(job_id, ContextEnvelope(prompt, tuple(memory), tuple(attachments)), sender_id, chat_id, message_id, username, ts, ExecutionPolicy(timeout_seconds=timeout)) def clear(self, adapter, job: AdapterJob, **overrides): privacy = self.classifier.classify(job.envelope).tier args = dict(job_id=job.job_id, backend=adapter.name, privacy=privacy, context_digest=adapter.outbound_digest(job), sender_id=job.sender_id, chat_id=job.chat_id) args.update(overrides) return self.issuer.issue(**args) def photo(local_path="/tmp/inbox/1-abc.jpg"): return Attachment(kind="photo", file_id="AgACphoto", size=1234, mime="image/jpeg", local_path=local_path) def document(name="bericht.pdf", data=None): return Attachment(kind="document", file_id="BQACdoc", size=len(data) if data is not None else 2048, mime="application/pdf", name=name, data=data) __all__ = [ "OWNER_ID", "OTHER_ID", "GROUP_ID", "FAKE_CODEX", "PYTHON", "fake_codex_template", "FakeClock", "write_key", "deep_merge", "config_dict", "make_config", "RecordingSink", "GatewayTestCase", "AdapterHarness", "GuardedSubprocessRunner", "RealBinaryGuardError", "DEFAULT_MATRIX", "photo", "document", "MemorySnippet", "Attachment", "ContextEnvelope", ]