25c613c5cb
- entity CRUD via fire_entity_event choke point (name resolved/sanitized; deletes pass name explicitly) - auth: failures + WS session establishment (no tokens logged); per-IP audit-record throttle - device: online/offline (health), discovered/lost (zeroconf), ADB connect/disconnect - capture/system: target start-stop, scenes, playlists, automations, backup/restore, update, restart, calibration, settings - security hardening: sanitize_display strips control/NUL/ANSI/newlines from untrusted strings; malformed-IPv6 origin guard - 129 instrumentation tests (incl. secret-leak, log-injection, throttle, best-effort) + autouse throttle-reset fixture
328 lines
10 KiB
Python
328 lines
10 KiB
Python
"""Pytest configuration and shared fixtures.
|
|
|
|
IMPORTANT: This conftest patches the global config singleton BEFORE any test
|
|
module can import ``ledgrab.main``. ``main.py`` reads ``get_config()``
|
|
at module level to open the database — if the singleton is not patched first,
|
|
the REAL production database (``data/ledgrab.db``) is opened and tests
|
|
read/write/delete production data.
|
|
"""
|
|
|
|
import tempfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ISOLATE ALL TESTS FROM PRODUCTION DATA — must happen before any test module
|
|
# imports ``ledgrab.main``.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import ledgrab.config as _config_mod # noqa: E402
|
|
|
|
_test_tmp = Path(tempfile.mkdtemp(prefix="wled_test_"))
|
|
_test_db_path = str(_test_tmp / "test_ledgrab.db")
|
|
_test_assets_dir = str(_test_tmp / "test_assets")
|
|
|
|
# Pre-create the test database file so main.py's legacy-data migration
|
|
# (which copies the user's production DB into the configured location when
|
|
# it doesn't exist) doesn't shovel real data into the test DB. Without this
|
|
# touch, tests see production state — settings, devices, notification
|
|
# preferences, history — and assertions about "default" state fail.
|
|
Path(_test_db_path).touch()
|
|
|
|
_original_config = _config_mod.Config.load()
|
|
_test_config = _original_config.model_copy(
|
|
update={
|
|
"storage": _config_mod.StorageConfig(database_file=_test_db_path),
|
|
"assets": _config_mod.AssetsConfig(
|
|
assets_dir=_test_assets_dir,
|
|
max_file_size_mb=_original_config.assets.max_file_size_mb,
|
|
),
|
|
# Inject a test API key so auth-enforcement tests have something
|
|
# concrete to authenticate against and reject when omitted.
|
|
"auth": _config_mod.AuthConfig(api_keys={"test": "test-api-key-12345"}),
|
|
},
|
|
)
|
|
_config_mod.config = _test_config
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from ledgrab.config import Config, StorageConfig, ServerConfig, AuthConfig # noqa: E402
|
|
from ledgrab.storage.database import Database # noqa: E402
|
|
from ledgrab.storage.device_store import Device, DeviceStore # noqa: E402
|
|
from ledgrab.storage.sync_clock import SyncClock # noqa: E402
|
|
from ledgrab.storage.sync_clock_store import SyncClockStore # noqa: E402
|
|
from ledgrab.storage.output_target_store import OutputTargetStore # noqa: E402
|
|
from ledgrab.storage.automation import Automation # noqa: E402
|
|
from ledgrab.storage.automation_store import AutomationStore # noqa: E402
|
|
from ledgrab.storage.value_source_store import ValueSourceStore # noqa: E402
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Directory / path fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def test_data_dir(tmp_path):
|
|
"""Provide a temporary directory for test data."""
|
|
d = tmp_path / "data"
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return d
|
|
|
|
|
|
@pytest.fixture
|
|
def test_config_dir(tmp_path):
|
|
"""Provide a temporary directory for test configuration."""
|
|
return tmp_path / "config"
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_store_dir(tmp_path):
|
|
"""Provide a temp directory for store files, cleaned up after tests."""
|
|
d = tmp_path / "stores"
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return d
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Database fixture
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_db(tmp_path):
|
|
"""Provide a temporary SQLite Database instance."""
|
|
db = Database(tmp_path / "test.db")
|
|
yield db
|
|
db.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def test_config(tmp_path):
|
|
"""A Config instance with temp directories for all store files."""
|
|
data_dir = tmp_path / "data"
|
|
data_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
storage = StorageConfig(
|
|
database_file=str(data_dir / "test.db"),
|
|
)
|
|
|
|
return Config(
|
|
server=ServerConfig(host="127.0.0.1", port=9999),
|
|
auth=AuthConfig(api_keys={"test": "test-api-key-12345"}),
|
|
storage=storage,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Store fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def device_store(tmp_db):
|
|
"""Provide a DeviceStore backed by a temp database."""
|
|
return DeviceStore(tmp_db)
|
|
|
|
|
|
@pytest.fixture
|
|
def sync_clock_store(tmp_db):
|
|
"""Provide a SyncClockStore backed by a temp database."""
|
|
return SyncClockStore(tmp_db)
|
|
|
|
|
|
@pytest.fixture
|
|
def output_target_store(tmp_db):
|
|
"""Provide an OutputTargetStore backed by a temp database."""
|
|
return OutputTargetStore(tmp_db)
|
|
|
|
|
|
@pytest.fixture
|
|
def automation_store(tmp_db):
|
|
"""Provide an AutomationStore backed by a temp database."""
|
|
return AutomationStore(tmp_db)
|
|
|
|
|
|
@pytest.fixture
|
|
def value_source_store(tmp_db):
|
|
"""Provide a ValueSourceStore backed by a temp database."""
|
|
return ValueSourceStore(tmp_db)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sample entity factories
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_device():
|
|
"""Provide a sample device configuration dict."""
|
|
return {
|
|
"id": "test_device_001",
|
|
"name": "Test WLED Device",
|
|
"url": "http://192.168.1.100",
|
|
"led_count": 150,
|
|
"enabled": True,
|
|
"device_type": "wled",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def make_device():
|
|
"""Factory fixture: call make_device(name=..., **overrides) to build a Device."""
|
|
_counter = 0
|
|
|
|
def _factory(name=None, **kwargs):
|
|
nonlocal _counter
|
|
_counter += 1
|
|
defaults = dict(
|
|
device_id=f"device_test_{_counter:04d}",
|
|
name=name or f"Device {_counter}",
|
|
url=f"http://192.168.1.{_counter}",
|
|
led_count=150,
|
|
)
|
|
defaults.update(kwargs)
|
|
return Device(**defaults)
|
|
|
|
return _factory
|
|
|
|
|
|
@pytest.fixture
|
|
def make_sync_clock():
|
|
"""Factory fixture: call make_sync_clock(name=..., **overrides)."""
|
|
_counter = 0
|
|
|
|
def _factory(name=None, **kwargs):
|
|
nonlocal _counter
|
|
_counter += 1
|
|
now = datetime.now(timezone.utc)
|
|
defaults = dict(
|
|
id=f"sc_test_{_counter:04d}",
|
|
name=name or f"Clock {_counter}",
|
|
speed=1.0,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
defaults.update(kwargs)
|
|
return SyncClock(**defaults)
|
|
|
|
return _factory
|
|
|
|
|
|
@pytest.fixture
|
|
def make_automation():
|
|
"""Factory fixture: call make_automation(name=..., **overrides)."""
|
|
_counter = 0
|
|
|
|
def _factory(name=None, **kwargs):
|
|
nonlocal _counter
|
|
_counter += 1
|
|
now = datetime.now(timezone.utc)
|
|
defaults = dict(
|
|
id=f"auto_test_{_counter:04d}",
|
|
name=name or f"Automation {_counter}",
|
|
enabled=True,
|
|
rule_logic="or",
|
|
rules=[],
|
|
scene_preset_id=None,
|
|
deactivation_mode="none",
|
|
deactivation_scene_preset_id=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
defaults.update(kwargs)
|
|
return Automation(**defaults)
|
|
|
|
return _factory
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Authenticated test client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def authenticated_client(test_config, monkeypatch):
|
|
"""Provide a FastAPI TestClient with auth header pre-set.
|
|
|
|
Patches global config so the app uses temp storage paths.
|
|
"""
|
|
import ledgrab.config as config_mod
|
|
|
|
monkeypatch.setattr(config_mod, "config", test_config)
|
|
|
|
from fastapi.testclient import TestClient
|
|
from ledgrab.main import app
|
|
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
client.headers["Authorization"] = "Bearer test-api-key-12345"
|
|
return client
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Calibration sample (kept from original conftest)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_calibration():
|
|
"""Provide a sample calibration configuration."""
|
|
return {
|
|
"layout": "clockwise",
|
|
"start_position": "bottom_left",
|
|
"segments": [
|
|
{"edge": "bottom", "led_start": 0, "led_count": 40, "reverse": False},
|
|
{"edge": "right", "led_start": 40, "led_count": 30, "reverse": False},
|
|
{"edge": "top", "led_start": 70, "led_count": 40, "reverse": True},
|
|
{"edge": "left", "led_start": 110, "led_count": 40, "reverse": True},
|
|
],
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth throttle isolation — reset per-IP throttle state between every test
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_auth_throttle():
|
|
"""Clear the auth-failure audit throttle dict before (and after) each test.
|
|
|
|
The module-global ``_auth_record_last`` dict in ``ledgrab.api.auth`` persists
|
|
across tests. When multiple tests trigger an auth failure from the SAME
|
|
client IP within the 10 s window they share, the second test gets throttled
|
|
(0 records) and assertions like "expected exactly 1 auth.rejected" fail.
|
|
|
|
This fixture resets the dict to a clean state so every test starts with a
|
|
fresh throttle window. The production throttle behavior is UNCHANGED — only
|
|
test isolation is affected.
|
|
"""
|
|
import ledgrab.api.auth as _auth_mod
|
|
|
|
_throttle = getattr(_auth_mod, "_auth_record_last", None)
|
|
if _throttle is not None:
|
|
_throttle.clear()
|
|
yield
|
|
if _throttle is not None:
|
|
_throttle.clear()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Session cleanup — remove temporary test directory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def _cleanup_test_tmp():
|
|
"""Remove the temporary test directory after all tests complete."""
|
|
import shutil
|
|
|
|
yield
|
|
shutil.rmtree(_test_tmp, ignore_errors=True)
|