Red Queen
ed374ac4ce
- C5A objects table: object_id nullable + reason_code column so HUMAN_REVIEW/ SECRET_DETECTED object changes (object_id=None) are persisted, not silently dropped (was: object_id TEXT NOT NULL, no reason_code field) - C5B poll_once: content_before now read from parent_sha (state BEFORE the change) instead of sha, so MODIFIED changes classify as CONTENT_UPDATE instead of being misclassified (Checker-Befund) - test_c5b: add test_modified_content_reads_parent regression test C5B 40/40, C5A 25/25, real-repo dry run: 233 object changes (109 IN_SCOPE with valid id, 124 HUMAN_REVIEW), idempotent.
560 lines
22 KiB
Python
560 lines
22 KiB
Python
#!/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_modified_content_reads_parent(self):
|
|
"""Regression: content_before muss aus dem PARENT-Commit gelesen werden,
|
|
nicht aus dem aktuellen Commit. Sonst wird ein MODIFIED-Change faelschlich
|
|
als METADATA_UPDATE/kein CONTENT_UPDATE klassifiziert (Checker-Befund)."""
|
|
fx = FixtureRepo()
|
|
try:
|
|
fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body v1")
|
|
fx.commit("c1")
|
|
fx.write("modul-09.md", _fm("object/77b02661-d67b-4bae-b612-01d1287cea6b") + "body v2")
|
|
sha2 = fx.commit("c2")
|
|
store = self._new_store()
|
|
poller = C5BPoller(store, fx.dir)
|
|
r = poller.poll_once()
|
|
self.assertEqual(r["commits_discovered"], 2)
|
|
objs = store.list_object_changes(sha2)
|
|
ops = [o["operation"] for o in objs]
|
|
self.assertIn(OP_CONTENT_UPDATE, ops, f"erwartet CONTENT_UPDATE, habe {ops}")
|
|
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)
|