File size: 2,743 Bytes
201dc77 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | """U11-11: `qwen3_parity.run` must not reuse saved tensors that another ckpt/code produced."""
import importlib.util
import json
import os
import pytest
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def _load():
spec = importlib.util.spec_from_file_location(
"parity_under_test", os.path.join(REPO, "scripts", "qwen3_parity.py")
)
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
return m
@pytest.fixture
def env(tmp_path, monkeypatch):
par = _load()
ckpt = tmp_path / "ckpt"
ckpt.mkdir()
(ckpt / "model-00001.safetensors").write_bytes(b"w" * 10)
(ckpt / "config.json").write_text("{}")
out = tmp_path / "out"
calls = []
def fake_run(cmd, timeout):
if "tokenize" in cmd:
d = cmd[cmd.index("tokenize") + 3]
os.makedirs(d, exist_ok=True)
json.dump({"input_ids": [[1, 2]]}, open(os.path.join(d, "ids.json"), "w"))
elif "worker-align" in cmd:
calls.append("align")
d = cmd[cmd.index("--out") + 1]
os.makedirs(d, exist_ok=True)
json.dump({}, open(os.path.join(d, "meta.json"), "w"))
else:
calls.append("hf")
d = cmd[cmd.index("hf") + 3]
os.makedirs(d, exist_ok=True)
json.dump({}, open(os.path.join(d, "meta.json"), "w"))
return ""
monkeypatch.setattr(par, "_run", fake_run)
monkeypatch.setattr(par, "compare_dirs", lambda *a, **k: {"verdict": {}})
return par, str(ckpt), str(out), calls
def test_identical_inputs_reuse_cached_tensors(env):
par, ckpt, out, calls = env
par.run(ckpt, out, "fp32")
assert calls == ["align", "hf"]
par.run(ckpt, out, "fp32")
assert calls == ["align", "hf"] # lock-race retry still skips finished workers
def test_changed_checkpoint_forces_rerun(env):
par, ckpt, out, calls = env
par.run(ckpt, out, "fp32")
with open(os.path.join(ckpt, "model-00001.safetensors"), "wb") as f:
f.write(b"other weights!")
par.run(ckpt, out, "fp32")
assert calls == ["align", "hf", "align", "hf"]
def test_changed_align_code_reruns_only_the_align_side(env, monkeypatch):
par, ckpt, out, calls = env
par.run(ckpt, out, "fp32")
monkeypatch.setattr(par, "_align_code_sha", lambda: "edited-qwen3.py")
par.run(ckpt, out, "fp32")
assert calls == ["align", "hf", "align"]
def test_unstamped_legacy_output_is_not_trusted(env):
par, ckpt, out, calls = env
for side in ("align_fp32", "hf_fp32"):
os.makedirs(os.path.join(out, side))
json.dump({}, open(os.path.join(out, side, "meta.json"), "w"))
par.run(ckpt, out, "fp32")
assert calls == ["align", "hf"]
|