#!/usr/bin/env python3 """ Red Queen — C5B Testsuite (Forgejo Polling & Change Detection Engine). Testet gegen ein kontrolliertes lokales Git-Fixture-Repo (read-only fuer C5B). Jeder Test nutzt eine frische temp-DB (tempfile.mkdtemp), nie die Produkt-DB. """ from __future__ import annotations import json import os import shutil import subprocess import sys import tempfile import unittest from pathlib import Path from typing import Dict # C5B importieren (aus demselben Verzeichnis) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from rq_c5a import C5AStore, ST_DISCOVERED, ST_VALIDATING, ST_HUMAN_REVIEW_REQUIRED, ST_APPLIED from rq_c5b import ( C5BPoller, GitReader, KnowledgeScope, ChangeClassifier, assert_no_downstream_write, parse_frontmatter, extract_object_id, content_hash, metadata_hash, detect_secret, SCOPE_IN_SCOPE, SCOPE_OUT_OF_SCOPE, SCOPE_LEGACY_SPECIAL, SCOPE_HUMAN_REVIEW, OP_CREATE, OP_CONTENT_UPDATE, OP_METADATA_UPDATE, OP_STATE_UPDATE, OP_TAGS_UPDATE, OP_SOURCE_CANONICAL_RELATION_UPDATE, OP_RENAME, OP_MOVE, OP_DELETE_REQUEST, OP_SUPERSEDE, RC_FORGEJO_UNAVAILABLE, RC_OUT_OF_ORDER_COMMIT, RC_UNKNOWN_OBJECT_ID, RC_SECRET_DETECTED, GitWriteBlockedError, ForgejoUnavailableError, ) # --------------------------------------------------------------------------- # Fixture: kontrolliertes Git-Repo # --------------------------------------------------------------------------- def _git(repo: str, *args: str) -> str: """Fuehrt einen git-Befehl im Fixture-Repo aus (Test-Helfer, darf schreiben).""" proc = subprocess.run( ["git", "-C", repo] + list(args), capture_output=True, text=True, ) if proc.returncode != 0: raise RuntimeError(f"git {' '.join(args)} fehlgeschlagen: {proc.stderr}") return proc.stdout def _write(repo: str, path: str, content: str) -> None: full = os.path.join(repo, path) os.makedirs(os.path.dirname(full), exist_ok=True) with open(full, "w", encoding="utf-8") as f: f.write(content) def _commit(repo: str, message: str) -> str: _git(repo, "add", "-A") _git(repo, "commit", "-m", message, "--no-verify") return _git(repo, "rev-parse", "HEAD").strip() def _fm(id_: str, **extra) -> str: """Baut Frontmatter fuer ein Knowledge-Objekt.""" lines = ["---", f"knowledge_schema: 1", f"id: {id_}"] for k, v in extra.items(): if isinstance(v, list): lines.append(f"{k}: [{', '.join(v)}]") else: lines.append(f"{k}: {v}") lines.append("---") return "\n".join(lines) + "\n" class FixtureRepo: """Erstellt ein kontrolliertes Git-Fixture-Repo fuer C5B-Tests.""" def __init__(self): self.dir = tempfile.mkdtemp(prefix="c5b_fixture_") _git(self.dir, "init", "-q", "-b", "main") _git(self.dir, "config", "user.email", "test@test") _git(self.dir, "config", "user.name", "Test") self.commits: Dict[str, str] = {} def write(self, path: str, content: str) -> None: _write(self.dir, path, content) def commit(self, message: str) -> str: sha = _commit(self.dir, message) self.commits[message] = sha return sha def cleanup(self) -> None: shutil.rmtree(self.dir, ignore_errors=True) # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- class TestFrontmatterAndHashes(unittest.TestCase): """Frontmatter-Parsing, Object-ID-Extraktion, Hash-Model.""" def test_parse_frontmatter(self): content = _fm("object/48bd264f-607b-15f1-5f73-3e922af9b19d", type="arch", role="module", representation="source", state="current", tags=["a", "b"]) fm, body = parse_frontmatter(content) self.assertEqual(fm["id"], "object/48bd264f-607b-15f1-5f73-3e922af9b19d") self.assertEqual(fm["type"], "arch") self.assertEqual(fm["tags"], ["a", "b"]) self.assertNotIn("---", body) def test_extract_object_id_valid(self): content = _fm("object/48bd264f-607b-15f1-5f73-3e922af9b19d") self.assertEqual( extract_object_id(content), "object/48bd264f-607b-15f1-5f73-3e922af9b19d", ) def test_extract_object_id_invalid(self): # Kein object/-Praefix, kein FULL_UUID self.assertIsNone(extract_object_id("---\nid: object/abc\n---\n")) self.assertIsNone(extract_object_id("---\nid: 12345\n---\n")) self.assertIsNone(extract_object_id("kein frontmatter")) def test_content_hash_deterministic(self): self.assertEqual(content_hash("abc"), content_hash("abc")) self.assertNotEqual(content_hash("abc"), content_hash("abd")) def test_metadata_hash_fields(self): fm1 = {"id": "object/77b02661-d67b-4bae-b612-01d1287cea6b", "type": "arch", "role": "module", "representation": "source", "state": "current", "knowledge_schema": "1"} fm2 = dict(fm1) fm2["state"] = "historical" self.assertNotEqual(metadata_hash(fm1), metadata_hash(fm2)) # Pfad ist NICHT Teil von metadata_hash self.assertEqual(metadata_hash(fm1), metadata_hash(fm1)) def test_detect_secret(self): self.assertIsNotNone(detect_secret("key: sk-abcdefghijklmnopqrstuvwxyz123456")) self.assertIsNotNone(detect_secret("-----BEGIN RSA PRIVATE KEY-----\n...")) self.assertIsNone(detect_secret("modul-07-risk-manager.md")) class TestKnowledgeScope(unittest.TestCase): """Knowledge-Scope-Klassifikation.""" def setUp(self): self.dir = tempfile.mkdtemp(prefix="c5b_scope_") self.scope = KnowledgeScope(self.dir) def tearDown(self): shutil.rmtree(self.dir, ignore_errors=True) def test_root_markdown_in_scope(self): self.assertEqual(self.scope.classify("modul-09-execution-service.md"), SCOPE_IN_SCOPE) def test_canonical_in_scope(self): self.assertEqual( self.scope.classify("notes/trading/system-docs/modul-09-execution-service.md"), SCOPE_IN_SCOPE, ) def test_out_of_scope_dirs(self): for p in ["tolaria/C5_SYNC_ARCHITECTURE_DESIGN.md", "a2/README.md", "red-queen-architecture/AGENT_CONTRACTS.md", "notion-safety-brain/README.md", "backup_patches/x.md", "notes/reference/vps-infrastruktur.md"]: self.assertEqual(self.scope.classify(p), SCOPE_OUT_OF_SCOPE, p) def test_non_markdown_out_of_scope(self): self.assertEqual(self.scope.classify("index_source.json"), SCOPE_OUT_OF_SCOPE) def test_legacy_special(self): self.assertEqual(self.scope.classify("README.md"), SCOPE_LEGACY_SPECIAL) self.assertEqual(self.scope.classify("notes/trading/system-docs/README.md"), SCOPE_LEGACY_SPECIAL) self.assertEqual(self.scope.classify("vps.md"), SCOPE_LEGACY_SPECIAL) def test_in_scope_with_valid_id(self): content = _fm("object/48bd264f-607b-15f1-5f73-3e922af9b19d") r = self.scope.classify_with_content("modul-09.md", content) self.assertEqual(r["scope"], SCOPE_IN_SCOPE) self.assertEqual(r["object_id"], "object/48bd264f-607b-15f1-5f73-3e922af9b19d") def test_in_scope_without_id_human_review(self): r = self.scope.classify_with_content("modul-09.md", "kein frontmatter") self.assertEqual(r["scope"], SCOPE_HUMAN_REVIEW) self.assertEqual(r["reason_code"], RC_UNKNOWN_OBJECT_ID) class TestChangeClassifier(unittest.TestCase): """Change-Klassifikation (BEFORE/AFTER -> Operationen).""" def setUp(self): self.dir = tempfile.mkdtemp(prefix="c5b_cls_") self.cls = ChangeClassifier(KnowledgeScope(self.dir)) def tearDown(self): shutil.rmtree(self.dir, ignore_errors=True) def test_create(self): ops = self.cls.classify(None, "modul-09.md", None, _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b")) self.assertEqual(ops[0]["operation"], OP_CREATE) def test_delete_request(self): ops = self.cls.classify("modul-09.md", None, _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b"), None) self.assertEqual(ops[0]["operation"], OP_DELETE_REQUEST) def test_content_update(self): before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="current") + "body v1" after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="current") + "body v2" ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) self.assertIn(OP_CONTENT_UPDATE, [o["operation"] for o in ops]) def test_state_update(self): before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="current") + "body" after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="historical") + "body" ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) self.assertIn(OP_STATE_UPDATE, [o["operation"] for o in ops]) def test_tags_update(self): before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", tags=["a"]) + "body" after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", tags=["a", "b"]) + "body" ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) self.assertIn(OP_TAGS_UPDATE, [o["operation"] for o in ops]) def test_relation_update(self): before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", derived_from="object/y") + "body" ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) self.assertIn(OP_SOURCE_CANONICAL_RELATION_UPDATE, [o["operation"] for o in ops]) def test_rename_same_id(self): before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" ops = self.cls.classify("modul-09.md", "modul-09-new.md", before, after) self.assertIn(OP_RENAME, [o["operation"] for o in ops]) def test_move_same_id(self): before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" ops = self.cls.classify("modul-09.md", "sub/modul-09.md", before, after) self.assertIn(OP_MOVE, [o["operation"] for o in ops]) def test_supersede(self): before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="current") + "body" after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b", state="superseded") + "body" ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) self.assertIn(OP_SUPERSEDE, [o["operation"] for o in ops]) def test_no_change(self): before = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" after = _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body" ops = self.cls.classify("modul-09.md", "modul-09.md", before, after) self.assertEqual(ops, []) class TestGitReader(unittest.TestCase): """GitReader read-only + Write-Blockade.""" def test_write_commands_blocked(self): fx = FixtureRepo() try: reader = GitReader(fx.dir) for cmd in ["push", "commit", "add", "reset", "checkout", "merge", "rebase"]: with self.assertRaises(GitWriteBlockedError): reader._run([cmd, "x"]) finally: fx.cleanup() def test_head_discovery(self): fx = FixtureRepo() try: fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body") sha = fx.commit("c1") reader = GitReader(fx.dir) self.assertEqual(reader.get_current_head(), sha) finally: fx.cleanup() def test_forgejo_unavailable(self): with self.assertRaises(ForgejoUnavailableError): GitReader("/nonexistent/path") class TestC5BPoller(unittest.TestCase): """End-to-End Poll-Zyklen gegen Fixture-Repo.""" def _new_store(self): db = os.path.join(tempfile.mkdtemp(prefix="c5b_db_"), "c5a.db") return C5AStore(db) def test_single_commit_discovery(self): fx = FixtureRepo() try: fx.write("modul-09.md", _fm("object/48bd264f-607b-15f1-5f73-3e922af9b19d") + "body") sha = fx.commit("c1") store = self._new_store() poller = C5BPoller(store, fx.dir) r = poller.poll_once() self.assertEqual(r["status"], "OK") self.assertEqual(r["commits_discovered"], 1) self.assertEqual(r["head"], sha) # Commit registriert, Status VALIDATING (nicht weiter) c = store.get_commit(sha) self.assertEqual(c["status"], ST_VALIDATING) store.close() finally: fx.cleanup() def test_multi_commit_ordering(self): fx = FixtureRepo() try: fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v1") a = fx.commit("A") fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v2") b = fx.commit("B") fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v3") c = fx.commit("C") store = self._new_store() poller = C5BPoller(store, fx.dir) r = poller.poll_once() self.assertEqual(r["commits_discovered"], 3) # Reihenfolge A, B, C (chronologisch) commits = store.list_commits() self.assertEqual([c["commit_sha"] for c in commits], [a, b, c]) store.close() finally: fx.cleanup() def test_duplicate_poll_idempotent(self): fx = FixtureRepo() try: fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body") fx.commit("c1") store = self._new_store() poller = C5BPoller(store, fx.dir) r1 = poller.poll_once() r2 = poller.poll_once() # gleicher HEAD self.assertEqual(r1["commits_discovered"], 1) self.assertEqual(r2["commits_discovered"], 0) # keine neuen # Keine doppelten ObjectChanges objs = store.list_object_changes(fx.commits["c1"]) self.assertEqual(len(objs), 1) store.close() finally: fx.cleanup() def test_lost_poll_recovery(self): fx = FixtureRepo() try: fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v1") a = fx.commit("A") store = self._new_store() poller = C5BPoller(store, fx.dir) poller.poll_once() # Poll bei A # Zwei Poll-Intervalle ausfallen: B, C, D entstehen fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v2") b = fx.commit("B") fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v3") c = fx.commit("C") fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v4") d = fx.commit("D") r = poller.poll_once() # naechster Poll bei D self.assertEqual(r["commits_discovered"], 3) self.assertEqual(r["new_commits"], [b, c, d]) store.close() finally: fx.cleanup() def test_out_of_scope_not_registered(self): fx = FixtureRepo() try: fx.write("tolaria/C5_SYNC_ARCHITECTURE_DESIGN.md", "# design") fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body") fx.commit("c1") store = self._new_store() poller = C5BPoller(store, fx.dir) r = poller.poll_once() self.assertEqual(r["commits_discovered"], 1) self.assertEqual(r["out_of_scope_count"], 1) # tolaria/ ist out-of-scope store.close() finally: fx.cleanup() def test_legacy_special_not_registered(self): fx = FixtureRepo() try: fx.write("README.md", "# README") fx.write("vps.md", "# vps") fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body") fx.commit("c1") store = self._new_store() poller = C5BPoller(store, fx.dir) r = poller.poll_once() self.assertEqual(r["out_of_scope_count"], 2) # README + vps store.close() finally: fx.cleanup() def test_unknown_object_id_human_review(self): fx = FixtureRepo() try: fx.write("modul-09.md", "kein frontmatter, keine id") fx.commit("c1") store = self._new_store() poller = C5BPoller(store, fx.dir) r = poller.poll_once() self.assertEqual(r["human_review_count"], 1) c = store.get_commit(fx.commits["c1"]) self.assertEqual(c["status"], ST_HUMAN_REVIEW_REQUIRED) store.close() finally: fx.cleanup() def test_secret_detection_fail_closed(self): fx = FixtureRepo() try: fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "key: sk-abcdefghijklmnopqrstuvwxyz123456") fx.commit("c1") store = self._new_store() poller = C5BPoller(store, fx.dir) r = poller.poll_once() self.assertEqual(r["human_review_count"], 1) c = store.get_commit(fx.commits["c1"]) self.assertEqual(c["status"], ST_HUMAN_REVIEW_REQUIRED) # Kein Secret-Inhalt in der DB objs = store.list_object_changes(fx.commits["c1"]) for o in objs: self.assertNotIn("sk-", json.dumps(o)) store.close() finally: fx.cleanup() def test_history_divergence_fail_closed(self): fx = FixtureRepo() try: fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v1") a = fx.commit("A") store = self._new_store() poller = C5BPoller(store, fx.dir) poller.poll_once() # last_applied = a. Jetzt History-Rewrite: neuer Root ohne a als Ancestor. # Wir wechseln auf einen Orphan-Branch (neuer Root), der a NICHT enthaelt. _git(fx.dir, "checkout", "--orphan", "newroot") _git(fx.dir, "rm", "-rf", ".") fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "v2") b = fx.commit("B") # HEAD ist jetzt b (newroot), a ist NICHT in dessen Ancestry self.assertNotEqual(a, b) r = poller.poll_once() self.assertEqual(r["status"], "FAIL_CLOSED") self.assertEqual(r["reason_code"], RC_OUT_OF_ORDER_COMMIT) store.close() finally: fx.cleanup() def test_c5a_persistence_integration(self): fx = FixtureRepo() try: fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body") sha = fx.commit("c1") db = os.path.join(tempfile.mkdtemp(prefix="c5b_db_"), "c5a.db") store = C5AStore(db) poller = C5BPoller(store, fx.dir) poller.poll_once() store.close() # Neuer Store auf gleicher DB: State restart-fest store2 = C5AStore(db) c = store2.get_commit(sha) self.assertEqual(c["status"], ST_VALIDATING) store2.close() finally: fx.cleanup() def test_no_downstream_write(self): r = assert_no_downstream_write() self.assertTrue(r["no_downstream_write"]) self.assertEqual(r["http_imports_found"], []) self.assertEqual(r["forbidden_endpoints_found"], []) self.assertEqual(r["git_write_in_whitelist"], []) class TestPollIntervalConfig(unittest.TestCase): """Poll-Intervall ist Environment-konfigurierbar.""" def test_default_60(self): fx = FixtureRepo() try: store = C5AStore(os.path.join(tempfile.mkdtemp(), "c5a.db")) poller = C5BPoller(store, fx.dir) self.assertEqual(poller.poll_interval_seconds, 60) store.close() finally: fx.cleanup() def test_env_override(self): fx = FixtureRepo() try: os.environ["C5_POLL_INTERVAL_SECONDS"] = "120" store = C5AStore(os.path.join(tempfile.mkdtemp(), "c5a.db")) poller = C5BPoller(store, fx.dir) self.assertEqual(poller.poll_interval_seconds, 120) store.close() del os.environ["C5_POLL_INTERVAL_SECONDS"] finally: fx.cleanup() if __name__ == "__main__": unittest.main(verbosity=2)