63 lines
2.6 KiB
Python
63 lines
2.6 KiB
Python
"""Model-free checks for the matched-reference input contract."""
|
|
import copy
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location("oracle", Path(__file__).with_name("mtplx-execution-reference.py"))
|
|
oracle = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(oracle)
|
|
|
|
|
|
class SourceContract(unittest.TestCase):
|
|
def test_progress_requires_new_tokens_and_limits_output_only(self):
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stderr(output), patch.object(oracle.time, "monotonic", side_effect=[0, .4, 1]):
|
|
report = oracle.token_report("chat_tokens", "progress", step=0)
|
|
report([])
|
|
report([10])
|
|
report([11, 12])
|
|
report([13])
|
|
events = [json.loads(line) for line in output.getvalue().splitlines()]
|
|
self.assertEqual([e["tokens"] for e in events], [1, 4])
|
|
self.assertTrue(all(e["event"] == "chat_decode_progress" and "ids" not in e for e in events))
|
|
|
|
def test_requires_matching_complete_warm_chat(self):
|
|
start = dict(event="start", model="qwen3.8-flash-next", plain_chat=True,
|
|
system_prompt="", canary=True, prompts=["summary", "story", "python"],
|
|
settings=dict(reasoning="low", power_percent=100, prefill_chunk=2048,
|
|
min_p=0, quality=False, seed=42, acceleration=dict(kind="mtp", enabled=True)),
|
|
warmup=dict(enabled=True))
|
|
records = [start, dict(event="warmup_result", ok=True)] + [
|
|
dict(event="result", turn=n, ok=True, finish_reason="stop") for n in (1, 2, 3)]
|
|
self.assertEqual(oracle.source_chat(records, True, True)[0], start)
|
|
for records_bad, mtp, canary in [
|
|
(records[:-1], True, True),
|
|
(records + [start], True, True),
|
|
(records, False, True),
|
|
(records, True, False),
|
|
]:
|
|
with self.assertRaises(ValueError):
|
|
oracle.source_chat(records_bad, mtp, canary)
|
|
for location, name, value in [
|
|
("settings", "power_percent", 75),
|
|
("settings", "seed", None),
|
|
("warmup", "enabled", False),
|
|
]:
|
|
bad = copy.deepcopy(records)
|
|
bad[0][location][name] = value
|
|
with self.assertRaises(ValueError):
|
|
oracle.source_chat(bad, True, True)
|
|
bad = copy.deepcopy(records)
|
|
bad[-1]["finish_reason"] = "length"
|
|
with self.assertRaises(ValueError):
|
|
oracle.source_chat(bad, True, True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|