tests/test_normal_split.py

"""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)