mirror of
https://github.com/MODSetter/SurfSense.git
synced 2026-05-29 19:35:20 +02:00
353755fd73
Top-level tests that span multiple submodules: - test_stores.py (7): the trigger + action registry contracts — register round-trip, unknown type → None (not raise), duplicate registration rejected, defensive snapshot from all_*. - test_definition_types.py (2): params_schema property on both ActionDefinition and TriggerDefinition reflects the Pydantic model. - test_persistence_enums.py (3): exact string values + member sets of AutomationStatus / RunStatus / TriggerType — the postgres-mirrored contract that breaks stored rows if drifted. - test_import_registrations.py (2): the bundled agent_task action and schedule trigger self-register on package import (canary for the side-effect import chain). conftest.py adds isolated_action_registry / isolated_trigger_registry fixtures: snapshot + restore of the module-level _REGISTRY dicts so tests that add their own definitions don't leak across the suite. 14 tests, pure unit.
39 lines
1.2 KiB
Python
39 lines
1.2 KiB
Python
"""Shared fixtures for the ``app.automations`` unit-test tree.
|
|
|
|
Provides registry isolation: the built-in ``schedule`` trigger and
|
|
``agent_task`` action self-register at import time. Tests that register
|
|
additional triggers/actions (or assert on the registry contents) must
|
|
not leak that state to other tests. These fixtures snapshot and restore
|
|
the module-level registry dicts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterator
|
|
|
|
import pytest
|
|
|
|
from app.automations.actions import store as action_store
|
|
from app.automations.triggers import store as trigger_store
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_action_registry() -> Iterator[None]:
|
|
"""Snapshot and restore the action registry around a test."""
|
|
snapshot = dict(action_store._REGISTRY)
|
|
try:
|
|
yield
|
|
finally:
|
|
action_store._REGISTRY.clear()
|
|
action_store._REGISTRY.update(snapshot)
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_trigger_registry() -> Iterator[None]:
|
|
"""Snapshot and restore the trigger registry around a test."""
|
|
snapshot = dict(trigger_store._REGISTRY)
|
|
try:
|
|
yield
|
|
finally:
|
|
trigger_store._REGISTRY.clear()
|
|
trigger_store._REGISTRY.update(snapshot)
|