File size: 4,306 Bytes
bc09816 c57b128 bc09816 c57b128 bc09816 | 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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | """Normal-baseline synthetic split tests (D-ANOM-02, Pattern 8b)."""
from __future__ import annotations
from pathlib import Path
import pyarrow.parquet as pq
import pytest
from numpy.random import PCG64, Generator, SeedSequence
from model.features import CLASSES
from model.normal_split import (
_COLUMNS,
N_NORMAL,
_flatten_frame,
generate_normal_split,
)
from model.synth.state_machines.normal_baseline import (
BASELINE_LABEL,
FRAMES_PER_WINDOW,
_normal_baseline_window,
)
def test_n_normal_is_one_thousand() -> None:
"""D-ANOM-02 sample size: ~1000 windows."""
assert N_NORMAL == 1000
def test_baseline_label_is_not_a_class() -> None:
"""The label MUST NOT collide with any of the 10 CLASSES."""
assert BASELINE_LABEL not in CLASSES
assert BASELINE_LABEL.startswith("_") # convention: leading underscore = baseline
def test_normal_baseline_window_shape() -> None:
"""One window emits FRAMES_PER_WINDOW frames, each with the 20-field allowlist."""
rng = Generator(PCG64(SeedSequence(20260601)))
frames = _normal_baseline_window(rng)
assert len(frames) == FRAMES_PER_WINDOW
# Spot-check a frame
f = frames[0]
assert f["class"] == BASELINE_LABEL
assert f["dhcp_event_class"] == "none"
assert f["auth_event_class"] == "8021x_success"
assert f["captive_portal_detected"] is False
assert f["network_mode"] in {"enterprise", "captive", "home", "unknown"}
assert "ping_continuity" in f and isinstance(f["ping_continuity"], dict)
def test_normal_baseline_healthy_ranges() -> None:
"""Sanity: numeric values fall in healthy bands (loose bounds, no failure injection)."""
rng = Generator(PCG64(SeedSequence(20260601)))
# Generate 50 windows (~1500 frames) for stable sanity stats
all_frames: list[dict] = []
for _ in range(50):
all_frames.extend(_normal_baseline_window(rng))
rssi = [f["rssi_dbm"] for f in all_frames]
ploss = [f["ping_continuity"]["packet_loss_pct"] for f in all_frames]
dns = [f["dns_resolution_ms"] for f in all_frames]
retries = [f["per_packet_retry_count"] for f in all_frames]
# Loose bounds — these are sanity ceilings, not distribution tests
assert min(rssi) > -75.0, f"healthy RSSI floor; saw min {min(rssi)}"
assert max(ploss) < 0.5, f"healthy packet loss; saw max {max(ploss)}"
assert max(dns) < 100.0, f"healthy DNS resolution; saw max {max(dns)}"
assert max(retries) < 10, f"healthy retry count; saw max {max(retries)}"
@pytest.mark.skipif(
not Path("data/normal.parquet").exists(),
reason="data/normal.parquet not generated (run `make synth-normal` first)",
)
def test_normal_split_shape() -> None:
"""data/normal.parquet has ~30,000 rows in 24-column Phase 1 schema."""
tbl = pq.read_table("data/normal.parquet")
assert tbl.num_rows == N_NORMAL * FRAMES_PER_WINDOW
assert tuple(tbl.column_names) == _COLUMNS
@pytest.mark.skipif(
not Path("data/normal.parquet").exists(),
reason="data/normal.parquet not generated (run `make synth-normal` first)",
)
def test_normal_split_class_column_is_baseline_marker() -> None:
"""Every row's `class` is BASELINE_LABEL — NOT one of the 10 CLASSES."""
tbl = pq.read_table("data/normal.parquet", columns=["class"])
unique_classes = set(tbl["class"].to_pylist())
assert unique_classes == {BASELINE_LABEL}
# And explicitly: none of the 10 trained slugs leaked in
assert unique_classes.isdisjoint(set(CLASSES))
def test_generate_normal_split_byte_identical(tmp_path: Path) -> None:
"""Same seed → same SHA-256 (mirrors Phase 1 byte-identicality contract)."""
import hashlib
a = tmp_path / "a.parquet"
b = tmp_path / "b.parquet"
generate_normal_split(seed=20260601, out_path=a)
generate_normal_split(seed=20260601, out_path=b)
h_a = hashlib.sha256(a.read_bytes()).hexdigest()
h_b = hashlib.sha256(b.read_bytes()).hexdigest()
assert h_a == h_b, f"byte-identicality broken: {h_a} != {h_b}"
def test_flatten_frame_preserves_columns() -> None:
"""_flatten_frame produces exactly the 24 _COLUMNS fields."""
rng = Generator(PCG64(SeedSequence(0)))
frame = _normal_baseline_window(rng)[0]
flat = _flatten_frame(frame)
assert set(flat.keys()) == set(_COLUMNS)
|