Needs AISSTREAM_API_KEY. Magenta = military · gold = law/gov · teal = underway. Click for type / dest / IMO.
+ Two feeds: AISStream (US live) and VesselAPI (Hormuz, 5×/day). Paste both keys on the Keys page. Magenta = military · gold = law/gov · teal = underway. Click for type / dest / IMO.
diff --git a/app/vesselapi.py b/app/vesselapi.py
new file mode 100644
index 0000000..2c41612
--- /dev/null
+++ b/app/vesselapi.py
@@ -0,0 +1,445 @@
+"""VesselAPI REST poller — quota-capped AIS for the Middle East (free tier 150 calls/mo).
+
+VesselAPI and AISStream are two independent, first-class vessel providers —
+not a primary/fallback pair. AISStream (WebSocket) owns live US-coast AIS;
+VesselAPI (REST) covers the Strait of Hormuz (default box) where AISStream
+has no coverage. Missing one key never disables the other. This worker polls
+the REST ``GET /v1/location/vessels/bounding-box`` endpoint at most
+``VESSELAPI_MAX_CALLS_PER_DAY`` (default 5) *successful 2xx* calls per UTC day
+and upserts the results into the shared ``vessel_last_known`` store.
+
+Idle (no crash) when VESSELAPI_API_KEY is unset. Never called from the GET
+/api/vessels path — map pans must not hit upstream. One request per poll,
+``pagination.limit=50``, never follow ``nextToken``, never send ``filter.sat``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import calendar
+import logging
+import os
+from datetime import date, datetime, timezone
+
+import httpx
+from sqlalchemy import Column, Date, DateTime, Integer, Table, func, select, text
+
+from config import (
+ OSINT_USER_AGENT,
+ VESSELAPI_API_KEY,
+ VESSELAPI_BBOX,
+ VESSELAPI_INTERVAL,
+ VESSELAPI_MAX_CALLS_PER_DAY,
+)
+from database import async_session, engine, metadata
+from live_layers import to_marker, upsert_vessel
+
+logger = logging.getLogger("osint.vesselapi")
+
+BASE_URL = "https://api.vesselapi.com/v1"
+ENDPOINT = f"{BASE_URL}/location/vessels/bounding-box"
+MAX_SPAN_DEG = 4.0 # |dLat| + |dLon| — VesselAPI 400s above this.
+PAGE_LIMIT = 50 # pagination.limit; never follow nextToken on the free tier.
+
+_client: httpx.AsyncClient | None = None
+_client_lock = asyncio.Lock()
+
+
+# ── Box parsing / span validation ─────────────────────────────────────────
+
+class BboxError(ValueError):
+ """A configured VesselAPI box violates the 4° span rule or is malformed."""
+
+
+def validate_bbox_span(
+ minlat: float, minlon: float, maxlat: float, maxlon: float,
+) -> None:
+ """Reject boxes VesselAPI would 400 on (span > 4°, bad order, bad range)."""
+ if not (-90 <= minlat <= 90 and -90 <= maxlat <= 90
+ and -180 <= minlon <= 180 and -180 <= maxlon <= 180):
+ raise BboxError("coordinates out of range")
+ if minlat >= maxlat or minlon >= maxlon:
+ raise BboxError("bbox must have min < max on both axes")
+ dlat = abs(maxlat - minlat)
+ dlon = abs(maxlon - minlon)
+ if dlat + dlon > MAX_SPAN_DEG:
+ raise BboxError(
+ f"span |dLat|+|dLon| = {dlat + dlon:.2f}° exceeds {MAX_SPAN_DEG}° cap"
+ )
+
+
+def parse_boxes(raw: str) -> list[tuple[float, float, float, float]]:
+ """Env format: ``minlat,minlon,maxlat,maxlon[; ...]`` (lat/lon order)."""
+ out: list[tuple[float, float, float, float]] = []
+ for chunk in (raw or "").split(";"):
+ parts = [p.strip() for p in chunk.split(",") if p.strip()]
+ if len(parts) != 4:
+ continue
+ try:
+ minlat = float(parts[0])
+ minlon = float(parts[1])
+ maxlat = float(parts[2])
+ maxlon = float(parts[3])
+ except ValueError:
+ continue
+ out.append((minlat, minlon, maxlat, maxlon))
+ return out
+
+
+def parse_boxes_validated(raw: str) -> list[tuple[float, float, float, float]]:
+ """Parse boxes, log + skip any that violate the span/order/range rules."""
+ valid: list[tuple[float, float, float, float]] = []
+ for box in parse_boxes(raw):
+ try:
+ validate_bbox_span(*box)
+ valid.append(box)
+ except BboxError as exc:
+ logger.warning("VesselAPI bbox %r skipped: %s", box, exc)
+ return valid
+
+
+# ── Position → marker transform ───────────────────────────────────────────
+
+def _f(value: object) -> float | None:
+ if value is None or value == "":
+ return None
+ try:
+ return float(value)
+ except (TypeError, ValueError):
+ return None
+
+
+def _s(value: object) -> str | None:
+ if value is None:
+ return None
+ text = str(value).strip()
+ return text or None
+
+
+def transform_vesselapi_position(obj: dict | None) -> dict | None:
+ """Map one VesselAPI position object to the shared marker contract.
+
+ Returns None for glitch rows, missing MMSI, or missing coordinates.
+ """
+ if not obj or not isinstance(obj, dict):
+ return None
+ if obj.get("suspected_glitch") is True:
+ return None
+ mmsi = obj.get("mmsi")
+ if mmsi is None:
+ return None
+ lat = _f(obj.get("latitude"))
+ lon = _f(obj.get("longitude"))
+ if lat is None or lon is None:
+ return None
+ mmsi_s = str(mmsi)
+ name = _s(obj.get("vessel_name") or obj.get("name"))
+ heading = _f(obj.get("heading"))
+ if heading is None:
+ heading = _f(obj.get("cog"))
+ sog = _f(obj.get("sog"))
+ extra: dict = {
+ "src": "vesselapi",
+ "mmsi": mmsi_s,
+ "cog": obj.get("cog"),
+ "sog": obj.get("sog"),
+ "navstat": obj.get("nav_status"),
+ }
+ imo = obj.get("imo")
+ if imo:
+ extra["imo"] = imo
+ dest = _s(obj.get("dest") or obj.get("destination"))
+ if dest:
+ extra["dest"] = dest
+ ts = obj.get("timestamp") or obj.get("processed_timestamp")
+ if ts:
+ extra["timestamp"] = ts
+ return to_marker(
+ mmsi_s, lat, lon,
+ heading=heading,
+ speed=sog,
+ label=name or mmsi_s,
+ extra=extra,
+ )
+
+
+def transform_vesselapi_payload(payload: dict | None) -> list[dict]:
+ """Flatten a bounding-box response ``{vessels: [...]}`` to markers."""
+ if not payload or not isinstance(payload, dict):
+ return []
+ rows = payload.get("vessels") or []
+ out = []
+ for row in rows:
+ marker = transform_vesselapi_position(row)
+ if marker:
+ out.append(marker)
+ return out
+
+
+# ── Durable daily quota (Postgres, survives restarts) ─────────────────────
+# Mirrors keystore.api_keys: lazy CREATE TABLE IF NOT EXISTS, no alembic fork.
+
+vesselapi_quota = Table(
+ "vesselapi_quota",
+ metadata,
+ Column("day", Date, primary_key=True),
+ Column("calls", Integer, nullable=False, server_default="0"),
+ Column("remaining", Integer, nullable=True),
+ Column("updated_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
+)
+
+_CREATE_QUOTA_SQL = text(
+ """
+ CREATE TABLE IF NOT EXISTS vesselapi_quota (
+ day DATE PRIMARY KEY,
+ calls INTEGER NOT NULL DEFAULT 0,
+ remaining INTEGER,
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+ )
+ """
+)
+
+_quota_lock = asyncio.Lock()
+_quota_ensured = False
+
+
+async def ensure_quota_table() -> None:
+ global _quota_ensured
+ if _quota_ensured:
+ return
+ async with _quota_lock:
+ if _quota_ensured:
+ return
+ async with engine.begin() as conn:
+ await conn.execute(_CREATE_QUOTA_SQL)
+ _quota_ensured = True
+
+
+class PgQuotaStore:
+ """Postgres-backed daily call counter. Injected for tests."""
+
+ async def calls_today(self, day: date) -> int:
+ await ensure_quota_table()
+ async with async_session() as session:
+ row = (await session.execute(
+ select(vesselapi_quota.c.calls).where(vesselapi_quota.c.day == day)
+ )).scalar()
+ return int(row) if row else 0
+
+ async def remaining_today(self, day: date) -> int | None:
+ await ensure_quota_table()
+ async with async_session() as session:
+ row = (await session.execute(
+ select(vesselapi_quota.c.remaining).where(vesselapi_quota.c.day == day)
+ )).scalar()
+ return int(row) if row is not None else None
+
+ async def bump(self, day: date, remaining: int | None) -> int:
+ await ensure_quota_table()
+ now = datetime.now(timezone.utc)
+ async with async_session() as session:
+ existing = (await session.execute(
+ select(vesselapi_quota.c.calls).where(vesselapi_quota.c.day == day)
+ )).scalar()
+ if existing is None:
+ await session.execute(
+ vesselapi_quota.insert().values(
+ day=day, calls=1, remaining=remaining, updated_at=now,
+ )
+ )
+ else:
+ await session.execute(
+ vesselapi_quota.update()
+ .where(vesselapi_quota.c.day == day)
+ .values(
+ calls=vesselapi_quota.c.calls + 1,
+ remaining=remaining,
+ updated_at=now,
+ )
+ )
+ await session.commit()
+ return (int(existing) if existing else 0) + 1
+
+
+# ── Budget / scheduling (pure, unit-testable) ─────────────────────────────
+
+def days_left_in_month(now: datetime) -> int:
+ """UTC days remaining in the current month, inclusive of today."""
+ _, last = calendar.monthrange(now.year, now.month)
+ return last - now.day + 1
+
+
+def budget_allows(
+ calls_today: int,
+ remaining: int | None,
+ days_left: int,
+ max_per_day: int,
+) -> bool:
+ """True if another poll is permitted today.
+
+ Local hard cap: fewer than ``max_per_day`` successful calls today.
+ Monthly floor: if ``X-RateLimit-Remaining`` is known, keep at least
+ ``max_per_day * days_left`` in reserve for the rest of the month.
+ """
+ if calls_today >= max_per_day:
+ return False
+ if remaining is not None and remaining <= max_per_day * days_left:
+ return False
+ return True
+
+
+def choose_box(
+ boxes: list[tuple[float, float, float, float]],
+ calls_today: int,
+ max_per_day: int,
+) -> int:
+ """Index into ``boxes`` for the next poll.
+
+ Prefer refreshing the first (primary) box rather than spraying one call
+ across every region — round-robin only when the remaining daily budget is
+ enough to cover all boxes.
+ """
+ if len(boxes) <= 1:
+ return 0
+ budget_left = max_per_day - calls_today
+ if budget_left >= len(boxes):
+ return calls_today % len(boxes)
+ return 0
+
+
+# ── HTTP / poll ───────────────────────────────────────────────────────────
+
+async def _resolve_key() -> str:
+ from keystore import get_api_key
+ return (
+ os.getenv("VESSELAPI_API_KEY")
+ or VESSELAPI_API_KEY
+ or (await get_api_key("VESSELAPI_API_KEY"))
+ or ""
+ ).strip()
+
+
+async def _get_client() -> httpx.AsyncClient:
+ global _client
+ if _client is None:
+ async with _client_lock:
+ if _client is None:
+ _client = httpx.AsyncClient(
+ timeout=httpx.Timeout(15.0, connect=5.0),
+ follow_redirects=True,
+ headers={"User-Agent": OSINT_USER_AGENT, "Accept": "application/json"},
+ limits=httpx.Limits(max_connections=1, max_keepalive_connections=1),
+ )
+ return _client
+
+
+async def close_client() -> None:
+ global _client
+ if _client is not None:
+ await _client.aclose()
+ _client = None
+
+
+def _int_header(value: str | None) -> int | None:
+ if value is None:
+ return None
+ try:
+ return int(value)
+ except (TypeError, ValueError):
+ return None
+
+
+async def poll_once(store, boxes: list[tuple[float, float, float, float]], key: str) -> bool:
+ """One quota-checked poll. Returns True if a successful 2xx was made.
+
+ Only successful 2xx responses count against the monthly quota; 4xx/5xx/429
+ are skipped without retry-storming (Retry-After respected by simply
+ sleeping the interval).
+ """
+ now = datetime.now(timezone.utc)
+ today = now.date()
+ calls = await store.calls_today(today)
+ remaining = await store.remaining_today(today)
+ days_left = days_left_in_month(now)
+ if not budget_allows(calls, remaining, days_left, VESSELAPI_MAX_CALLS_PER_DAY):
+ logger.info(
+ "VesselAPI quota reached (calls_today=%d, remaining=%s, days_left=%d) — skip poll",
+ calls, remaining, days_left,
+ )
+ return False
+
+ idx = choose_box(boxes, calls, VESSELAPI_MAX_CALLS_PER_DAY)
+ minlat, minlon, maxlat, maxlon = boxes[idx]
+ client = await _get_client()
+ params = {
+ "filter.latBottom": str(minlat),
+ "filter.latTop": str(maxlat),
+ "filter.lonLeft": str(minlon),
+ "filter.lonRight": str(maxlon),
+ "pagination.limit": str(PAGE_LIMIT),
+ }
+ headers = {"Authorization": f"Bearer {key}"}
+ try:
+ resp = await client.get(ENDPOINT, params=params, headers=headers)
+ except httpx.HTTPError as exc:
+ logger.warning("VesselAPI request failed: %s", exc)
+ return False
+
+ if resp.status_code == 429:
+ logger.warning(
+ "VesselAPI rate-limited (Retry-After=%s) — skip poll",
+ resp.headers.get("Retry-After"),
+ )
+ return False
+ if resp.status_code >= 400:
+ logger.warning("VesselAPI HTTP %d — not counted against quota", resp.status_code)
+ return False
+
+ # 2xx success — counts against the monthly quota.
+ remaining = _int_header(resp.headers.get("X-RateLimit-Remaining"))
+ calls = await store.bump(today, remaining)
+ try:
+ data = resp.json()
+ except ValueError:
+ logger.warning("VesselAPI 2xx with non-JSON body — counted but ignored")
+ return True
+ markers = transform_vesselapi_payload(data)
+ for m in markers:
+ await upsert_vessel(m)
+ logger.info(
+ "VesselAPI poll OK: %d vessels (remaining=%s, calls_today=%d)",
+ len(markers), remaining, calls,
+ )
+ return True
+
+
+# ── Worker loop ───────────────────────────────────────────────────────────
+
+async def run_vesselapi_worker(store: PgQuotaStore | None = None) -> None:
+ """Long-lived poll loop. Idle when the key is unset; never crashes the app."""
+ if store is None:
+ store = PgQuotaStore()
+ boxes = parse_boxes_validated(VESSELAPI_BBOX)
+ if not boxes:
+ logger.warning(
+ "VESSELAPI_BBOX has no valid boxes (span ≤ %.1f°) — poller idle", MAX_SPAN_DEG,
+ )
+ while True:
+ try:
+ if not boxes:
+ await asyncio.sleep(VESSELAPI_INTERVAL)
+ continue
+ key = await _resolve_key()
+ if not key:
+ logger.warning(
+ "VESSELAPI_API_KEY not set — VesselAPI poller idle. "
+ "Create a free key at https://dashboard.vesselapi.com/"
+ )
+ await asyncio.sleep(VESSELAPI_INTERVAL)
+ continue
+ await poll_once(store, boxes, key)
+ except asyncio.CancelledError:
+ raise
+ except Exception: # noqa: BLE001 — keep the loop alive across transient failures
+ logger.exception("VesselAPI poll error")
+ await asyncio.sleep(VESSELAPI_INTERVAL)
diff --git a/docker-compose.yml b/docker-compose.yml
index 00fd71c..f908975 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -97,6 +97,11 @@ services:
AISSTREAM_API_KEY: ${AISSTREAM_API_KEY:-}
AISSTREAM_BBOX: ${AISSTREAM_BBOX:-24,-125,50,-66}
AISSTREAM_IN_INGEST: ${AISSTREAM_IN_INGEST:-0}
+ VESSELAPI_API_KEY: ${VESSELAPI_API_KEY:-}
+ VESSELAPI_BBOX: ${VESSELAPI_BBOX:-25.5,55.4,27.3,57.2}
+ VESSELAPI_INTERVAL: ${VESSELAPI_INTERVAL:-17280}
+ VESSELAPI_MAX_CALLS_PER_DAY: ${VESSELAPI_MAX_CALLS_PER_DAY:-5}
+ VESSELAPI_IN_INGEST: ${VESSELAPI_IN_INGEST:-0}
command: ["python", "app/run_ingester.py"]
entrypoint: ["python", "app/run_ingester.py"]
@@ -131,6 +136,11 @@ services:
AISSTREAM_API_KEY: ${AISSTREAM_API_KEY:-}
AISSTREAM_BBOX: ${AISSTREAM_BBOX:-24,-125,50,-66}
AISSTREAM_IN_APP: ${AISSTREAM_IN_APP:-1}
+ VESSELAPI_API_KEY: ${VESSELAPI_API_KEY:-}
+ VESSELAPI_BBOX: ${VESSELAPI_BBOX:-25.5,55.4,27.3,57.2}
+ VESSELAPI_INTERVAL: ${VESSELAPI_INTERVAL:-17280}
+ VESSELAPI_MAX_CALLS_PER_DAY: ${VESSELAPI_MAX_CALLS_PER_DAY:-5}
+ VESSELAPI_IN_APP: ${VESSELAPI_IN_APP:-1}
ports:
- "127.0.0.1:8000:8000"
deploy:
diff --git a/tests/test_vesselapi.py b/tests/test_vesselapi.py
new file mode 100644
index 0000000..292bed4
--- /dev/null
+++ b/tests/test_vesselapi.py
@@ -0,0 +1,351 @@
+"""Unit tests for the VesselAPI poller (no network, no DB).
+
+Covers box span validation, position → marker mapping, glitch skipping,
+and the daily-quota gate (6th 2xx attempt skipped). ``poll_once`` is driven
+with an in-memory fake quota store + fake HTTP client; ``upsert_vessel``'s
+DB/WS side effects are monkeypatched to no-ops so markers can be asserted in
+``vessel_last_known``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timezone
+
+import pytest
+
+import vesselapi
+from vesselapi import (
+ BboxError,
+ budget_allows,
+ choose_box,
+ days_left_in_month,
+ parse_boxes,
+ parse_boxes_validated,
+ poll_once,
+ transform_vesselapi_payload,
+ transform_vesselapi_position,
+ validate_bbox_span,
+)
+from live_layers import vessel_last_known
+
+
+# ── Fake quota store (in-memory, injected) ────────────────────────────────
+
+class FakeStore:
+ def __init__(self, calls: int = 0, remaining: int | None = None):
+ self.calls = calls
+ self.remaining = remaining
+ self.bumps = 0
+
+ async def calls_today(self, day):
+ return self.calls
+
+ async def remaining_today(self, day):
+ return self.remaining
+
+ async def bump(self, day, remaining):
+ self.calls += 1
+ self.remaining = remaining
+ self.bumps += 1
+ return self.calls
+
+
+# ── Fake HTTP client ───────────────────────────────────────────────────────
+
+class FakeResp:
+ def __init__(self, status: int = 200, headers: dict | None = None, data: dict | None = None):
+ self.status_code = status
+ self.headers = headers or {}
+ self._data = data or {}
+
+ def json(self):
+ return self._data
+
+
+class FakeClient:
+ def __init__(self, *responses: FakeResp):
+ self.responses = list(responses)
+ self.calls: list[tuple] = []
+
+ async def get(self, url, params=None, headers=None):
+ self.calls.append((url, params, headers))
+ return self.responses.pop(0)
+
+ async def aclose(self):
+ pass
+
+
+def _sample_payload() -> dict:
+ return {
+ "vessels": [
+ {
+ "mmsi": 422050100,
+ "imo": 9321483,
+ "vessel_name": "HORMUZ STAR",
+ "latitude": 26.5,
+ "longitude": 56.3,
+ "cog": 88.0,
+ "sog": 12.4,
+ "heading": 90,
+ "nav_status": 0,
+ "timestamp": "2026-08-29T12:00:00Z",
+ "suspected_glitch": False,
+ },
+ {
+ "mmsi": 422050101,
+ "vessel_name": "GLITCHY",
+ "latitude": 26.6,
+ "longitude": 56.4,
+ "cog": 45.0,
+ "sog": 5.0,
+ "heading": 45,
+ "nav_status": 5,
+ "suspected_glitch": True,
+ },
+ {"mmsi": 422050102, "vessel_name": "NOFIX"}, # no coords → drop
+ ],
+ "nextToken": "deadbeef",
+ }
+
+
+# ── Box span / parsing ─────────────────────────────────────────────────────
+
+def test_validate_bbox_span_accepts_hormuz():
+ validate_bbox_span(25.5, 55.4, 27.3, 57.2) # span 3.6 — no raise
+
+
+def test_validate_bbox_span_rejects_conus():
+ with pytest.raises(BboxError):
+ validate_bbox_span(24.0, -125.0, 50.0, -66.0) # span 85
+
+
+def test_validate_bbox_span_rejects_marine_regions_gazetteer_box():
+ # 25.2732–27.3713 N, 55.1647–57.3419 E → span 4.28 > 4.0.
+ with pytest.raises(BboxError):
+ validate_bbox_span(25.2732, 55.1647, 27.3713, 57.3419)
+
+
+def test_validate_bbox_span_rejects_inverted_axes():
+ with pytest.raises(BboxError):
+ validate_bbox_span(27.0, 55.0, 25.0, 57.0)
+
+
+def test_parse_boxes_semicolon_and_skip_malformed():
+ boxes = parse_boxes("25.5,55.4,27.3,57.2; 10,20,11,21; garbage")
+ assert boxes == [(25.5, 55.4, 27.3, 57.2), (10.0, 20.0, 11.0, 21.0)]
+
+
+def test_parse_boxes_validated_skips_over_span():
+ # Second box is CONUS-sized → dropped, first kept.
+ valid = parse_boxes_validated("25.5,55.4,27.3,57.2;24,-125,50,-66")
+ assert valid == [(25.5, 55.4, 27.3, 57.2)]
+
+
+# ── Position → marker mapping ─────────────────────────────────────────────
+
+def test_transform_position_maps_shared_marker_contract():
+ m = transform_vesselapi_position({
+ "mmsi": 422050100, "imo": 9321483, "vessel_name": "HORMUZ STAR",
+ "latitude": 26.5, "longitude": 56.3, "heading": 90, "cog": 88.0,
+ "sog": 12.4, "nav_status": 0, "timestamp": "2026-08-29T12:00:00Z",
+ "suspected_glitch": False,
+ })
+ assert m is not None
+ assert m["id"] == "422050100"
+ assert m["lat"] == 26.5
+ assert m["lon"] == 56.3
+ assert m["label"] == "HORMUZ STAR"
+ assert m["heading"] == 90
+ assert m["speed"] == 12.4
+ assert m["extra"]["src"] == "vesselapi"
+ assert m["extra"]["mmsi"] == "422050100"
+ assert m["extra"]["imo"] == 9321483
+ assert m["extra"]["navstat"] == 0
+ assert m["extra"]["cog"] == 88.0
+ assert m["extra"]["sog"] == 12.4
+ assert m["extra"]["timestamp"] == "2026-08-29T12:00:00Z"
+
+
+def test_transform_position_heading_falls_back_to_cog():
+ m = transform_vesselapi_position({
+ "mmsi": 123456789, "vessel_name": "X", "latitude": 1.0, "longitude": 2.0,
+ "heading": None, "cog": 123.4, "sog": 5.0,
+ })
+ assert m["heading"] == 123.4
+
+
+def test_transform_position_skips_glitch():
+ assert transform_vesselapi_position({
+ "mmsi": 123456789, "latitude": 1.0, "longitude": 2.0,
+ "suspected_glitch": True,
+ }) is None
+
+
+def test_transform_position_skips_missing_coords():
+ assert transform_vesselapi_position({"mmsi": 123456789, "vessel_name": "NOFIX"}) is None
+
+
+def test_transform_payload_skips_glitch_and_nofix_rows():
+ rows = transform_vesselapi_payload(_sample_payload())
+ assert [r["id"] for r in rows] == ["422050100"]
+
+
+# ── Quota budget / scheduling ─────────────────────────────────────────────
+
+def test_days_left_in_month():
+ assert days_left_in_month(datetime(2026, 8, 29, tzinfo=timezone.utc)) == 3
+
+
+def test_budget_allows_local_daily_cap():
+ # 5 calls already made → 6th is blocked regardless of remaining.
+ assert budget_allows(5, remaining=1000, days_left=3, max_per_day=5) is False
+
+
+def test_budget_allows_monthly_floor():
+ # remaining 14 ≤ 5*3=15 → skip; 16 > 15 → allow.
+ assert budget_allows(2, remaining=14, days_left=3, max_per_day=5) is False
+ assert budget_allows(2, remaining=16, days_left=3, max_per_day=5) is True
+
+
+def test_budget_allows_unknown_remaining():
+ assert budget_allows(2, remaining=None, days_left=3, max_per_day=5) is True
+
+
+def test_choose_box_prefers_primary_when_budget_tight():
+ boxes = [(1, 1, 2, 2), (3, 3, 4, 4), (5, 5, 6, 6)]
+ # 4 calls made, 1 left → always box 0.
+ assert choose_box(boxes, 4, max_per_day=5) == 0
+
+
+def test_choose_box_round_robins_when_budget_covers_all():
+ boxes = [(1, 1, 2, 2), (3, 3, 4, 4), (5, 5, 6, 6)]
+ # 0 calls made, 5 left ≥ 3 boxes → round-robin.
+ assert choose_box(boxes, 0, max_per_day=5) == 0
+ assert choose_box(boxes, 1, max_per_day=5) == 1
+ assert choose_box(boxes, 2, max_per_day=5) == 2
+
+
+# ── poll_once integration (fake store + fake client) ─────────────────────
+
+def _patch_side_effects(monkeypatch):
+ async def _noop(*a, **k):
+ return None
+
+ monkeypatch.setattr("tracks.record_position", _noop)
+ monkeypatch.setattr("geofence.record_and_notify", _noop)
+
+
+def test_poll_once_lands_markers_in_vessel_last_known(monkeypatch):
+ _patch_side_effects(monkeypatch)
+ vessel_last_known.clear()
+ client = FakeClient(
+ FakeResp(200, {"X-RateLimit-Remaining": "140"}, _sample_payload()),
+ )
+ monkeypatch.setattr(vesselapi, "_get_client", _make_get_client(client))
+ store = FakeStore()
+
+ ok = asyncio.run(poll_once(store, [(25.5, 55.4, 27.3, 57.2)], "test-key"))
+
+ assert ok is True
+ assert store.calls == 1
+ assert "422050100" in vessel_last_known
+ assert vessel_last_known["422050100"]["extra"]["src"] == "vesselapi"
+ assert "422050101" not in vessel_last_known # glitch skipped
+ # One HTTP call, bounding-box params, no sat / no nextToken follow.
+ assert len(client.calls) == 1
+ _url, params, headers = client.calls[0]
+ assert params["filter.latBottom"] == "25.5"
+ assert params["filter.latTop"] == "27.3"
+ assert params["filter.lonLeft"] == "55.4"
+ assert params["filter.lonRight"] == "57.2"
+ assert params["pagination.limit"] == "50"
+ assert "sat" not in params
+ assert headers["Authorization"] == "Bearer test-key"
+
+
+def test_poll_once_sixth_2xx_is_skipped_with_zero_http(monkeypatch):
+ client = FakeClient()
+ monkeypatch.setattr(vesselapi, "_get_client", _make_get_client(client))
+ # 5 successful calls already today → 6th poll makes no HTTP request.
+ store = FakeStore(calls=5, remaining=1000)
+
+ ok = asyncio.run(poll_once(store, [(25.5, 55.4, 27.3, 57.2)], "test-key"))
+
+ assert ok is False
+ assert store.bumps == 0
+ assert client.calls == []
+
+
+def test_poll_once_4xx_not_counted_and_no_upsert(monkeypatch):
+ _patch_side_effects(monkeypatch)
+ vessel_last_known.clear()
+ client = FakeClient(FakeResp(400, {}, {"error": {}}))
+ monkeypatch.setattr(vesselapi, "_get_client", _make_get_client(client))
+ store = FakeStore()
+
+ ok = asyncio.run(poll_once(store, [(25.5, 55.4, 27.3, 57.2)], "test-key"))
+
+ assert ok is False
+ assert store.bumps == 0 # 4xx does not count against quota
+ assert vessel_last_known == {}
+
+
+def _make_get_client(client):
+ async def _get_client():
+ return client
+
+ return _get_client
+
+
+# ── Independent-provider idle behaviour ───────────────────────────────────
+# VesselAPI must never be gated on AISStream (or vice versa): a missing key on
+# one provider leaves the other running. ``_StopLoop`` is a BaseException so
+# the worker's ``except Exception`` handler can't swallow it — the first
+# ``asyncio.sleep`` aborts the loop after exactly one decision.
+
+class _StopLoop(BaseException):
+ pass
+
+
+async def _stop_sleep(*_a, **_k):
+ raise _StopLoop()
+
+
+async def _resolve(value: str):
+ return value
+
+
+def test_worker_idles_without_vesselapi_key_even_if_aisstream_set(monkeypatch):
+ # AISStream key present, VesselAPI key absent -> no poll, still idles.
+ monkeypatch.setenv("AISSTREAM_API_KEY", "unused-aisstream-key")
+ monkeypatch.setenv("VESSELAPI_API_KEY", "")
+ monkeypatch.setattr(vesselapi, "_resolve_key", lambda: _resolve(""))
+ monkeypatch.setattr(vesselapi.asyncio, "sleep", _stop_sleep)
+ poll_calls: list = []
+ async def _spy_poll(store, boxes, key):
+ poll_calls.append(key)
+ return True
+ monkeypatch.setattr(vesselapi, "poll_once", _spy_poll)
+
+ with pytest.raises(_StopLoop):
+ asyncio.run(vesselapi.run_vesselapi_worker(FakeStore()))
+
+ assert poll_calls == []
+
+
+def test_worker_polls_with_vesselapi_key_even_if_aisstream_unset(monkeypatch):
+ # AISStream key absent, VesselAPI key present -> still polls exactly once.
+ monkeypatch.setenv("AISSTREAM_API_KEY", "")
+ monkeypatch.setattr(vesselapi, "_resolve_key", lambda: _resolve("vesselapi-key"))
+ monkeypatch.setattr(vesselapi.asyncio, "sleep", _stop_sleep)
+ poll_calls: list = []
+ async def _spy_poll(store, boxes, key):
+ poll_calls.append(key)
+ return True
+ monkeypatch.setattr(vesselapi, "poll_once", _spy_poll)
+
+ with pytest.raises(_StopLoop):
+ asyncio.run(vesselapi.run_vesselapi_worker(FakeStore()))
+
+ assert poll_calls == ["vesselapi-key"]