ledgrab/server/tests/test_release_review_2026_06_23.py

"""Regression tests for the 2026-06-23 pre-release review fixes.

Covers the roadmap-batch (per-pixel smart-lights + integrations) findings fixed
before release:

* solar timezone offset must not crash on a malformed ``timezone`` string
  (DoS of the automation evaluation tick / 500 on manual trigger);
* the webhook action's ``content_type`` must reject CRLF / control chars
  (outbound HTTP header injection);
* the MQTT ``discovery_prefix`` must reject wildcard / control chars
  (HA-discovery topic injection);
* the effect ``reactive_mode`` must reject unknown values (silent no-op);
* game-integration ``get_stats`` must return an independent copy
  (cross-thread ``dict changed size during iteration``).
"""

import datetime
import types

import pytest
from pydantic import ValidationError

from ledgrab.api.routes.automations import _action_from_schema
from ledgrab.api.schemas.automations import ActionSchema
from ledgrab.api.schemas.color_strip_sources import EffectCSSCreate
from ledgrab.api.schemas.mqtt import MQTTSourceCreate
from ledgrab.core.game_integration import runtime_state
from ledgrab.utils.solar import utc_offset_hours_for


# ── solar timezone offset hardening ──────────────────────────────────────────


@pytest.mark.parametrize(
    "bad_tz",
    [
        "../../../etc/passwd",  # path traversal -> ValueError
        "foo\x00bar",  # embedded null -> ValueError
        "x" * 5000,  # over-long -> OSError on some platforms
        "Not/A/Real/Zone",  # plausible-but-unknown -> ZoneInfoNotFoundError
    ],
)
def test_solar_offset_does_not_raise_on_malformed_timezone(bad_tz):
    """A crafted SolarRule.timezone must fall back, never crash the eval tick."""
    when = datetime.datetime(2026, 1, 15, 12, 0, 0)
    offset = utc_offset_hours_for(bad_tz, when)
    assert isinstance(offset, float)


def test_solar_offset_valid_timezone_still_resolves():
    when = datetime.datetime(2026, 1, 15, 12, 0, 0)  # winter -> EST = -5
    assert utc_offset_hours_for("America/New_York", when) == pytest.approx(-5.0)


# ── webhook action Content-Type header injection ─────────────────────────────


@pytest.mark.parametrize(
    "bad_ct",
    [
        "application/json\r\nX-Injected: evil",
        "text/plain\nX-Injected: evil",
        "application/json\x00",
        "application/jsön",  # non-ASCII
    ],
)
def test_webhook_action_rejects_unsafe_content_type(bad_ct):
    schema = ActionSchema(
        action_type="webhook",
        webhook_url="http://10.0.0.5/hook",
        method="POST",
        content_type=bad_ct,
        fire_on="activate",
    )
    with pytest.raises(ValueError, match="content_type"):
        _action_from_schema(schema)


def test_webhook_action_accepts_normal_content_type():
    schema = ActionSchema(
        action_type="webhook",
        webhook_url="http://10.0.0.5/hook",  # LAN IP is allowed by SSRF policy
        method="POST",
        content_type="application/json; charset=utf-8",
        fire_on="activate",
    )
    action = _action_from_schema(schema)
    assert action.content_type == "application/json; charset=utf-8"


# ── MQTT HA-discovery topic injection ────────────────────────────────────────


@pytest.mark.parametrize(
    "bad_prefix",
    [
        "homeassistant/+/evil",  # MQTT wildcard
        "homeassistant/#",  # MQTT wildcard
        "home\nassistant",  # control char
        "x" * 65,  # over max_length
    ],
)
def test_mqtt_discovery_prefix_rejects_unsafe_value(bad_prefix):
    with pytest.raises(ValidationError):
        MQTTSourceCreate(name="src", broker_host="broker.local", discovery_prefix=bad_prefix)


def test_mqtt_discovery_prefix_defaults_to_homeassistant():
    src = MQTTSourceCreate(name="src", broker_host="broker.local")
    assert src.discovery_prefix == "homeassistant"


# ── effect reactive_mode validation ──────────────────────────────────────────


def test_reactive_mode_rejects_unknown_value():
    with pytest.raises(ValidationError):
        EffectCSSCreate(name="fx", reactive_mode="invalid")


@pytest.mark.parametrize("mode", ["brightness", "saturation", "both"])
def test_reactive_mode_accepts_known_values(mode):
    src = EffectCSSCreate(name="fx", reactive_mode=mode)
    assert src.reactive_mode == mode


# ── game-integration get_stats returns an independent snapshot ───────────────


def test_get_stats_returns_independent_copy():
    integration_id = "test-int-copy"
    runtime_state.cleanup_state(integration_id)
    try:
        event = types.SimpleNamespace(event_type="kill", timestamp="2026-06-23T00:00:00Z")
        runtime_state.record_events(integration_id, [event])

        snapshot = runtime_state.get_stats(integration_id)
        assert snapshot["event_count"] == 1
        assert snapshot["event_counts_by_type"] == {"kill": 1}

        # Mutating the returned snapshot must not corrupt the live state.
        snapshot["event_counts_by_type"]["kill"] = 999
        snapshot["event_counts_by_type"]["injected"] = 1

        fresh = runtime_state.get_stats(integration_id)
        assert fresh["event_counts_by_type"] == {"kill": 1}
        assert "injected" not in fresh["event_counts_by_type"]
    finally:
        runtime_state.cleanup_state(integration_id)