"""Plugin background jobs: registration, cron parsing, singleton lock (SPEC ยง11.2).""" from __future__ import annotations from typing import Any import pytest from app.plugins import jobs, loader from app.plugins.jobs import job_name, parse_cron, plugin_arq_functions, plugin_cron_jobs from app.plugins.sdk import JobSpec, Plugin, sdk from app.redis_client import get_redis async def _noop(ctx: dict[str, Any], **kwargs: Any) -> None: return None def test_job_name_is_namespaced() -> None: assert job_name("calendar", JobSpec(func=_noop)) == "plg.calendar._noop" def test_parse_cron() -> None: assert parse_cron("*/15 * * * *") == { "minute": {0, 15, 30, 45}, "hour": None, "day": None, "month": None, "weekday": None, } assert parse_cron("0 9-11 1 * 1,3") == { "minute": {0}, "hour": {9, 10, 11}, "day": {1}, "month": None, "weekday": {1, 3}, } with pytest.raises(ValueError, match="5 fields"): parse_cron("* * *") def test_functions_and_cron_registered() -> None: async def refresh(ctx: dict[str, Any], **kwargs: Any) -> None: return None async def poll(ctx: dict[str, Any], **kwargs: Any) -> None: return None loader.set_plugins_for_test( [ Plugin( name="cal", version="1", jobs=[JobSpec(func=refresh), JobSpec(func=poll, schedule="*/5 * * * *")], ) ] ) fn_names = {f.name for f in plugin_arq_functions()} assert fn_names == {"plg.cal.refresh", "plg.cal.poll"} # both enqueueable crons = plugin_cron_jobs() assert [c.name for c in crons] == ["plg.cal.poll.cron"] # only the scheduled one async def test_singleton_lock_prevents_overlap() -> None: calls: list[str] = [] async def work(ctx: dict[str, Any], **kwargs: Any) -> None: calls.append(str(kwargs.get("user_id"))) plugin = Plugin(name="cal", version="1", jobs=[JobSpec(func=work, singleton=True)]) runner, _ = jobs._make_runner(plugin, plugin.jobs[0]) await runner({}, user_id="u1") # runs; lock acquired then released assert calls == ["u1"] # Simulate an instance already holding the lock for u2 -> the run is skipped. held = jobs._lock_key("cal", "work", "u2") await get_redis().set(held, b"1", nx=True, ex=60) await runner({}, user_id="u2") assert calls == ["u1"] # unchanged await get_redis().delete(held) await runner({}, user_id="u2") # lock free now -> runs assert calls == ["u1", "u2"] async def test_non_singleton_always_runs() -> None: calls: list[int] = [] async def work(ctx: dict[str, Any], **kwargs: Any) -> None: calls.append(1) plugin = Plugin(name="cal", version="1", jobs=[JobSpec(func=work)]) runner, _ = jobs._make_runner(plugin, plugin.jobs[0]) await runner({}, user_id="u1") await runner({}, user_id="u1") assert calls == [1, 1] # no lock, both run async def test_sdk_enqueue_uses_namespaced_name(monkeypatch: pytest.MonkeyPatch) -> None: captured: dict[str, Any] = {} async def fake_enqueue(name: str, **kwargs: Any) -> None: captured["name"] = name captured["kwargs"] = kwargs monkeypatch.setattr("app.plugins.sdk.enqueue_job", fake_enqueue) async def refresh(ctx: dict[str, Any], **kwargs: Any) -> None: return None await sdk("cal").enqueue(refresh, user_id="u1") assert captured["name"] == "plg.cal.refresh" assert captured["kwargs"] == {"user_id": "u1"}