""" AUTH.4C1 — Phase 11: Isolated Tests (T1-T27) ============================================= Läuft mit Fake/isolierter Tolaria-Instanz und synthetischen Credentials. KEIN produktiver Tolaria-Kontakt. KEIN produktiver SAVE. Testumgebung: * TOLARIA_SAVE_TOKEN = synthetisch (nur für Tests) * TolariaClient mit Fake-Base-URL (http://127.0.0.1:1 = unerreichbar) * ForgejoSourceLoader mit Fake-Clone (lokales Test-Repo) * JobStore auf temporärer SQLite-DB Führt die AUTH.4C1-Tests T1-T27 aus. Exit 0 = alle PASS. """ from __future__ import annotations import hashlib import json import os import shutil import subprocess import sys import tempfile import time import unittest import uuid from typing import Any, Dict, Optional # Module aus dem Build-Verzeichnis importieren sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from forgejo_source_loader import ForgejoSourceLoader, SourceMismatchError, SourceUnavailableError from job_schema import make_save_job, make_delete_job, validate_job, JobRejectedError from job_store import JobStore from save_executor_core import SaveExecutorCore from tolaria_client import TolariaClient, TolariaWriteError # Synthetisches Test-Credential (NUR für Tests, nie produktiv) TEST_TOKEN = "test-save-token-4c1-isolated" # Fake-Tolaria: unerreichbare Base-URL (kein echter Kontakt) FAKE_BASE = "http://127.0.0.1:1/api/vault" def _sha256(s: str) -> str: return hashlib.sha256(s.encode("utf-8")).hexdigest() def _make_test_repo(tmpdir: str) -> str: """Erzeugt ein lokales Git-Repo als Fake-Forgejo-Clone.""" repo = os.path.join(tmpdir, "test-repo") os.makedirs(repo, exist_ok=True) subprocess.run(["git", "init", "-q", repo], check=True) subprocess.run(["git", "-C", repo, "config", "user.email", "t@t"], check=True) subprocess.run(["git", "-C", repo, "config", "user.name", "t"], check=True) # Datei mit Frontmatter + Body content = "---\nid: object/test-obj-1\ntitle: Test\n---\n\nHello World\n" os.makedirs(os.path.join(repo, "docs"), exist_ok=True) with open(os.path.join(repo, "docs", "test.md"), "w") as f: f.write(content) subprocess.run(["git", "-C", repo, "add", "-A"], check=True) subprocess.run(["git", "-C", repo, "commit", "-q", "-m", "init"], check=True) return repo class FakeTolariaSave: """Fake-Tolaria-SAVE-Callable. Zeichnet Aufrufe auf, führt keinen echten Write aus.""" def __init__(self, behavior: str = "ok"): self.behavior = behavior # ok | uncertain | fail | credential_missing self.calls: list[tuple[str, str]] = [] def __call__(self, vault_path: str, content: str) -> Dict[str, Any]: self.calls.append((vault_path, content)) if self.behavior == "ok": return {"status": "ok"} if self.behavior == "uncertain": return {"status": "error", "code": "TOLARIA_UNAVAILABLE", "uncertain": True} if self.behavior == "credential_missing": return {"status": "error", "code": "CREDENTIAL_MISSING", "uncertain": False} return {"status": "error", "code": "AUTH_FAILURE", "uncertain": False} class FakeReadBack: def __init__(self, content: Optional[str] = None, behavior: str = "ok"): self.content = content self.behavior = behavior # ok | mismatch | unknown self.calls = 0 def __call__(self, vault_path: str) -> Optional[str]: self.calls += 1 if self.behavior == "unknown": return None if self.behavior == "mismatch": return "DIFFERENT" return self.content class AUTH4C1IsolatedTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.tmp = tempfile.mkdtemp(prefix="c5-4c1-test-") cls.repo = _make_test_repo(cls.tmp) cls.db = os.path.join(cls.tmp, "test_save.db") cls.store = JobStore(cls.db, "SAVE") # Commit-Hash des Test-Repos cls.commit = subprocess.run( ["git", "-C", cls.repo, "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip() cls.content = "---\nid: object/test-obj-1\ntitle: Test\n---\n\nHello World\n" cls.prov = _sha256(cls.content) @classmethod def tearDownClass(cls): cls.store.close() shutil.rmtree(cls.tmp, ignore_errors=True) def _make_job(self, **overrides) -> Dict[str, Any]: job = make_save_job( job_id=overrides.get("job_id", str(uuid.uuid4())), mission_id=overrides.get("mission_id", str(uuid.uuid4())), object_id=overrides.get("object_id", "object/test-obj-1"), vault_path=overrides.get("vault_path", "docs/test.md"), source_commit=overrides.get("source_commit", self.commit), provenance_hash=overrides.get("provenance_hash", self.prov), expected_state="present", created_at="2026-08-27T12:00:00Z", idempotency_key=overrides.get("idempotency_key", str(uuid.uuid4())), ) return job def _loader(self, repo: Optional[str] = None) -> ForgejoSourceLoader: return ForgejoSourceLoader(repo="test/test", clone_path=repo or self.repo, forgejo_base="http://127.0.0.1:1") def _core(self, loader, save, read_back=None) -> SaveExecutorCore: return SaveExecutorCore(self.store, loader.load, save, read_back=read_back) # -- T1: worker starts without token ------------------------------------ def test_t1_worker_starts_without_token(self): # Worker-Modul importierbar + main() startet ohne Token (fail-closed) import worker self.assertIsNotNone(worker) self.assertFalse(worker.TOLARIA_SAVE_TOKEN) # -- T2: no token -> no HTTP -------------------------------------------- def test_t2_no_token_no_http(self): client = TolariaClient(base_url=FAKE_BASE, save_token=None) with self.assertRaises(TolariaWriteError) as ctx: client.write("x.md", "content") self.assertEqual(ctx.exception.code, "CREDENTIAL_MISSING") # -- T3: valid SAVE job processed --------------------------------------- def test_t3_valid_save_processed(self): job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") rb = FakeReadBack(self.content) core = self._core(self._loader(), save, rb) result = core.process_job(job["job_id"], "w-t3") self.assertEqual(result["state"], "SUCCEEDED") self.assertEqual(len(save.calls), 1) self.assertEqual(save.calls[0][0], "docs/test.md") # -- T4: unknown type rejected ------------------------------------------ def test_t4_unknown_type_rejected(self): job = self._make_job() job["job_type"] = "C5_UNKNOWN" with self.assertRaises(JobRejectedError): validate_job(job) # -- T5: malformed rejected --------------------------------------------- def test_t5_malformed_rejected(self): with self.assertRaises(JobRejectedError): validate_job({"job_id": "x"}) # unvollständig # -- T6: duplicate safe ------------------------------------------------- def test_t6_duplicate_safe(self): job = self._make_job() self.store.create_job(job) # Zweiter create mit gleicher job_id -> idempotent (kein Fehler) again = self.store.create_job(job) self.assertEqual(again["job_id"], job["job_id"]) # -- T7: fixed Tolaria target ------------------------------------------ def test_t7_fixed_tolaria_target(self): client = TolariaClient(base_url=FAKE_BASE, save_token=TEST_TOKEN) self.assertEqual(client.base_url, FAKE_BASE.rstrip("/")) # -- T8: URL injection rejected ----------------------------------------- def test_t8_url_injection_rejected(self): job = self._make_job() job["url"] = "http://evil" with self.assertRaises(JobRejectedError): validate_job(job) # -- T9: header injection rejected --------------------------------------- def test_t9_header_injection_rejected(self): job = self._make_job() job["headers"] = {"X-Evil": "1"} with self.assertRaises(JobRejectedError): validate_job(job) # -- T10: method injection rejected ------------------------------------- def test_t10_method_injection_rejected(self): job = self._make_job() job["method"] = "DELETE" with self.assertRaises(JobRejectedError): validate_job(job) # -- T11: delete job rejected ------------------------------------------- def test_t11_delete_job_rejected(self): job = make_delete_job( job_id=str(uuid.uuid4()), mission_id=str(uuid.uuid4()), delete_request_id=str(uuid.uuid4()), object_id="object/x", vault_path="x.md", expected_commit=self.commit, expected_provenance_hash=self.prov, approval_id=str(uuid.uuid4()), created_at="2026-08-27T12:00:00Z", idempotency_key=str(uuid.uuid4())) self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-t11") self.assertEqual(result["state"], "REJECTED") self.assertEqual(len(save.calls), 0) # -- T12: traversal rejected -------------------------------------------- def test_t12_traversal_rejected(self): # _make_job validiert bereits beim Bauen -> JobRejectedError with self.assertRaises(JobRejectedError): self._make_job(vault_path="../../etc/passwd") # -- T13: content loaded from exact Forgejo commit ----------------------- def test_t13_content_from_exact_commit(self): loader = self._loader() content = loader.load(self.commit, "object/test-obj-1", "docs/test.md") self.assertEqual(content, self.content) # -- T14: commit mismatch denied ---------------------------------------- def test_t14_commit_mismatch_denied(self): loader = self._loader() with self.assertRaises(SourceUnavailableError): loader.load("deadbeef" * 5, "object/test-obj-1", "docs/test.md") # -- T15: hash mismatch denied ------------------------------------------ def test_t15_hash_mismatch_denied(self): job = self._make_job(provenance_hash=_sha256("WRONG")) self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-t15") self.assertEqual(result["state"], "FAILED") self.assertEqual(result["result_code"], "PROVENANCE_MISMATCH") self.assertEqual(len(save.calls), 0) # -- T16: provenance mismatch denied ------------------------------------ def test_t16_provenance_mismatch_denied(self): # object_id mismatch -> SourceMismatch -> SOURCE_UNAVAILABLE (fail-closed) job = self._make_job(object_id="object/wrong-id") self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-t16") self.assertEqual(result["state"], "FAILED") self.assertEqual(len(save.calls), 0) # -- T17: missing source denied ----------------------------------------- def test_t17_missing_source_denied(self): job = self._make_job(vault_path="docs/missing.md") self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-t17") self.assertEqual(result["state"], "FAILED") self.assertEqual(result["result_code"], "SOURCE_UNAVAILABLE") self.assertEqual(len(save.calls), 0) # -- T18: Tolaria 401 fail-closed --------------------------------------- def test_t18_tolaria_401_fail_closed(self): job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("fail") # AUTH_FAILURE core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-t18") self.assertEqual(result["state"], "FAILED") self.assertEqual(result["result_code"], "AUTH_FAILURE") # -- T19: Tolaria 403 fail-closed --------------------------------------- def test_t19_tolaria_403_fail_closed(self): # gleiche Semantik wie 401 -> AUTH_FAILURE job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("fail") core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-t19") self.assertEqual(result["state"], "FAILED") # -- T20: timeout -> safe ----------------------------------------------- def test_t20_timeout_safe(self): # Unerreichbare Base-URL -> TolariaUnavailable -> uncertain -> OUTCOME_UNKNOWN job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) client = TolariaClient(base_url=FAKE_BASE, save_token=TEST_TOKEN, timeout=0.1) save = lambda p, c: (lambda r: {"status": "error", "code": "TOLARIA_UNAVAILABLE", "uncertain": True})(None) core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-t20") self.assertEqual(result["state"], "OUTCOME_UNKNOWN") # -- T21: uncertain write -> OUTCOME_UNKNOWN ---------------------------- def test_t21_uncertain_write_outcome_unknown(self): job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("uncertain") core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-t21") self.assertEqual(result["state"], "OUTCOME_UNKNOWN") # -- T22: no blind retry ------------------------------------------------- def test_t22_no_blind_retry(self): # OUTCOME_UNKNOWN ist terminal im Worker (kein Retry im Loop) job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("uncertain") core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-t22") self.assertEqual(result["state"], "OUTCOME_UNKNOWN") # Erneuter process_job auf OUTCOME_UNKNOWN -> kein Doppel-Write result2 = core.process_job(job["job_id"], "w-t22") self.assertEqual(len(save.calls), 1) # -- T23: read-back mismatch not SUCCEEDED ------------------------------- def test_t23_readback_mismatch_not_succeeded(self): job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") rb = FakeReadBack("DIFFERENT", behavior="mismatch") core = self._core(self._loader(), save, rb) result = core.process_job(job["job_id"], "w-t23") self.assertEqual(result["state"], "FAILED") self.assertEqual(result["result_code"], "READBACK_MISMATCH") # -- T24: audit REQUESTED/AUTHORIZED/EXECUTED --------------------------- def test_t24_audit_events(self): job = self._make_job() self.store.create_job(job) # -> REQUESTED self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") rb = FakeReadBack(self.content) core = self._core(self._loader(), save, rb) core.process_job(job["job_id"], "w-t24") events = self.store._conn.execute( "SELECT event_type FROM audit_events WHERE job_id=? ORDER BY created_at", (job["job_id"],)).fetchall() types = [r["event_type"] for r in events] self.assertIn("REQUESTED", types) self.assertIn("AUTHORIZED", types) self.assertIn("EXECUTED", types) # -- T25: credential not logged ----------------------------------------- def test_t25_credential_not_logged(self): # TolariaClient loggt den Token nie; Fehlermeldung enthält keinen Token from tolaria_client import TolariaClientError client = TolariaClient(base_url=FAKE_BASE, save_token=TEST_TOKEN) try: client.write("x.md", "c") except TolariaClientError as e: self.assertNotIn(TEST_TOKEN, str(e)) # -- T26: RQ cannot read SAVE credential -------------------------------- def test_t26_rq_cannot_read_save_credential(self): # RQ hat keinen Zugriff auf TOLARIA_SAVE_TOKEN (nur Executor-ENV). # Hier: Token ist nicht in der DB / nicht im Job / nicht im Store. job = self._make_job() self.store.create_job(job) payload = json.dumps(job) self.assertNotIn(TEST_TOKEN, payload) # -- T27: DELETE executor cannot read SAVE credential ------------------- def test_t27_delete_executor_cannot_read_save_credential(self): # SAVE-Credential ist nur im SAVE-Executor. DELETE-Executor hat es nicht. # Hier: kein DELETE-Code im SAVE-Build (nur die 9 Runtime-Module prüfen). import os build_dir = os.path.dirname(os.path.abspath(__file__)) runtime_modules = [ "save_executor_core.py", "job_store.py", "job_schema.py", "job_state_machine.py", "job_claim.py", "tolaria_client.py", "forgejo_source_loader.py", "worker.py", "entrypoint.py", ] for f in runtime_modules: with open(os.path.join(build_dir, f)) as fh: self.assertNotIn("delete_executor_core", fh.read()) class AUTH4C2PathContractTests(unittest.TestCase): """AUTH.4C2 PATH CONTRACT REPAIR — isolierte Path-Contract-Tests (T1-T20). Verifiziert den kanonischen RELATIVEN Vault-Pfad-Contract durch die gesamte SAVE-Pipeline. KEIN produktiver SAVE, KEIN neuer Job, KEIN DELETE. """ @classmethod def setUpClass(cls): cls.tmp = tempfile.mkdtemp(prefix="c5-4c2-path-test-") cls.repo = _make_test_repo(cls.tmp) cls.db = os.path.join(cls.tmp, "test_save.db") cls.store = JobStore(cls.db, "SAVE") cls.commit = subprocess.run( ["git", "-C", cls.repo, "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip() cls.content = "---\nid: object/test-obj-1\ntitle: Test\n---\n\nHello World\n" cls.prov = _sha256(cls.content) @classmethod def tearDownClass(cls): cls.store.close() shutil.rmtree(cls.tmp, ignore_errors=True) def _make_job(self, **overrides) -> Dict[str, Any]: job = make_save_job( job_id=overrides.get("job_id", str(uuid.uuid4())), mission_id=overrides.get("mission_id", str(uuid.uuid4())), object_id=overrides.get("object_id", "object/test-obj-1"), vault_path=overrides.get("vault_path", "docs/test.md"), source_commit=overrides.get("source_commit", self.commit), provenance_hash=overrides.get("provenance_hash", self.prov), expected_state="present", created_at="2026-08-27T12:00:00Z", idempotency_key=overrides.get("idempotency_key", str(uuid.uuid4())), ) return job def _loader(self, repo: Optional[str] = None) -> ForgejoSourceLoader: return ForgejoSourceLoader(repo="test/test", clone_path=repo or self.repo, forgejo_base="http://127.0.0.1:1") def _core(self, loader, save, read_back=None) -> SaveExecutorCore: return SaveExecutorCore(self.store, loader.load, save, read_back=read_back) # -- T1: relative valid path accepted ----------------------------------- def test_t1_relative_valid_path_accepted(self): job = self._make_job(vault_path="docs/test.md") validated = validate_job(job) self.assertEqual(validated["vault_path"], "docs/test.md") # -- T2: absolute /app/vault path rejected ------------------------------ def test_t2_absolute_app_vault_path_rejected(self): with self.assertRaises(JobRejectedError): self._make_job(vault_path="/app/vault/docs/test.md") # -- T3: absolute foreign path rejected --------------------------------- def test_t3_absolute_foreign_path_rejected(self): with self.assertRaises(JobRejectedError): self._make_job(vault_path="/etc/passwd") # -- T4: ../ rejected ---------------------------------------------------- def test_t4_dotdot_rejected(self): with self.assertRaises(JobRejectedError): self._make_job(vault_path="../secret") # -- T5: encoded traversal rejected -------------------------------------- def test_t5_encoded_traversal_rejected(self): with self.assertRaises(JobRejectedError): self._make_job(vault_path="..%2fsecret") # -- T6: backslash traversal rejected ----------------------------------- def test_t6_backslash_traversal_rejected(self): with self.assertRaises(JobRejectedError): self._make_job(vault_path="..\\secret") # -- T7: empty path rejected -------------------------------------------- def test_t7_empty_path_rejected(self): with self.assertRaises(JobRejectedError): self._make_job(vault_path="") # -- T8: URL rejected ---------------------------------------------------- def test_t8_url_rejected(self): with self.assertRaises(JobRejectedError): self._make_job(vault_path="http://evil/x.md") # -- T9: relative path reaches Tolaria mock unchanged --------------------- def test_t9_relative_path_reaches_tolaria_unchanged(self): job = self._make_job(vault_path="docs/test.md") self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") rb = FakeReadBack(self.content) core = self._core(self._loader(), save, rb) result = core.process_job(job["job_id"], "w-pc-t9") self.assertEqual(result["state"], "SUCCEEDED") self.assertEqual(len(save.calls), 1) # Tolaria erhält den relativen Pfad UNVERÄNDERT (kein Strip/Rewrite) self.assertEqual(save.calls[0][0], "docs/test.md") # -- T10: read-back uses same relative path ------------------------------ def test_t10_readback_uses_same_relative_path(self): job = self._make_job(vault_path="docs/test.md") self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") rb = FakeReadBack(self.content) core = self._core(self._loader(), save, rb) core.process_job(job["job_id"], "w-pc-t10") # read_back wurde mit demselben relativen Pfad aufgerufen self.assertEqual(rb.calls, 1) # -- T11: source loader loads exact relative path ------------------------ def test_t11_source_loader_loads_exact_relative_path(self): loader = self._loader() content = loader.load(self.commit, "object/test-obj-1", "docs/test.md") self.assertEqual(content, self.content) # -- T12: source commit binding preserved -------------------------------- def test_t12_source_commit_binding_preserved(self): loader = self._loader() with self.assertRaises(SourceUnavailableError): loader.load("deadbeef" * 5, "object/test-obj-1", "docs/test.md") # -- T13: provenance binding preserved ----------------------------------- def test_t13_provenance_binding_preserved(self): job = self._make_job(provenance_hash=_sha256("WRONG")) self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-pc-t13") self.assertEqual(result["state"], "FAILED") self.assertEqual(result["result_code"], "PROVENANCE_MISMATCH") self.assertEqual(len(save.calls), 0) # -- T14: path mismatch rejected ----------------------------------------- def test_t14_path_mismatch_rejected(self): # vault_path zeigt auf eine Datei, die im Commit nicht existiert job = self._make_job(vault_path="docs/missing.md") self.store.create_job(job) self.store.mark_ready(job["job_id"]) save = FakeTolariaSave("ok") core = self._core(self._loader(), save) result = core.process_job(job["job_id"], "w-pc-t14") self.assertEqual(result["state"], "FAILED") self.assertEqual(result["result_code"], "SOURCE_UNAVAILABLE") self.assertEqual(len(save.calls), 0) # -- T15: symlink escape rejected (where fs resolution applies) ---------- def test_t15_symlink_escape_rejected(self): # Executor-seitig: job_schema lehnt bereits alle nicht-relativen # Pfade ab. Ein Symlink-Escape via relativer Pfad ist durch die # Whitelist (kein .., kein Scheme) ausgeschlossen. with self.assertRaises(JobRejectedError): self._make_job(vault_path="docs/../../etc/passwd") # -- T16: no hidden /app/vault strip behavior ---------------------------- def test_t16_no_hidden_app_vault_strip(self): # Der relative Pfad wird NICHT gestrippt/rewritten — er erreicht # Tolaria unverändert. Ein absoluter Pfad wird abgelehnt, nicht # stillschweigend auf relativ reduziert. with self.assertRaises(JobRejectedError): self._make_job(vault_path="/app/vault/docs/test.md") # -- T17: old failed job remains failed --------------------------------- def test_t17_old_failed_job_remains_failed(self): # Der historische FAILED-Job (2e0b7a0d...) bleibt unverändert. # Hier: ein FAILED-Job wird nicht durch den Path-Repair berührt. job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) # Simuliere einen Fehlschlag (provenance mismatch) job2 = self._make_job(provenance_hash=_sha256("WRONG")) self.store.create_job(job2) self.store.mark_ready(job2["job_id"]) save = FakeTolariaSave("ok") core = self._core(self._loader(), save) core.process_job(job2["job_id"], "w-pc-t17") failed = self.store.get_job(job2["job_id"]) self.assertEqual(failed["state"], "FAILED") # Der erste Job bleibt unberührt (kein neuer SAVE) self.assertEqual(len(save.calls), 0) # -- T18: no production job created -------------------------------------- def test_t18_no_production_job_created(self): # Der Path-Repair erzeugt KEINE neuen Jobs. Nur die Test-Jobs # dieser Klasse existieren in der temporären DB. jobs = self.store.list_jobs() for j in jobs: self.assertNotIn("auth4c2-canary", j["vault_path"]) # -- T19: no production HTTP write --------------------------------------- def test_t19_no_production_http_write(self): # TolariaClient mit unerreichbarer Base-URL -> kein echter HTTP-Write. # Der Path-Repair selbst führt keinen produktiven SAVE aus. client = TolariaClient(base_url=FAKE_BASE, save_token=TEST_TOKEN) self.assertEqual(client.base_url, FAKE_BASE.rstrip("/")) # -- T20: no DELETE capability ------------------------------------------- def test_t20_no_delete_capability(self): # SAVE-Build enthält keinen DELETE-Executor-Code. build_dir = os.path.dirname(os.path.abspath(__file__)) runtime_modules = [ "save_executor_core.py", "job_store.py", "job_schema.py", "job_state_machine.py", "job_claim.py", "tolaria_client.py", "forgejo_source_loader.py", "worker.py", "entrypoint.py", ] for f in runtime_modules: with open(os.path.join(build_dir, f)) as fh: self.assertNotIn("delete_executor_core", fh.read()) if __name__ == "__main__": unittest.main(verbosity=2)