# -*- coding: utf-8 -*- """Regression — task-2553+49 MICRO-HARDENING: executor self-collector / self-dispatch / self-adjudication 구조적 차단 + write-back role/fallback binding conflict 코드화. Covers 회장 §4 verbatim regression 1~15, §3 8-classification coverage, §2/§8 owner-pin invariants, §5/9-R byte-0 carve-out, and 9-R.1 Layer A: §4.1 executor → ANU key normal callback = PASS §4.2 executor → self key normal callback = FAIL §4.3 executor → self key fallback callback = FAIL §4.4 executor_key == collector_key = FAIL §4.5 collector_role != ANU = FAIL §4.6 prompt says ANU collector but owner key is executor = FAIL §4.7 executor self-adjudication (Codex audit/adjudication) = FAIL §4.8 executor self-dispatch (follow-up) = FAIL §4.9 +47 self-chain fixture = FAIL §4.10 +47/+48 independent ANU verification fixture = PASS §4.11 idempotency same collector but role/fallback mismatch -> WRITEBACK_ BINDING_CONFLICT §4.12 valid duplicate write-back -> idempotent SKIP §4.13 +32 callback mandatory regression 무회귀 §4.14 +44/+46 4-tuple registry regression 무회귀 §4.15 +45/+48 cancel-on-success regression 무회귀 §5/9-R byte-0 carve-out: anu_v3/callback_4tuple_registry.py and dispatch/executor_completion_contract.py are byte-0 (pinned by the existing +44/+48 FROZEN_SHA256 autouse invariant; additive patching them would regress 13/14/15 and violate §5 "기존 +47/+48 산출물 수정 금지"). The §2 owner fields (executor_key/collector_key/collector_owner_key/collector_role=ANU) are carried by the NEW dispatch.callback_owner_enforcer record + schema, NOT by mutating the frozen legacy record (byte-0 우선·불가피 시 additive; the §9 allowlist is a max-write set, not a must-write set; +42/+43/+45 선례 동형). 9-R.1 Layer A: callback_owner_enforcer / normal_fallback_callback_helper perform ZERO cron register/remove, ZERO dispatch/merge/subprocess/cokacdir exec — pure validation only. """ import ast import hashlib import importlib.util import json import subprocess import sys import unittest from pathlib import Path import jsonschema _ROOT = Path(__file__).resolve().parents[2] if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) def _load(modname: str, relpath: str): """Hermetic file-path import (collision-proof vs tests/dispatch shadow).""" spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath) assert spec is not None and spec.loader is not None mod = importlib.util.module_from_spec(spec) sys.modules[modname] = mod spec.loader.exec_module(mod) return mod # Pre-seed canonical dotted names so internal `from dispatch.* import` and # `from anu_v3.* import` resolve to the real workspace modules. _ecc = _load( "dispatch.executor_completion_contract", "dispatch/executor_completion_contract.py", ) _stv = _load( "dispatch.spec_template_validator", "dispatch/spec_template_validator.py", ) _reg = _load( "anu_v3.callback_4tuple_registry", "anu_v3/callback_4tuple_registry.py", ) _enf = _load( "dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py", ) _hlp = _load( "dispatch.normal_fallback_callback_helper", "dispatch/normal_fallback_callback_helper.py", ) _grd = _load( "dispatch.cron_dispatch_guard", "dispatch/cron_dispatch_guard.py", ) enforce_callback_owner = _enf.enforce_callback_owner assert_not_self_adjudication = _enf.assert_not_self_adjudication assert_not_self_dispatch = _enf.assert_not_self_dispatch audit_writeback_binding_conflict = _enf.audit_writeback_binding_conflict CallbackOwner4Tuple = _enf.CallbackOwner4Tuple PASS = _enf.PASS FAIL = _enf.FAIL HOLD = _enf.HOLD SELF_COLLECTOR_FORBIDDEN = _enf.SELF_COLLECTOR_FORBIDDEN EXECUTOR_SELF_ADJUDICATION_FORBIDDEN = _enf.EXECUTOR_SELF_ADJUDICATION_FORBIDDEN SELF_DISPATCH_FORBIDDEN = _enf.SELF_DISPATCH_FORBIDDEN CALLBACK_OWNER_MISMATCH = _enf.CALLBACK_OWNER_MISMATCH CALLBACK_COLLECTOR_NOT_ANU = _enf.CALLBACK_COLLECTOR_NOT_ANU CALLBACK_4TUPLE_INVALID = _enf.CALLBACK_4TUPLE_INVALID DISPATCH_PATH_BYPASSED_CONTRACT = _enf.DISPATCH_PATH_BYPASSED_CONTRACT WRITEBACK_BINDING_CONFLICT = _enf.WRITEBACK_BINDING_CONFLICT WRITEBACK_IDEMPOTENT_SKIP = _enf.WRITEBACK_IDEMPOTENT_SKIP WRITEBACK_NO_CONFLICT = _enf.WRITEBACK_NO_CONFLICT ALL_CLASSIFICATIONS = _enf.ALL_CLASSIFICATIONS build_anu_owned_callback_request = _hlp.build_anu_owned_callback_request verify_post_registration_owner = _hlp.verify_post_registration_owner guard_dispatch_with_owner = _grd.guard_dispatch_with_owner Callback4Tuple = _ecc.Callback4Tuple make_record = _reg.make_record ENF_SRC = _ROOT / "dispatch" / "callback_owner_enforcer.py" HLP_SRC = _ROOT / "dispatch" / "normal_fallback_callback_helper.py" SCHEMA = _ROOT / "schemas" / "callback_owner_enforcement.schema.json" FX_SELF = (_ROOT / "memory" / "fixtures" / "task-2553plus47.self-chain-violation.json") FX_ANU = (_ROOT / "memory" / "fixtures" / "task-2553plus47-48.independent-anu.json") EXEC_KEY = "1e41a2324a3ccdd0" # dev6 페룬 executor self key (발사 금지 대상) ANU_KEY = "c119085addb0f8b7" # 독립 ANU key (회장 §10) CHAT = "6937032012" GIT_HEAD_PRE = "20456b5f83fc039f2fd6f50f4b94095c29b41bfb" GIT_BRANCH_PRE = "task/task-2553p1-f1-clean-replacement" # §5/9-R byte-0 carve-out — frozen modules pinned by the existing +44/+48 # FROZEN_SHA256 autouse invariant. Literal pins (within-run tautology 금지): # a change here means 13/14/15 regress and §5 is violated. REG_SHA = "774d550628410d36962c23a7663c4b6dbf72789de7c7fd940871e9ad8280e5ab" ECC_SHA = "364caa11904285657abd716d78c5493b1f8b519318387d0f864fb6a136dca0b4" CET_SHA = "352ad0f570e55040e7c1e4a32cbfe0f076cbd53529b4db6222a8da1a4bee9cc5" _CLAUSE = ( "[EXECUTOR COMPLETION CALLBACK — MANDATORY] executor 는 ANU 에 normal " "completion callback cron 을 반드시 발사해야 한다." ) def _sha(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() def _git(*args): return subprocess.run( ["git", "-C", str(_ROOT), *args], capture_output=True, text=True, check=True, ).stdout.strip() def _enf_call(**over): base = dict( task_id="task-2553+49", executor_key=EXEC_KEY, collector_key=ANU_KEY, collector_owner_key=ANU_KEY, collector_role="ANU", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", dispatch_cron_id="D", chat_id=CHAT, prompt_claims_anu_collector=True, entry_path="cokacdir_cron_direct", ) base.update(over) return enforce_callback_owner(**base) class OwnerPinRegression(unittest.TestCase): """§4 회장 verbatim regression 1~12 + §3 classification coverage.""" # §4.1 — executor → ANU key normal callback = PASS def test_01_executor_to_anu_normal_pass(self): r = _enf_call() self.assertEqual(r.verdict, PASS) self.assertEqual(r.classifications, []) self.assertTrue(r.owner_is_independent_anu) # §4.2 — executor → self key normal callback = FAIL def test_02_executor_self_normal_fail(self): r = _enf_call(collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY) self.assertEqual(r.verdict, FAIL) self.assertIn(SELF_COLLECTOR_FORBIDDEN, r.classifications) self.assertFalse(r.owner_is_independent_anu) # §4.3 — executor → self key fallback callback = FAIL def test_03_executor_self_fallback_fail(self): # fallback owned by executor self key (normal absent here). r = _enf_call( collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY, normal_collector_cron_id="NC", fallback_callback_cron_id="FB-self", ) self.assertEqual(r.verdict, FAIL) self.assertIn(SELF_COLLECTOR_FORBIDDEN, r.classifications) # §4.4 — executor_key == collector_key = FAIL def test_04_executor_eq_collector_fail(self): r = _enf_call(collector_key=EXEC_KEY, collector_owner_key=ANU_KEY) self.assertEqual(r.verdict, FAIL) self.assertIn(SELF_COLLECTOR_FORBIDDEN, r.classifications) # §4.5 — collector_role != ANU = FAIL def test_05_collector_role_not_anu_fail(self): r = _enf_call(collector_role="executor") self.assertEqual(r.verdict, FAIL) self.assertIn(CALLBACK_COLLECTOR_NOT_ANU, r.classifications) # §4.6 — prompt says ANU collector but owner key is executor = FAIL def test_06_prompt_anu_but_owner_executor_fail(self): r = _enf_call( collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY, prompt_claims_anu_collector=True, ) self.assertEqual(r.verdict, FAIL) self.assertIn(CALLBACK_OWNER_MISMATCH, r.classifications) self.assertIn(SELF_COLLECTOR_FORBIDDEN, r.classifications) # §4.7 — executor self-adjudication = FAIL def test_07_executor_self_adjudication_fail(self): g = assert_not_self_adjudication( executor_key=EXEC_KEY, actor_key=EXEC_KEY, is_codex_audit=True, is_adjudication=True) self.assertEqual(g.verdict, FAIL) self.assertEqual(g.classification, EXECUTOR_SELF_ADJUDICATION_FORBIDDEN) # independent ANU actor -> PASS ok = assert_not_self_adjudication( executor_key=EXEC_KEY, actor_key=ANU_KEY, is_codex_audit=True, is_adjudication=True) self.assertEqual(ok.verdict, PASS) # §4.8 — executor self-dispatch = FAIL def test_08_executor_self_dispatch_fail(self): g = assert_not_self_dispatch( executor_key=EXEC_KEY, actor_key=EXEC_KEY, is_followup_dispatch=True) self.assertEqual(g.verdict, FAIL) self.assertEqual(g.classification, SELF_DISPATCH_FORBIDDEN) ok = assert_not_self_dispatch( executor_key=EXEC_KEY, actor_key=ANU_KEY, is_followup_dispatch=True) self.assertEqual(ok.verdict, PASS) # §4.9 — +47 self-chain fixture = FAIL (every limb) def test_09_plus47_self_chain_fixture_fail(self): fx = json.loads(FX_SELF.read_text("utf-8")) c = fx["cases"] rn = enforce_callback_owner( **c["normal_callback_self_owned"]["enforce_input"], anu_keys=tuple(fx["anu_keys"])) self.assertEqual(rn.verdict, FAIL) self.assertIn(SELF_COLLECTOR_FORBIDDEN, rn.classifications) rf = enforce_callback_owner( **c["fallback_callback_self_owned"]["enforce_input"], anu_keys=tuple(fx["anu_keys"])) self.assertEqual(rf.verdict, FAIL) ga = assert_not_self_adjudication( **c["dev3_self_codex_audit"]["input"]) self.assertEqual(ga.verdict, FAIL) self.assertEqual(ga.classification, EXECUTOR_SELF_ADJUDICATION_FORBIDDEN) gd = assert_not_self_dispatch( **c["dev3_self_dispatch_plus48"]["input"]) self.assertEqual(gd.verdict, FAIL) self.assertEqual(gd.classification, SELF_DISPATCH_FORBIDDEN) # §4.10 — +47/+48 independent ANU verification fixture = PASS def test_10_independent_anu_fixture_pass(self): fx = json.loads(FX_ANU.read_text("utf-8")) c = fx["cases"] rn = enforce_callback_owner( **c["normal_callback_anu_owned"]["enforce_input"], anu_keys=tuple(fx["anu_keys"])) self.assertEqual(rn.verdict, PASS) self.assertEqual(rn.classifications, []) self.assertTrue(rn.owner_is_independent_anu) self.assertEqual( assert_not_self_adjudication( **c["anu_session_codex_audit_ok"]["input"]).verdict, PASS) self.assertEqual( assert_not_self_dispatch( **c["anu_session_followup_dispatch_ok"]["input"]).verdict, PASS) # §4.11 — idempotency same collector but role/fallback mismatch -> # WRITEBACK_BINDING_CONFLICT (recorded, NOT silent skip) def test_11_writeback_binding_conflict(self): fx = json.loads(FX_ANU.read_text("utf-8")) case = fx["cases"]["idempotency_role_fallback_conflict"] hist = [make_record( task_id="task-2553+49", dispatch_id=h["dispatch_id"], dispatch_cron_id=h["dispatch_id"], executor="dev", chat_id=h["chat_id"], normal_collector_cron_id=h["normal_collector_cron_id"], fallback_callback_cron_id=h["fallback_callback_cron_id"], role=h["role"], status=h["status"]) for h in case["history"]] a = audit_writeback_binding_conflict( hist, task_id="task-2553+49", **{k: v for k, v in case["candidate"].items() if k != "task_id"}) self.assertEqual(a.classification, WRITEBACK_BINDING_CONFLICT) self.assertEqual(a.verdict, FAIL) self.assertTrue(a.conflict) self.assertTrue(a.matched_idempotency_key) self.assertTrue(a.conflicting_fields) # recorded, not silent # §4.12 — valid duplicate write-back -> idempotent SKIP def test_12_writeback_idempotent_skip(self): fx = json.loads(FX_ANU.read_text("utf-8")) case = fx["cases"]["idempotency_true_duplicate"] hist = [make_record( task_id="task-2553+49", dispatch_id=h["dispatch_id"], dispatch_cron_id=h["dispatch_id"], executor="dev", chat_id=h["chat_id"], normal_collector_cron_id=h["normal_collector_cron_id"], fallback_callback_cron_id=h["fallback_callback_cron_id"], role=h["role"], status=h["status"]) for h in case["history"]] a = audit_writeback_binding_conflict( hist, task_id="task-2553+49", **{k: v for k, v in case["candidate"].items() if k != "task_id"}) self.assertEqual(a.classification, WRITEBACK_IDEMPOTENT_SKIP) self.assertEqual(a.verdict, PASS) self.assertFalse(a.conflict) # no idempotency match at all -> NO_CONFLICT a2 = audit_writeback_binding_conflict( [], task_id="task-2553+49", dispatch_id="X", chat_id=CHAT, normal_collector_cron_id="NC", candidate_role="executor", candidate_fallback_cron_id="FB") self.assertEqual(a2.classification, WRITEBACK_NO_CONFLICT) class ClassificationCoverage(unittest.TestCase): """§3 — all 8 classifications are reachable & schema-valid.""" def test_all_8_classifications_reachable(self): seen = set() seen |= set(_enf_call(collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY).classifications) seen |= set(_enf_call(collector_role="x").classifications) seen |= set(_enf_call( normal_collector_cron_id=None).classifications) seen |= set(_enf_call(entry_path="mystery_path").classifications) seen |= set(_enf_call(collector_owner_key="deadbeef", collector_key="deadbeef").classifications) seen.add(assert_not_self_adjudication( executor_key=EXEC_KEY, actor_key=EXEC_KEY, is_adjudication=True).classification) seen.add(assert_not_self_dispatch( executor_key=EXEC_KEY, actor_key=EXEC_KEY, is_followup_dispatch=True).classification) seen.add(audit_writeback_binding_conflict( [make_record(task_id="t", dispatch_id="d", dispatch_cron_id="d", executor="e", chat_id=CHAT, normal_collector_cron_id="n", fallback_callback_cron_id="f1", role="executor", status="COMPLETED")], task_id="t", dispatch_id="d", chat_id=CHAT, normal_collector_cron_id="n", candidate_role="fallback", candidate_fallback_cron_id="f2").classification) for c in ALL_CLASSIFICATIONS: self.assertIn(c, seen, f"{c} not reachable") def test_schema_validation(self): schema = json.loads(SCHEMA.read_text("utf-8")) jsonschema.validate(_enf_call().to_json(), schema) jsonschema.validate( _enf_call(collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY).to_json(), schema) jsonschema.validate( assert_not_self_dispatch( executor_key=EXEC_KEY, actor_key=EXEC_KEY, is_followup_dispatch=True).to_json(), schema) jsonschema.validate( audit_writeback_binding_conflict( [], task_id="t", dispatch_id="d", chat_id=CHAT, normal_collector_cron_id="n", candidate_role="executor", candidate_fallback_cron_id="f").to_json(), schema) t = CallbackOwner4Tuple( task_id="task-2553+49", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", executor_key=EXEC_KEY, collector_key=ANU_KEY, collector_owner_key=ANU_KEY, collector_role="ANU", chat_id=CHAT) jsonschema.validate(t.to_json(), schema) def test_hold_is_conditional_only(self): # §6/9-R.2 — HOLD ONLY when the ANU key set is genuinely # un-resolvable; never a blanket pre-declaration. r = _enf_call(anu_keys_resolvable=False) self.assertEqual(r.verdict, HOLD) r2 = _enf_call(anu_keys=()) self.assertEqual(r2.verdict, HOLD) # normal path never HOLDs self.assertEqual(_enf_call().verdict, PASS) class HelperAndCompositeGate(unittest.TestCase): """§8/§10 helper fail-closed + composite gate.""" def test_helper_failclosed_no_argv_on_self_owner(self): cr = build_anu_owned_callback_request( kind="normal", task_id="task-2553+49", executor_key=EXEC_KEY, owner_key=EXEC_KEY, chat_id=CHAT, prompt="p", at="10s", cron_id="NC", dispatch_cron_id="D", fallback_callback_cron_id="FB") self.assertEqual(cr.verdict, FAIL) self.assertIsNone(cr.argv) # cron-direct CANNOT register self-owned def test_helper_anu_owner_produces_anukeyed_argv(self): cr = build_anu_owned_callback_request( kind="normal", task_id="task-2553+49", executor_key=EXEC_KEY, owner_key=ANU_KEY, chat_id=CHAT, prompt="p", at="10s", cron_id="NC", dispatch_cron_id="D", fallback_callback_cron_id="FB") self.assertEqual(cr.verdict, PASS) self.assertIn("--key", cr.argv) self.assertEqual(cr.argv[cr.argv.index("--key") + 1], ANU_KEY) self.assertNotIn(EXEC_KEY, cr.argv) def test_post_registration_owner_crosscheck(self): ok = verify_post_registration_owner( task_id="t", executor_key=EXEC_KEY, observed_owner_key=ANU_KEY, observed_chat_id=CHAT, observed_role="ANU", expected_chat_id=CHAT) self.assertEqual(ok.verdict, PASS) bad = verify_post_registration_owner( task_id="t", executor_key=EXEC_KEY, observed_owner_key=EXEC_KEY, observed_chat_id=CHAT, observed_role="ANU", expected_chat_id=CHAT) self.assertEqual(bad.verdict, FAIL) def test_composite_gate_strictest_wins(self): t = Callback4Tuple(task_id="task-2553+49", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB") b, o, c = guard_dispatch_with_owner( spec_text=_CLAUSE, entry_path="cokacdir_cron_direct", tuple_=t, executor_key=EXEC_KEY, collector_key=ANU_KEY, collector_owner_key=ANU_KEY, collector_role="ANU", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", chat_id=CHAT, prompt_claims_anu_collector=True) self.assertEqual((b.verdict, o.verdict, c), (PASS, PASS, PASS)) b, o, c = guard_dispatch_with_owner( spec_text=_CLAUSE, entry_path="cokacdir_cron_direct", tuple_=t, executor_key=EXEC_KEY, collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY, collector_role="ANU", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", chat_id=CHAT) self.assertEqual(c, FAIL) class FrozenByte0AndGitInvariant(unittest.TestCase): """§5/9-R byte-0 carve-out + git invariant (read-only regression).""" def test_registry_ecc_cet_byte0(self): self.assertEqual( _sha(_ROOT / "anu_v3/callback_4tuple_registry.py"), REG_SHA) self.assertEqual( _sha(_ROOT / "dispatch/executor_completion_contract.py"), ECC_SHA) self.assertEqual( _sha(_ROOT / "anu_v3/callback_event_trigger.py"), CET_SHA) def test_git_invariant(self): self.assertEqual(_git("rev-parse", "HEAD"), GIT_HEAD_PRE) self.assertEqual( _git("rev-parse", "--abbrev-ref", "HEAD"), GIT_BRANCH_PRE) class LayerANoCronInvariant(unittest.TestCase): """9-R.1 Layer A — ZERO cron/dispatch/subprocess/cokacdir exec.""" def _assert_no_exec(self, src: Path): """No subprocess/os.system/Popen import or call anywhere (the real Layer A guarantee — prose docstrings are never executed).""" tree = ast.parse(src.read_text("utf-8")) for node in ast.walk(tree): if isinstance(node, ast.Import): for a in node.names: self.assertNotEqual(a.name.split(".")[0], "subprocess", f"Layer A: import subprocess {src}") if isinstance(node, ast.ImportFrom): self.assertNotEqual(node.module, "subprocess", f"Layer A: from subprocess {src}") if isinstance(node, ast.Call) and isinstance( node.func, ast.Attribute): base = node.func.value if isinstance(base, ast.Name): self.assertNotEqual((base.id, node.func.attr), ("os", "system"), str(src)) self.assertNotEqual(base.id, "subprocess", str(src)) self.assertNotEqual(node.func.attr, "Popen", str(src)) def test_enforcer_layer_a(self): self._assert_no_exec(ENF_SRC) # the enforcer is pure validation — it never even names a cokacdir # command string (not in code, not in docstrings). tree = ast.parse(ENF_SRC.read_text("utf-8")) for node in ast.walk(tree): if isinstance(node, ast.Constant) and isinstance( node.value, str): self.assertNotIn("cokacdir --", node.value.lower()) def test_helper_layer_a(self): # the helper BUILDS a cokacdir argv as DATA (ast.List elements) and # never execs it; a "cokacdir"/"--cron" token may legitimately appear # (a) in a docstring (prose, never executed) or (b) as an element of # a list literal (the returned argv). It must NEVER be an argument to # a call. self._assert_no_exec(HLP_SRC) tree = ast.parse(HLP_SRC.read_text("utf-8")) call_arg_ids = set() for node in ast.walk(tree): if isinstance(node, ast.Call): for a in list(node.args) + [ k.value for k in node.keywords]: if isinstance(a, ast.Constant): call_arg_ids.add(id(a)) # A "cokacdir"/"--cron" literal may legitimately exist as argv DATA # (returned list) or prose; it must NEVER be passed to a call (which # would be the only way to exec it). _assert_no_exec already bans # subprocess/os.system/Popen; this pins the data-only invariant. for node in ast.walk(tree): if isinstance(node, ast.Constant) and isinstance( node.value, str) and ( "cokacdir" in node.value.lower() or "--cron" in node.value.lower()): self.assertNotIn( id(node), call_arg_ids, "Layer A: a cokacdir/--cron string is a call argument " "(would be an exec); it must be argv DATA only.") class CrossRegressionNoRegression(unittest.TestCase): """§4.13/§4.14/§4.15 — +32 / +44+46 / +45+48 무회귀 (subprocess).""" def _run(self, *files): r = subprocess.run( [sys.executable, "-m", "pytest", "-q", "--no-header", *[str(_ROOT / "tests" / "regression" / f) for f in files]], cwd=str(_ROOT), capture_output=True, text=True, timeout=900) self.assertEqual( r.returncode, 0, f"regression 무회귀 실패 ({files}):\n{r.stdout[-2000:]}\n" f"{r.stderr[-1000:]}") def test_13_plus32_mandatory_no_regression(self): self._run("test_executor_completion_callback_mandatory_2553plus32.py") def test_14_plus44_plus46_registry_no_regression(self): self._run("test_callback_4tuple_registry_2553plus44.py", "test_artifact_root_resolver_2553plus46.py") def test_15_plus45_plus48_cancel_no_regression(self): self._run("test_cancel_on_success_live_wiring_2553plus45.py", "test_cancel_on_success_live_e2e_2553plus48.py", "test_callback_event_trigger_2553plus47.py") if __name__ == "__main__": unittest.main(verbosity=2)