From 9213b26f7869dab5bc4e7ea35802277a84e11422 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 05:07:59 +0000 Subject: [PATCH] Add read-only Forge API and Live Map /api/events feed A stdlib WSGI app serves objects, history, links, neighborhoods and search from the ontology store, plus /api/events in the envelope Live Map already reads. A shared conformance fixture pins the feed shape; tests reproduce it on SQLite and PostgreSQL. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01GcagdBzPRAWDhhc9Yipm4d --- .github/workflows/ci.yml | 4 +- README.md | 7 + docs/API.md | 50 ++ docs/ONTOLOGY.md | 3 +- osint_forge/api/__init__.py | 6 + osint_forge/api/app.py | 171 +++++ osint_forge/api/projection.py | 285 ++++++++ pyproject.toml | 1 + tests/fixtures/contracts/live_map_events.json | 686 ++++++++++++++++++ tests/test_api.py | 229 ++++++ 10 files changed, 1439 insertions(+), 3 deletions(-) create mode 100644 docs/API.md create mode 100644 osint_forge/api/__init__.py create mode 100644 osint_forge/api/app.py create mode 100644 osint_forge/api/projection.py create mode 100644 tests/fixtures/contracts/live_map_events.json create mode 100644 tests/test_api.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 87652f4..86c2f9e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,5 +44,5 @@ jobs: python-version: "3.12" - name: Install PostgreSQL driver run: python -m pip install "psycopg[binary]>=3.1" - - name: Run ontology tests against PostgreSQL - run: python -m unittest tests.test_ontology -v + - name: Run ontology and feed tests against PostgreSQL + run: python -m unittest tests.test_ontology tests.test_api -v diff --git a/README.md b/README.md index f5a32c8..19b47e9 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,9 @@ osint_forge/ | +-- model.py | +-- store.py | +-- sql/ ++-- api/ +| +-- app.py +| +-- projection.py +-- dashboard/ | +-- events_api.py | +-- index.html @@ -74,6 +77,10 @@ osint_forge/ `osint_forge.ontology` holds typed objects (events, places, organizations, vessels, facilities, documents, claims, assessments) and typed, time-bounded links that must cite evidence. It stores full revision history and an audit log in SQLite or PostgreSQL. This is the record store Live Map will read from. See `docs/ONTOLOGY.md`. +## Forge API + +`osint-forge-api --db forge.db` serves the store read-only over HTTP: `/objects`, `/events`, `/claims`, `/assessments`, `/search`, object history, links and neighborhoods, and `/api/events`, a public feed in the shape Live Map already reads. A shared conformance fixture keeps the feed and Live Map's contract in step. See `docs/API.md`. + ## Agentic Intelligence Layers - `osint_forge.ingestion.live_adapters` defines `IngestionAdapter`, `NormalizedOSINTEvent`, health tracking, raw-payload storage, deduplication, rate limiting, retries, and source-specific adapters for GDELT, Reddit, Telegram, and X. diff --git a/docs/API.md b/docs/API.md new file mode 100644 index 0000000..ae47735 --- /dev/null +++ b/docs/API.md @@ -0,0 +1,50 @@ +# Forge API + +`osint_forge.api` serves the ontology store over HTTP. It is read-only: writes go through the Python store until authentication exists. It is a plain WSGI app, so it needs no extra packages. + +```bash +osint-forge-api --db forge.db # SQLite +osint-forge-api --database-url postgresql://... # PostgreSQL +``` + +It listens on `127.0.0.1:8765` by default. For production, mount `ForgeApi(store)` in any WSGI server. + +## Routes + +All routes are `GET`, return JSON, and send `Access-Control-Allow-Origin: *`. `OPTIONS` returns `204`. Errors are `{"error", "message"}` with status `400`, `404`, `405` or `500`, and never include stack traces. + +| Route | Returns | +|---|---| +| `/health` | `{"status": "ok"}` | +| `/objects?type=&q=&bbox=&limit=` | Objects filtered by type, label text and `bbox=minLon,minLat,maxLon,maxLat`. `limit` is 1 to 500, default 100. | +| `/events`, `/claims`, `/assessments` | The same list, fixed to one type. | +| `/search?q=&type=` | Label search. `q` is required. | +| `/objects/{id}` | One object. | +| `/objects/{id}/history` | Every revision. | +| `/objects/{id}/links?direction=&type=` | Links in, out or both, optionally limited to a comma-separated list of link types. | +| `/objects/{id}/neighborhood?depth=&max=` | Breadth-first graph for link analysis. `depth` is 0 to 4. | +| `/api/events?hours=&recordKind=&verification=&domain=&country=` | The Live Map feed, described below. | + +Served objects drop provenance `metadata`, which can carry extractor detail. Source, locator, observation time and content hash stay. + +## Live Map feed + +`/api/events` returns the envelope Live Map documents in its `docs/API_CONTRACTS.md`: `events` (legacy, epoch-millisecond clocks), `canonicalEvents` (ISO clocks), `generatedAt`, `filters`, `sources`, `sourceStatus`, the source registries, `providerResults`, `systemStatus`, `mode`, `errors` and `changeCursor`. Live Map can point at Forge without changing its client. + +How an `event` object becomes a feed event: + +- Only `event` objects with a `valid_from` inside the window are included. The window is `hours`, clamped to 24 to 720, default 168. +- An event with `properties.public` set to `false` is never published. +- `label` becomes `title`, `valid_from` becomes `startedAt`, and `lat`/`lon` become the coordinates. `(0, 0)` stays a mapped point. +- These properties map directly: `provider`, `source_name`, `source_url`, `source_type`, `source_tier`, `category`, `domain`, `description`, `country_code`, `country_name`, `region`, `location_name`, `severity` (0 to 100), `status`, `verification_status`, `source_published_at`, `last_verified_at`, `updated_at`, `tags`, `strategic_areas`, `affected_chokepoints` and `impact_types`. +- `confidence` is stored as 0 to 1, like link confidence, and published as 0 to 100. A missing value stays `null`, and zero stays zero. +- `source_url` is dropped unless it is absolute HTTP(S) with no embedded credentials. +- `ingestedAt` is the latest provenance retrieval time. It describes processing, never when the event happened. +- `independentSourceCount` counts distinct provenance sources on the event. Entity resolution will later group re-acquisitions of one origin. +- `metadata.forge` carries the object id and content hash so Live Map can link back. + +Invalid filters are ignored and reported in `warnings`, as Live Map does. + +## Conformance fixture + +`tests/fixtures/contracts/live_map_events.json` holds fixture objects, a fixed `now`, the exact expected envelope, and a type schema for the envelope, canonical events and legacy events. Forge's tests reproduce the expected envelope on SQLite and PostgreSQL. Live Map can validate its own `/api/events` responses against the same `schema` block so the two sides can't drift apart. The fixture is test data, not live records. diff --git a/docs/ONTOLOGY.md b/docs/ONTOLOGY.md index bc95be7..f1201b3 100644 --- a/docs/ONTOLOGY.md +++ b/docs/ONTOLOGY.md @@ -50,8 +50,9 @@ The PostgreSQL schema lives in `osint_forge/ontology/sql/postgres/` and is appli Set `FORGE_TEST_DATABASE_URL` to run the PostgreSQL tests locally. CI runs them against a PostgreSQL 16 service. +The read-only HTTP API and the Live Map feed are described in [API.md](API.md). + ## Not yet -- An HTTP API and the public projection feed Live Map will consume. - Moving the existing `NormalizedOSINTEvent` and memory-graph data onto these objects. - Entity resolution that proposes `same_as` links. diff --git a/osint_forge/api/__init__.py b/osint_forge/api/__init__.py new file mode 100644 index 0000000..926809e --- /dev/null +++ b/osint_forge/api/__init__.py @@ -0,0 +1,6 @@ +"""Read-only HTTP API and the public projection feed Live Map consumes.""" + +from osint_forge.api.app import ForgeApi, public_object +from osint_forge.api.projection import canonical_event, events_feed, legacy_event + +__all__ = ["ForgeApi", "canonical_event", "events_feed", "legacy_event", "public_object"] diff --git a/osint_forge/api/app.py b/osint_forge/api/app.py new file mode 100644 index 0000000..567ad9d --- /dev/null +++ b/osint_forge/api/app.py @@ -0,0 +1,171 @@ +"""Read-only HTTP API over the ontology store. + +A plain WSGI application, so it needs no extra packages and runs under any WSGI +server (``wsgiref`` for local use, gunicorn or uWSGI in production). Writes go +through the Python store, not HTTP, until authentication exists. +""" + +from __future__ import annotations + +import argparse +import json +from typing import Any, Callable, Iterable +from urllib.parse import parse_qs, unquote + +from osint_forge.api.projection import events_feed +from osint_forge.ontology import OBJECT_TYPES, OntologyError, OntologyObject, OntologyStore + +JSON_HEADERS = [ + ("Content-Type", "application/json; charset=utf-8"), + ("Access-Control-Allow-Origin", "*"), + ("X-Content-Type-Options", "nosniff"), +] +MAX_LIMIT = 500 +TYPE_ROUTES = {"/events": "event", "/claims": "claim", "/assessments": "assessment"} +_STATUS_TEXT = {200: "OK", 204: "No Content", 400: "Bad Request", 404: "Not Found", 405: "Method Not Allowed", 500: "Internal Server Error"} + + +class ApiError(Exception): + def __init__(self, status: int, error: str, message: str) -> None: + super().__init__(message) + self.status, self.error, self.message = status, error, message + + +def public_object(obj: OntologyObject) -> dict[str, Any]: + """Object as served over HTTP. Provenance metadata can hold extractor detail, so it is dropped.""" + body = obj.to_dict() + body["provenance"] = [{k: v for k, v in ref.items() if k != "metadata"} for ref in body["provenance"]] + return body + + +def _int(query: dict[str, str], key: str, default: int, low: int, high: int) -> int: + raw = query.get(key) + if raw in (None, ""): + return default + try: + return max(low, min(high, int(raw))) + except ValueError: + raise ApiError(400, "invalid-parameter", f"{key} must be an integer") from None + + +def _bbox(query: dict[str, str]) -> tuple[float, float, float, float] | None: + raw = query.get("bbox") + if not raw: + return None + try: + min_lon, min_lat, max_lon, max_lat = (float(part) for part in raw.split(",")) + except ValueError: + raise ApiError(400, "invalid-parameter", "bbox must be minLon,minLat,maxLon,maxLat") from None + if not (-180 <= min_lon <= max_lon <= 180 and -90 <= min_lat <= max_lat <= 90): + raise ApiError(400, "invalid-parameter", "bbox is out of range or inverted") + return min_lon, min_lat, max_lon, max_lat + + +def _object_type(query: dict[str, str]) -> str | None: + value = query.get("type") or None + if value and value not in OBJECT_TYPES: + raise ApiError(400, "invalid-parameter", f"unknown object type: {value}") + return value + + +class ForgeApi: + """WSGI callable: ``ForgeApi(store)``.""" + + def __init__(self, store: OntologyStore) -> None: + self.store = store + + def __call__(self, environ: dict[str, Any], start_response: Callable[..., Any]) -> Iterable[bytes]: + method = environ.get("REQUEST_METHOD", "GET") + path = environ.get("PATH_INFO") or "/" + query = {k: v[-1] for k, v in parse_qs(environ.get("QUERY_STRING", ""), keep_blank_values=True).items()} + if method == "OPTIONS": + return self._respond(start_response, 204, None, [("Access-Control-Allow-Methods", "GET, OPTIONS")]) + try: + if method != "GET": + raise ApiError(405, "method-not-allowed", "This API is read-only; use GET.") + status, body = 200, self.route(path, query) + except ApiError as exc: + status, body = exc.status, {"error": exc.error, "message": exc.message} + except OntologyError as exc: + status, body = 400, {"error": "invalid-request", "message": str(exc)} + except Exception: # noqa: BLE001 - never leak stack traces + status, body = 500, {"error": "internal-error", "message": "The request could not be completed."} + extra = [("Cache-Control", "public, max-age=30")] if path == "/api/events" and status == 200 else [("Cache-Control", "no-store")] + return self._respond(start_response, status, body, extra) + + def route(self, path: str, query: dict[str, str]) -> dict[str, Any]: + if path == "/health": + return {"status": "ok"} + if path == "/api/events": + return events_feed(self.store, query) + if path == "/objects": + return self._list(query, _object_type(query)) + if path in TYPE_ROUTES: + return self._list(query, TYPE_ROUTES[path]) + if path == "/search": + if not (query.get("q") or "").strip(): + raise ApiError(400, "invalid-parameter", "q is required") + return self._list(query, _object_type(query)) + if path.startswith("/objects/"): + return self._object_route(path[len("/objects/"):], query) + raise ApiError(404, "not-found", "No such route.") + + def _list(self, query: dict[str, str], object_type: str | None) -> dict[str, Any]: + limit = _int(query, "limit", 100, 1, MAX_LIMIT) + objects = self.store.list_objects( + object_type=object_type, + text=(query.get("q") or "").strip() or None, + bbox=_bbox(query), + limit=limit, + ) + return {"objects": [public_object(o) for o in objects], "count": len(objects), "limit": limit} + + def _object_route(self, rest: str, query: dict[str, str]) -> dict[str, Any]: + object_id, _, view = unquote(rest).partition("/") + obj = self.store.get_object(object_id) + if obj is None: + raise ApiError(404, "not-found", f"unknown object: {object_id}") + if view == "": + return {"object": public_object(obj)} + if view == "history": + return {"id": object_id, "revisions": self.store.object_history(object_id)} + if view == "links": + types = [t for t in (query.get("type") or "").split(",") if t] + links = self.store.links_for(object_id, direction=query.get("direction") or "both", link_types=types) + return {"id": object_id, "links": links, "count": len(links)} + if view == "neighborhood": + return self.store.neighborhood( + object_id, + depth=_int(query, "depth", 1, 0, 4), + max_objects=_int(query, "max", 200, 1, MAX_LIMIT), + ) + raise ApiError(404, "not-found", "No such route.") + + @staticmethod + def _respond(start_response: Callable[..., Any], status: int, body: Any, extra: list[tuple[str, str]]) -> list[bytes]: + start_response(f"{status} {_STATUS_TEXT[status]}", JSON_HEADERS + extra) + if body is None: + return [b""] + return [json.dumps(body, ensure_ascii=False, default=str).encode("utf-8")] + + +def main(argv: list[str] | None = None) -> None: + from wsgiref.simple_server import make_server + + from osint_forge.ontology import PostgresOntologyStore, SQLiteOntologyStore + + parser = argparse.ArgumentParser(description="Serve the read-only Forge API.") + target = parser.add_mutually_exclusive_group(required=True) + target.add_argument("--db", help="SQLite ontology database path") + target.add_argument("--database-url", help="PostgreSQL connection string") + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8765) + args = parser.parse_args(argv) + store = SQLiteOntologyStore(args.db) if args.db else PostgresOntologyStore(args.database_url) + with make_server(args.host, args.port, ForgeApi(store)) as server: + print(f"Forge API on http://{args.host}:{args.port}") + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/osint_forge/api/projection.py b/osint_forge/api/projection.py new file mode 100644 index 0000000..c4a212d --- /dev/null +++ b/osint_forge/api/projection.py @@ -0,0 +1,285 @@ +"""Public projection of Forge ontology events in Live Map's ``/api/events`` shape. + +Live Map documents the envelope in ``docs/API_CONTRACTS.md`` and builds events +with ``src/events/normalized-event.js``. This module produces the same envelope +from ``event`` objects so Live Map can read Forge instead of its own providers. +``tests/fixtures/contracts/live_map_events.json`` is the shared conformance +fixture both sides check against. + +Only ``event`` objects are projected, and an event whose ``properties.public`` +is ``False`` is left out. Raw payloads and provenance metadata never appear. +""" + +from __future__ import annotations + +import re +from datetime import datetime, timedelta, timezone +from typing import Any +from urllib.parse import urlsplit + +from osint_forge.ontology import OntologyObject, OntologyStore + +FEED_PROVIDER = "osint-forge" +FEED_MODE = "forge-ontology" +HOURS_DEFAULT, HOURS_MIN, HOURS_MAX = 168, 24, 720 +FEED_LIMIT = 1000 + +RECORD_KINDS = ("event", "observation", "discovery-lead", "moving-object") +VALID_STATUS = {"active", "monitoring", "resolved", "unknown"} +_COUNTRY_RE = re.compile(r"^[A-Z]{2,3}$") +_LIST_FIELDS = ("tags", "strategicAreas", "affectedChokepoints", "impactTypes") +_PROPERTY_LIST_KEYS = { + "tags": "tags", + "strategicAreas": "strategic_areas", + "affectedChokepoints": "affected_chokepoints", + "impactTypes": "impact_types", +} + + +# -- value semantics (mirrors Live Map's src/events/value-semantics.js) ----- +def _finite(value: Any) -> float | None: + if isinstance(value, bool) or not isinstance(value, (int, float, str)): + return None + if isinstance(value, str) and not value.strip(): + return None + try: + number = float(value) + except ValueError: + return None + return number if number == number and abs(number) != float("inf") else None + + +def _parse_time(value: Any) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + return None + text = value.strip().replace("Z", "+00:00") + try: + parsed = datetime.fromisoformat(text if "T" in text or len(text) > 10 else text + "T00:00:00+00:00") + except ValueError: + return None + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def _iso(value: Any) -> str | None: + parsed = _parse_time(value) + return parsed.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") if parsed else None + + +def _ms(value: Any) -> int | None: + parsed = _parse_time(value) + return int(parsed.timestamp() * 1000) if parsed else None + + +def safe_source_url(value: Any) -> str | None: + """Absolute HTTP(S) without embedded credentials, else None.""" + if not isinstance(value, str) or not re.match(r"^https?://", value.strip(), re.I): + return None + try: + parts = urlsplit(value.strip()) + except ValueError: + return None + if parts.scheme.lower() not in {"http", "https"} or parts.username or parts.password or not parts.hostname: + return None + return value.strip() + + +def severity_label(score: float) -> str: + if score >= 85: + return "critical" + if score >= 65: + return "high" + if score >= 35: + return "moderate" + return "low" + + +# -- event projection ---------------------------------------------------- +def canonical_event(obj: OntologyObject) -> dict[str, Any]: + """Project one ``event`` object to Live Map's canonical normalized event. + + Forge stores ``confidence`` in 0..1 like link confidence; Live Map expects 0..100. + """ + props = obj.properties + provenance = obj.provenance + first_source = provenance[0].source if provenance else FEED_PROVIDER + severity = _finite(props.get("severity")) + severity = 25.0 if severity is None else max(0.0, min(100.0, severity)) + confidence = _finite(props.get("confidence")) + geographic = obj.lat is not None + ingested = max((ref.observedAt for ref in provenance if _parse_time(ref.observedAt)), default=None, key=_ms) + source_name = props.get("source_name") or first_source + event = { + "id": obj.id, + "provider": props.get("provider") or first_source, + "providerEventId": obj.id.split(":", 1)[1], + "recordKind": props.get("record_kind") if props.get("record_kind") in RECORD_KINDS else "event", + "title": obj.label, + "description": props.get("description"), + "category": str(props.get("category") or "other"), + "domain": props.get("domain"), + "latitude": obj.lat, + "longitude": obj.lon, + "geographic": geographic, + "mapDisplayStatus": "mapped" if geographic else "not-mapped", + "nonGeographicReason": None if geographic else "Geography not supplied", + "countryCode": props.get("country_code"), + "countryName": props.get("country_name"), + "region": props.get("region"), + "locationName": props.get("location_name"), + "startedAt": _iso(obj.valid_from), + "updatedAt": _iso(props.get("updated_at")), + "ingestedAt": _iso(ingested), + "severity": severity, + "severityLabel": severity_label(severity), + "confidence": None if confidence is None else max(0.0, min(100.0, confidence * 100)), + "independentSourceCount": len({ref.source for ref in provenance}), + "verificationStatus": props.get("verification_status"), + "status": props.get("status") if props.get("status") in VALID_STATUS else "unknown", + "sourceName": source_name, + "sourceUrl": safe_source_url(props.get("source_url")), + "sourceType": props.get("source_type") or "Unknown", + "sourceTier": props.get("source_tier"), + "publisher": props.get("publisher") or source_name, + "language": props.get("language"), + "verification": None, + "publicationPolicy": None, + "clusterId": None, + "sourcePublishedAt": _iso(props.get("source_published_at")), + "lastVerifiedAt": _iso(props.get("last_verified_at")), + "geometry": None, + "chokepointCorrelations": [], + "metadata": {"forge": {"objectId": obj.id, "contentHash": obj.content_hash()}}, + } + for field_name in _LIST_FIELDS: + values = props.get(_PROPERTY_LIST_KEYS[field_name]) or [] + event[field_name] = list(dict.fromkeys(str(v) for v in values if v)) if isinstance(values, list) else [] + return event + + +def legacy_event(event: dict[str, Any]) -> dict[str, Any]: + """Mirror of Live Map's ``toLegacyEvent``: epoch-millisecond clocks and UI aliases.""" + ui_severity = "medium" if event["severityLabel"] == "moderate" else event["severityLabel"] + place = event["locationName"] or event["countryName"] or "Unknown location" + return { + **event, + "providerId": event["providerEventId"], + "severityScore": event["severity"], + "severity": ui_severity, + "uiSeverity": ui_severity, + "lat": event["latitude"], + "lon": event["longitude"], + "country": event["countryName"] or "Multiple/unknown", + "place": place, + "location": place, + "occurredAt": _ms(event["startedAt"]), + "updatedAt": _ms(event["updatedAt"]), + "firstReportedAt": _ms(event["sourcePublishedAt"]), + "ingestedAt": _ms(event["ingestedAt"]), + "lastVerifiedAt": _ms(event["lastVerifiedAt"]), + "summary": event["description"] or "", + "source": event["sourceName"], + "providerName": event["sourceName"], + "severityReason": "Severity is recorded on the Forge event object.", + "coordinateMethod": "Precision not supplied", + "details": {}, + } + + +# -- request parsing ----------------------------------------------------- +def parse_feed_params(query: dict[str, str]) -> tuple[int, dict[str, Any], list[str]]: + """Return ``(hours, filters, warnings)``. Invalid values are ignored with a warning, like Live Map.""" + warnings: list[str] = [] + hours = HOURS_DEFAULT + raw_hours = query.get("hours") + if raw_hours not in (None, ""): + number = _finite(raw_hours) + if number is None: + warnings.append("Invalid hours ignored.") + else: + hours = int(max(HOURS_MIN, min(HOURS_MAX, number))) + filters: dict[str, Any] = {"recordKind": None, "verification": None, "domain": None, "country": None} + record_kind = query.get("recordKind") + if record_kind: + if record_kind in RECORD_KINDS: + filters["recordKind"] = record_kind + else: + warnings.append("Invalid recordKind filter ignored.") + for key in ("verification", "domain"): + value = (query.get(key) or "").strip() + if value: + if re.match(r"^[a-z][a-z0-9-]{0,40}$", value): + filters[key] = value + else: + warnings.append(f"Invalid {key} filter ignored.") + country = (query.get("country") or "").strip().upper() + if country: + if _COUNTRY_RE.match(country): + filters["country"] = country + else: + warnings.append("Invalid country filter ignored.") + return hours, filters, warnings + + +def _matches(event: dict[str, Any], filters: dict[str, Any], cutoff: datetime) -> bool: + started = _parse_time(event["startedAt"]) + if started is None or started < cutoff: + return False + if filters["recordKind"] and event["recordKind"] != filters["recordKind"]: + return False + if filters["domain"] and event["domain"] != filters["domain"]: + return False + if filters["verification"] and (event["verificationStatus"] or "single-source") != filters["verification"]: + return False + if filters["country"] and (event["countryCode"] or "").upper() != filters["country"]: + return False + return True + + +# -- envelope ------------------------------------------------------------ +def events_feed(store: OntologyStore, query: dict[str, str] | None = None, *, now: datetime | None = None) -> dict[str, Any]: + """Build the ``/api/events`` envelope from the store's public event objects. + + Events need a ``valid_from`` inside the requested window; undated events are not projected. + """ + hours, filters, warnings = parse_feed_params(query or {}) + now = now or datetime.now(timezone.utc) + cutoff = now - timedelta(hours=hours) + objects = store.list_objects(object_type="event", limit=FEED_LIMIT) + canonical = [ + canonical_event(obj) + for obj in objects + if obj.properties.get("public") is not False + ] + canonical = [event for event in canonical if _matches(event, filters, cutoff)] + canonical.sort(key=lambda e: (e["startedAt"], e["id"]), reverse=True) + generated_at = int(now.timestamp() * 1000) + sources = sorted({event["sourceName"] for event in canonical}) + status = { + "ok": True, + "mode": FEED_MODE, + "count": len(canonical), + "lastSuccessAt": generated_at, + "stale": False, + "truncated": len(objects) >= FEED_LIMIT, + } + return { + "events": [legacy_event(event) for event in canonical], + "canonicalEvents": canonical, + "generatedAt": generated_at, + "filters": filters, + "hours": hours, + "sources": sources, + "sourceStatus": {FEED_PROVIDER: status}, + "sourceRegistry": {}, + "providerSourceRegistry": {}, + "domainSourceStatus": {}, + "providerResults": [{"provider": FEED_PROVIDER, "ok": True, "count": len(canonical), "mode": FEED_MODE}], + "observationSummary": {"total": 0, "retained": "metadata-only", "coverageGaps": []}, + "storylines": [], + "observations": [], + "systemStatus": "operational", + "changeCursor": None, + "mode": FEED_MODE, + "errors": [], + "warnings": warnings, + } diff --git a/pyproject.toml b/pyproject.toml index 34614b2..29f6355 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,7 @@ postgres = ["psycopg[binary]>=3.1"] [project.scripts] osint-forge = "osint_forge.main:main" osint-forge-review = "osint_forge.review.review_cli:main" +osint-forge-api = "osint_forge.api.app:main" [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/tests/fixtures/contracts/live_map_events.json b/tests/fixtures/contracts/live_map_events.json new file mode 100644 index 0000000..5231bdf --- /dev/null +++ b/tests/fixtures/contracts/live_map_events.json @@ -0,0 +1,686 @@ +{ + "contract": "live-map /api/events envelope, as documented in Zacharon/live-map docs/API_CONTRACTS.md and src/events/normalized-event.js", + "version": 1, + "notes": [ + "Fixture data, not live records.", + "Both Forge and Live Map should validate responses against `schema`.", + "Forge must reproduce `expected` exactly from `objects` at `now` with `query`." + ], + "now": "2026-10-19T00:00:00Z", + "query": {}, + "objects": [ + { + "id": "event:usgs-us7000abcd", + "type": "event", + "label": "M 5.8 - 40 km SSE of Bandar Abbas, Iran", + "lat": 26.84, + "lon": 56.42, + "valid_from": "2026-10-18T03:12:00Z", + "properties": { + "provider": "usgs", + "source_name": "USGS Earthquake Hazards Program", + "source_url": "https://earthquake.usgs.gov/earthquakes/eventpage/us7000abcd", + "source_type": "Official", + "source_tier": "official", + "category": "earthquake", + "domain": "natural-hazards", + "country_code": "IRN", + "country_name": "Iran", + "location_name": "40 km SSE of Bandar Abbas", + "severity": 70, + "confidence": 0.95, + "status": "active", + "verification_status": "official", + "source_published_at": "2026-10-18T03:20:00Z", + "tags": [ + "earthquake", + "earthquake", + "hormuz" + ], + "affected_chokepoints": [ + "strait-of-hormuz" + ] + }, + "provenance": [ + { + "source": "usgs", + "locator": "usgs:us7000abcd", + "observedAt": "2026-10-18T03:25:00Z", + "contentHash": "fixture-usgs-1" + } + ] + }, + { + "id": "event:notice-hormuz-advisory", + "type": "event", + "label": "Maritime security advisory issued for the Strait of Hormuz", + "valid_from": "2026-10-17T12:00:00Z", + "properties": { + "category": "maritime-advisory", + "domain": "maritime", + "description": "Advisory issued; issued is not closed.", + "severity": 45, + "source_url": "https://user:secret@example.org/notice", + "public": true + }, + "provenance": [ + { + "source": "maritime-notices", + "locator": "notice:2026-117", + "observedAt": "2026-10-17T13:00:00Z", + "contentHash": "fixture-notice-1" + }, + { + "source": "port-authority", + "locator": "pa:bulletin-88", + "observedAt": "2026-10-17T14:30:00Z", + "contentHash": "fixture-pa-1" + } + ] + }, + { + "id": "event:restricted-draft", + "type": "event", + "label": "Analyst draft not cleared for publication", + "lat": 25.0, + "lon": 55.0, + "valid_from": "2026-10-18T00:00:00Z", + "properties": { + "public": false + }, + "provenance": [ + { + "source": "analyst", + "locator": "draft:1", + "observedAt": "2026-10-18T01:00:00Z" + } + ] + }, + { + "id": "event:old-port-delay", + "type": "event", + "label": "Port delay outside the default window", + "lat": 0.0, + "lon": 0.0, + "valid_from": "2026-10-01T00:00:00Z", + "properties": { + "severity": 10 + }, + "provenance": [ + { + "source": "fixture", + "locator": "old:1", + "observedAt": "2026-10-01T01:00:00Z" + } + ] + }, + { + "id": "place:strait-of-hormuz", + "type": "place", + "label": "Strait of Hormuz", + "lat": 26.57, + "lon": 56.25, + "properties": { + "kind": "chokepoint" + }, + "provenance": [ + { + "source": "fixture", + "locator": "place:hormuz", + "observedAt": "2026-10-01T00:00:00Z" + } + ] + } + ], + "schema": { + "envelope": { + "events": [ + "array" + ], + "canonicalEvents": [ + "array" + ], + "generatedAt": [ + "number" + ], + "filters": [ + "object" + ], + "sources": [ + "array" + ], + "sourceStatus": [ + "object" + ], + "sourceRegistry": [ + "object" + ], + "providerSourceRegistry": [ + "object" + ], + "domainSourceStatus": [ + "object" + ], + "providerResults": [ + "array" + ], + "systemStatus": [ + "string" + ], + "mode": [ + "string" + ], + "errors": [ + "array" + ], + "changeCursor": [ + "string", + "null" + ] + }, + "filters": { + "recordKind": [ + "string", + "null" + ], + "verification": [ + "string", + "null" + ], + "domain": [ + "string", + "null" + ], + "country": [ + "string", + "null" + ] + }, + "canonicalEvent": { + "id": [ + "string" + ], + "provider": [ + "string" + ], + "providerEventId": [ + "string", + "null" + ], + "recordKind": [ + "string" + ], + "title": [ + "string" + ], + "description": [ + "string", + "null" + ], + "category": [ + "string" + ], + "latitude": [ + "number", + "null" + ], + "longitude": [ + "number", + "null" + ], + "geographic": [ + "boolean" + ], + "mapDisplayStatus": [ + "string" + ], + "countryCode": [ + "string", + "null" + ], + "countryName": [ + "string", + "null" + ], + "locationName": [ + "string", + "null" + ], + "startedAt": [ + "string", + "null" + ], + "updatedAt": [ + "string", + "null" + ], + "ingestedAt": [ + "string" + ], + "severity": [ + "number" + ], + "severityLabel": [ + "string" + ], + "confidence": [ + "number", + "null" + ], + "status": [ + "string" + ], + "sourceName": [ + "string" + ], + "sourceUrl": [ + "string", + "null" + ], + "sourcePublishedAt": [ + "string", + "null" + ], + "lastVerifiedAt": [ + "string", + "null" + ], + "tags": [ + "array" + ], + "metadata": [ + "object" + ] + }, + "legacyEvent": { + "id": [ + "string" + ], + "title": [ + "string" + ], + "lat": [ + "number", + "null" + ], + "lon": [ + "number", + "null" + ], + "severity": [ + "string" + ], + "severityScore": [ + "number" + ], + "uiSeverity": [ + "string" + ], + "occurredAt": [ + "number", + "null" + ], + "updatedAt": [ + "number", + "null" + ], + "ingestedAt": [ + "number", + "null" + ], + "firstReportedAt": [ + "number", + "null" + ], + "lastVerifiedAt": [ + "number", + "null" + ], + "country": [ + "string" + ], + "place": [ + "string" + ], + "summary": [ + "string" + ], + "source": [ + "string" + ], + "recordKind": [ + "string" + ], + "geographic": [ + "boolean" + ] + }, + "enums": { + "recordKind": [ + "event", + "observation", + "discovery-lead", + "moving-object" + ], + "severityLabel": [ + "low", + "moderate", + "high", + "critical" + ], + "uiSeverity": [ + "low", + "medium", + "high", + "critical" + ], + "status": [ + "active", + "monitoring", + "resolved", + "unknown" + ], + "mapDisplayStatus": [ + "mapped", + "not-mapped" + ] + } + }, + "expected": { + "events": [ + { + "id": "event:usgs-us7000abcd", + "provider": "usgs", + "providerEventId": "usgs-us7000abcd", + "recordKind": "event", + "title": "M 5.8 - 40 km SSE of Bandar Abbas, Iran", + "description": null, + "category": "earthquake", + "domain": "natural-hazards", + "latitude": 26.84, + "longitude": 56.42, + "geographic": true, + "mapDisplayStatus": "mapped", + "nonGeographicReason": null, + "countryCode": "IRN", + "countryName": "Iran", + "region": null, + "locationName": "40 km SSE of Bandar Abbas", + "startedAt": "2026-10-18T03:12:00.000Z", + "updatedAt": null, + "ingestedAt": 1792293900000, + "severity": "high", + "severityLabel": "high", + "confidence": 95.0, + "independentSourceCount": 1, + "verificationStatus": "official", + "status": "active", + "sourceName": "USGS Earthquake Hazards Program", + "sourceUrl": "https://earthquake.usgs.gov/earthquakes/eventpage/us7000abcd", + "sourceType": "Official", + "sourceTier": "official", + "publisher": "USGS Earthquake Hazards Program", + "language": null, + "verification": null, + "publicationPolicy": null, + "clusterId": null, + "sourcePublishedAt": "2026-10-18T03:20:00.000Z", + "lastVerifiedAt": null, + "geometry": null, + "chokepointCorrelations": [], + "metadata": { + "forge": { + "objectId": "event:usgs-us7000abcd", + "contentHash": "7fb5f590d582823723e7921f0ed02664a5cb61c47a1e90717f9df6bb319954c5" + } + }, + "tags": [ + "earthquake", + "hormuz" + ], + "strategicAreas": [], + "affectedChokepoints": [ + "strait-of-hormuz" + ], + "impactTypes": [], + "providerId": "usgs-us7000abcd", + "severityScore": 70.0, + "uiSeverity": "high", + "lat": 26.84, + "lon": 56.42, + "country": "Iran", + "place": "40 km SSE of Bandar Abbas", + "location": "40 km SSE of Bandar Abbas", + "occurredAt": 1792293120000, + "firstReportedAt": 1792293600000, + "summary": "", + "source": "USGS Earthquake Hazards Program", + "providerName": "USGS Earthquake Hazards Program", + "severityReason": "Severity is recorded on the Forge event object.", + "coordinateMethod": "Precision not supplied", + "details": {} + }, + { + "id": "event:notice-hormuz-advisory", + "provider": "maritime-notices", + "providerEventId": "notice-hormuz-advisory", + "recordKind": "event", + "title": "Maritime security advisory issued for the Strait of Hormuz", + "description": "Advisory issued; issued is not closed.", + "category": "maritime-advisory", + "domain": "maritime", + "latitude": null, + "longitude": null, + "geographic": false, + "mapDisplayStatus": "not-mapped", + "nonGeographicReason": "Geography not supplied", + "countryCode": null, + "countryName": null, + "region": null, + "locationName": null, + "startedAt": "2026-10-17T12:00:00.000Z", + "updatedAt": null, + "ingestedAt": 1792247400000, + "severity": "medium", + "severityLabel": "moderate", + "confidence": null, + "independentSourceCount": 2, + "verificationStatus": null, + "status": "unknown", + "sourceName": "maritime-notices", + "sourceUrl": null, + "sourceType": "Unknown", + "sourceTier": null, + "publisher": "maritime-notices", + "language": null, + "verification": null, + "publicationPolicy": null, + "clusterId": null, + "sourcePublishedAt": null, + "lastVerifiedAt": null, + "geometry": null, + "chokepointCorrelations": [], + "metadata": { + "forge": { + "objectId": "event:notice-hormuz-advisory", + "contentHash": "6421d7bf4a30be77391e13f5030ae23c5ccbb7d2756b4af23ebf855256e9e3b6" + } + }, + "tags": [], + "strategicAreas": [], + "affectedChokepoints": [], + "impactTypes": [], + "providerId": "notice-hormuz-advisory", + "severityScore": 45.0, + "uiSeverity": "medium", + "lat": null, + "lon": null, + "country": "Multiple/unknown", + "place": "Unknown location", + "location": "Unknown location", + "occurredAt": 1792238400000, + "firstReportedAt": null, + "summary": "Advisory issued; issued is not closed.", + "source": "maritime-notices", + "providerName": "maritime-notices", + "severityReason": "Severity is recorded on the Forge event object.", + "coordinateMethod": "Precision not supplied", + "details": {} + } + ], + "canonicalEvents": [ + { + "id": "event:usgs-us7000abcd", + "provider": "usgs", + "providerEventId": "usgs-us7000abcd", + "recordKind": "event", + "title": "M 5.8 - 40 km SSE of Bandar Abbas, Iran", + "description": null, + "category": "earthquake", + "domain": "natural-hazards", + "latitude": 26.84, + "longitude": 56.42, + "geographic": true, + "mapDisplayStatus": "mapped", + "nonGeographicReason": null, + "countryCode": "IRN", + "countryName": "Iran", + "region": null, + "locationName": "40 km SSE of Bandar Abbas", + "startedAt": "2026-10-18T03:12:00.000Z", + "updatedAt": null, + "ingestedAt": "2026-10-18T03:25:00.000Z", + "severity": 70.0, + "severityLabel": "high", + "confidence": 95.0, + "independentSourceCount": 1, + "verificationStatus": "official", + "status": "active", + "sourceName": "USGS Earthquake Hazards Program", + "sourceUrl": "https://earthquake.usgs.gov/earthquakes/eventpage/us7000abcd", + "sourceType": "Official", + "sourceTier": "official", + "publisher": "USGS Earthquake Hazards Program", + "language": null, + "verification": null, + "publicationPolicy": null, + "clusterId": null, + "sourcePublishedAt": "2026-10-18T03:20:00.000Z", + "lastVerifiedAt": null, + "geometry": null, + "chokepointCorrelations": [], + "metadata": { + "forge": { + "objectId": "event:usgs-us7000abcd", + "contentHash": "7fb5f590d582823723e7921f0ed02664a5cb61c47a1e90717f9df6bb319954c5" + } + }, + "tags": [ + "earthquake", + "hormuz" + ], + "strategicAreas": [], + "affectedChokepoints": [ + "strait-of-hormuz" + ], + "impactTypes": [] + }, + { + "id": "event:notice-hormuz-advisory", + "provider": "maritime-notices", + "providerEventId": "notice-hormuz-advisory", + "recordKind": "event", + "title": "Maritime security advisory issued for the Strait of Hormuz", + "description": "Advisory issued; issued is not closed.", + "category": "maritime-advisory", + "domain": "maritime", + "latitude": null, + "longitude": null, + "geographic": false, + "mapDisplayStatus": "not-mapped", + "nonGeographicReason": "Geography not supplied", + "countryCode": null, + "countryName": null, + "region": null, + "locationName": null, + "startedAt": "2026-10-17T12:00:00.000Z", + "updatedAt": null, + "ingestedAt": "2026-10-17T14:30:00.000Z", + "severity": 45.0, + "severityLabel": "moderate", + "confidence": null, + "independentSourceCount": 2, + "verificationStatus": null, + "status": "unknown", + "sourceName": "maritime-notices", + "sourceUrl": null, + "sourceType": "Unknown", + "sourceTier": null, + "publisher": "maritime-notices", + "language": null, + "verification": null, + "publicationPolicy": null, + "clusterId": null, + "sourcePublishedAt": null, + "lastVerifiedAt": null, + "geometry": null, + "chokepointCorrelations": [], + "metadata": { + "forge": { + "objectId": "event:notice-hormuz-advisory", + "contentHash": "6421d7bf4a30be77391e13f5030ae23c5ccbb7d2756b4af23ebf855256e9e3b6" + } + }, + "tags": [], + "strategicAreas": [], + "affectedChokepoints": [], + "impactTypes": [] + } + ], + "generatedAt": 1792368000000, + "filters": { + "recordKind": null, + "verification": null, + "domain": null, + "country": null + }, + "hours": 168, + "sources": [ + "USGS Earthquake Hazards Program", + "maritime-notices" + ], + "sourceStatus": { + "osint-forge": { + "ok": true, + "mode": "forge-ontology", + "count": 2, + "lastSuccessAt": 1792368000000, + "stale": false, + "truncated": false + } + }, + "sourceRegistry": {}, + "providerSourceRegistry": {}, + "domainSourceStatus": {}, + "providerResults": [ + { + "provider": "osint-forge", + "ok": true, + "count": 2, + "mode": "forge-ontology" + } + ], + "observationSummary": { + "total": 0, + "retained": "metadata-only", + "coverageGaps": [] + }, + "storylines": [], + "observations": [], + "systemStatus": "operational", + "changeCursor": null, + "mode": "forge-ontology", + "errors": [], + "warnings": [] + } +} diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..988918d --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +import io +import json +import os +import unittest +import uuid +from datetime import datetime, timezone +from pathlib import Path + +from osint_forge.api import ForgeApi, canonical_event, events_feed +from osint_forge.api.projection import safe_source_url +from osint_forge.ontology import OntologyLink, OntologyObject, SQLiteOntologyStore +from osint_forge.provenance import ProvenanceRef + +FIXTURE = Path(__file__).parent / "fixtures" / "contracts" / "live_map_events.json" +JSON_TYPES = {"string": str, "number": (int, float), "boolean": bool, "array": list, "object": dict, "null": type(None)} + + +def load_fixture() -> dict: + return json.loads(FIXTURE.read_text(encoding="utf-8")) + + +def fixture_store(store=None): + store = store or SQLiteOntologyStore(":memory:") + for item in load_fixture()["objects"]: + store.upsert_object(OntologyObject.from_dict(item)) + return store + + +def fixture_now() -> datetime: + return datetime.fromisoformat(load_fixture()["now"].replace("Z", "+00:00")) + + +def type_errors(record: dict, spec: dict, label: str) -> list[str]: + errors = [] + for key, allowed in spec.items(): + if key not in record: + errors.append(f"{label}.{key} missing") + continue + value = record[key] + ok = any( + isinstance(value, JSON_TYPES[t]) and not (t == "number" and isinstance(value, bool)) + for t in allowed + ) + if not ok: + errors.append(f"{label}.{key} is {type(value).__name__}, expected {allowed}") + return errors + + +class LiveMapContractTests(unittest.TestCase): + def make_store(self): + return SQLiteOntologyStore(":memory:") + + def setUp(self) -> None: + self.fixture = load_fixture() + self.store = fixture_store(self.make_store()) + + def tearDown(self) -> None: + self.store.close() + + def test_feed_reproduces_conformance_fixture(self) -> None: + feed = events_feed(self.store, self.fixture["query"], now=fixture_now()) + self.assertEqual(json.loads(json.dumps(feed)), self.fixture["expected"]) + + def test_feed_matches_contract_schema_and_enums(self) -> None: + feed = events_feed(self.store, {}, now=fixture_now()) + schema = self.fixture["schema"] + errors = type_errors(feed, schema["envelope"], "envelope") + errors += type_errors(feed["filters"], schema["filters"], "filters") + for event in feed["canonicalEvents"]: + errors += type_errors(event, schema["canonicalEvent"], event["id"]) + for key in ("recordKind", "severityLabel", "status", "mapDisplayStatus"): + if event[key] not in schema["enums"][key]: + errors.append(f"{event['id']}.{key}={event[key]!r}") + for event in feed["events"]: + errors += type_errors(event, schema["legacyEvent"], "legacy " + event["id"]) + if event["uiSeverity"] not in schema["enums"]["uiSeverity"]: + errors.append(f"legacy {event['id']}.uiSeverity") + self.assertEqual(errors, []) + + def test_restricted_old_and_non_event_objects_are_not_projected(self) -> None: + ids = [e["id"] for e in events_feed(self.store, {}, now=fixture_now())["canonicalEvents"]] + self.assertEqual(ids, ["event:usgs-us7000abcd", "event:notice-hormuz-advisory"]) + wide = events_feed(self.store, {"hours": "720"}, now=fixture_now()) + self.assertIn("event:old-port-delay", [e["id"] for e in wide["canonicalEvents"]]) + self.assertNotIn("event:restricted-draft", [e["id"] for e in wide["canonicalEvents"]]) + + def test_zero_coordinates_stay_mapped(self) -> None: + wide = events_feed(self.store, {"hours": "720"}, now=fixture_now()) + old = next(e for e in wide["canonicalEvents"] if e["id"] == "event:old-port-delay") + self.assertEqual((old["latitude"], old["longitude"], old["geographic"]), (0.0, 0.0, True)) + + def test_filters_and_invalid_params(self) -> None: + feed = events_feed(self.store, {"domain": "maritime"}, now=fixture_now()) + self.assertEqual([e["id"] for e in feed["canonicalEvents"]], ["event:notice-hormuz-advisory"]) + feed = events_feed(self.store, {"country": "irn"}, now=fixture_now()) + self.assertEqual([e["id"] for e in feed["canonicalEvents"]], ["event:usgs-us7000abcd"]) + feed = events_feed(self.store, {"hours": "5", "recordKind": "bogus", "country": ""}, now=fixture_now()) + self.assertEqual(feed["hours"], 24) + self.assertEqual(feed["filters"]["recordKind"], None) + self.assertEqual(len(feed["warnings"]), 2) + + def test_source_urls_with_credentials_or_other_schemes_are_dropped(self) -> None: + self.assertIsNone(safe_source_url("https://user:pw@example.org/x")) + self.assertIsNone(safe_source_url("javascript:alert(1)")) + self.assertIsNone(safe_source_url("ftp://example.org")) + self.assertEqual(safe_source_url("https://example.org/a"), "https://example.org/a") + + def test_confidence_is_scaled_and_zero_is_kept(self) -> None: + obj = OntologyObject( + id="event:zero", + type="event", + label="Zero confidence", + properties={"confidence": 0}, + provenance=[ProvenanceRef(source="t", locator="z", observedAt="2026-10-18T00:00:00Z")], + ) + self.assertEqual(canonical_event(obj)["confidence"], 0.0) + + +@unittest.skipUnless(os.environ.get("FORGE_TEST_DATABASE_URL"), "set FORGE_TEST_DATABASE_URL to run PostgreSQL tests") +class PostgresLiveMapContractTests(LiveMapContractTests): + def make_store(self): + import psycopg + + from osint_forge.ontology import PostgresOntologyStore + + base = os.environ["FORGE_TEST_DATABASE_URL"] + self.schema = "forge_test_" + uuid.uuid4().hex[:12] + with psycopg.connect(base, autocommit=True) as admin: + admin.execute(f"CREATE SCHEMA {self.schema}") + sep = "&" if "?" in base else "?" + return PostgresOntologyStore(f"{base}{sep}options=-csearch_path%3D{self.schema}") + + def tearDown(self) -> None: + import psycopg + + super().tearDown() + with psycopg.connect(os.environ["FORGE_TEST_DATABASE_URL"], autocommit=True) as admin: + admin.execute(f"DROP SCHEMA {self.schema} CASCADE") + + +class ForgeApiTests(unittest.TestCase): + def setUp(self) -> None: + self.store = fixture_store() + self.store.upsert_object( + OntologyObject( + id="document:usgs-page", + type="document", + label="USGS event page", + provenance=[ProvenanceRef(source="usgs", locator="page", observedAt="2026-10-18T03:25:00Z", metadata={"extractor_note": "internal"})], + ) + ) + self.store.add_link( + OntologyLink( + type="located_at", + from_id="event:usgs-us7000abcd", + to_id="place:strait-of-hormuz", + provenance=[ProvenanceRef(source="usgs", locator="geo", observedAt="2026-10-18T03:25:00Z")], + ) + ) + self.api = ForgeApi(self.store) + + def tearDown(self) -> None: + self.store.close() + + def call(self, path: str, query: str = "", method: str = "GET") -> tuple[int, dict, dict | None]: + captured: dict = {} + + def start_response(status: str, headers: list) -> None: + captured["status"] = int(status.split()[0]) + captured["headers"] = dict(headers) + + environ = {"REQUEST_METHOD": method, "PATH_INFO": path, "QUERY_STRING": query, "wsgi.input": io.BytesIO()} + body = b"".join(self.api(environ, start_response)) + return captured["status"], captured["headers"], json.loads(body) if body else None + + def test_health_and_cors(self) -> None: + status, headers, body = self.call("/health") + self.assertEqual((status, body), (200, {"status": "ok"})) + self.assertEqual(headers["Access-Control-Allow-Origin"], "*") + + def test_list_routes_filter_by_type_text_and_bbox(self) -> None: + _, _, events = self.call("/events") + self.assertEqual(events["count"], 4) + _, _, places = self.call("/objects", "type=place") + self.assertEqual([o["id"] for o in places["objects"]], ["place:strait-of-hormuz"]) + _, _, boxed = self.call("/objects", "bbox=56,26,57,27") + self.assertEqual({o["id"] for o in boxed["objects"]}, {"event:usgs-us7000abcd", "place:strait-of-hormuz"}) + _, _, found = self.call("/search", "q=hormuz") + self.assertEqual(found["count"], 2) + self.assertEqual(self.call("/claims")[2]["count"], 0) + + def test_object_views(self) -> None: + status, _, body = self.call("/objects/event%3Ausgs-us7000abcd") + self.assertEqual((status, body["object"]["label"]), (200, "M 5.8 - 40 km SSE of Bandar Abbas, Iran")) + _, _, history = self.call("/objects/event:usgs-us7000abcd/history") + self.assertEqual([r["revision"] for r in history["revisions"]], [1]) + _, _, links = self.call("/objects/event:usgs-us7000abcd/links", "direction=out") + self.assertEqual(links["count"], 1) + _, _, hood = self.call("/objects/place:strait-of-hormuz/neighborhood", "depth=1") + self.assertEqual({o["id"] for o in hood["objects"]}, {"place:strait-of-hormuz", "event:usgs-us7000abcd"}) + + def test_provenance_metadata_is_not_served(self) -> None: + _, _, body = self.call("/objects/document:usgs-page") + self.assertNotIn("metadata", body["object"]["provenance"][0]) + + def test_feed_route_serves_live_map_envelope(self) -> None: + status, headers, body = self.call("/api/events", "hours=720") + self.assertEqual(status, 200) + self.assertIn("public", headers["Cache-Control"]) + self.assertEqual(body["mode"], "forge-ontology") + self.assertEqual(len(body["events"]), len(body["canonicalEvents"])) + + def test_errors_are_structured_json(self) -> None: + self.assertEqual(self.call("/objects/event:missing")[0], 404) + self.assertEqual(self.call("/nope")[2]["error"], "not-found") + self.assertEqual(self.call("/objects", "type=person")[0], 400) + self.assertEqual(self.call("/objects", "bbox=1,2,3")[0], 400) + self.assertEqual(self.call("/objects", "limit=abc")[0], 400) + self.assertEqual(self.call("/search")[0], 400) + self.assertEqual(self.call("/objects/event:usgs-us7000abcd/links", "direction=sideways")[0], 400) + status, _, body = self.call("/objects", method="POST") + self.assertEqual((status, body["error"]), (405, "method-not-allowed")) + self.assertEqual(self.call("/api/events", method="OPTIONS")[0], 204) + + +if __name__ == "__main__": + unittest.main()