"""Application behavior tests with synthetic fixtures; no network or credentials.""" import asyncio import copy import importlib.util import json from pathlib import Path import subprocess import sys import tempfile import unittest def load(name): spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(name + ".py")) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module typed = load("typed_questions_lab") lab = load("reliability_lab") class TypeInterpretationTests(unittest.TestCase): def test_one_state_keeps_three_semantics_distinct(self): r = typed.interpret(typed.FIXTURE) self.assertEqual(r["severity_position_0_to_2"], 1.8) self.assertEqual(r["choice_confidence"], 0.8) self.assertEqual(r["winning_option_probability"], 0.85) self.assertEqual(r["ongoing_disruption_probability"], 0.9) self.assertEqual(r["action"], "review_before_any_ticket_update") def test_malformed_values_do_not_pass_as_typed_decisions(self): for key, field, value in [("route", "choice", "invented"), ("route", "confidence", True), ("severity", "score", 3), ("urgent", "noul", float("nan"))]: with self.subTest(key=key, field=field): bad = copy.deepcopy(typed.FIXTURE) bad[key][field] = value with self.assertRaises(ValueError): typed.interpret(bad) def test_distribution_must_cover_declared_options(self): bad = copy.deepcopy(typed.FIXTURE) del bad["route"]["probabilities"]["human"] with self.assertRaises(ValueError): typed.interpret(bad) class ReliabilityTests(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) self.db = str(Path(self.directory.name) / "lab.sqlite") def ingest(self, scenario="ok", **kwargs): return asyncio.run(lab.ingest(self.db, "ticket-7", 1, "Checkout unavailable", lab.fixture(scenario), **kwargs)) def test_transient_failure_retries_then_saves_one_outbox_event(self): self.assertEqual(self.ingest("transient")["attempts"], 2) self.assertEqual(len(lab.snapshot(self.db)["outbox"]), 1) def test_documented_cli_distinguishes_cached_auth_from_fresh_401(self): def run(*args): result = subprocess.run([sys.executable, str(Path(__file__).with_name("reliability_lab.py")), "ingest", "--db", self.db, *args], capture_output=True, text=True, check=True) return json.loads(result.stdout)["result"] first = run("--scenario", "transient") cached_auth = run("--scenario", "auth") fresh_auth = run("--ticket", "auth-case", "--scenario", "auth") self.assertEqual((first["action"], first["attempts"]), ("queue_assignment", 2)) self.assertEqual(cached_auth, first) self.assertEqual((fresh_auth["action"], fresh_auth["attempts"]), ("review", 1)) self.assertEqual(len(lab.snapshot(self.db)["outbox"]), 1) def test_auth_and_invalid_answers_are_not_retried_or_enqueued(self): for scenario in ("auth", "invalid"): r = asyncio.run(lab.ingest(self.db, scenario, 1, "Input", lab.fixture(scenario))) self.assertEqual((r["action"], r["attempts"]), ("review", 1)) self.assertEqual(lab.snapshot(self.db)["outbox"], []) def test_timeouts_have_a_finite_attempt_budget(self): r = self.ingest("timeout", attempt_timeout=0.005, total_timeout=0.05, max_attempts=2) self.assertEqual((r["action"], r["reason"], r["attempts"]), ("review", "timeout", 2)) self.assertEqual(lab.snapshot(self.db)["assignments"], []) def test_exhausted_total_budget_never_calls_provider(self): r = self.ingest(total_timeout=0) self.assertEqual((r["action"], r["reason"], r["attempts"]), ("review", "budget_exhausted", 0)) def test_cached_decision_does_not_reevaluate_and_key_binds_input(self): first = self.ingest() async def must_not_call(_): self.fail("cached request must not call provider") self.assertEqual(asyncio.run(lab.ingest(self.db, "ticket-7", 1, "Checkout unavailable", must_not_call)), first) with self.assertRaises(ValueError): asyncio.run(lab.ingest(self.db, "ticket-7", 1, "Different input", must_not_call)) def test_concurrent_evaluations_share_one_commit_and_one_effect(self): async def run(): async def slow(_): await asyncio.sleep(0.005) return {"type": "choice", "choice": "technical", "confidence": 0.8} return await asyncio.gather(*(lab.ingest(self.db, "t", 1, "Input", slow) for _ in range(2))) asyncio.run(run()) self.assertEqual(len(lab.snapshot(self.db)["outbox"]), 1) self.assertEqual(lab.deliver(self.db), 1) self.assertEqual(lab.deliver(self.db), 0) def test_crash_rolls_back_effect_marker_and_delivery_together(self): self.ingest() with self.assertRaises(RuntimeError): lab.deliver(self.db, crash_before_commit=True) state = lab.snapshot(self.db) self.assertEqual(state["assignments"], []) self.assertEqual(state["applied_events"], []) self.assertEqual(state["outbox"][0]["delivered"], 0) self.assertEqual(lab.deliver(self.db), 1) def test_older_version_cannot_overwrite_newer_assignment(self): asyncio.run(lab.ingest(self.db, "t", 2, "Input v2", lab.fixture("ok"))) lab.deliver(self.db) asyncio.run(lab.ingest(self.db, "t", 1, "Input v1", lab.fixture("ok"))) lab.deliver(self.db) self.assertEqual(lab.snapshot(self.db)["assignments"][0]["version"], 2) def test_restart_and_crash_across_real_processes(self): script = str(Path(__file__).with_name("reliability_lab.py")) def run(command, *args, check=True): r = subprocess.run([sys.executable, script, command, "--db", self.db, *args], capture_output=True, text=True) if check: self.assertEqual(r.returncode, 0, r.stderr) return json.loads(r.stdout)["result"] return r run("ingest") # A second process with failing provider fixtures reuses the saved decision. self.assertEqual(run("ingest", "--scenario", "auth")["action"], "queue_assignment") self.assertEqual(run("deliver", "--hard-crash-before-commit", check=False).returncode, 86) self.assertEqual(run("inspect")["assignments"], []) self.assertEqual(run("deliver")["applied"], 1) self.assertEqual(run("deliver")["applied"], 0) self.assertEqual(len(run("inspect")["assignments"]), 1) if __name__ == "__main__": unittest.main()