| """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("_") |
|
|
|
|
| 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 |
| |
| 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))) |
| |
| 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] |
| |
| 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} |
| |
| 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) |
|
|
