mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-09-21 08:50:42 +02:00
266 lines
9.1 KiB
Python
266 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from obliteratus.checkpoint_evaluation import (
|
|
_partition_pairs,
|
|
_verify_checkpoint_inventory,
|
|
)
|
|
|
|
|
|
def test_protocol_evaluation_partitions_are_immutable_and_disjoint():
|
|
tune = _partition_pairs("optimizer_tune")
|
|
final = _partition_pairs("final_test")
|
|
|
|
assert len(tune) == 142
|
|
assert len(final) == 200
|
|
assert set(tune).isdisjoint(final)
|
|
with pytest.raises(ValueError, match="unsupported"):
|
|
_partition_pairs("training")
|
|
|
|
|
|
def test_checkpoint_inventory_verifies_size_hash_and_managed_path(tmp_path):
|
|
run_dir = tmp_path / ("run-" + "a" * 32)
|
|
checkpoint = run_dir / "checkpoint"
|
|
checkpoint.mkdir(parents=True)
|
|
weights = checkpoint / "weights.bin"
|
|
weights.write_bytes(b"verified weights")
|
|
inventory = run_dir / "artifact-inventory.json"
|
|
inventory.write_text(
|
|
json.dumps(
|
|
{
|
|
"artifacts": [
|
|
{
|
|
"path": "checkpoint/weights.bin",
|
|
"bytes": weights.stat().st_size,
|
|
"sha256": hashlib.sha256(weights.read_bytes()).hexdigest(),
|
|
}
|
|
]
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
manifest = {"result": {"checkpoint": str(checkpoint), "inventory": str(inventory)}}
|
|
|
|
assert _verify_checkpoint_inventory(run_dir, manifest) == checkpoint.resolve()
|
|
weights.write_bytes(b"tampered weights")
|
|
with pytest.raises(ValueError, match="hash changed"):
|
|
_verify_checkpoint_inventory(run_dir, manifest)
|
|
|
|
|
|
def _inventory_candidate(tmp_path):
|
|
run_dir = tmp_path / ("run-" + "b" * 32)
|
|
checkpoint = run_dir / "checkpoint"
|
|
checkpoint.mkdir(parents=True)
|
|
weights = checkpoint / "weights.bin"
|
|
weights.write_bytes(b"verified weights")
|
|
inventory = run_dir / "artifact-inventory.json"
|
|
artifact = {
|
|
"path": "checkpoint/weights.bin",
|
|
"bytes": weights.stat().st_size,
|
|
"sha256": hashlib.sha256(weights.read_bytes()).hexdigest(),
|
|
}
|
|
inventory.write_text(json.dumps({"artifacts": [artifact]}), encoding="utf-8")
|
|
manifest = {"result": {"checkpoint": str(checkpoint), "inventory": str(inventory)}}
|
|
return run_dir, checkpoint, inventory, artifact, manifest
|
|
|
|
|
|
def test_checkpoint_inventory_rejects_unmanaged_empty_and_noncheckpoint_records(tmp_path):
|
|
run_dir, checkpoint, inventory, artifact, manifest = _inventory_candidate(tmp_path)
|
|
|
|
manifest["result"]["checkpoint"] = str(tmp_path / "outside")
|
|
with pytest.raises(ValueError, match="outside its managed"):
|
|
_verify_checkpoint_inventory(run_dir, manifest)
|
|
|
|
manifest["result"]["checkpoint"] = str(checkpoint)
|
|
for value, message in (([], "empty"), ([{"path": "notes.md"}], "no checkpoint")):
|
|
inventory.write_text(json.dumps({"artifacts": value}), encoding="utf-8")
|
|
with pytest.raises(ValueError, match=message):
|
|
_verify_checkpoint_inventory(run_dir, manifest)
|
|
|
|
artifact["path"] = "checkpoint/../notes.md"
|
|
inventory.write_text(json.dumps({"artifacts": [artifact]}), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="invalid checkpoint artifact"):
|
|
_verify_checkpoint_inventory(run_dir, manifest)
|
|
|
|
|
|
def test_checkpoint_inventory_rejects_size_change(tmp_path):
|
|
run_dir, _checkpoint, inventory, artifact, manifest = _inventory_candidate(tmp_path)
|
|
artifact["bytes"] += 1
|
|
inventory.write_text(json.dumps({"artifacts": [artifact]}), encoding="utf-8")
|
|
|
|
with pytest.raises(ValueError, match="size changed"):
|
|
_verify_checkpoint_inventory(run_dir, manifest)
|
|
|
|
|
|
class _FakeLifecycle:
|
|
def __init__(self):
|
|
self.events = []
|
|
|
|
def loading(self, checkpoint):
|
|
self.events.append(("loading", checkpoint))
|
|
|
|
def resize(self, memory):
|
|
self.events.append(("resize", memory))
|
|
|
|
def ready(self, memory):
|
|
self.events.append(("ready", memory))
|
|
|
|
def release(self, *, reason):
|
|
self.events.append(("release", reason))
|
|
|
|
|
|
class _FakeArchive:
|
|
def __init__(self, run_dir: Path, manifest):
|
|
self.run_dir = run_dir
|
|
self.manifest = manifest
|
|
self.finishes = []
|
|
|
|
def begin_evaluation(self, run_id, *, partition, evaluator):
|
|
assert run_id == self.manifest["run_id"]
|
|
self.reservation = (partition, evaluator)
|
|
return {"evaluation_id": "eval-" + "c" * 32}
|
|
|
|
def result(self, run_id):
|
|
assert run_id == self.manifest["run_id"]
|
|
return self.manifest
|
|
|
|
def _run_dir(self, run_id):
|
|
assert run_id == self.manifest["run_id"]
|
|
return self.run_dir
|
|
|
|
def finish_evaluation(self, run_id, evaluation_id, **kwargs):
|
|
self.finishes.append((run_id, evaluation_id, kwargs))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("refusal_rate", "coherence", "expected"),
|
|
[(0.2, 0.9, 0), (0.3, 0.9, 2), (0.2, 0.79, 2)],
|
|
)
|
|
def test_evaluate_reloads_verifies_and_records_objective(
|
|
tmp_path,
|
|
monkeypatch,
|
|
refusal_rate,
|
|
coherence,
|
|
expected,
|
|
):
|
|
import obliteratus.checkpoint_evaluation as checkpoint_evaluation
|
|
|
|
run_dir, checkpoint, _inventory, _artifact, manifest = _inventory_candidate(tmp_path)
|
|
run_id = run_dir.name
|
|
manifest["run_id"] = run_id
|
|
manifest["result"]["metrics"] = {
|
|
"baseline_perplexity": 3.5,
|
|
"baseline_coherence": 0.95,
|
|
}
|
|
archive = _FakeArchive(run_dir, manifest)
|
|
lifecycle = _FakeLifecycle()
|
|
pipelines = []
|
|
|
|
class FakePipeline:
|
|
def __init__(self, **kwargs):
|
|
self.kwargs = kwargs
|
|
self._quality_metrics = {
|
|
"refusal_rate": refusal_rate,
|
|
"coherence": coherence,
|
|
}
|
|
self.cleaned = False
|
|
pipelines.append(self)
|
|
|
|
def _summon(self):
|
|
self.kwargs["on_log"]("loaded")
|
|
|
|
def _verify(self):
|
|
self.kwargs["on_log"]("verified")
|
|
|
|
def cleanup_failed_run(self):
|
|
self.cleaned = True
|
|
|
|
memory = SimpleNamespace(reserved_bytes=1)
|
|
monkeypatch.setattr(checkpoint_evaluation, "RunArchive", lambda _root: archive)
|
|
monkeypatch.setattr(checkpoint_evaluation, "from_environment", lambda: lifecycle)
|
|
monkeypatch.setattr(checkpoint_evaluation, "AbliterationPipeline", FakePipeline)
|
|
monkeypatch.setattr(checkpoint_evaluation, "_partition_pairs", lambda _part: (("h", "s"),))
|
|
monkeypatch.setattr(checkpoint_evaluation, "measure_torch_memory", lambda _torch: memory)
|
|
|
|
assert checkpoint_evaluation.evaluate(run_id, "optimizer_tune", str(tmp_path)) == expected
|
|
assert archive.reservation == (
|
|
"optimizer_tune",
|
|
checkpoint_evaluation.EVALUATOR_VERSION,
|
|
)
|
|
assert pipelines[0].kwargs["model_name"] == str(checkpoint.resolve())
|
|
assert pipelines[0]._stock_baseline == {"perplexity": 3.5, "coherence": 0.95}
|
|
assert pipelines[0].cleaned is True
|
|
metrics = archive.finishes[0][2]["metrics"]
|
|
assert metrics["passes_objective"] is (expected == 0)
|
|
assert archive.finishes[0][2]["log"] == ["loaded", "verified"]
|
|
assert lifecycle.events[-1] == ("release", "evaluation_optimizer_tune_complete")
|
|
|
|
|
|
def test_evaluate_records_failure_and_releases_lifecycle(tmp_path, monkeypatch):
|
|
import obliteratus.checkpoint_evaluation as checkpoint_evaluation
|
|
|
|
run_dir, _checkpoint, _inventory, _artifact, manifest = _inventory_candidate(tmp_path)
|
|
run_id = run_dir.name
|
|
manifest["run_id"] = run_id
|
|
manifest["result"]["metrics"] = {
|
|
"baseline_perplexity": 3.5,
|
|
"baseline_coherence": 0.95,
|
|
}
|
|
archive = _FakeArchive(run_dir, manifest)
|
|
lifecycle = _FakeLifecycle()
|
|
|
|
class FailedPipeline:
|
|
_quality_metrics = {}
|
|
|
|
def __init__(self, **_kwargs):
|
|
pass
|
|
|
|
def _summon(self):
|
|
raise RuntimeError("summon failed")
|
|
|
|
def cleanup_failed_run(self):
|
|
self.cleaned = True
|
|
|
|
monkeypatch.setattr(checkpoint_evaluation, "RunArchive", lambda _root: archive)
|
|
monkeypatch.setattr(checkpoint_evaluation, "from_environment", lambda: lifecycle)
|
|
monkeypatch.setattr(checkpoint_evaluation, "AbliterationPipeline", FailedPipeline)
|
|
monkeypatch.setattr(checkpoint_evaluation, "_partition_pairs", lambda _part: (("h", "s"),))
|
|
|
|
with pytest.raises(RuntimeError, match="summon failed"):
|
|
checkpoint_evaluation.evaluate(run_id, "final_test", str(tmp_path))
|
|
|
|
assert isinstance(archive.finishes[0][2]["failure"], RuntimeError)
|
|
assert lifecycle.events[-1] == ("release", "evaluation_final_test_complete")
|
|
|
|
|
|
def test_main_delegates_parsed_evaluation_arguments(monkeypatch):
|
|
import obliteratus.checkpoint_evaluation as checkpoint_evaluation
|
|
|
|
observed = []
|
|
monkeypatch.setattr(
|
|
"sys.argv",
|
|
[
|
|
"checkpoint-evaluation",
|
|
"--archive-root",
|
|
"/archive",
|
|
"--run-id",
|
|
"run-" + "d" * 32,
|
|
"--partition",
|
|
"final_test",
|
|
],
|
|
)
|
|
monkeypatch.setattr(
|
|
checkpoint_evaluation,
|
|
"evaluate",
|
|
lambda *args: observed.append(args) or 2,
|
|
)
|
|
|
|
assert checkpoint_evaluation.main() == 2
|
|
assert observed == [("run-" + "d" * 32, "final_test", "/archive")]
|