qtalker---/backend/tests/test_device_anomaly.py

import uuid

import pytest

import app.ws as ws_module
from app.device_anomaly import (
    BOOLEAN_ANOMALY_MAGNITUDE,
    detect_boolean_transition,
    detect_sensor_spike,
    frequency_for_sensor_type,
    is_boolean_sensor,
    process_device_reading_for_summon,
    reset_state,
)
from app.models.contact_session import ContactSession
from app.models.user import User
from app.ws import SeanceState, get_active_session, register_active_session

from .conftest import TestSessionLocal


@pytest.fixture(autouse=True)
def _clear_device_anomaly_state():
    reset_state()
    yield
    reset_state()


# ---------------------------------------------------------------------------
# detect_sensor_spike — same three-guard shape as detect_wire_spike
# ---------------------------------------------------------------------------


def test_numeric_spike_needs_history():
    assert detect_sensor_spike([], 99.0, "temperature") is None
    assert detect_sensor_spike([21.0, 21.1, 21.0], 30.0, "temperature") is None


def test_numeric_spike_fires_on_real_surge():
    history = [21.0, 21.1, 20.9, 21.2, 21.0, 21.1, 20.8]
    deviation = detect_sensor_spike(history, 30.0, "temperature")
    assert deviation is not None
    assert deviation > 8.0


def test_numeric_spike_ignores_normal_fluctuation():
    history = [21.0, 21.1, 20.9, 21.2, 21.0, 21.1, 20.8]
    assert detect_sensor_spike(history, 21.3, "temperature") is None


def test_numeric_spike_has_absolute_floor_for_quiet_sensors():
    # A rock-steady baseline with a tiny wobble must never cry ghost, even
    # if that wobble technically clears a 3-sigma bar against near-zero std.
    history = [20.00, 20.01, 19.99, 20.00, 20.02, 19.98, 20.01]
    assert detect_sensor_spike(history, 20.30, "temperature") is None


def test_numeric_spike_adapts_to_loud_baseline():
    # Once the sensor is already swinging widely, the same absolute jump is
    # no longer anomalous relative to its own noisy baseline.
    history = [15.0, 22.0, 14.0, 23.0, 16.0, 21.0, 15.5]
    assert detect_sensor_spike(history, 24.0, "temperature") is None


def test_numeric_spike_uses_known_floor_per_sensor_type():
    # Humidity's floor (3.0) is looser than temperature's (0.8) — a
    # deviation that would fire for temperature must not fire for humidity.
    history = [45.0, 46.0, 44.5, 45.5, 45.0, 44.8, 45.2]
    assert detect_sensor_spike(history, 47.6, "humidity") is None
    assert detect_sensor_spike(history, 47.6, "temperature") is not None


def test_numeric_spike_unknown_sensor_type_uses_relative_fallback_floor():
    # No hand-picked floor for "voltage" — falls back to a fraction of the
    # rolling mean, but a genuine multi-fold surge still fires.
    history = [5.0, 5.1, 4.9, 5.0, 5.05, 4.95, 5.02]
    assert detect_sensor_spike(history, 5.1, "voltage") is None
    assert detect_sensor_spike(history, 12.0, "voltage") is not None


# ---------------------------------------------------------------------------
# detect_boolean_transition
# ---------------------------------------------------------------------------


def test_boolean_transition_false_to_true_is_anomalous():
    assert detect_boolean_transition(False, True) is True


def test_boolean_transition_no_prior_state_is_not_anomalous():
    assert detect_boolean_transition(None, True) is False
    assert detect_boolean_transition(None, False) is False


def test_boolean_transition_true_to_true_is_not_anomalous():
    assert detect_boolean_transition(True, True) is False


def test_boolean_transition_true_to_false_is_not_anomalous():
    assert detect_boolean_transition(True, False) is False


# ---------------------------------------------------------------------------
# is_boolean_sensor classification rule
# ---------------------------------------------------------------------------


def test_is_boolean_sensor_by_unit():
    assert is_boolean_sensor("bool", 1) is True
    assert is_boolean_sensor("BOOL", 0) is True


def test_is_boolean_sensor_by_python_bool_value():
    assert is_boolean_sensor("pct", True) is True


def test_is_boolean_sensor_numeric_zero_one_with_non_bool_unit_stays_numeric():
    # A duty-cycle percentage reading exactly 0 or 1 must not be
    # misclassified as boolean just because its value looks bool-like.
    assert is_boolean_sensor("pct", 1) is False
    assert is_boolean_sensor("c", 0) is False


# ---------------------------------------------------------------------------
# frequency_for_sensor_type
# ---------------------------------------------------------------------------


def test_frequency_stable_for_known_sensor_types():
    assert frequency_for_sensor_type("temperature") == frequency_for_sensor_type("temperature")
    assert frequency_for_sensor_type("temperature") != frequency_for_sensor_type("humidity")


