osint-dashboard/tests/conftest.py

"""Shared test fixtures.

Unit tests (parsing/mapping) run anywhere. DB-backed tests (idempotency, API)
are marked `integration` and auto-skip when the test database is unreachable —
set DB_HOST/DB_PORT/DB_USER/DB_PASSWORD/DB_NAME to point at a TimescaleDB+PostGIS
instance (e.g. the `localhost/osint-dashboard-pg:test` image) to run them.
"""

from __future__ import annotations

import os
import sys
from pathlib import Path

import pytest

# Make the app package importable from tests (repo root/app).
APP_DIR = Path(__file__).resolve().parent.parent / "app"
if str(APP_DIR) not in sys.path:
    sys.path.insert(0, str(APP_DIR))

# Point the app's config at the test database BEFORE any app module is imported
# (database.py builds DATABASE_URL from env at import time).
os.environ.setdefault("DB_HOST", "127.0.0.1")
os.environ.setdefault("DB_PORT", "55432")
os.environ.setdefault("DB_USER", "osint")
os.environ.setdefault("DB_PASSWORD", "osint")
os.environ.setdefault("DB_NAME", "osint_data")
# Each test opens a fresh event loop (asyncio.run); a pooled connection can't be
# reused across loops, so force a NullPool (new connection per session).
os.environ.setdefault("DB_NULL_POOL", "1")

import asyncpg  # noqa: E402

pytestmark = []


def _db_reachable() -> bool:
    try:
        import asyncio

        async def _ping() -> bool:
            try:
                conn = await asyncpg.connect(
                    host=os.environ["DB_HOST"],
                    port=int(os.environ["DB_PORT"]),
                    user=os.environ["DB_USER"],
                    password=os.environ["DB_PASSWORD"],
                    database=os.environ["DB_NAME"],
                    timeout=3,
                )
                await conn.close()
                return True
            except Exception:
                return False

        return asyncio.run(_ping())
    except Exception:
        return False


requires_db = pytest.mark.skipif(
    not _db_reachable(),
    reason="test database unreachable (set DB_* env or start the PG container)",
)


@pytest.fixture()
def clean_fires():
    """Truncate the fires table before and after a DB-backed test."""
    import asyncio

    async def _truncate():
        conn = await asyncpg.connect(
            host=os.environ["DB_HOST"],
            port=int(os.environ["DB_PORT"]),
            user=os.environ["DB_USER"],
            password=os.environ["DB_PASSWORD"],
            database=os.environ["DB_NAME"],
        )
        try:
            await conn.execute("TRUNCATE fires")
        finally:
            await conn.close()

    asyncio.run(_truncate())
    yield
    asyncio.run(_truncate())


def make_fire_msg(**overrides) -> dict:
    """A realistic VIIRS hotspot NATS message (as fire_sources publishes it)."""
    msg = {
        "source_type": "fire",
        "latitude": 39.45678,
        "longitude": -121.12345,
        "brightness": 341.40,
        "confidence": "h",
        "acq_time": "2026-08-24T18:10:00+00:00",
        "satellite": "N",
        "instrument": "VIIRS",
        "bright_ti5": 310.20,
        "frp": 12.4,
        "daynight": "D",
        "scan": 0.45,
        "track": 0.47,
        "version": "2.0NRT",
        "raw": None,
    }
    msg.update(overrides)
    return msg