KOKO\Mimi
269 lines
13 KiB
Python
269 lines
13 KiB
Python
"""Exercise process boundaries and preserved raw evidence, not solver numerics."""
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SPEC = importlib.util.spec_from_file_location("measure", ROOT / "scripts/performance/measure.py")
|
|
M = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(M)
|
|
|
|
|
|
@pytest.fixture
|
|
def source(tmp_path):
|
|
path = tmp_path / "model.inp"
|
|
path.write_text("from pathlib import Path\nimport time\ntime.sleep(0.08)\n"
|
|
"Path(__file__).with_name('model_results.h5').write_bytes(b'fresh')\n")
|
|
return path
|
|
|
|
|
|
def test_real_spawn_isolated_bytes_and_pending_comparison(tmp_path, source, monkeypatch):
|
|
monkeypatch.setattr(M, "validate_result", lambda *a: {"frames": 101})
|
|
before = source.read_bytes()
|
|
rows = [M.run_once(Path(sys.executable), source, "fixture", tmp_path / str(i), i)
|
|
for i in range(2)]
|
|
assert source.read_bytes() == before
|
|
for i, row in enumerate(rows):
|
|
assert row["native_exit"] == 0 and row["execution_state"] == "completed"
|
|
assert row["elapsed_seconds"] >= 0.08
|
|
assert row["comparison_status"] == "pending"
|
|
assert row["measurement_index"] == i
|
|
native = json.loads(Path(row["native_record"]).read_text())
|
|
assert len(native["command"]) == 2
|
|
assert native["original_sha256"] == native["scratch_sha256"]
|
|
assert native["scratch_unchanged"] and native["source_unchanged"]
|
|
assert rows[0]["scratch_input"] != rows[1]["scratch_input"]
|
|
|
|
|
|
def test_timer_excludes_preparation_and_validation(tmp_path, source, monkeypatch):
|
|
events = []
|
|
real_copy, real_spawn = M.shutil.copyfile, M.subprocess.Popen
|
|
monkeypatch.setattr(M.shutil, "copyfile", lambda *a: (events.append("copy"), real_copy(*a))[1])
|
|
monkeypatch.setattr(M.time, "perf_counter", lambda: (events.append("clock"), float(len(events)))[1])
|
|
monkeypatch.setattr(M.subprocess, "Popen", lambda *a, **kw: (events.append("spawn"), real_spawn(*a, **kw))[1])
|
|
monkeypatch.setattr(M, "validate_result", lambda *a: (events.append("validate"), {})[1])
|
|
M.run_once(Path(sys.executable), source, "fixture", tmp_path / "run", 0)
|
|
assert events == ["copy", "clock", "spawn", "clock", "validate"]
|
|
|
|
|
|
def test_nonzero_with_output_is_not_success(tmp_path, source, monkeypatch):
|
|
source.write_text(source.read_text() + "raise SystemExit(7)\n")
|
|
monkeypatch.setattr(M, "validate_result", lambda *a: pytest.fail("invalid native exit validated"))
|
|
row = M.run_once(Path(sys.executable), source, "fixture", tmp_path / "run", 0)
|
|
assert row["native_exit"] == 7 and not row["execution_valid"]
|
|
assert row["result_sha256"] is not None
|
|
|
|
|
|
def test_existing_directory_and_old_output_rejected(tmp_path, source):
|
|
run = tmp_path / "run"
|
|
run.mkdir()
|
|
old = run / "model_results.h5"
|
|
old.write_bytes(b"old")
|
|
with pytest.raises(FileExistsError):
|
|
M.run_once(Path(sys.executable), source, "fixture", run, 0)
|
|
assert old.read_bytes() == b"old"
|
|
|
|
|
|
def test_launch_failure_never_invents_result_hash(tmp_path, source):
|
|
row = M.run_once(tmp_path / "absent.exe", source, "fixture", tmp_path / "run", 0)
|
|
assert row["execution_state"] == "launch-failed" and row["native_exit"] is None
|
|
assert row["result_sha256"] is None and not row["execution_valid"]
|
|
|
|
|
|
def test_input_mutation_invalidates_result(tmp_path, source, monkeypatch):
|
|
source.write_text(source.read_text() + "Path(__file__).write_text('changed')\n")
|
|
monkeypatch.setattr(M, "validate_result", lambda *a: {})
|
|
row = M.run_once(Path(sys.executable), source, "fixture", tmp_path / "run", 0)
|
|
assert not row["execution_valid"]
|
|
assert not json.loads(Path(row["native_record"]).read_text())["scratch_unchanged"]
|
|
|
|
|
|
def test_interrupted_run_and_prior_case_are_preserved(tmp_path, source, monkeypatch):
|
|
real = M.subprocess.Popen
|
|
calls = 0
|
|
class Interrupted:
|
|
def __init__(self, *a, **kw):
|
|
self.proc = real(*a, **kw)
|
|
def wait(self, timeout=None):
|
|
raise KeyboardInterrupt
|
|
def kill(self):
|
|
self.proc.kill()
|
|
def communicate(self):
|
|
return self.proc.communicate()
|
|
@property
|
|
def returncode(self):
|
|
return self.proc.returncode
|
|
def spawn(*a, **kw):
|
|
nonlocal calls
|
|
calls += 1
|
|
return real(*a, **kw) if calls == 1 else Interrupted(*a, **kw)
|
|
monkeypatch.setattr(M.subprocess, "Popen", spawn)
|
|
monkeypatch.setattr(M, "validate_result", lambda *a: {})
|
|
monkeypatch.setattr(M, "collect_environment", lambda *a: {"fixed": True})
|
|
out = tmp_path / "artifacts"
|
|
assert M.measure(Path(sys.executable), out, cases=[("first", source, 1), ("second", source, 1)]) == 1
|
|
report = json.loads((out / "measurements.json").read_text())
|
|
assert len(report["cases"]) == 2
|
|
assert report["cases"][0]["runs"][0]["execution_state"] == "completed"
|
|
assert report["cases"][1]["runs"][0]["execution_state"] == "interrupted"
|
|
assert not report["cases"][1]["all_measured_runs_within_target"]
|
|
|
|
|
|
def test_timeout_preserves_actual_exit(tmp_path, source, monkeypatch):
|
|
monkeypatch.setattr(M, "validate_result", lambda *a: pytest.fail("timed out"))
|
|
row = M.run_once(Path(sys.executable), source, "fixture", tmp_path / "run", 0, timeout=0.001)
|
|
assert row["execution_state"] == "timed-out"
|
|
assert type(row["native_exit"]) is int and not row["execution_valid"]
|
|
|
|
|
|
def test_native_zero_missing_or_invalid_result_fails(tmp_path, source, monkeypatch):
|
|
source.write_text("pass\n")
|
|
row = M.run_once(Path(sys.executable), source, "fixture", tmp_path / "missing", 0)
|
|
assert row["native_exit"] == 0 and not row["execution_valid"]
|
|
assert row["result_sha256"] is None
|
|
source.write_text("from pathlib import Path\nPath(__file__).with_name('model_results.h5').write_bytes(b'bad schema')\n")
|
|
def reject(*a):
|
|
raise ValueError("invalid inventory")
|
|
monkeypatch.setattr(M, "validate_result", reject)
|
|
row = M.run_once(Path(sys.executable), source, "fixture", tmp_path / "invalid", 0)
|
|
assert row["native_exit"] == 0 and not row["execution_valid"]
|
|
assert row["error"] == "invalid inventory"
|
|
|
|
|
|
def test_slow_runs_are_retained_and_exit_zero_is_not_time_pass(tmp_path, source, monkeypatch):
|
|
monkeypatch.setattr(M, "validate_result", lambda *a: {})
|
|
monkeypatch.setattr(M, "collect_environment", lambda *a: {"fixed": True})
|
|
out = tmp_path / "artifacts"
|
|
assert M.measure(Path(sys.executable), out, cases=[("fixture", source, 0.001)]) == 0
|
|
report = json.loads((out / "measurements.json").read_text())
|
|
case = report["cases"][0]
|
|
assert report["execution_complete"] and len(case["runs"]) == 1
|
|
assert not case["all_measured_runs_within_target"]
|
|
assert case["runs"][0]["kind"] == "measurement"
|
|
assert case["runs"][0]["measurement_index"] == 1
|
|
assert all(r["comparison_status"] == "pending" for r in case["runs"])
|
|
|
|
|
|
def test_changed_binary_or_source_blocks_group(tmp_path, source, monkeypatch):
|
|
monkeypatch.setattr(M, "validate_result", lambda *a: {})
|
|
captures = iter([{"sha": "before"}, {"sha": "before"}, {"sha": "after"}])
|
|
monkeypatch.setattr(M, "collect_environment", lambda *a: next(captures))
|
|
out = tmp_path / "artifacts"
|
|
assert M.measure(Path(sys.executable), out, cases=[("fixture", source, 1)]) == 1
|
|
row = json.loads((out / "measurements.json").read_text())["cases"][0]["runs"][0]
|
|
assert row["native_exit"] == 0 and not row["execution_valid"]
|
|
|
|
|
|
def test_cli_bad_arguments_are_exit_two(tmp_path):
|
|
proc = subprocess.run([sys.executable, str(ROOT / "scripts/performance/measure.py"),
|
|
"--exe", str(tmp_path / "missing.exe"), "--output-dir", str(tmp_path)],
|
|
capture_output=True)
|
|
assert proc.returncode == 2
|
|
|
|
|
|
def test_comparison_link_preserves_raw_time_and_rejects_wrong_result(tmp_path, source, monkeypatch):
|
|
monkeypatch.setattr(M, "validate_result", lambda *a: {})
|
|
monkeypatch.setattr(M, "collect_environment", lambda *a: {"fixed": True})
|
|
out = tmp_path / "artifacts"
|
|
assert M.measure(Path(sys.executable), out, cases=[("fixture", source, 1)]) == 0
|
|
before = json.loads((out / "measurements.json").read_text())
|
|
row = before["cases"][0]["runs"][0]
|
|
comparison_path = Path(row["native_record"]).parent / "comparison.json"
|
|
comparison = {"case": "fixture", "identities": {
|
|
"input": {"sha256": row["scratch_sha256"]}, "results": {"sha256": row["result_sha256"]}},
|
|
"precheck": {"passed": True}, "blocking_passed": True, "verdict": "passed"}
|
|
M.publish(comparison_path, comparison)
|
|
assert M.link_comparisons(out) == 0
|
|
after = json.loads((out / "measurements.json").read_text())
|
|
assert after["cases"][0]["runs"][0]["comparison_status"] == "passed"
|
|
assert [r["elapsed_seconds"] for r in after["cases"][0]["runs"]] == [
|
|
r["elapsed_seconds"] for r in before["cases"][0]["runs"]]
|
|
comparison["identities"]["results"]["sha256"] = "other run"
|
|
M.publish(comparison_path, comparison)
|
|
with pytest.raises(ValueError):
|
|
M.link_comparisons(out)
|
|
assert json.loads((out / "measurements.json").read_text()) == after
|
|
|
|
|
|
def test_result_validation_interrupt_preserves_completed_native_and_prior_case(tmp_path, source, monkeypatch):
|
|
calls = 0
|
|
def validate(*a):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 2:
|
|
raise KeyboardInterrupt
|
|
return {}
|
|
monkeypatch.setattr(M, "validate_result", validate)
|
|
monkeypatch.setattr(M, "collect_environment", lambda *a: {"fixed": True})
|
|
out = tmp_path / "artifacts"
|
|
try:
|
|
code = M.measure(Path(sys.executable), out, cases=[("first", source, 1), ("second", source, 1)])
|
|
except KeyboardInterrupt:
|
|
pytest.fail("Result validation interrupt escaped without a completed native row")
|
|
assert code == 1
|
|
report = json.loads((out / "measurements.json").read_text())
|
|
assert len(report["cases"]) == 2 and not report["execution_complete"]
|
|
assert report["cases"][0]["runs"][0]["execution_valid"]
|
|
row = report["cases"][1]["runs"][0]
|
|
native = json.loads(Path(row["native_record"]).read_text())
|
|
assert row["native_exit"] == native["native_exit"] == 0
|
|
assert row["execution_state"] == native["execution_state"] == "completed"
|
|
assert row["elapsed_seconds"] == native["elapsed_seconds"]
|
|
assert row["result_sha256"] == native["result_sha256"] is not None
|
|
assert row["result_validation_state"] == "interrupted" and not row["execution_valid"]
|
|
assert row["comparison_status"] == "pending"
|
|
|
|
|
|
@pytest.mark.parametrize("exception,state", [(KeyboardInterrupt, "interrupted"), (OSError, "failed")])
|
|
def test_post_native_environment_failure_preserves_current_and_prior_case(tmp_path, source, monkeypatch, exception, state):
|
|
monkeypatch.setattr(M, "validate_result", lambda *a: {})
|
|
out = tmp_path / "artifacts"
|
|
def collect(*a):
|
|
if (out / "second/run-1/native.json").is_file():
|
|
raise exception("post-native environment unavailable")
|
|
return {"fixed": True}
|
|
monkeypatch.setattr(M, "collect_environment", collect)
|
|
try:
|
|
code = M.measure(Path(sys.executable), out, cases=[("first", source, 1), ("second", source, 1)])
|
|
except (KeyboardInterrupt, OSError):
|
|
pytest.fail("Post-native environment failure escaped without preserving the current row")
|
|
assert code == 1
|
|
report = json.loads((out / "measurements.json").read_text())
|
|
assert report["cases"][0]["runs"][0]["execution_valid"]
|
|
row = report["cases"][1]["runs"][0]
|
|
native = json.loads(Path(row["native_record"]).read_text())
|
|
assert row["native_exit"] == native["native_exit"] == 0
|
|
assert row["execution_state"] == native["execution_state"] == "completed"
|
|
assert row["elapsed_seconds"] == native["elapsed_seconds"]
|
|
assert row["result_sha256"] == native["result_sha256"] is not None
|
|
assert row["environment_validation_state"] == state
|
|
assert not row["execution_valid"] and row["error"]
|
|
assert not row["within_target"] and not report["execution_complete"]
|
|
assert json.loads((out / "second/run-1/run.json").read_text()) == row
|
|
|
|
|
|
@pytest.mark.parametrize("case,value,code", [
|
|
("hexa_nonlinearstatic", float("nan"), "nonfinite-result-value"),
|
|
("hexa_nonlinearstatic", float("inf"), "nonfinite-result-value"),
|
|
("hexa_dynnonlinear", 1e-12, "nonzero-initial-result-state"),
|
|
])
|
|
def test_hex8_precheck_rejects_invalid_physical_values(tmp_path, case, value, code):
|
|
import h5py
|
|
helper_spec = importlib.util.spec_from_file_location("hex8_fixture", ROOT / "tests/test_compare_hex8.py")
|
|
helper = importlib.util.module_from_spec(helper_spec)
|
|
helper_spec.loader.exec_module(helper)
|
|
scratch = tmp_path / f"{case}.inp"
|
|
M.shutil.copyfile(ROOT / "reference/hexa" / scratch.name, scratch)
|
|
result = tmp_path / f"{case}_results.h5"
|
|
model = helper.COMPARE.input_inventory(scratch, case)
|
|
helper.make_result(result, model)
|
|
with h5py.File(result, "r+") as file:
|
|
file["steps/Step-1/frames/0/nodal/displacement"][0, 0] = value
|
|
with pytest.raises(Exception) as error:
|
|
M.validate_result(case, scratch, result)
|
|
assert error.value.code == code
|