张宗平
8aeddebce9
- tests/test_instruct_check.py: cover load_by_category (bucketing/cap/seed
determinism/short-pool/blank-line) and _forecast_list (nested/flat/
dict/garbage). 8 tests pass.
- train_stage{1,2}.sh: export PYTHONPATH=src + PYTHONUNBUFFERED=1 so future
background runs stream logs live (current stage2 run is block-buffered).
92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
"""Unit tests for eval.instruct_check (T4.2) — CPU helpers only.
|
|
|
|
The model-driven ``evaluate`` path is exercised live in M4 exit validation.
|
|
"""
|
|
import json
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
from tsmm.eval.instruct_check import (
|
|
CATEGORIES,
|
|
_forecast_list,
|
|
load_by_category,
|
|
)
|
|
|
|
|
|
def _write_eval(tmpdir, rows):
|
|
path = os.path.join(tmpdir, "eval.jsonl")
|
|
with open(path, "w") as f:
|
|
for r in rows:
|
|
f.write(json.dumps(r) + "\n")
|
|
return path
|
|
|
|
|
|
def test_load_by_category_buckets_and_caps(tmp_path):
|
|
rows = []
|
|
for cat in CATEGORIES:
|
|
for i in range(5):
|
|
rows.append({"category": cat, "series": [[1.0]], "answer": "x", "labels": [0]})
|
|
# plus an unknown-category row that must be ignored
|
|
rows.append({"category": "mystery", "series": [[1.0]], "answer": "x"})
|
|
path = _write_eval(str(tmp_path), rows)
|
|
out = load_by_category(path, per_cat=3, seed=0)
|
|
assert set(out.keys()) == set(CATEGORIES)
|
|
for cat in CATEGORIES:
|
|
assert len(out[cat]) == 3
|
|
assert "mystery" not in out
|
|
|
|
|
|
def test_load_by_category_deterministic_with_seed(tmp_path):
|
|
rows = [{"category": "anomaly", "series": [[float(i)]], "answer": str(i), "labels": [0]}
|
|
for i in range(20)]
|
|
path = _write_eval(str(tmp_path), rows)
|
|
a = load_by_category(path, per_cat=5, seed=42)
|
|
b = load_by_category(path, per_cat=5, seed=42)
|
|
assert [s["answer"] for s in a["anomaly"]] == [s["answer"] for s in b["anomaly"]]
|
|
# different seed → (very likely) different selection
|
|
c = load_by_category(path, per_cat=5, seed=7)
|
|
assert [s["answer"] for s in a["anomaly"]] != [s["answer"] for s in c["anomaly"]]
|
|
|
|
|
|
def test_load_by_category_handles_short_pool(tmp_path):
|
|
rows = [{"category": "event", "series": [[1.0]], "answer": "x"} for _ in range(2)]
|
|
path = _write_eval(str(tmp_path), rows)
|
|
out = load_by_category(path, per_cat=10, seed=0)
|
|
assert len(out["event"]) == 2
|
|
# missing categories are empty lists, not errors
|
|
assert out["anomaly"] == []
|
|
|
|
|
|
def test_load_by_category_skips_blank_lines(tmp_path):
|
|
path = os.path.join(str(tmp_path), "eval.jsonl")
|
|
with open(path, "w") as f:
|
|
f.write(json.dumps({"category": "anomaly", "series": [[1.0]], "answer": "x"}) + "\n")
|
|
f.write("\n")
|
|
f.write(" \n")
|
|
f.write(json.dumps({"category": "anomaly", "series": [[2.0]], "answer": "y"}) + "\n")
|
|
out = load_by_category(path, per_cat=5, seed=0)
|
|
assert len(out["anomaly"]) == 2
|
|
|
|
|
|
def test_forecast_list_from_nested_json():
|
|
ref = json.dumps({"forecast": [[1.0, 2.0, 3.0]]})
|
|
assert _forecast_list(ref) == [1.0, 2.0, 3.0]
|
|
|
|
|
|
def test_forecast_list_from_flat_json():
|
|
ref = json.dumps({"forecast": [4.0, 5.0]})
|
|
assert _forecast_list(ref) == [4.0, 5.0]
|
|
|
|
|
|
def test_forecast_list_accepts_dict_input():
|
|
assert _forecast_list({"forecast": [[1.0, 2.0]]}) == [1.0, 2.0]
|
|
|
|
|
|
def test_forecast_list_returns_empty_on_garbage():
|
|
assert _forecast_list("not json") == []
|
|
assert _forecast_list(json.dumps({"no_forecast": 1})) == []
|
|
assert _forecast_list("") == []
|
|
assert _forecast_list(None) == []
|