osint-dashboard/tests/test_bg_jobs.py

"""ffmpeg snapshots stay off the request path (asyncio.create_task)."""

from __future__ import annotations

import asyncio

import bg_jobs


def test_bg_jobs_has_no_pps_cap():
    assert not any(name.endswith("_PPS_CAP") for name in dir(bg_jobs))


def test_camera_preview_has_no_public_feed_probe():
    import camera_preview

    assert not hasattr(camera_preview, "probe_public_feed")
    assert not hasattr(camera_preview, "_http_feed_url")


def test_ingest_routes_exclude_active_discovery():
    from main import app

    ingest = [
        getattr(r, "path", "")
        for r in app.routes
        if getattr(r, "path", "").startswith("/api/ingest/")
    ]
    assert "/api/ingest/fires" in ingest
    assert all("scan" not in path for path in ingest)


def test_schedule_ffmpeg_snapshot_is_a_task_not_inline(monkeypatch):
    calls = {"n": 0}

    async def fake_grab(url, timeout=8.0):
        calls["n"] += 1
        await asyncio.sleep(5)
        return b"\xff\xd8fakejpeg"

    monkeypatch.setattr(bg_jobs, "_ffmpeg_grab", fake_grab)
    bg_jobs._ffmpeg_tasks.clear()
    bg_jobs._ffmpeg_cache.clear()

    async def run():
        task = bg_jobs.schedule_ffmpeg_snapshot("rtsp://10.0.0.1/")
        assert isinstance(task, asyncio.Task)
        assert not task.done()
        task.cancel()
        try:
            await task
        except (asyncio.CancelledError, Exception):
            pass

    asyncio.run(run())