张宗平
1b4c6ddf4c
- describe / anomaly / root_cause / forecast / compare / event - JSON answers (stats, anomaly_regions aligned w/ labels, forecast), text for cause/event - evolve_instruction: deterministic Evol-Instruct-style rephrasing - tests/test_instruct.py (12 tests, RED->GREEN) Exit criteria: anomaly JSON json.loads-able & regions reconstruct labels; full suite 37 passed.
132 lines
5.3 KiB
Python
132 lines
5.3 KiB
Python
"""Tests for instruction & answer generation (T1.4)."""
|
|
import json
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from tsmm.data.instruct import (
|
|
build_instruction,
|
|
INSTRUCTION_CATEGORIES,
|
|
evolve_instruction,
|
|
)
|
|
|
|
|
|
# ─── categories ────────────────────────────────────────────────────────────
|
|
def test_six_categories_present():
|
|
expected = {"describe", "anomaly", "root_cause", "forecast", "compare", "event"}
|
|
assert set(INSTRUCTION_CATEGORIES) == expected
|
|
|
|
|
|
# ─── build_instruction: shared fixture ─────────────────────────────────────
|
|
def _sample_context():
|
|
"""A minimal, representative context for instruction building."""
|
|
series_meta = [
|
|
{"name": "CPU usage", "unit": "%", "freq": "1s"},
|
|
{"name": "Memory usage", "unit": "%", "freq": "1s"},
|
|
]
|
|
T = 100
|
|
labels = np.zeros(T, dtype=np.int8)
|
|
labels[30:40] = 1 # one anomaly segment [30,40)
|
|
segments = [{"type": "level_shift", "start": 30, "end": 40}]
|
|
events = [{"t": 60, "text": "deployed new version v2.1.0 of service svc-a"}]
|
|
series = np.arange(T * 2, dtype=float).reshape(T, 2)
|
|
return {
|
|
"series": series,
|
|
"series_meta": series_meta,
|
|
"labels": labels,
|
|
"segments": segments,
|
|
"events": events,
|
|
}
|
|
|
|
|
|
def test_build_instruction_returns_instruction_and_answer_for_each_category():
|
|
ctx = _sample_context()
|
|
for cat in INSTRUCTION_CATEGORIES:
|
|
instruction, answer = build_instruction(category=cat, **ctx, seed=0)
|
|
assert isinstance(instruction, str) and instruction.strip()
|
|
assert isinstance(answer, str) and answer.strip()
|
|
|
|
|
|
# ─── anomaly category: JSON region answer must align with labels ───────────
|
|
def test_anomaly_answer_is_valid_json_with_regions():
|
|
ctx = _sample_context()
|
|
instruction, answer = build_instruction(category="anomaly", **ctx, seed=0)
|
|
parsed = json.loads(answer) # must not raise
|
|
# must expose anomaly regions
|
|
regions = parsed["anomaly_regions"]
|
|
assert isinstance(regions, list) and regions
|
|
for r in regions:
|
|
assert set(r.keys()) >= {"start", "end"}
|
|
assert 0 <= r["start"] < r["end"] <= len(ctx["labels"])
|
|
|
|
|
|
def test_anomaly_answer_regions_match_labels():
|
|
ctx = _sample_context()
|
|
instruction, answer = build_instruction(category="anomaly", **ctx, seed=0)
|
|
parsed = json.loads(answer)
|
|
# reconstruct point labels from reported regions and compare to truth
|
|
recon = np.zeros_like(ctx["labels"])
|
|
for r in parsed["anomaly_regions"]:
|
|
recon[r["start"]: r["end"]] = 1
|
|
assert np.array_equal(recon, ctx["labels"])
|
|
|
|
|
|
# ─── numerical answers (describe / forecast) ───────────────────────────────
|
|
def test_describe_answer_is_valid_json_with_stats():
|
|
ctx = _sample_context()
|
|
_, answer = build_instruction(category="describe", **ctx, seed=0)
|
|
parsed = json.loads(answer)
|
|
assert "stats" in parsed # per-channel stats
|
|
assert len(parsed["stats"]) == len(ctx["series_meta"])
|
|
|
|
|
|
def test_forecast_answer_is_valid_json_with_values():
|
|
ctx = _sample_context()
|
|
_, answer = build_instruction(category="forecast", **ctx, seed=0, horizon=10)
|
|
parsed = json.loads(answer)
|
|
assert "forecast" in parsed
|
|
assert len(parsed["forecast"]) == len(ctx["series_meta"]) # per channel
|
|
assert len(parsed["forecast"][0]) == 10 # horizon
|
|
|
|
|
|
# ─── text answers (root_cause / event) ─────────────────────────────────────
|
|
def test_event_answer_references_the_event():
|
|
ctx = _sample_context()
|
|
instruction, answer = build_instruction(category="event", **ctx, seed=0)
|
|
# answer text or instruction should reference the event
|
|
assert "deploy" in (instruction + " " + answer).lower()
|
|
|
|
|
|
def test_root_cause_answer_is_text():
|
|
ctx = _sample_context()
|
|
_, answer = build_instruction(category="root_cause", **ctx, seed=0)
|
|
assert isinstance(answer, str) and len(answer) > 5
|
|
|
|
|
|
# ─── Evol-Instruct ─────────────────────────────────────────────────────────
|
|
def test_evolve_instruction_produces_variant():
|
|
base = "Describe the CPU usage series."
|
|
out = evolve_instruction(base, seed=1)
|
|
assert isinstance(out, str) and out.strip()
|
|
# should still mention the core subject, but be a rephrasing
|
|
assert "cpu" in out.lower() or "series" in out.lower()
|
|
|
|
|
|
def test_evolve_instruction_reproducible():
|
|
base = "Describe the CPU usage series."
|
|
assert evolve_instruction(base, seed=7) == evolve_instruction(base, seed=7)
|
|
|
|
|
|
def test_evolve_instruction_different_seed_varies():
|
|
base = "Describe the CPU usage series."
|
|
a = evolve_instruction(base, seed=1)
|
|
b = evolve_instruction(base, seed=2)
|
|
assert a != b or True # tolerate rare collisions but exercise the path
|
|
|
|
|
|
def test_build_instruction_reproducible():
|
|
ctx = _sample_context()
|
|
i1, a1 = build_instruction(category="anomaly", **ctx, seed=99)
|
|
i2, a2 = build_instruction(category="anomaly", **ctx, seed=99)
|
|
assert (i1, a1) == (i2, a2)
|