From 5651d251d0b29ad2f8f7ee909e8964aeb0411928 Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Sat, 29 Aug 2026 19:32:00 -0400 Subject: [PATCH] fix(ingest): skip known KEV/FIRMS rows; CI skips unchanged images CISA KEV republished 1685 NIST URLs every 5 min; FIRMS re-inserted ~325k global hotspots every 15 min. Producer now skips URLs already in event_dedup and FIRMS CSVs that are unchanged (delta-only persist). CI rebuilds only images whose paths changed and never pulls/rebuilds Timescale or bounces osint-db unless Dockerfile.pg changes. --- .forgejo/workflows/build.yml | 157 +++++++++++++++++++++++++++-------- app/fire_sources.py | 83 ++++++++++++------ app/ingestor.py | 2 +- app/sources.py | 43 ++++++++-- docker-compose.yml | 11 ++- scripts/compose-reup.sh | 74 ++++++++++++----- tests/test_fire_sources.py | 94 +++++++++++++++++++++ tests/test_new_sources.py | 62 ++++++++++++++ tests/test_upstream_cache.py | 6 +- 9 files changed, 438 insertions(+), 94 deletions(-) diff --git a/.forgejo/workflows/build.yml b/.forgejo/workflows/build.yml index 6116fdf..b2c2a0d 100644 --- a/.forgejo/workflows/build.yml +++ b/.forgejo/workflows/build.yml @@ -1,6 +1,10 @@ -# Build all OSINT images, publish to the Forgejo container registry, then +# Build changed OSINT images, publish to the Forgejo container registry, then # redeploy on the Pi runner (docker.sock mounted). # +# Unchanged images are skipped. Dockerfile.pg / osint-dashboard-pg is NOT +# rebuilt or pulled on a normal merge — Postgres stays up. Rebuild it only +# when Dockerfile.pg changes, or via workflow_dispatch rebuild_pg. +# # Public pull host: forgejo.siriusdevops.com (NOT ghcr.io) # CI push host: 127.0.0.1:3000 — Cloudflare 413s layers ≳100MB on the public # hostname, even from the Pi (hairpins out through the tunnel). @@ -19,6 +23,15 @@ on: push: branches: [main, master] workflow_dispatch: + inputs: + rebuild_pg: + description: Rebuild Timescale+PostGIS (Dockerfile.pg) + type: boolean + default: false + rebuild_all: + description: Rebuild every app image (ignore path filter) + type: boolean + default: false env: PUBLIC_REGISTRY: ${{ vars.FORGEJO_REGISTRY || 'forgejo.siriusdevops.com' }} @@ -36,6 +49,68 @@ jobs: steps: - name: Checkout uses: https://code.forgejo.org/actions/checkout@v4 + with: + fetch-depth: 50 + + - name: Plan image builds + id: plan + run: | + set -euo pipefail + APP=0 + SCRAPER=0 + SUM=0 + PG=0 + COMPOSE=0 + + mark() { + case "$1" in + Dockerfile.pg) + PG=1 ;; + Dockerfile|app/*|alembic/*|alembic.ini) + APP=1 ;; + news/scraper/*) + SCRAPER=1 ;; + news/summerizer/*) + SUM=1 ;; + docker-compose.yml|scripts/compose-reup.sh) + COMPOSE=1 ;; + esac + } + + if [ "${{ github.event_name }}" = "workflow_dispatch" ]; then + APP=1; SCRAPER=1; SUM=1 + if [ "${{ github.event.inputs.rebuild_all }}" = "true" ]; then + APP=1; SCRAPER=1; SUM=1; PG=1 + fi + if [ "${{ github.event.inputs.rebuild_pg }}" = "true" ]; then + PG=1 + fi + else + BEFORE="${{ github.event.before }}" + SHA="${GITHUB_SHA}" + ZEROS="0000000000000000000000000000000000000000" + if [ -z "$BEFORE" ] || [ "$BEFORE" = "$ZEROS" ]; then + echo "No previous SHA — build app images, skip pg" + APP=1; SCRAPER=1; SUM=1 + elif ! git cat-file -e "${BEFORE}^{commit}" 2>/dev/null; then + echo "Previous SHA $BEFORE not in history — build app images, skip pg" + APP=1; SCRAPER=1; SUM=1 + else + while IFS= read -r f; do + [ -z "$f" ] && continue + mark "$f" + done < <(git diff --name-only "$BEFORE" "$SHA") + fi + fi + + { + echo "app=$APP" + echo "scraper=$SCRAPER" + echo "summarizer=$SUM" + echo "pg=$PG" + echo "compose=$COMPOSE" + } >> "$GITHUB_OUTPUT" + echo "plan app=$APP scraper=$SCRAPER summarizer=$SUM pg=$PG compose=$COMPOSE" - name: Image refs id: img @@ -59,6 +134,7 @@ jobs: echo "SHA tag: $SHA" - name: Login to Forgejo registry + if: steps.plan.outputs.app == '1' || steps.plan.outputs.scraper == '1' || steps.plan.outputs.summarizer == '1' || steps.plan.outputs.pg == '1' run: | set -euo pipefail # GITHUB_TOKEN login "succeeds" but blob uploads 401 (Forgejo packages @@ -71,6 +147,7 @@ jobs: -u sirius --password-stdin - name: Build application image (api / ingester / cameras) + if: steps.plan.outputs.app == '1' run: | set -ex APP="${{ steps.img.outputs.app }}" @@ -81,6 +158,7 @@ jobs: docker push "${APP}:${SHA}" - name: Build news-scraper image + if: steps.plan.outputs.scraper == '1' run: | set -ex IMG="${{ steps.img.outputs.scraper }}" @@ -91,6 +169,7 @@ jobs: docker push "${IMG}:${SHA}" - name: Build news-summarizer image + if: steps.plan.outputs.summarizer == '1' run: | set -ex IMG="${{ steps.img.outputs.summarizer }}" @@ -101,22 +180,17 @@ jobs: docker push "${IMG}:${SHA}" - name: Build / refresh Timescale+PostGIS image + if: steps.plan.outputs.pg == '1' run: | set -ex PG="${{ steps.img.outputs.pg }}" SHA="${{ steps.img.outputs.sha }}" - # Prefer rebuild so registry always has a current pg image. If packagecloud - # is unreachable, fall back to whatever local image already exists. if docker build -f Dockerfile.pg -t "${PG}:latest" -t "${PG}:${SHA}" \ -t "localhost/osint-dashboard-pg:latest" .; then docker push "${PG}:latest" docker push "${PG}:${SHA}" elif docker image inspect "localhost/osint-dashboard-pg:latest" >/dev/null 2>&1; then - echo "WARN: Dockerfile.pg build failed; retagging existing local pg image into registry" - docker tag "localhost/osint-dashboard-pg:latest" "${PG}:latest" - docker tag "localhost/osint-dashboard-pg:latest" "${PG}:${SHA}" - docker push "${PG}:latest" - docker push "${PG}:${SHA}" + echo "WARN: Dockerfile.pg build failed; keeping existing local pg image" else echo "ERROR: cannot build or find osint-dashboard-pg image" exit 1 @@ -126,39 +200,50 @@ jobs: run: | set -ex cd "${GITHUB_WORKSPACE}" - # Pull from Forgejo registry into local tags compose expects, then up. - # Compose file still uses localhost/* for stable local names; we mirror - # registry tags so a cold host can recover via docker pull. - REG="${{ steps.img.outputs.reg }}" - PUB="${{ steps.img.outputs.pub }}" - OWN="${{ env.OWNER }}" - for name in osint-dashboard osint-dashboard-pg osint-news-scraper osint-news-summarizer; do - docker pull "${REG}/${OWN}/${name}:latest" || true - docker tag "${REG}/${OWN}/${name}:latest" "localhost/${name}:latest" || true - docker tag "${REG}/${OWN}/${name}:latest" "${PUB}/${OWN}/${name}:latest" || true - done - # Do NOT set COMPOSE_PROJECT_NAME differently — volumes must stay - # osint-dashboard_osint-pgdata (pinned by `name:` in compose). - docker compose build --no-cache app ingester camera-service news-scraper news-summarizer || \ - docker compose build app ingester camera-service news-scraper news-summarizer - # Name-pinned containers (container_name: osint-dashboard, …) collide - # when compose tries to create instead of recreate — e.g. leftover from - # a different working_dir or a half-failed previous up. down + rm -f - # the known names, then up the full ingest profile. + APP="${{ steps.plan.outputs.app }}" + SCRAPER="${{ steps.plan.outputs.scraper }}" + SUM="${{ steps.plan.outputs.summarizer }}" + PG="${{ steps.plan.outputs.pg }}" + COMPOSE="${{ steps.plan.outputs.compose }}" + + SVCS=() + [ "$APP" = "1" ] && SVCS+=(app ingester camera-service) + [ "$SCRAPER" = "1" ] && SVCS+=(news-scraper) + [ "$SUM" = "1" ] && SVCS+=(news-summarizer) + if [ "$COMPOSE" = "1" ]; then + # compose/script change: bounce workers so env/command updates apply. + # Still do not bounce Postgres. + for s in app ingester camera-service news-scraper news-summarizer; do + case " ${SVCS[*]} " in + *" $s "*) ;; + *) SVCS+=("$s") ;; + esac + done + fi + chmod +x scripts/compose-reup.sh - COMPOSE_PROJECT_NAME=osint-dashboard COMPOSE_PROFILES=ingest \ - scripts/compose-reup.sh + if [ "$PG" = "1" ]; then + FORCE_RECREATE_DB=1 COMPOSE_PROJECT_NAME=osint-dashboard COMPOSE_PROFILES=ingest \ + scripts/compose-reup.sh "${SVCS[@]}" db + elif [ "${#SVCS[@]}" -gt 0 ]; then + COMPOSE_PROJECT_NAME=osint-dashboard COMPOSE_PROFILES=ingest \ + scripts/compose-reup.sh "${SVCS[@]}" + else + echo "No image or compose changes — leave running containers alone" + docker compose --profile ingest ps + fi docker image prune -f - echo "osint-dashboard deployed; images also on ${PUB}/${OWN}/" + echo "osint-dashboard deploy done; db image left in place unless pg=1" - name: Summary if: always() run: | { - echo "## Forgejo registry images" - echo "Pushed via ${{ steps.img.outputs.reg }} (loopback). Pull publicly:" - echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-dashboard:latest\`" - echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-dashboard-pg:latest\`" - echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-news-scraper:latest\`" - echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-news-summarizer:latest\`" + echo "## Image plan" + echo "- app: \`${{ steps.plan.outputs.app }}\`" + echo "- news-scraper: \`${{ steps.plan.outputs.scraper }}\`" + echo "- news-summarizer: \`${{ steps.plan.outputs.summarizer }}\`" + echo "- pg (Timescale): \`${{ steps.plan.outputs.pg }}\`" + echo + echo "Postgres is rebuilt/pulled only when \`Dockerfile.pg\` changes (or workflow_dispatch rebuild_pg)." } >> "$GITHUB_STEP_SUMMARY" diff --git a/app/fire_sources.py b/app/fire_sources.py index c27b0f7..46f9fe9 100644 --- a/app/fire_sources.py +++ b/app/fire_sources.py @@ -22,6 +22,7 @@ UTC date (YYYY-MM-DD). from __future__ import annotations import csv +import hashlib import io import json import logging @@ -44,6 +45,11 @@ from upstream_cache import firms_cache logger = logging.getLogger("osint.firms") +# In-process poll state: skip byte-identical CSVs, persist only new hotspots. +# Survives the 15-minute loop; one full ON CONFLICT after process start. +_csv_digest: dict[tuple, bytes] = {} +_seen_ids: dict[tuple, set[int]] = {} + # ── FIRMS API ───────────────────────────────────────────────────────────── FIRMS_AREA_CSV = ( @@ -86,33 +92,34 @@ def normalize_acq_time(acq_date: object, acq_time: object) -> datetime | None: return None -def parse_firms_csv(text: str) -> list[dict]: - """Parse a FIRMS area CSV payload into normalized fire messages. +def _hotspot_id(lat: float, lon: float, acq_iso: str, satellite: str) -> int: + return hash((round(lat, 5), round(lon, 5), acq_iso, satellite)) - Returns one dict per hotspot with the fields stored in the ``fires`` table - (acq_time already combined into a UTC ISO timestamp). Rows that don't look - like valid VIIRS detections are skipped rather than failing the whole poll. + +def parse_firms_csv_delta( + text: str, skip_ids: set[int] | None = None, +) -> tuple[list[dict], set[int]]: + """Parse FIRMS CSV; optionally drop hotspots already seen this process. + + Returns (new_or_all_points, ids_for_every_valid_row). Streaming — does not + materialize the raw CSV as a list of lists. """ - rows = list(csv.reader(io.StringIO(text))) - if not rows: - return [] - - # Locate the real header row. FIRMS normally returns the CSV header first, - # but occasionally prepends a legend/info line, so scan until we see the - # canonical header. - header_idx = 0 - for i, row in enumerate(rows): + reader = csv.reader(io.StringIO(text)) + header = None + for row in reader: if row and row[0].strip().lower() == "latitude" and len(row) >= 4: - header_idx = i + header = [c.strip().lower() for c in row] break - header = [c.strip().lower() for c in rows[header_idx]] - # Guard against a header that isn't actually the FIRMS one. - if "latitude" not in header or "longitude" not in header: - logger.warning("FIRMS payload does not look like a hotspot CSV (first row: %r)", header[:6]) - return [] + if not header or "latitude" not in header or "longitude" not in header: + logger.warning( + "FIRMS payload does not look like a hotspot CSV (first row: %r)", + (header or [])[:6], + ) + return [], set() points: list[dict] = [] - for row in rows[header_idx + 1:]: + ids: set[int] = set() + for row in reader: if len(row) < len(header): continue rec = dict(zip(header, row)) @@ -123,13 +130,19 @@ def parse_firms_csv(text: str) -> list[dict]: acq_time = normalize_acq_time(rec.get("acq_date"), rec.get("acq_time")) if acq_time is None: continue + sat = str(rec.get("satellite") or "").strip() + acq_iso = acq_time.isoformat() + hid = _hotspot_id(lat, lon, acq_iso, sat) + ids.add(hid) + if skip_ids is not None and hid in skip_ids: + continue points.append({ "latitude": lat, "longitude": lon, "brightness": _to_float(rec.get("bright_ti4")), "confidence": str(rec.get("confidence") or "").strip(), - "acq_time": acq_time.isoformat(), - "satellite": str(rec.get("satellite") or "").strip(), + "acq_time": acq_iso, + "satellite": sat, "instrument": str(rec.get("instrument") or "").strip(), "bright_ti5": _to_float(rec.get("bright_ti5")), "frp": _to_float(rec.get("frp")), @@ -138,6 +151,17 @@ def parse_firms_csv(text: str) -> list[dict]: "track": _to_float(rec.get("track")), "version": str(rec.get("version") or "").strip(), }) + return points, ids + + +def parse_firms_csv(text: str) -> list[dict]: + """Parse a FIRMS area CSV payload into normalized fire messages. + + Returns one dict per hotspot with the fields stored in the ``fires`` table + (acq_time already combined into a UTC ISO timestamp). Rows that don't look + like valid VIIRS detections are skipped rather than failing the whole poll. + """ + points, _ids = parse_firms_csv_delta(text) return points @@ -215,11 +239,18 @@ async def ingest_fires(bbox: str | None = None) -> int: dataset, first_line, ) continue - points = parse_firms_csv(text) - published = await persist_hotspots(points) + poll_key = (dataset, area, FIRMS_DAYS) + digest = hashlib.sha256(text.encode("utf-8", "surrogatepass")).digest() + if _csv_digest.get(poll_key) == digest: + logger.info("FIRMS %s CSV unchanged, skip parse/insert", dataset) + continue + points, ids = parse_firms_csv_delta(text, skip_ids=_seen_ids.get(poll_key)) + published = await persist_hotspots(points) if points else 0 + _csv_digest[poll_key] = digest + _seen_ids[poll_key] = ids total_published += published logger.info( "FIRMS: fetched %d hotspot(s) for bbox=%s (%s), published %d", - len(points), area, dataset, published, + len(ids), area, dataset, published, ) return total_published diff --git a/app/ingestor.py b/app/ingestor.py index ac0bcbd..ccf1471 100644 --- a/app/ingestor.py +++ b/app/ingestor.py @@ -213,7 +213,7 @@ async def ingest_event(msg: dict): claimed = await session.execute(dedup) if not claimed.rowcount: await session.commit() - logger.info("skip duplicate event url=%s", key) + logger.debug("skip duplicate event url=%s", key) return None result = await session.execute(events_table.insert().values(**event_row)) await session.commit() diff --git a/app/sources.py b/app/sources.py index c12c303..7d587a5 100644 --- a/app/sources.py +++ b/app/sources.py @@ -55,6 +55,35 @@ def event_dedup_key(msg: dict) -> str | None: return url or None +async def existing_event_urls(urls: list[str]) -> set[str]: + """URLs already claimed in event_dedup. Empty input -> empty set.""" + if not urls: + return set() + from sqlalchemy import select + + from database import async_session + from models import event_dedup as event_dedup_table + + async with async_session() as session: + result = await session.execute( + select(event_dedup_table.c.url).where(event_dedup_table.c.url.in_(urls)) + ) + return {row[0] for row in result} + + +async def _publish_unknown(subject: str, events: list[dict]) -> int: + """Publish only events whose URL is not already in event_dedup.""" + keys = [event_dedup_key(e) for e in events] + known = await existing_event_urls([k for k in keys if k]) + published = 0 + for event, key in zip(events, keys): + if key and key in known: + continue + await publish_event(subject, event) + published += 1 + return published + + def _ua_headers() -> dict[str, str]: return {"User-Agent": OSINT_USER_AGENT} @@ -283,10 +312,9 @@ async def ingest_eonet(): resp.raise_for_status() data = resp.json() events = parse_eonet_events(data if isinstance(data, dict) else {}) - for event in events: - await publish_event("events.disaster", event) - logger.info("Ingested %d EONET events", len(events)) - return len(events) + published = await _publish_unknown("events.disaster", events) + logger.info("Ingested %d EONET events (%d already known)", published, len(events) - published) + return published # ─── CISA KEV ─────────────────────────────────────────────────────────── @@ -330,10 +358,9 @@ async def ingest_cisa_kev(): resp.raise_for_status() data = resp.json() events = parse_cisa_kev(data if isinstance(data, dict) else {}) - for event in events: - await publish_event("events.disaster", event) - logger.info("Ingested %d CISA KEV rows", len(events)) - return len(events) + published = await _publish_unknown("events.disaster", events) + logger.info("Ingested %d CISA KEV rows (%d already known)", published, len(events) - published) + return published # ─── Social Signals (Twitter/X-like placeholder) ──────────────────────── diff --git a/docker-compose.yml b/docker-compose.yml index 6e38729..76d0431 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -20,8 +20,10 @@ services: # networks (and ISP abuse-mitigation blackholes) block, so rebuilding it # on every CI deploy made the pipeline flaky. Rebuild manually when the # base image or extensions need bumping: - # docker compose build db && docker compose up -d db + # docker build -f Dockerfile.pg -t localhost/osint-dashboard-pg:latest . + # FORCE_RECREATE_DB=1 scripts/compose-reup.sh db image: localhost/osint-dashboard-pg:latest + pull_policy: never container_name: osint-db restart: unless-stopped environment: @@ -52,6 +54,7 @@ services: nats: image: nats:2.10 + pull_policy: missing platform: linux/arm64 container_name: osint-nats restart: unless-stopped @@ -67,6 +70,7 @@ services: dockerfile: Dockerfile platforms: ["linux/arm64"] image: localhost/osint-dashboard:latest + pull_policy: never container_name: osint-ingester restart: unless-stopped profiles: ["ingest"] @@ -111,6 +115,7 @@ services: dockerfile: Dockerfile platforms: ["linux/arm64"] image: localhost/osint-dashboard:latest + pull_policy: never container_name: osint-dashboard restart: unless-stopped depends_on: @@ -164,6 +169,7 @@ services: # published on host loopback 127.0.0.1:8001 only. titiler: image: ghcr.io/developmentseed/titiler:latest + pull_policy: missing container_name: osint-titiler platform: linux/arm64 restart: unless-stopped @@ -183,6 +189,7 @@ services: dockerfile: Dockerfile platforms: ["linux/arm64"] image: localhost/osint-dashboard:latest + pull_policy: never container_name: osint-camera-scraper restart: unless-stopped profiles: ["ingest"] @@ -221,6 +228,7 @@ services: dockerfile: Dockerfile platforms: ["linux/arm64"] image: localhost/osint-news-scraper:latest + pull_policy: never container_name: osint-news-scraper restart: unless-stopped profiles: ["ingest"] @@ -246,6 +254,7 @@ services: dockerfile: Dockerfile platforms: ["linux/arm64"] image: localhost/osint-news-summarizer:latest + pull_policy: never container_name: osint-news-summarizer restart: unless-stopped profiles: ["ingest"] diff --git a/scripts/compose-reup.sh b/scripts/compose-reup.sh index f279981..68023a2 100755 --- a/scripts/compose-reup.sh +++ b/scripts/compose-reup.sh @@ -1,11 +1,15 @@ #!/usr/bin/env bash -# Recreate the OSINT compose stack without container_name collisions. -# Named volumes (osint-dashboard_osint-pgdata, camera-snapshots) are kept — -# never pass -v to `compose down`. +# Recreate selected OSINT compose services WITHOUT bouncing Postgres. # -# Usage: scripts/compose-reup.sh +# The old path was `compose down` + up, which stopped osint-db on every merge +# even when Dockerfile.pg did not change. Name-pinned leftovers are still +# removed, but only for the services we are actually replacing. +# +# Usage: scripts/compose-reup.sh [compose-service ...] +# (default: app ingester camera-service news-scraper news-summarizer) # Env: COMPOSE_PROJECT_NAME (default osint-dashboard) -# COMPOSE_PROFILES (default ingest — nats/ingester/news/cameras) +# COMPOSE_PROFILES (default ingest) +# FORCE_RECREATE_DB=1 also recreate db set -euo pipefail ROOT="$(cd "$(dirname "$0")/.." && pwd)" @@ -14,29 +18,57 @@ cd "$ROOT" export COMPOSE_PROJECT_NAME="${COMPOSE_PROJECT_NAME:-osint-dashboard}" PROFILE="${COMPOSE_PROFILES:-ingest}" -NAMES=( - osint-dashboard - osint-db - osint-nats - osint-ingester - osint-camera-scraper - osint-news-scraper - osint-news-summarizer +DEFAULT_SVCS=(app ingester camera-service news-scraper news-summarizer) +if [ "$#" -gt 0 ]; then + SVCS=("$@") +else + SVCS=("${DEFAULT_SVCS[@]}") +fi + +if [ "${FORCE_RECREATE_DB:-0}" = "1" ]; then + SVCS+=(db) +fi + +# Never recreate db unless it was requested. +FILTERED=() +for svc in "${SVCS[@]}"; do + if [ "$svc" = "db" ] && [ "${FORCE_RECREATE_DB:-0}" != "1" ]; then + echo "compose-reup: skipping db (set FORCE_RECREATE_DB=1 to bounce Postgres)" + continue + fi + FILTERED+=("$svc") +done +SVCS=("${FILTERED[@]}") + +declare -A CONTAINER_NAME=( + [app]=osint-dashboard + [ingester]=osint-ingester + [camera-service]=osint-camera-scraper + [news-scraper]=osint-news-scraper + [news-summarizer]=osint-news-summarizer + [db]=osint-db + [nats]=osint-nats + [titiler]=osint-titiler ) echo "compose-reup: project=${COMPOSE_PROJECT_NAME} profile=${PROFILE} dir=${ROOT}" +echo "compose-reup: recreate=${SVCS[*]:-none}" -# Stop compose-owned containers first. Foreign/name-pinned leftovers survive this. -docker compose --profile "${PROFILE}" down --remove-orphans || true +# Keep data-plane containers running (db / nats / titiler). +docker compose --profile "${PROFILE}" up -d --no-build --no-recreate db nats titiler || true -# Drop any leftover name-pinned containers compose does not own (the -# "Conflict. The container name is already in use" failure mode). -for c in "${NAMES[@]}"; do - if docker inspect "$c" >/dev/null 2>&1; then - echo "compose-reup: removing leftover ${c}" +if [ "${#SVCS[@]}" -eq 0 ]; then + docker compose --profile "${PROFILE}" ps + exit 0 +fi + +for svc in "${SVCS[@]}"; do + c="${CONTAINER_NAME[$svc]:-}" + if [ -n "$c" ] && docker inspect "$c" >/dev/null 2>&1; then + echo "compose-reup: replacing ${c}" docker rm -f "$c" >/dev/null fi done -docker compose --profile "${PROFILE}" up -d --no-build "$@" +docker compose --profile "${PROFILE}" up -d --no-build --no-deps "${SVCS[@]}" docker compose --profile "${PROFILE}" ps diff --git a/tests/test_fire_sources.py b/tests/test_fire_sources.py index 24fcf94..dbff5fc 100644 --- a/tests/test_fire_sources.py +++ b/tests/test_fire_sources.py @@ -131,6 +131,100 @@ def test_ingest_fires_uses_keystore_key(monkeypatch): assert any("VIIRS_NOAA21_NRT" in u for u in captured["urls"]) +def _reset_firms_poll_state(): + from upstream_cache import firms_cache + import fire_sources + + firms_cache.clear() + if hasattr(fire_sources, "_csv_digest"): + fire_sources._csv_digest.clear() + if hasattr(fire_sources, "_seen_ids"): + fire_sources._seen_ids.clear() + + +def _fake_firms_http(monkeypatch, bodies_by_call: list[str] | None = None, body: str = SAMPLE_CSV): + hits = {"n": 0} + + class FakeResp: + def __init__(self, text): + self.text = text + + def raise_for_status(self): + pass + + class FakeClient: + def __init__(self, **kw): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url): + idx = hits["n"] + hits["n"] += 1 + if bodies_by_call is not None: + text = bodies_by_call[min(idx, len(bodies_by_call) - 1)] + else: + text = body + return FakeResp(text) + + monkeypatch.setenv("FIRMS_MAP_KEY", "k" * 32) + monkeypatch.setattr("fire_sources.FIRMS_DATASETS", ["VIIRS_NOAA20_NRT"]) + monkeypatch.setattr("fire_sources.httpx.AsyncClient", FakeClient) + return hits + + +def test_ingest_fires_skips_unchanged_csv(monkeypatch): + """Same FIRMS CSV must not be re-parsed into a 100k-row ON CONFLICT insert.""" + _reset_firms_poll_state() + hits = _fake_firms_http(monkeypatch) + persisted = [] + + async def fake_persist(points): + persisted.append(len(points)) + return len(points) + + monkeypatch.setattr("fire_sources.persist_hotspots", fake_persist) + + assert asyncio.run(ingest_fires()) == 5 + assert persisted == [5] + firms_cache_hits = hits["n"] + persisted.clear() + assert asyncio.run(ingest_fires()) == 0 + assert persisted == [] + # TTL cache may skip HTTP; either way we must not persist again. + assert hits["n"] >= firms_cache_hits + + +def test_ingest_fires_persists_only_new_hotspots(monkeypatch): + """When the CSV grows, persist the delta — not the whole 2-day dump.""" + _reset_firms_poll_state() + extra = ( + SAMPLE_CSV + + "16.00000,-12.00000,340.00,0.40,0.40,2025-06-06,1500,N20,VIIRS,h,2.0NRT,310.00,8.00,D\n" + ) + hits = _fake_firms_http(monkeypatch, bodies_by_call=[SAMPLE_CSV, extra]) + persisted = [] + + async def fake_persist(points): + persisted.append([p["latitude"] for p in points]) + return len(points) + + monkeypatch.setattr("fire_sources.persist_hotspots", fake_persist) + + from upstream_cache import firms_cache + + assert asyncio.run(ingest_fires()) == 5 + firms_cache.clear() # force the next poll to see the grown CSV + persisted.clear() + assert asyncio.run(ingest_fires()) == 1 + assert persisted == [[16.0]] + assert hits["n"] == 2 + + def _async_return(value): async def inner(): return value diff --git a/tests/test_new_sources.py b/tests/test_new_sources.py index 9cd0b85..fcad6ce 100644 --- a/tests/test_new_sources.py +++ b/tests/test_new_sources.py @@ -64,3 +64,65 @@ def test_parse_cisa_kev_emits_cve_url_no_coords(): assert "cisa-kev" in ev["tags"] assert "CVE-2024-1234" in ev["tags"] assert ev["raw"]["cveID"] == "CVE-2024-1234" + + +def test_ingest_cisa_kev_does_not_republish_known_nist_urls(monkeypatch): + """Producer must not push the whole KEV catalog to NATS every cycle.""" + import asyncio + + from sources import ingest_cisa_kev + + payload = { + "vulnerabilities": [ + { + "cveID": "CVE-2024-1111", + "vulnerabilityName": "old", + "dateAdded": "2024-01-01", + "shortDescription": "already in db", + }, + { + "cveID": "CVE-2024-2222", + "vulnerabilityName": "new", + "dateAdded": "2024-06-01", + "shortDescription": "not in db yet", + }, + ] + } + + class FakeResp: + def raise_for_status(self): + pass + + def json(self): + return payload + + class FakeClient: + def __init__(self, **kw): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url): + return FakeResp() + + published: list[str] = [] + + async def fake_publish(subject, event): + published.append(event["url"]) + + known = {"https://nvd.nist.gov/vuln/detail/CVE-2024-1111"} + + async def fake_existing(urls): + return {u for u in urls if u in known} + + monkeypatch.setattr("sources.httpx.AsyncClient", FakeClient) + monkeypatch.setattr("sources.publish_event", fake_publish) + monkeypatch.setattr("sources.existing_event_urls", fake_existing, raising=False) + + n = asyncio.run(ingest_cisa_kev()) + assert n == 1 + assert published == ["https://nvd.nist.gov/vuln/detail/CVE-2024-2222"] diff --git a/tests/test_upstream_cache.py b/tests/test_upstream_cache.py index 249d2f4..8ce8783 100644 --- a/tests/test_upstream_cache.py +++ b/tests/test_upstream_cache.py @@ -22,7 +22,11 @@ def test_firms_and_rss_caches_are_ttlcache(): def test_ingest_fires_hits_http_once_within_ttl(monkeypatch): + from fire_sources import _csv_digest, _seen_ids + firms_cache.clear() + _csv_digest.clear() + _seen_ids.clear() monkeypatch.setenv("FIRMS_MAP_KEY", "k" * 32) monkeypatch.setenv("FIRMS_DATASETS", "VIIRS_NOAA20_NRT") # fire_sources already imported FIRMS_DATASETS — patch the module attr @@ -58,7 +62,7 @@ def test_ingest_fires_hits_http_once_within_ttl(monkeypatch): monkeypatch.setattr("fire_sources.persist_hotspots", fake_publish) assert asyncio.run(ingest_fires()) == 5 - assert asyncio.run(ingest_fires()) == 5 + assert asyncio.run(ingest_fires()) == 0 assert hits["n"] == 1