152 lines
10 KiB
Python
152 lines
10 KiB
Python
"""Pure immutable r26 driver/mapper review. No auth or production import."""
|
|
import ast
|
|
import contextlib
|
|
import copy
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
OUT = Path(__file__).resolve().parents[2] / "app/artifacts/ai_concurrency"
|
|
spec = importlib.util.spec_from_file_location("r26_driver_independent", OUT / "private_tcm_r26_acceptance_driver.py")
|
|
driver = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(driver)
|
|
|
|
|
|
def summary(status="stopped", *, identified=True):
|
|
terminal = identified and status in {"stopped", "succeeded", "failed"}
|
|
lifecycle = dict(producer_terminal_status="stopped", persisted_run_status=status, independently_terminal=terminal,
|
|
generation_transfer_closed=True, lifecycle_conflict=status != "stopped", stop_accepted=True,
|
|
verification_attempts=1, verification_elapsed_ms=25,
|
|
observations=[dict(status=status, observed_at=12.0, within_budget=True)],
|
|
producer_terminal_observed_at=10.0, persisted_run_observed_at=12.0, stop_accepted_at=9.0,
|
|
transfer_closed_at=11.0, verification_started_at=12345.0, clock_domains=dict(wall="unix_seconds_microtime", verification_started_at="monotonic_seconds_hrtime", verification_elapsed_ms="monotonic_milliseconds")) if identified else None
|
|
return dict(audit_schema="v43r26b_submissions", generation_count=1, submitted_attempt_count=1, submitted_transfer_closed_count=1,
|
|
submitted_pending_attempt_count=int(not terminal), submitted_without_identity_count=int(not identified),
|
|
pre_execution_closed_attempt_count=0, not_submitted_attempt_count=0, rejected_before_execution_attempt_count=0,
|
|
never_submitted_generation_count=0, no_execution_generation_count=0, persisted=True, serialized_request_count=1, original_task_count=int(identified), original_run_count=int(identified),
|
|
audit_bytes=1200, audit_sha256="b" * 64, closed_generation_count=1,
|
|
independently_terminal_generation_count=int(terminal), pending_generation_count=int(not terminal),
|
|
unidentified_generation_count=int(not identified), lifecycle_consistent=True, independent_cleanup_confirmed=terminal,
|
|
generations=[dict(generation=1, serialized_request=True, submitted=True, not_submitted=False, no_execution_confirmed=False,
|
|
original_task_captured=identified, submission=dict(submitted_at=8.0, transfer_closed=True, transfer_closed_at=11.0, disposition="unconfirmed"), original_pair_captured=identified, transfer_closed=True, closed_at_request_ms=10.0,
|
|
transfer_exit="natural_eof", independently_terminal=terminal, lifecycle=lifecycle)])
|
|
|
|
|
|
def result(status="stopped", *, identified=True, controlled=False):
|
|
return with_counts(dict(preparation_isolation_verified=True, production_acceptance_path=True, diagnostic_only=False,
|
|
ok=not controlled, error_code="UPSTREAM_TIMEOUT" if controlled else "", private_wire_audit=summary(status, identified=identified),
|
|
state=dict(transport_closed=True, upstream_terminal_confirmed=identified and status != "running",
|
|
upstream_stop_confirmed=True, cancel_requested=False, timeout_reached=controlled,
|
|
upstream_pending_task_count=int(not identified or status == "running"))))
|
|
|
|
|
|
def with_counts(row):
|
|
# Derivative fixtures supply the canonical new Control counters; assertions
|
|
# below still test all original pending/ownership/denominator outcomes.
|
|
row = copy.deepcopy(row)
|
|
audit = row["private_wire_audit"]
|
|
for key in ("submitted_attempt_count", "submitted_transfer_closed_count", "submitted_pending_attempt_count",
|
|
"submitted_without_identity_count", "pre_execution_closed_attempt_count", "not_submitted_attempt_count", "rejected_before_execution_attempt_count"):
|
|
row["state"][key] = audit[key]
|
|
row["state"].setdefault("upstream_pending_task_count", audit["pending_generation_count"])
|
|
if "terminal_recovery_after_state" in row:
|
|
row["terminal_recovery_after_state"] = copy.deepcopy(row["state"])
|
|
return row
|
|
|
|
|
|
class IndependentDriver(unittest.TestCase):
|
|
def test_fully_unknown_generation_is_durable_but_never_clean(self):
|
|
value = summary(identified=False)
|
|
self.assertEqual(driver.safe_private_wire_summary(value), value)
|
|
row = result(identified=False)
|
|
row["state"].update(upstream_terminal_confirmed=True, upstream_pending_task_count=0)
|
|
self.assertFalse(driver.independent_cleanup_confirmed(with_counts(row)))
|
|
self.assertEqual(value["pending_generation_count"], 1)
|
|
|
|
def test_safe_summary_rejects_raw_nested_fields_and_false_terminal(self):
|
|
for status in ("running", "paused", "partial-succeeded"):
|
|
value = summary(status)
|
|
self.assertEqual(driver.safe_private_wire_summary(value), value)
|
|
value["independent_cleanup_confirmed"] = True
|
|
self.assertEqual(driver.safe_private_wire_summary(value), {"persisted": False})
|
|
for path in ([], ["generations", 0], ["generations", 0, "lifecycle"], ["generations", 0, "lifecycle", "observations", 0]):
|
|
value = summary(); target = value
|
|
for key in path:
|
|
target = target[key]
|
|
target["original_task_id"] = "private-identity-must-not-export"
|
|
self.assertEqual(driver.safe_private_wire_summary(value), {"persisted": False})
|
|
|
|
def test_full90_pending_holds_remaining_profile_slots(self):
|
|
writes = []; original_open = Path.open
|
|
@contextlib.contextmanager
|
|
def memory_open(path, mode="r", *args, **kwargs):
|
|
if mode == "x":
|
|
stream = io.StringIO(); yield stream; stream.close()
|
|
else:
|
|
with original_open(path, mode, *args, **kwargs) as stream:
|
|
yield stream
|
|
replies = [result(controlled=True), result("running")]
|
|
with patch.object(driver, "run", side_effect=replies) as dispatch, \
|
|
patch.object(driver, "_local_fcgi", side_effect=AssertionError("auth forbidden")) as auth, \
|
|
patch.object(Path, "exists", return_value=False), patch.object(Path, "open", new=memory_open), \
|
|
patch.object(Path, "write_text", side_effect=lambda text, **kw: writes.append(json.loads(text))), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
driver.run_deadline_sequence("independent_full_pending", ["qwen", "openai"], driver.argparse.ArgumentParser())
|
|
self.assertEqual(dispatch.call_count, 2)
|
|
auth.assert_not_called()
|
|
final = writes[-1]
|
|
self.assertEqual((len(final["expected_requests"]), final["completed_requests"], len(final["unexecuted_requests"])), (4, 2, 2))
|
|
self.assertFalse(final["suite_complete"])
|
|
self.assertTrue(final["rows"][0]["ordinary_deadline_pass"])
|
|
self.assertFalse(final["rows"][1]["next_full_production_acceptance_pass"])
|
|
self.assertEqual({row["reason"] for row in final["unexecuted_requests"]}, {"PREDECESSOR_CLEANUP_UNCONFIRMED"})
|
|
|
|
def test_unknown_deadline_predecessor_never_dispatches_full90(self):
|
|
writes = []; original_open = Path.open
|
|
@contextlib.contextmanager
|
|
def memory_open(path, mode="r", *args, **kwargs):
|
|
if mode == "x":
|
|
stream = io.StringIO(); yield stream; stream.close()
|
|
else:
|
|
with original_open(path, mode, *args, **kwargs) as stream:
|
|
yield stream
|
|
with patch.object(driver, "run", return_value=result(identified=False, controlled=True)) as dispatch, \
|
|
patch.object(driver, "_local_fcgi", side_effect=AssertionError("auth forbidden")) as auth, \
|
|
patch.object(Path, "exists", return_value=False), patch.object(Path, "open", new=memory_open), \
|
|
patch.object(Path, "write_text", side_effect=lambda text, **kw: writes.append(json.loads(text))), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
driver.run_deadline_sequence("independent_unknown", ["qwen"], driver.argparse.ArgumentParser())
|
|
self.assertEqual(dispatch.call_count, 1); auth.assert_not_called()
|
|
self.assertEqual(writes[-1]["completed_requests"], 1)
|
|
self.assertEqual(len(writes[-1]["unexecuted_requests"]), 1)
|
|
self.assertFalse(writes[-1]["rows"][0]["ordinary_deadline_pass"])
|
|
|
|
def test_export_auth_question_and_clinical_callback_preserved(self):
|
|
for kind in ("acceptance", "deadline"):
|
|
old = (OUT / f"private_tcm_r25_{kind}_probe.php").read_text(encoding="utf-8")
|
|
new = (OUT / f"private_tcm_r26_{kind}_probe.php").read_text(encoding="utf-8").replace("r26", "r25").replace("R26", "R25")
|
|
self.assertEqual(old[:old.index(" $controlDir =")], new[:new.index(" $controlDir =")])
|
|
for start, end in [(" // No real preparation", " $testedHashes="), (" $reply = ", " if ($reply === null)")]:
|
|
self.assertEqual(old[old.index(start):old.index(end)], new[new.index(start):new.index(end)])
|
|
self.assertLess(new.index("$control->markTerminal($taskId,$status,$workflowRunId)"), new.index("$privateWireAudit->captureControl($taskId)", new.index("'on_terminal'")))
|
|
self.assertIn("$control->closeGeneration($taskId, [\\app\\common\\service\\DifyWorkflowRunStatus::class, 'lookup'])", new)
|
|
old = (OUT / "private_tcm_r25_acceptance_driver.py").read_text(encoding="utf-8")
|
|
new = (OUT / "private_tcm_r26_acceptance_driver.py").read_text(encoding="utf-8")
|
|
start = ' if result.get("preparation_isolation_verified") is True and save_body:'
|
|
self.assertEqual(old[old.index(start):old.index(" return result", old.index(start))], new[new.index(start):new.index(" return result", new.index(start))])
|
|
|
|
def test_mapper_is_exact_revision_and_pin_transformation(self):
|
|
old = (OUT / "clinical_ttft_r25_review.py").read_text(encoding="utf-8")
|
|
new = (OUT / "clinical_ttft_r26_review.py").read_text(encoding="utf-8")
|
|
new = new.replace("9c264a7f2871e62998b1328d2e892e35d8038a976120c591705ca887e83dcc10", "263a1a10fa0285dabffe26bb05ebc9683919e7f6b50552949a52d7ed2205e849")
|
|
new = new.replace("80e8eb0b4feac04d4c5d6837e89e9a4c67942a257986bbb3c5e19bd87678773f", "8858793fe6f189782749345867613a8aa0d9a8f622089173448034874b87a0de")
|
|
new = new.replace("r26", "r25").replace("R26", "R25")
|
|
self.assertEqual(ast.dump(ast.parse(old)), ast.dump(ast.parse(new)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|