Files
alexei.dolgolyov 25c613c5cb feat(activity-log): phase 3 - event instrumentation (4 categories)
- 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
2026-06-09 19:20:57 +03:00

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)