3ef0990ffc
The music side's 'Clean Search History' automation now has a video counterpart so it appears on the video Automations page too. Distinct action_type video_clean_search_history (the system seeder keys on action_type, so reusing the music key would collide), registered to the SAME shared handler so behaviour is identical, scope='video' block (registry — users can build their own), and an owned_by='video' system automation on the same 1h cadence. The music action/row is untouched. Seam tests: video-scoped only (not on music), music action still music-scoped, exactly one video-owned system row at the 1h cadence, and it reuses the music handler. Registration contract (EXPECTED_ACTION_NAMES) updated.
69 lines
2.8 KiB
Python
69 lines
2.8 KiB
Python
"""Video twins of the music maintenance automations.
|
|
|
|
The music side has "Clean Search History" / "Clean Completed Downloads" / "Full
|
|
Cleanup" / "Auto-Backup Database". The video side gets its own copies so they
|
|
appear on the video Automations page too — distinct ``video_*`` action_type
|
|
(the system seeder keys on action_type, so a shared key would collide), the SAME
|
|
shared handler where the behaviour is identical, and ``owned_by='video'`` so the
|
|
music page is never disrupted.
|
|
|
|
These pin the registry + scope + seed contract, one twin per phase.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from core.automation.blocks import blocks_for_scope
|
|
from core.automation_engine import SYSTEM_AUTOMATIONS
|
|
|
|
|
|
def _action_types(scope):
|
|
return {a["type"] for a in blocks_for_scope(scope)["actions"]}
|
|
|
|
|
|
def _system_by_action(action_type):
|
|
return [s for s in SYSTEM_AUTOMATIONS if s.get("action_type") == action_type]
|
|
|
|
|
|
# ── Phase 2: Clean Search History ───────────────────────────────────────────
|
|
|
|
def test_video_clean_search_history_is_video_scoped_only():
|
|
# Appears on the video builder, NOT the music builder (music block untouched).
|
|
assert "video_clean_search_history" in _action_types("video")
|
|
assert "video_clean_search_history" not in _action_types("music")
|
|
|
|
|
|
def test_music_clean_search_history_is_untouched():
|
|
# The original music action still exists and is still music-scoped — no disruption.
|
|
assert "clean_search_history" in _action_types("music")
|
|
assert "clean_search_history" not in _action_types("video")
|
|
|
|
|
|
def test_video_clean_search_history_seeds_one_video_owned_system_row():
|
|
rows = _system_by_action("video_clean_search_history")
|
|
assert len(rows) == 1
|
|
row = rows[0]
|
|
assert row["owned_by"] == "video"
|
|
assert row["trigger_type"] == "schedule"
|
|
# mirrors the music cadence (every 1 hour)
|
|
assert row["trigger_config"] == {"interval": 1, "unit": "hours"}
|
|
|
|
|
|
def test_video_clean_search_history_reuses_the_music_handler():
|
|
# Registered to the SAME handler function as the music action (identical behaviour).
|
|
from core.automation.handlers import register_all
|
|
|
|
class _Eng:
|
|
def __init__(self):
|
|
self.handlers = {}
|
|
def register_action_handler(self, t, fn, guard_fn=None):
|
|
self.handlers[t] = fn
|
|
def register_progress_callbacks(self, *a, **k):
|
|
pass
|
|
|
|
import core.automation.handlers.registration as reg
|
|
eng = _Eng()
|
|
# Build minimal deps via the test helper the registration suite uses.
|
|
from tests.automation.test_handler_registration import _build_deps
|
|
register_all(_build_deps(eng))
|
|
assert "video_clean_search_history" in eng.handlers
|
|
assert "clean_search_history" in eng.handlers
|