def test_frequency_stable_and_deterministic_for_unknown_sensor_type():
    freq1 = frequency_for_sensor_type("cosmic_ray_flux")
    freq2 = frequency_for_sensor_type("cosmic_ray_flux")
    assert freq1 == freq2
    assert freq1 != frequency_for_sensor_type("other_unknown_sensor")


# ---------------------------------------------------------------------------
# process_device_reading_for_summon — integration
# ---------------------------------------------------------------------------


async def _make_user_and_session(db_session) -> tuple[uuid.UUID, uuid.UUID]:
    user = User(username=f"devowner-{uuid.uuid4().hex[:8]}", password_hash="x")
    db_session.add(user)
    await db_session.flush()

    session = ContactSession(user_id=user.id)
    db_session.add(session)
    await db_session.commit()
    await db_session.refresh(session)
    return user.id, session.id


@pytest.mark.asyncio
async def test_process_reading_pushes_anomaly_into_active_session(db_session, monkeypatch):
    monkeypatch.setattr(ws_module, "session_maker", TestSessionLocal)
    user_id, session_id = await _make_user_and_session(db_session)

    state = SeanceState(user_id=user_id, session_id=session_id, client_ip="127.0.0.1")
    register_active_session(user_id, state)
    device_id = uuid.uuid4()

    try:
        # Warm up the baseline with unremarkable readings — none of these
        # should produce an anomaly.
        for value in [21.0, 21.1, 20.9, 21.2, 21.0, 21.1]:
            await process_device_reading_for_summon(
                user_id, device_id, "temperature", value, "c"
            )
        assert state.anomalies == []

        # A genuine surge fires and gets pushed into state.anomalies in the
        # same shape _handle_anomaly uses for the browser-based modes.
        await process_device_reading_for_summon(
            user_id, device_id, "temperature", 30.0, "c"
        )
        assert len(state.anomalies) == 1
        anomaly = state.anomalies[0]
        assert anomaly["source"] == "temperature"
        assert anomaly["frequency"] == frequency_for_sensor_type("temperature")
        assert anomaly["magnitude"] > 0
    finally:
        from app.ws import unregister_active_session

        unregister_active_session(user_id, state)


@pytest.mark.asyncio
async def test_process_reading_boolean_presence_transition(db_session, monkeypatch):
    monkeypatch.setattr(ws_module, "session_maker", TestSessionLocal)
    user_id, session_id = await _make_user_and_session(db_session)

    state = SeanceState(user_id=user_id, session_id=session_id, client_ip="127.0.0.1")
    register_active_session(user_id, state)
    device_id = uuid.uuid4()

    try:
        # First reading (False) establishes state, no prior value to
        # transition from either way.
        await process_device_reading_for_summon(user_id, device_id, "presence", 0, "bool")
        assert state.anomalies == []

        # false -> true is the anomaly.
        await process_device_reading_for_summon(user_id, device_id, "presence", 1, "bool")
        assert len(state.anomalies) == 1
        anomaly = state.anomalies[0]
        assert anomaly["source"] == "presence"
        assert anomaly["magnitude"] == BOOLEAN_ANOMALY_MAGNITUDE

        # true -> true is not.
        await process_device_reading_for_summon(user_id, device_id, "presence", 1, "bool")
        assert len(state.anomalies) == 1
    finally:
        from app.ws import unregister_active_session

        unregister_active_session(user_id, state)


@pytest.mark.asyncio
async def test_process_reading_no_active_session_is_a_noop(db_session, monkeypatch):
    monkeypatch.setattr(ws_module, "session_maker", TestSessionLocal)
    user_id, _session_id = await _make_user_and_session(db_session)
    device_id = uuid.uuid4()

    assert get_active_session(user_id) is None

    # Should not raise even though there's no active session and no
    # baseline yet — feeding a single reading can't be anomalous anyway.
    await process_device_reading_for_summon(user_id, device_id, "temperature", 21.0, "c")
    assert get_active_session(user_id) is None


@pytest.mark.asyncio
async def test_process_reading_non_anomalous_value_does_not_touch_session(db_session, monkeypatch):
    monkeypatch.setattr(ws_module, "session_maker", TestSessionLocal)
    user_id, session_id = await _make_user_and_session(db_session)

    state = SeanceState(user_id=user_id, session_id=session_id, client_ip="127.0.0.1")
    register_active_session(user_id, state)
    device_id = uuid.uuid4()

    try:
        for value in [21.0, 21.1, 20.9, 21.2, 21.0, 21.1, 21.0]:
            await process_device_reading_for_summon(
                user_id, device_id, "temperature", value, "c"
            )
        assert state.anomalies == []
    finally:
        from app.ws import unregister_active_session

        unregister_active_session(user_id, state)