commit 26a027d1e8739c00c295fdfb30c2c671abc0b920 Author: Jonas Bengtsson Date: Mon Aug 31 21:13:38 2026 +0200 🎉 Initial commit diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..56e3495 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,7 @@ +.git +.venv +__pycache__ +*.py[cod] +.env +tools/ +README.md \ No newline at end of file diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..18fa8d4 --- /dev/null +++ b/.env.example @@ -0,0 +1,47 @@ +# --------------------------------------------------------------------------- +# Copy to .env (docker compose) or paste into Portainer's "Environment +# variables" section when deploying the stack. Everything has a working +# default except POSTGRES_PASSWORD. +# --------------------------------------------------------------------------- + +# PostgreSQL (db container). Pick a strong password; avoid characters that are +# reserved in URLs (@ : / ? # % &) or URL-encode them, since the password is +# embedded in the DB_DSN connection string. +POSTGRES_USER=nodeledge +POSTGRES_PASSWORD=change-me-strong-password +POSTGRES_DB=nodeledge + +# MQTT broker the listener subscribes to. The default points at the optional +# "broker" test service; in production set this to your real broker host. +MQTT_BROKER_HOST=broker +MQTT_BROKER_PORT=1883 +MQTT_USERNAME= +MQTT_PASSWORD= + +# Subscription filter. Default "/+/+" matches topics on the form +# //. Use "#", "/up/#", ... to widen it (the IMEI is then +# always taken from the LAST topic segment). +MQTT_TOPIC_FILTER=/+/+ +# 0 = at most once, 1 = at least once (duplicate rows possible after reconnects) +MQTT_QOS=1 +MQTT_CLIENT_ID=nodeledge-mqtt-listener + +# Keys the listener injects into every payload. +INJECT_IMEI_KEY=imei +INJECT_DEVICE_KEY=device + +# Comma-separated payload keys searched for a device timestamp (first +# parseable one wins). Accepts ISO 8601 strings, epoch seconds or epoch +# milliseconds. Rows without a parseable timestamp keep payload_time NULL. +PAYLOAD_TIME_KEYS=timestamp,time,ts + +# Host ports. On Windows, 1883 often falls into a Hyper-V excluded port range +# (netsh interface ipv4 show excludedportrange protocol=tcp); 2883 avoids it. +# 8080 frequently collides with other dev tools; 18080 avoids that. +API_HOST_PORT=18080 +MQTT_HOST_PORT=2883 + +# API: hard cap for the limit query parameter. +API_MAX_LIMIT=10000 + +LOG_LEVEL=INFO \ No newline at end of file diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..700dd7d --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +# local secrets / environment +.env + +# python +__pycache__/ +*.py[cod] +.venv/ +*.egg-info/ + +# editors +.vscode/ +.idea/ \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..4b4b81d --- /dev/null +++ b/README.md @@ -0,0 +1,133 @@ +# nodeledge-mqtt-listener + +A Portainer / docker compose stack that: + +1. **mqtt-listener** — subscribes to MQTT topics on the form `//`, injects `imei` and a `device` name into the JSON payload, and stores it in PostgreSQL. +2. **api** — a FastAPI read API over the database, protected by static app tokens. +3. **db** — PostgreSQL 16 storing the messages (JSONB payloads). + +``` +┌─────────────┐ /xxx/IMEI ┌───────────────┐ INSERT ┌────┐ +│ MQTT broker │ ──────────────► │ mqtt-listener │ ──────────► │ db │ +└─────────────┘ JSON payloads └───────────────┘ └────┘ + injects imei + device ▲ +┌─────────────┐ GET /messages ┌───────────────┐ SELECT │ +│ API clients │ ──────────────► │ api │ ──────────────┘ +└─────────────┘ X-App-Token └───────────────┘ +``` + +## Layout + +``` +docker-compose.yml stack definition (paste into Portainer) +.env.example environment variables (POSTGRES_PASSWORD etc.) +init/01_schema.sql creates the mqtt_messages table on first db start +configs/listener/imei_map.json IMEI -> device name mapping +configs/api/app_tokens.json app tokens for the API +listener/ mqtt -> db service (paho-mqtt, psycopg) +api/ FastAPI service (uvicorn) +mosquitto/ config for the optional test broker (profile "test") +tools/ host-side test scripts and pytest suite +``` + +## Quick start (local, with the test broker) + +```bash +cp .env.example .env # then edit POSTGRES_PASSWORD +docker compose --profile test up -d --build +``` + +This starts db, mqtt-listener, api and an anonymous Mosquitto broker (ports 1883 → host 2883, API 18080 → host 18080). Publish some messages: + +```bash +pip install -r tools/requirements.txt +python tools/publish_test_messages.py --broker localhost:1883 +``` + +Fetch data from the API: + +```bash +python tools/smoke_test_api.py --url http://localhost:18080 \ + --token replace-me-with-a-long-random-secret-1 +``` + +Expected smoke test output: + +``` +PASS: missing token rejected with 401 +PASS: wrong token rejected with 401 +PASS: valid token accepted, N message(s) returned +PASS: count endpoint returned {'count': N} +``` + +## Deploying in Portainer + +1. Create a stack from **Stacks → Add stack**, choose "Web editor", paste the contents of `docker-compose.yml`. +2. Under **Environment variables**, add at minimum: + - `POSTGRES_PASSWORD` — **required** (see the password note in `.env.example`). + - `MQTT_BROKER_HOST` — your real broker hostname/IP (not the test broker). + - `MQTT_USERNAME` / `MQTT_PASSWORD` — broker credentials, if required. + - Optional: `MQTT_TOPIC_FILTER`, `MQTT_QOS`, `API_MAX_LIMIT`, `LOG_LEVEL`, ports. +3. Because the stack references local files (`init/01_schema.sql`, `configs/...`, Dockerfiles), the stack files must be present on the Docker host. Two options: + - **Recommended:** put the repository in a folder on the host (e.g. `/opt/nodeledge`) and use **Stacks → Add stack → Repository** pointing at your git repo, or + - Upload the files via "Custom template" including all files. +4. **Secrets:** edit `configs/api/app_tokens.json` and `configs/listener/imei_map.json` on the host to hold your real values before deploying (the defaults are placeholders). +5. Deploy the stack. The `db` service runs `init/01_schema.sql` on its first start (an empty volume), creating the `mqtt_messages` table. +6. Verify: Stacks → your stack → check that db, mqtt-listener and api are all healthy (green). + +> The `broker` service only runs when the compose profile `test` is active (Portainer: enable the `test` profile in the stack settings), so production deploys never spin up an extra broker. + +## How the IMEI injection works + +- The IMEI is the **last segment** of the topic, e.g. `/vivab/350123456789012` → `350123456789012`. +- It must be exactly **15 digits**; anything else (letters, separators, other lengths) is skipped. +- It looks the IMEI up in `configs/listener/imei_map.json`; on a match the device name is injected, otherwise the IMEI itself is used as the device name. +- The injected keys are `imei` and `device` (configurable via `INJECT_IMEI_KEY` / `INJECT_DEVICE_KEY`). Any existing values in the payload are **overwritten** — the topic is authoritative. +- A **device timestamp** is extracted at ingest time from the first parseable payload key in `PAYLOAD_TIME_KEYS` (default `timestamp,time,ts`; ISO 8601 strings, epoch seconds and epoch milliseconds all accepted) and stored in the `payload_time` column. A missing or unparseable timestamp does **not** skip the message — `payload_time` is simply NULL and `received_at` (server receipt time) remains authoritative. +- Messages whose payload is not a JSON object, or whose last topic segment is not a 15-digit IMEI, are skipped (with a warning log). + +## API reference + +Authentication: pass one of the configured app tokens either as the `X-App-Token` header or as the `app_token` query parameter. Missing/invalid tokens get `401`. + +| Endpoint | Auth | Description | +| -------- | ---- | ----------- | +| `GET /health` | none | Liveness probe (public) | +| `GET /messages/count` | token | Total number of stored messages | +| `GET /messages` | token | List messages, newest first. Query parameters: `imei`, `device` (exact-match filters), `from`/`to` (ISO 8601 range on `received_at` — server receipt time; inclusive; naive values treated as UTC), `payload_from`/`payload_to` (same but on `payload_time` — the device timestamp extracted from the JSON; messages without one are excluded; ordering switches to `payload_time DESC`), `limit` (default 100, capped at `API_MAX_LIMIT`), `offset` for paging. Reversed ranges return 422 | +| `GET /docs`, `GET /redoc` | none | OpenAPI docs (public) | + +Example — everything device `Water tank sensor B` reported on August 31st, by device clock: + +``` +GET /messages?device=Water%20tank%20sensor%20B&payload_from=2026-08-31T00:00:00Z&payload_to=2026-08-31T23:59:59Z +``` + +Example response: + +```json +{ + "limit": 100, + "offset": 0, + "from": null, + "to": null, + "payload_from": "2026-08-31T00:00:00+00:00", + "payload_to": "2026-08-31T23:59:59+00:00", + "items": [ + { + "id": 1, + "received_at": "2026-08-31T10:00:00.123456+00:00", + "topic": "/vivab/350123456789012", + "imei": "350123456789012", + "device": "Greenhouse sensor A", + "payload": { + "temperature": 21.5, + "timestamp": "2026-08-31T09:59:58+00:00", + "imei": "350123456789012", + "device": "Greenhouse sensor A" + }, + "payload_time": "2026-08-31T09:59:58+00:00" + } + ] +} +``` \ No newline at end of file diff --git a/api/Dockerfile b/api/Dockerfile new file mode 100644 index 0000000..eb4778b --- /dev/null +++ b/api/Dockerfile @@ -0,0 +1,20 @@ +# syntax=docker/dockerfile:1 +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /app + +# Copy into /app/api/ (not /app/): `COPY api/ /app/` would flatten the directory +# contents into /app and `uvicorn api.app:app` would find no api package. +COPY api/requirements.txt /app/api/requirements.txt +RUN pip install --no-cache-dir -r /app/api/requirements.txt + +COPY api/ /app/api/ + +# Run as an unprivileged user. +RUN useradd --system --create-home --uid 10001 appuser +USER appuser + +CMD ["uvicorn", "api.app:app", "--host", "0.0.0.0", "--port", "8000"] \ No newline at end of file diff --git a/api/__init__.py b/api/__init__.py new file mode 100644 index 0000000..270ec7e --- /dev/null +++ b/api/__init__.py @@ -0,0 +1 @@ +"""nodeledge api package.""" \ No newline at end of file diff --git a/api/app.py b/api/app.py new file mode 100644 index 0000000..5ae651b --- /dev/null +++ b/api/app.py @@ -0,0 +1,234 @@ +"""FastAPI application exposing read access to the mqtt_messages table. + +Authentication: static app tokens from a JSON config file, passed either as +the `X-App-Token` header or as an `app_token` query parameter. Tokens are +compared in constant time; invalid tokens yield 401. +""" + +from __future__ import annotations + +import hmac +import json +import logging +import os +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import psycopg +from fastapi import Depends, FastAPI, HTTPException, Query, Request, status +from psycopg import sql + +logger = logging.getLogger("api") + + +class AppConfig: + """Runtime configuration loaded once at startup.""" + + def __init__(self, dsn: str, tokens: dict[str, str], max_limit: int) -> None: + self.dsn = dsn + self.tokens = tokens + self.max_limit = max_limit + + +def load_tokens(path: str) -> dict[str, str]: + """Load {"tokens": {"app-name": "secret"}} from a JSON file.""" + data = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(data, dict) or not isinstance(data.get("tokens"), dict): + raise ValueError(f"{path}: expected JSON on the form {{\"tokens\": {{name: secret}}}}") + tokens = data["tokens"] + for name, secret in tokens.items(): + if not isinstance(secret, str) or not secret: + raise ValueError(f"{path}: token {name!r} must be a non-empty string") + if not tokens: + logger.warning("%s contains no tokens; the API will reject all requests", path) + return dict(tokens) + + +def _env(name: str, default: str | None = None) -> str: + value = os.environ.get(name) + if not value: + if default is None: + raise RuntimeError(f"missing required environment variable {name}") + return default + return value + + +def _as_utc(value: datetime) -> datetime: + """Interpret naive timestamps as UTC (received_at is TIMESTAMPTZ).""" + return value if value.tzinfo is not None else value.replace(tzinfo=timezone.utc) + + +def create_app() -> FastAPI: + app = FastAPI( + title="nodeledge MQTT data API", + version="1.0.0", + description="Read access to MQTT-collected JSON payloads.", + ) + + tokens_path = _env("APP_TOKENS_FILE") + dsn = _env("DB_DSN") + max_limit = int(_env("API_MAX_LIMIT", "10000")) + config = AppConfig(dsn=dsn, tokens=load_tokens(tokens_path), max_limit=max_limit) + + def authenticate(request: Request) -> None: + token = request.headers.get("X-App-Token") + if not token: + token = request.query_params.get("app_token") + if not token: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="missing app token: pass the X-App-Token header", + ) + for candidate in config.tokens.values(): + if hmac.compare_digest(candidate, token): + return + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="invalid app token", + ) + + app.state.config = config + app.state.authenticate = authenticate + + @app.get("/health", tags=["meta"]) + def health() -> dict[str, Any]: + return {"status": "ok"} + + @app.get("/messages/count", tags=["meta"], dependencies=[Depends(authenticate)]) + def message_count() -> dict[str, int]: + conn = psycopg.connect(config.dsn) + try: + with conn.cursor() as cur: + cur.execute("SELECT count(*) FROM mqtt_messages") + row = cur.fetchone() + finally: + conn.close() + count = row[0] if row is not None else 0 + return {"count": count} + + @app.get("/messages", tags=["data"], dependencies=[Depends(authenticate)]) + def list_messages( + imei: str | None = Query(None, description="Filter by IMEI (15 digits)"), + device: str | None = Query(None, description="Filter by device name"), + from_time: datetime | None = Query( + None, + alias="from", + description=( + "Only messages received at or after this time (ISO 8601, inclusive; " + "naive values are treated as UTC)" + ), + ), + to_time: datetime | None = Query( + None, + alias="to", + description=( + "Only messages received at or before this time (ISO 8601, inclusive; " + "naive values are treated as UTC)" + ), + ), + payload_from_time: datetime | None = Query( + None, + alias="payload_from", + description=( + "Frame on the device timestamp extracted from the JSON payload " + "(ISO 8601, inclusive; naive values are treated as UTC). Messages " + "without a parseable payload timestamp are excluded." + ), + ), + payload_to_time: datetime | None = Query( + None, + alias="payload_to", + description=( + "Frame on the device timestamp extracted from the JSON payload " + "(ISO 8601, inclusive; naive values are treated as UTC). Messages " + "without a parseable payload timestamp are excluded." + ), + ), + limit: int = Query(100, gt=0, description="Page size"), + offset: int = Query(0, ge=0, description="Page offset"), + ) -> dict[str, Any]: + if from_time is not None and to_time is not None and from_time > to_time: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="'from' must be earlier than or equal to 'to'", + ) + if ( + payload_from_time is not None + and payload_to_time is not None + and payload_from_time > payload_to_time + ): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="'payload_from' must be earlier than or equal to 'payload_to'", + ) + limit = min(limit, config.max_limit) + conditions: list[sql.Composable] = [] + params: list[Any] = [] + if imei: + conditions.append(sql.SQL("imei = %s")) + params.append(imei) + if device: + conditions.append(sql.SQL("device = %s")) + params.append(device) + if from_time is not None: + conditions.append(sql.SQL("received_at >= %s")) + params.append(_as_utc(from_time)) + if to_time is not None: + conditions.append(sql.SQL("received_at <= %s")) + params.append(_as_utc(to_time)) + if payload_from_time is not None: + conditions.append(sql.SQL("payload_time >= %s")) + params.append(_as_utc(payload_from_time)) + if payload_to_time is not None: + conditions.append(sql.SQL("payload_time <= %s")) + params.append(_as_utc(payload_to_time)) + + where = sql.SQL("") + if conditions: + where = sql.SQL(" WHERE ") + sql.SQL(" AND ").join(conditions) + + # Order by device timestamp when framing on it (NULLS LAST matches the + # idx_mqtt_messages_ptime index); otherwise newest-first by insert id. + if payload_from_time is not None or payload_to_time is not None: + order_by = sql.SQL(" ORDER BY payload_time DESC NULLS LAST LIMIT %s OFFSET %s") + else: + order_by = sql.SQL(" ORDER BY id DESC LIMIT %s OFFSET %s") + + query = sql.SQL( + "SELECT id, received_at, topic, imei, device, payload, payload_time FROM mqtt_messages" + ) + where + order_by + + conn = psycopg.connect(config.dsn) + try: + with conn.cursor() as cur: + cur.execute(query, (*params, limit, offset)) + rows = cur.fetchall() + finally: + conn.close() + + return { + "limit": limit, + "offset": offset, + "from": from_time.isoformat() if from_time is not None else None, + "to": to_time.isoformat() if to_time is not None else None, + "payload_from": payload_from_time.isoformat() if payload_from_time is not None else None, + "payload_to": payload_to_time.isoformat() if payload_to_time is not None else None, + "items": [ + { + "id": row[0], + "received_at": row[1].isoformat(), + "topic": row[2], + "imei": row[3], + "device": row[4], + "payload": row[5], + "payload_time": row[6].isoformat() if row[6] is not None else None, + } + for row in rows + ], + } + + return app + + +app = create_app() \ No newline at end of file diff --git a/api/healthcheck.py b/api/healthcheck.py new file mode 100644 index 0000000..1e6c63b --- /dev/null +++ b/api/healthcheck.py @@ -0,0 +1,24 @@ +"""Container healthcheck for the api service: probes GET /health.""" + +from __future__ import annotations + +import sys +import urllib.request + + +def main() -> int: + try: + with urllib.request.urlopen("http://127.0.0.1:8000/health", timeout=5) as response: + body = response.read().decode("utf-8", errors="replace") + if response.status == 200: + print(body) + return 0 + print(f"unexpected status {response.status}: {body}") + return 1 + except Exception as exc: # noqa: BLE001 - report any failure as unhealthy + print(f"health check failed: {exc}") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/api/requirements.txt b/api/requirements.txt new file mode 100644 index 0000000..6f74fc6 --- /dev/null +++ b/api/requirements.txt @@ -0,0 +1,3 @@ +fastapi==0.115.12 +uvicorn[standard]==0.34.2 +psycopg[binary]==3.2.3 \ No newline at end of file diff --git a/configs/api/app_tokens.json b/configs/api/app_tokens.json new file mode 100644 index 0000000..208a69c --- /dev/null +++ b/configs/api/app_tokens.json @@ -0,0 +1,6 @@ +{ + "tokens": { + "monitoring-app": "replace-me-with-a-long-random-secret-1", + "grafana-dashboard": "replace-me-with-a-long-random-secret-2" + } +} \ No newline at end of file diff --git a/configs/listener/imei_map.json b/configs/listener/imei_map.json new file mode 100644 index 0000000..c9df518 --- /dev/null +++ b/configs/listener/imei_map.json @@ -0,0 +1,6 @@ +{ + "350123456789012": "Greenhouse sensor A", + "490154203237518": "Water tank sensor B", + "356938035643809": "Weather station roof", + "861234012345678": "Soil moisture plot 12" +} \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..82d706e --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,111 @@ +# Nodeledge MQTT listener stack (Portainer / docker compose compatible). +# +# Services: +# db PostgreSQL holding the mqtt_messages table +# mqtt-listener subscribes to MQTT, injects IMEI + device into JSON payloads, writes rows +# api FastAPI read API over the database, protected by static app tokens +# broker OPTIONAL test broker (compose profile "test") for trying the stack out +# +# Portainer: Stacks -> Add stack -> paste this file, then set the environment +# variables (see .env.example) in the stack's "Environment variables" section. +# Locally: cp .env.example .env && docker compose --profile test up -d --build + +services: + db: + image: postgres:16-alpine + restart: unless-stopped + environment: + POSTGRES_USER: ${POSTGRES_USER:-nodeledge} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in the stack environment} + POSTGRES_DB: ${POSTGRES_DB:-nodeledge} + volumes: + - db_data:/var/lib/postgresql/data + # Creates the mqtt_messages table on the first database start only. + - ./init/01_schema.sql:/docker-entrypoint-initdb.d/01_schema.sql:ro + healthcheck: + test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"] + interval: 10s + timeout: 5s + retries: 10 + start_period: 10s + + mqtt-listener: + build: + context: . + dockerfile: listener/Dockerfile + init: true + restart: unless-stopped + depends_on: + db: + condition: service_healthy + environment: + MQTT_BROKER_HOST: ${MQTT_BROKER_HOST:-broker} + MQTT_BROKER_PORT: ${MQTT_BROKER_PORT:-1883} + MQTT_USERNAME: ${MQTT_USERNAME:-} + MQTT_PASSWORD: ${MQTT_PASSWORD:-} + MQTT_CLIENT_ID: ${MQTT_CLIENT_ID:-nodeledge-mqtt-listener} + # Matches topics on the form // (see README to widen it). + MQTT_TOPIC_FILTER: ${MQTT_TOPIC_FILTER:-/+/+} + MQTT_QOS: ${MQTT_QOS:-1} + DB_DSN: postgresql://${POSTGRES_USER:-nodeledge}:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in the stack environment}@db:5432/${POSTGRES_DB:-nodeledge} + IMEI_MAP_FILE: /app/config/imei_map.json + INJECT_IMEI_KEY: ${INJECT_IMEI_KEY:-imei} + INJECT_DEVICE_KEY: ${INJECT_DEVICE_KEY:-device} + # Comma-separated payload keys searched for a device timestamp. + PAYLOAD_TIME_KEYS: ${PAYLOAD_TIME_KEYS:-timestamp,time,ts} + LOG_LEVEL: ${LOG_LEVEL:-INFO} + volumes: + # IMEI -> device name mapping, see configs/listener/imei_map.json + - ./configs/listener:/app/config:ro + healthcheck: + test: ["CMD", "python", "/app/listener/healthcheck.py"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 30s + + api: + build: + context: . + dockerfile: api/Dockerfile + init: true + restart: unless-stopped + depends_on: + db: + condition: service_healthy + environment: + DB_DSN: postgresql://${POSTGRES_USER:-nodeledge}:${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in the stack environment}@db:5432/${POSTGRES_DB:-nodeledge} + APP_TOKENS_FILE: /app/config/app_tokens.json + API_MAX_LIMIT: ${API_MAX_LIMIT:-10000} + LOG_LEVEL: ${LOG_LEVEL:-INFO} + volumes: + # Static app tokens for API authentication, see configs/api/app_tokens.json + - ./configs/api:/app/config:ro + ports: + - "${API_HOST_PORT:-18080}:8000" + healthcheck: + test: ["CMD", "python", "/app/api/healthcheck.py"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 30s + + broker: + # Optional MQTT broker for local testing; excluded unless the "test" + # profile is active (docker compose --profile test ...). Point + # MQTT_BROKER_HOST at your real broker in production. + image: eclipse-mosquitto:2 + profiles: ["test"] + restart: unless-stopped + ports: + - "${MQTT_HOST_PORT:-1883}:1883" + volumes: + - ./mosquitto/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro + healthcheck: + test: ["CMD", "mosquitto_pub", "-h", "127.0.0.1", "-t", "broker/healthcheck", "-m", "ping"] + interval: 30s + timeout: 5s + retries: 3 + +volumes: + db_data: \ No newline at end of file diff --git a/init/01_schema.sql b/init/01_schema.sql new file mode 100644 index 0000000..7c6f0b7 --- /dev/null +++ b/init/01_schema.sql @@ -0,0 +1,35 @@ +-- Schema for the nodeledge MQTT data pipeline. +-- Executed automatically by the postgres image on the FIRST database start +-- (files in /docker-entrypoint-initdb.d are only run against an empty data dir). + +CREATE TABLE IF NOT EXISTS mqtt_messages ( + id BIGSERIAL PRIMARY KEY, + received_at TIMESTAMPTZ NOT NULL DEFAULT now(), + topic TEXT NOT NULL, + imei TEXT NOT NULL, + device TEXT NOT NULL, + payload JSONB NOT NULL, + payload_time TIMESTAMPTZ +); + +-- "newest messages for a device" and "recent messages overall" +-- (received_at = server receipt time) +CREATE INDEX IF NOT EXISTS idx_mqtt_messages_imei_time + ON mqtt_messages (imei, received_at DESC); +CREATE INDEX IF NOT EXISTS idx_mqtt_messages_time + ON mqtt_messages (received_at DESC); +-- device-timestamp framing (payload_time extracted from the JSON payload; +-- NULL when the payload carries no parseable timestamp) +CREATE INDEX IF NOT EXISTS idx_mqtt_messages_imei_ptime + ON mqtt_messages (imei, payload_time DESC); +CREATE INDEX IF NOT EXISTS idx_mqtt_messages_ptime + ON mqtt_messages (payload_time DESC); +-- exact topic lookups +CREATE INDEX IF NOT EXISTS idx_mqtt_messages_topic + ON mqtt_messages (topic); + +COMMENT ON COLUMN mqtt_messages.payload_time IS + 'Device timestamp extracted from the JSON payload at ingest time; NULL when absent or unparseable. received_at is the server receipt time.'; + +COMMENT ON TABLE mqtt_messages IS + 'JSON payloads received over MQTT; payload contains the original JSON with imei and device injected by the listener.'; \ No newline at end of file diff --git a/listener/Dockerfile b/listener/Dockerfile new file mode 100644 index 0000000..f8c948e --- /dev/null +++ b/listener/Dockerfile @@ -0,0 +1,20 @@ +# syntax=docker/dockerfile:1 +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /app + +# Copy into /app/listener/ (not /app/): `COPY listener/ /app/` would flatten the +# directory contents into /app and `python -m listener.main` would find no package. +COPY listener/requirements.txt /app/listener/requirements.txt +RUN pip install --no-cache-dir -r /app/listener/requirements.txt + +COPY listener/ /app/listener/ + +# Run as an unprivileged user. +RUN useradd --system --create-home --uid 10001 appuser +USER appuser + +CMD ["python", "-m", "listener.main"] \ No newline at end of file diff --git a/listener/__init__.py b/listener/__init__.py new file mode 100644 index 0000000..a642164 --- /dev/null +++ b/listener/__init__.py @@ -0,0 +1 @@ +"""nodeledge mqtt-listener package.""" \ No newline at end of file diff --git a/listener/db.py b/listener/db.py new file mode 100644 index 0000000..33f1171 --- /dev/null +++ b/listener/db.py @@ -0,0 +1,85 @@ +"""Database access for the MQTT listener. + +The network thread writes rows; writes happen only on that thread so a plain +(no pool) blocking connection is used, with reconnection on failure. +""" + +from __future__ import annotations + +import logging +import threading +from datetime import datetime +from typing import TYPE_CHECKING + +import psycopg +from psycopg import sql +from psycopg.types.json import Jsonb + +if TYPE_CHECKING: + from psycopg import Connection + +logger = logging.getLogger("listener.db") + +# Row insert. Values are %s placeholders passed as parameters; no manual +# string concatenation is involved, so this does not trip the +# sql.SQL-LiteralString diagnostics for dynamically-built queries. +_INSERT = sql.SQL( + """ + INSERT INTO mqtt_messages (topic, imei, device, payload, payload_time) + VALUES (%s, %s, %s, %s, %s) + """ +) + + +class Inserter: + """Single-threaded blocking writer for mqtt_messages. + + Not thread-safe by itself; the listener funnels all writes through the + MQTT network thread and guards the connection with a lock so shutdown + can safely disconnect from the main thread. + """ + + def __init__(self, dsn: str) -> None: + self._dsn = dsn + self._lock = threading.Lock() + self._conn: Connection | None = None + + def _connection(self) -> Connection: + if self._conn is None or self._conn.closed: + logger.info("connecting to database") + self._conn = psycopg.connect(self._dsn, autocommit=True) + return self._conn + + def insert( + self, + topic: str, + imei: str, + device: str, + payload: dict, + payload_time: datetime | None = None, + ) -> None: + """Insert one message. Reconnects once on a broken connection.""" + with self._lock: + try: + conn = self._connection() + with conn.cursor() as cur: + cur.execute( + _INSERT, + (topic, imei, device, Jsonb(payload), payload_time), + ) + except psycopg.Error: + logger.exception("database write failed (topic=%s imei=%s)", topic, imei) + self._close_locked() + raise + + def _close_locked(self) -> None: + if self._conn is not None: + try: + self._conn.close() + except psycopg.Error: + pass + self._conn = None + + def close(self) -> None: + with self._lock: + self._close_locked() \ No newline at end of file diff --git a/listener/healthcheck.py b/listener/healthcheck.py new file mode 100644 index 0000000..654252a --- /dev/null +++ b/listener/healthcheck.py @@ -0,0 +1,40 @@ +"""Container healthcheck for the mqtt-listener. + +Exits 0 when the listener is alive: the process is running and its MQTT client +is either connected or actively reconnecting. A missing pidfile means the main +process crashed and docker should restart it (restart: unless-stopped). +""" + +from __future__ import annotations + +import os +import sys + +PID_FILE = "/tmp/listener.pid" + + +def main() -> int: + try: + with open(PID_FILE, encoding="utf-8") as f: + pid = int(f.read().strip()) + except (FileNotFoundError, ValueError): + print("pidfile missing or unreadable") + return 1 + + # Docker kills the shell running the healthcheck, not the app, so the app + # pid must still exist. os.kill with signal 0 only probes for existence. + try: + os.kill(pid, 0) + except ProcessLookupError: + print(f"listener pid {pid} is gone") + return 1 + except PermissionError: + print(f"listener pid {pid} running (owned by another uid)") + return 0 + + print(f"listener pid {pid} is alive") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/listener/main.py b/listener/main.py new file mode 100644 index 0000000..dce047f --- /dev/null +++ b/listener/main.py @@ -0,0 +1,151 @@ +"""MQTT listener: subscribes to //, injects IMEI and device +name into the JSON payload and stores the result in PostgreSQL. + +The paho network loop runs in its own thread. Writes go straight to the +database from the network thread; the insert re-raises on failure so paho's +automatic reconnect logic keeps the subscriber alive while a single message +may be retried or dropped depending on QoS. +""" + +from __future__ import annotations + +import json +import logging +import os +import signal +import threading +from pathlib import Path + +import paho.mqtt.client as mqtt +from paho.mqtt.enums import CallbackAPIVersion + +from listener.db import Inserter +from listener.process import DEFAULT_TIME_KEYS, normalize_imei, process_message + +logger = logging.getLogger("listener") + +_stop_event = threading.Event() +_shutdown_complete = threading.Event() + + +def load_imei_map(path: str) -> dict[str, str]: + """Load the IMEI -> device name map from a JSON file. + + Keys must be 15 digit IMEIs; values must be non-empty strings. + """ + data = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise ValueError(f"{path}: expected a JSON object mapping IMEI to device name") + result: dict[str, str] = {} + for key, value in data.items(): + if not isinstance(value, str) or not value: + raise ValueError(f"{path}: values must be non-empty strings (key {key!r})") + normalized = key.strip() + if normalize_imei(normalized) is None: + raise ValueError(f"{path}: key {key!r} is not a 15-digit IMEI") + result[normalized] = value + logger.info("loaded %d IMEI mappings from %s", len(result), path) + return result + + +def _env(name: str, default: str | None = None) -> str: + value = os.environ.get(name) + if not value: + if default is None: + raise RuntimeError(f"missing required environment variable {name}") + return default + return value + + +def main() -> None: + log_level = _env("LOG_LEVEL", "INFO").upper() + logging.basicConfig( + level=getattr(logging, log_level, logging.INFO), + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + + broker_host = _env("MQTT_BROKER_HOST") + broker_port = int(_env("MQTT_BROKER_PORT", "1883")) + username = os.environ.get("MQTT_USERNAME") or None + password = os.environ.get("MQTT_PASSWORD") or None + client_id = _env("MQTT_CLIENT_ID", "nodeledge-mqtt-listener") + topic_filter = _env("MQTT_TOPIC_FILTER", "/+/+") + qos = int(_env("MQTT_QOS", "1")) + + map_file = _env("IMEI_MAP_FILE") + imei_map = load_imei_map(map_file) + imei_key = _env("INJECT_IMEI_KEY", "imei") + device_key = _env("INJECT_DEVICE_KEY", "device") + time_keys = tuple( + key.strip() + for key in _env("PAYLOAD_TIME_KEYS", ",".join(DEFAULT_TIME_KEYS)).split(",") + if key.strip() + ) + + inserter = Inserter(_env("DB_DSN")) + + # Healthcheck probe target: container health is "main process alive". + Path("/tmp/listener.pid").write_text(str(os.getpid()), encoding="utf-8") + logger.info("listener starting, pid %d", os.getpid()) + + def on_connect(client, userdata, flags, reason_code, properties): # noqa: ANN001 + if reason_code.is_failure: + logger.error("broker refused connection: %s", reason_code) + return + logger.info("connected to broker, subscribing to %s (qos %d)", topic_filter, qos) + client.subscribe(topic_filter, qos=qos) + + def on_message(client, userdata, message): # noqa: ANN001 + try: + topic = message.topic + result = process_message( + topic, + bytes(message.payload), + imei_map, + imei_key=imei_key, + device_key=device_key, + time_keys=time_keys, + ) + if result is None: + logger.warning( + "skipping message on %r: payload is not a JSON object or last topic " + "segment is not a 15-digit IMEI", + topic, + ) + return + inserter.insert(topic, result.imei, result.device, result.payload, result.payload_time) + logger.info("stored %s/%s -> %s", result.imei, result.device, topic) + except Exception: + logger.exception("message handling failed for topic %s", getattr(message, "topic", "?")) + + client = mqtt.Client( + callback_api_version=CallbackAPIVersion.VERSION2, + client_id=client_id, + ) + if username: + client.username_pw_set(username, password) + client.on_connect = on_connect + client.on_message = on_message + client.reconnect_delay_set(min_delay=1, max_delay=60) + client.connect_async(broker_host, broker_port, keepalive=60) + client.loop_start() # retries initial connection failures automatically + + def _handle_signal(signum, frame): # noqa: ANN001 + logger.info("received signal %s, shutting down", signum) + _stop_event.set() + + signal.signal(signal.SIGTERM, _handle_signal) + signal.signal(signal.SIGINT, _handle_signal) + + while not _stop_event.is_set(): + _stop_event.wait(timeout=5.0) + + client.loop_stop() + client.disconnect() + inserter.close() + Path("/tmp/listener.pid").unlink(missing_ok=True) + logger.info("shutdown complete") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/listener/process.py b/listener/process.py new file mode 100644 index 0000000..c07eb63 --- /dev/null +++ b/listener/process.py @@ -0,0 +1,123 @@ +"""Message handling: JSON parsing, IMEI extraction and payload injection.""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from datetime import datetime, timezone + +# An IMEI is a 15 digit decimal number. The topic segment must be exactly +# that: 15 digits, no separators, no letters, no other characters. +_IMEI_RE = re.compile(r"^[0-9]{15}$") + +# Epoch values above this many seconds must be milliseconds (seconds only +# reach this magnitude in the year ~5138, so the heuristic is safe). +_EPOCH_MS_THRESHOLD = 1e11 + +# Default payload keys searched for a device timestamp; first parseable +# value wins. Configurable via PAYLOAD_TIME_KEYS in the listener container. +DEFAULT_TIME_KEYS = ("timestamp", "time", "ts") + + +def normalize_imei(value: str) -> str | None: + """Return the IMEI as a plain 15 digit string, or None if not one.""" + if not _IMEI_RE.match(value): + return None + return value + + +def _as_utc(value: datetime) -> datetime: + """Interpret naive timestamps as UTC.""" + return value if value.tzinfo is not None else value.replace(tzinfo=timezone.utc) + + +def parse_payload_time(value: object) -> datetime | None: + """Best-effort device-timestamp parse. + + Accepts epoch seconds or milliseconds (number or numeric string) and + ISO 8601 strings. Returns an aware UTC datetime or None when the value + cannot be interpreted. Booleans are rejected (bool is an int subclass). + """ + if isinstance(value, bool) or value is None: + return None + if isinstance(value, str): + try: + return _as_utc(datetime.fromisoformat(value)) + except ValueError: + try: + value = float(value) # numeric string -> epoch path + except ValueError: + return None + if isinstance(value, (int, float)): + seconds = float(value) + if seconds > _EPOCH_MS_THRESHOLD: + seconds /= 1000.0 # milliseconds -> seconds + try: + return datetime.fromtimestamp(seconds, tz=timezone.utc) + except (OverflowError, OSError, ValueError): + return None + return None + + +@dataclass(frozen=True) +class InjectionResult: + imei: str + device: str + payload: dict + payload_time: datetime | None = None + + +def process_message( + topic: str, + payload_bytes: bytes, + imei_map: dict[str, str], + *, + imei_key: str = "imei", + device_key: str = "device", + time_keys: tuple[str, ...] = DEFAULT_TIME_KEYS, +) -> InjectionResult | None: + """Parse a payload and inject IMEI + device into it, extracting the + device timestamp. + + Returns None when the payload is not valid JSON or not a JSON object — + those messages are skipped (logged by the caller). The original payload's + own imei/device fields are overwritten by the values derived from the + topic, so the topic is always authoritative. + + A missing or unparseable timestamp does NOT skip the message: the + payload_time is simply None and only received_at (server receipt time) + exists for the row. + """ + parts = topic.split("/") + # IMEI is the last topic segment, e.g. /xxx/ -> imei segment. + raw_imei = parts[-1] if parts else "" + + normalized = normalize_imei(raw_imei) + if normalized is None: + return None + + try: + payload = json.loads(payload_bytes.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + return None + if not isinstance(payload, dict): + return None + + payload = dict(payload) # do not mutate shared objects + payload[imei_key] = normalized + payload[device_key] = imei_map.get(normalized, normalized) + + payload_time = None + for key in time_keys: + if key in payload: + payload_time = parse_payload_time(payload[key]) + if payload_time is not None: + break + + return InjectionResult( + imei=normalized, + device=payload[device_key], + payload=payload, + payload_time=payload_time, + ) \ No newline at end of file diff --git a/listener/requirements.txt b/listener/requirements.txt new file mode 100644 index 0000000..1f0267a --- /dev/null +++ b/listener/requirements.txt @@ -0,0 +1,2 @@ +paho-mqtt==2.1.0 +psycopg[binary]==3.2.3 \ No newline at end of file diff --git a/mosquitto/mosquitto.conf b/mosquitto/mosquitto.conf new file mode 100644 index 0000000..a73aee1 --- /dev/null +++ b/mosquitto/mosquitto.conf @@ -0,0 +1,3 @@ +# Minimal broker configuration for the optional "test" profile service. +listener 1883 +allow_anonymous true \ No newline at end of file diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..b818b4f --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +testpaths = tools +pythonpath = . \ No newline at end of file diff --git a/tools/publish_test_messages.py b/tools/publish_test_messages.py new file mode 100644 index 0000000..9029ccb --- /dev/null +++ b/tools/publish_test_messages.py @@ -0,0 +1,76 @@ +"""Publish sample JSON messages to // topics. + +Usage: + python tools/publish_test_messages.py --broker host:1883 \ + [--prefix "vivab"] [--count 5] [--no-timestamp] + +Requires paho-mqtt (see tools/requirements.txt). +""" + +from __future__ import annotations + +import argparse +import json +import random +import time +from datetime import datetime, timezone +from typing import Any + +import paho.mqtt.client as mqtt +from paho.mqtt.enums import CallbackAPIVersion + +# Sample IMEIs; the first two exist in configs/listener/imei_map.json. +SAMPLE_IMEIS = [ + ("350123456789012", "Greenhouse sensor A"), + ("490154203237518", "Water tank sensor B"), + ("356938035643809", "Weather station roof"), + ("861234012345678", "Soil moisture plot 12"), +] + + +def build_payload(imei: str, seq: int, with_timestamp: bool = True) -> dict[str, Any]: + payload: dict[str, Any] = { + "temperature": round(random.uniform(15, 30), 2), + "humidity": round(random.uniform(40, 90), 2), + "battery_mv": random.randint(2800, 4200), + "seq": seq, + } + if with_timestamp: + # ISO 8601 with explicit UTC offset, like a real device clock + payload["timestamp"] = datetime.now(timezone.utc).isoformat() + return payload + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--broker", default="localhost:1883", help="host:port of the MQTT broker") + parser.add_argument("--prefix", default="vivab", help="topic prefix: //") + parser.add_argument("--count", type=int, default=5, help="messages per IMEI") + parser.add_argument("--qos", type=int, default=1, choices=(0, 1, 2)) + parser.add_argument("--no-timestamp", action="store_true", help="publish without a timestamp property") + args = parser.parse_args() + + host, _, port = args.broker.partition(":") + port = port or "1883" + + client = mqtt.Client(callback_api_version=CallbackAPIVersion.VERSION2, client_id="test-publisher") + client.connect(host, int(port), keepalive=60) + client.loop_start() + try: + seq = 0 + for imei, _name in SAMPLE_IMEIS: + for _ in range(args.count): + seq += 1 + topic = f"/{args.prefix}/{imei}" + payload = json.dumps(build_payload(imei, seq, with_timestamp=not args.no_timestamp)) + info = client.publish(topic, payload, qos=args.qos) + info.wait_for_publish(timeout=5) + print(f"published {topic}: {payload}") + time.sleep(0.1) + finally: + client.loop_stop() + client.disconnect() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tools/requirements.txt b/tools/requirements.txt new file mode 100644 index 0000000..0074298 --- /dev/null +++ b/tools/requirements.txt @@ -0,0 +1,8 @@ +# Local development / test tooling. The container images only install their +# own service requirements; these extras are for running tests and tools on +# the host. +fastapi==0.115.12 +uvicorn[standard]==0.34.2 +psycopg[binary]==3.2.3 +paho-mqtt==2.1.0 +pytest==8.3.5 \ No newline at end of file diff --git a/tools/smoke_test_api.py b/tools/smoke_test_api.py new file mode 100644 index 0000000..68703f5 --- /dev/null +++ b/tools/smoke_test_api.py @@ -0,0 +1,101 @@ +"""End-to-end smoke test against a running stack. + +Checks: + 1. API rejects requests without a token (401) + 2. API rejects requests with a wrong token (401) + 3. API accepts a valid token and returns stored messages + 4. Time framing: a wide from/to window is accepted, a reversed one is rejected + +Usage: + python tools/smoke_test_api.py --url http://localhost:18080 --token + +Only uses the standard library. +""" + +from __future__ import annotations + +import argparse +import json +import sys +import urllib.error +import urllib.request +from urllib.parse import quote + + +def request(url: str, token: str | None) -> tuple[int, str]: + req = urllib.request.Request(url) + if token is not None: + req.add_header("X-App-Token", token) + try: + with urllib.request.urlopen(req, timeout=10) as response: + return response.status, response.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as exc: + return exc.code, exc.read().decode("utf-8", errors="replace") + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--url", default="http://localhost:8080") + parser.add_argument("--token", required=True, help="valid app token") + args = parser.parse_args() + + failures: list[str] = [] + + status, _ = request(f"{args.url}/messages?limit=1", token=None) + if status == 401: + print("PASS: missing token rejected with 401") + else: + failures.append(f"missing token -> expected 401, got {status}") + + status, _ = request(f"{args.url}/messages?limit=1", token="definitely-wrong-token") + if status == 401: + print("PASS: wrong token rejected with 401") + else: + failures.append(f"wrong token -> expected 401, got {status}") + + status, body = request(f"{args.url}/messages?limit=3", token=args.token) + if status == 200: + data = json.loads(body) + print(f"PASS: valid token accepted, {len(data['items'])} message(s) returned") + for item in data["items"]: + print(f" - {item['received_at']} {item['device']}: {json.dumps(item['payload'])}") + else: + failures.append(f"valid token -> expected 200, got {status}: {body}") + + status, body = request(f"{args.url}/messages/count", token=args.token) + if status == 200: + print(f"PASS: count endpoint returned {json.loads(body)}") + else: + failures.append(f"count endpoint -> expected 200, got {status}: {body}") + + wide_from = "2000-01-01T00:00:00Z" + wide_to = "2100-01-01T00:00:00Z" + status, body = request( + f"{args.url}/messages?limit=1&from={quote(wide_from)}&to={quote(wide_to)}", + token=args.token, + ) + if status == 200: + print("PASS: wide from/to window accepted") + else: + failures.append(f"wide from/to window -> expected 200, got {status}: {body}") + + status, body = request( + f"{args.url}/messages?limit=1&from={quote(wide_to)}&to={quote(wide_from)}", + token=args.token, + ) + if status == 422: + print("PASS: reversed from/to window rejected with 422") + else: + failures.append(f"reversed from/to window -> expected 422, got {status}: {body}") + + if failures: + print("\nFAILURES:") + for failure in failures: + print(f" - {failure}") + return 1 + print("\nAll smoke tests passed.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/tools/test_listener.py b/tools/test_listener.py new file mode 100644 index 0000000..5e2f023 --- /dev/null +++ b/tools/test_listener.py @@ -0,0 +1,292 @@ +"""Unit tests for listener.process and listener.main config loading.""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from listener.main import load_imei_map # noqa: E402 +from listener.process import ( # noqa: E402 + InjectionResult, + parse_payload_time, + process_message, +) +from datetime import datetime, timezone # noqa: E402 + +IMEI_MAP = { + "350123456789012": "Greenhouse sensor A", + "490154203237518": "Water tank sensor B", +} + + +class TestNormalizeImei: + @pytest.mark.parametrize( + ("value", "expected"), + [ + ("350123456789012", "350123456789012"), + ("490154203237518", "490154203237518"), + ], + ) + def test_valid(self, value: str, expected: str) -> None: + result = process_message(f"/up/{value}", b"{}", IMEI_MAP) + assert result is not None + assert result.imei == expected + + @pytest.mark.parametrize( + "value", + [ + "", + "too-short", + "3501234567890123", # 16 digits + "0011223344556677", # hex (old dev EUI form) - no longer valid + "35012345678901a", # letter in last position + "35012345678901", # 14 digits + "350-123-456-789-012", # separators not allowed + ], + ) + def test_invalid(self, value: str) -> None: + assert process_message(f"/up/{value}", b"{}", IMEI_MAP) is None + + +class TestProcessMessage: + def test_injects_imei_and_mapped_device(self) -> None: + payload = json.dumps({"temperature": 21.5}).encode() + result = process_message("/vivab/350123456789012", payload, IMEI_MAP) + assert result == InjectionResult( + imei="350123456789012", + device="Greenhouse sensor A", + payload={"temperature": 21.5, "imei": "350123456789012", "device": "Greenhouse sensor A"}, + ) + + def test_unmapped_imei_falls_back_to_imei(self) -> None: + payload = json.dumps({"temperature": 21.5}).encode() + result = process_message("/vivab/356938035643809", payload, IMEI_MAP) + assert result is not None + assert result.device == "356938035643809" + assert result.payload["device"] == "356938035643809" + + def test_overwrites_existing_keys(self) -> None: + payload = json.dumps({"imei": "fake", "device": "fake", "seq": 1}).encode() + result = process_message("/up/490154203237518", payload, IMEI_MAP) + assert result is not None + assert result.payload["imei"] == "490154203237518" + assert result.payload["device"] == "Water tank sensor B" + + def test_does_not_mutate_caller_payload(self) -> None: + original = {"seq": 1} + result = process_message("/up/350123456789012", json.dumps(original).encode(), IMEI_MAP) + assert result is not None + assert original == {"seq": 1} + + def test_invalid_json_returns_none(self) -> None: + assert process_message("/up/350123456789012", b"not json", IMEI_MAP) is None + + def test_non_object_json_returns_none(self) -> None: + assert process_message("/up/350123456789012", b"[1,2,3]", IMEI_MAP) is None + + def test_custom_keys(self) -> None: + payload = json.dumps({"seq": 1}).encode() + result = process_message( + "/up/350123456789012", payload, IMEI_MAP, imei_key="id", device_key="name" + ) + assert result is not None + assert result.payload["id"] == "350123456789012" + assert result.payload["name"] == "Greenhouse sensor A" + + +class TestParsePayloadTime: + def test_iso_string_with_offset(self) -> None: + parsed = parse_payload_time("2026-08-31T12:34:56+02:00") + assert parsed == datetime(2026, 8, 31, 10, 34, 56, tzinfo=timezone.utc) + + def test_iso_string_naive_treated_as_utc(self) -> None: + parsed = parse_payload_time("2026-08-31T12:34:56") + assert parsed == datetime(2026, 8, 31, 12, 34, 56, tzinfo=timezone.utc) + + def test_iso_string_with_z(self) -> None: + parsed = parse_payload_time("2026-08-31T12:34:56Z") + assert parsed == datetime(2026, 8, 31, 12, 34, 56, tzinfo=timezone.utc) + + def test_epoch_seconds(self) -> None: + parsed = parse_payload_time(1756646096) + assert parsed == datetime.fromtimestamp(1756646096, tz=timezone.utc) + + def test_epoch_milliseconds_detected(self) -> None: + # ms magnitude (>1e11) must be divided by 1000 + parsed = parse_payload_time(1756646096123) + assert parsed == datetime.fromtimestamp(1756646096.123, tz=timezone.utc) + + def test_numeric_string_epoch(self) -> None: + parsed = parse_payload_time("1756646096") + assert parsed == datetime.fromtimestamp(1756646096, tz=timezone.utc) + + def test_rejects_bool(self) -> None: + # bool is an int subclass; True must not parse as epoch 1 + assert parse_payload_time(True) is None + assert parse_payload_time(False) is None + + @pytest.mark.parametrize("value", [None, "", "not-a-date", "12:34", [], {}]) + def test_rejects_garbage(self, value: object) -> None: + assert parse_payload_time(value) is None + + +class TestPayloadTimeExtraction: + def test_timestamp_extracted_from_payload(self) -> None: + payload = json.dumps({"temperature": 21.5, "timestamp": "2026-08-31T09:59:58Z"}).encode() + result = process_message("/up/350123456789012", payload, IMEI_MAP) + assert result is not None + assert result.payload_time == datetime(2026, 8, 31, 9, 59, 58, tzinfo=timezone.utc) + # the original payload value stays untouched + assert result.payload["timestamp"] == "2026-08-31T09:59:58Z" + + def test_first_parseable_key_wins(self) -> None: + payload = json.dumps({"timestamp": "garbage", "time": 1756646096}).encode() + result = process_message("/up/350123456789012", payload, IMEI_MAP) + assert result is not None + assert result.payload_time == datetime.fromtimestamp(1756646096, tz=timezone.utc) + + def test_custom_time_keys(self) -> None: + payload = json.dumps({"measured_at": "2026-08-31T09:59:58Z"}).encode() + result = process_message( + "/up/350123456789012", payload, IMEI_MAP, time_keys=("measured_at",) + ) + assert result is not None + assert result.payload_time == datetime(2026, 8, 31, 9, 59, 58, tzinfo=timezone.utc) + + def test_missing_timestamp_yields_none_not_skip(self) -> None: + payload = json.dumps({"temperature": 21.5}).encode() + result = process_message("/up/350123456789012", payload, IMEI_MAP) + assert result is not None # message still stored + assert result.payload_time is None + + def test_unparseable_timestamp_yields_none_not_skip(self) -> None: + payload = json.dumps({"timestamp": "yesterday-ish"}).encode() + result = process_message("/up/350123456789012", payload, IMEI_MAP) + assert result is not None + assert result.payload_time is None + + +class TestLoadImeiMap: + def test_loads_valid_map(self, tmp_path: Path) -> None: + path = tmp_path / "map.json" + path.write_text( + json.dumps({"350123456789012": "Device X"}), encoding="utf-8" + ) + loaded = load_imei_map(str(path)) + assert loaded == {"350123456789012": "Device X"} + + def test_rejects_non_object(self, tmp_path: Path) -> None: + path = tmp_path / "map.json" + path.write_text("[]", encoding="utf-8") + with pytest.raises(ValueError, match="expected a JSON object"): + load_imei_map(str(path)) + + def test_rejects_non_string_values(self, tmp_path: Path) -> None: + path = tmp_path / "map.json" + path.write_text(json.dumps({"350123456789012": 42}), encoding="utf-8") + with pytest.raises(ValueError, match="non-empty strings"): + load_imei_map(str(path)) + + def test_rejects_non_imei_key(self, tmp_path: Path) -> None: + path = tmp_path / "map.json" + # Old dev EUI-style key: no longer accepted + path.write_text(json.dumps({"0011223344556677": "Device X"}), encoding="utf-8") + with pytest.raises(ValueError, match="15-digit IMEI"): + load_imei_map(str(path)) + + +class TestApiAuth: + """Light tests of the API auth dependency using the ASGI app directly.""" + + @pytest.fixture() + def app_module(self, tmp_path): + import importlib + import os + + # api.app runs create_app() at import time, so the environment must be + # prepared before importing. connect_timeout keeps the DB-connecting + # tests fast even when localhost:5432 is silently dropped by a firewall + # (psycopg otherwise waits out the OS-level TCP timeout). + tokens = tmp_path / "app_tokens.json" + tokens.write_text( + json.dumps({"tokens": {"test-app": "secret-abc"}}), encoding="utf-8" + ) + env = { + "APP_TOKENS_FILE": str(tokens), + "DB_DSN": "postgresql://user:pass@localhost:5432/db?connect_timeout=2", + } + old = {key: os.environ.get(key) for key in env} + os.environ.update(env) + try: + import api.app as app_module + + importlib.reload(app_module) + yield app_module + finally: + for key, value in old.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + def test_health_is_public(self, app_module) -> None: + from fastapi.testclient import TestClient + + client = TestClient(app_module.app) + assert client.get("/health").status_code == 405 or client.get("/health").json() == {"status": "ok"} + + def test_missing_token_rejected(self, app_module) -> None: + from fastapi.testclient import TestClient + + client = TestClient(app_module.app) + response = client.get("/messages?limit=1") + assert response.status_code == 401 + + def test_wrong_token_rejected(self, app_module) -> None: + from fastapi.testclient import TestClient + + client = TestClient(app_module.app) + response = client.get("/messages?limit=1", headers={"X-App-Token": "nope"}) + assert response.status_code == 401 + + def test_valid_token_accepted(self, app_module) -> None: + from fastapi.testclient import TestClient + + # The DB is not reachable in unit tests, so a valid token proceeds past + # auth and then fails on the database connection with a 500. That + # proves authentication passed (401 would mean it did not). + client = TestClient(app_module.app, raise_server_exceptions=False) + response = client.get("/messages?limit=1", headers={"X-App-Token": "secret-abc"}) + assert response.status_code != 401 + + def test_reversed_time_frame_rejected(self, app_module) -> None: + from fastapi.testclient import TestClient + + # Range validation runs before any database access, so this works + # without a reachable DB. + client = TestClient(app_module.app) + response = client.get( + "/messages", + params={"from": "2026-08-31T12:00:00Z", "to": "2026-08-31T11:00:00Z"}, + headers={"X-App-Token": "secret-abc"}, + ) + assert response.status_code == 422 + + def test_naive_and_offset_timestamps_accepted(self, app_module) -> None: + from fastapi.testclient import TestClient + + # Both forms must parse; the request fails later on DB access (500), + # never with a 422 validation error. + client = TestClient(app_module.app, raise_server_exceptions=False) + for value in ("2026-08-31T12:00:00", "2026-08-31T12:00:00+02:00"): + response = client.get( + "/messages", + params={"from": value}, + headers={"X-App-Token": "secret-abc"}, + ) + assert response.status_code != 422 \ No newline at end of file