diff --git a/README.md b/README.md index ffbf2fb..9bcb7b2 100644 --- a/README.md +++ b/README.md @@ -94,13 +94,17 @@ Authentication: pass one of the configured app tokens either as the `X-App-Token | -------- | ---- | ----------- | | `GET /health` | none | Liveness probe (public) | | `GET /messages/count` | token | Total number of stored messages | -| `GET /messages` | token | List messages, newest first. The device's **current settings document** (from `imei_map`) is merged into each item's `payload`; on duplicate keys the stored payload wins. 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 /messages` | token | List messages, newest first. The device's **current settings document** (from `imei_map`) is merged into each item's `payload`; on duplicate keys the stored payload wins. Each item also carries an **`attributes`** sibling field with the device's time-varying attributes **resolved as of the message's device time** (`payload_time`, `received_at` as the fallback) — apply the offsets to the payload values yourself (e.g. `payload[k] + attributes[k]`). 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 /configs/{name}` | token | Read a config as JSON. `name`: `app_tokens` (secrets are not recoverable — app names with `null` placeholders) or `imei_map` (all device settings documents) | | `PUT/POST /configs/{name}` | token | **Bulk replace** a config in the database (delete + insert, transactional). Body **must be a JSON object**; per-config schema is validated (`imei_map` keys must be 15-digit IMEIs, values are settings documents — flat `"name"` values are promoted to `{"device": "name"}`; `app_tokens` must keep the `{"tokens": {name: secret}}` shape). Invalid payloads → 422, unknown names → 404. `app_tokens` changes take effect immediately; `imei_map` changes are picked up by the listener within ~5 seconds | | `GET /devices` | token | All devices, each as `{"imei": ..., **settings}` | | `GET /devices/{imei}` | token | One device's settings document; unknown IMEI → 404, non-15-digit IMEI → 422 | | `PUT /devices/{imei}` | token | **Upsert one device's settings document** (create or replace). Body is the settings object (must have a non-empty `device` key; extra keys are arbitrary JSON); a body `imei` key is tolerated when it matches the URL. Returns the saved document, so a GET response can be edited and PUT straight back | | `DELETE /devices/{imei}` | token | Remove one device (its messages stay); unknown IMEI → 404 | +| `GET /devices/{imei}/attributes` | token | Time-varying attributes (e.g. calibration offsets) **resolved as of a time**: `?at=ISO8601` (default now) returns `{key: value}` — for each key the version with the greatest `effective_from <= at`. Keys whose first version lies after `at` are absent | +| `GET /devices/{imei}/attributes/history` | token | The full timeline per key, oldest version first: `{key: [{value, effective_from, note}, ...]}` | +| `PUT/POST /devices/{imei}/attributes` | token | **Append one version** to a key's timeline. Body: `{"key": ..., "value": ..., "effective_from": ISO8601?, "note"?}`; `effective_from` defaults to now, backdating allowed (a sensor swap discovered after the fact becomes a backdated row). Re-using an exact existing `(key, effective_from)` edits that version in place | +| `DELETE /devices/{imei}/attributes/{key}` | token | Delete **only the latest** version of a key (typo/undo); earlier history stays immutable and the previous value is back in effect. Unknown key → 404 | | `GET /tokens` | token | App names with a token (secrets are unrecoverable, shown as `null`) | | `PUT /tokens/{app}` | token | **Add or rotate ONE app token** (single-row upsert — never touches the other apps' rows). Body: `{"token": "secret"}`. Takes effect immediately | | `DELETE /tokens/{app}` | token | Remove one app token; unknown name → 404. Deleting the last token locks the API until a new one is added | @@ -139,6 +143,16 @@ curl -X PUT -H "X-App-Token: $TOKEN" -H "Content-Type: application/json" \ **Read-time settings merge:** `GET /messages` joins `imei_map` and merges the device's *current* settings document into each item's `payload` (`COALESCE(im.meta, '{}'::jsonb) || m.payload` — the stored payload wins on duplicate keys). Historical messages therefore always reflect the *latest* device settings; the ingest-time `imei`/`device` snapshots in the stored payload and columns are unaffected. +**Time-varying attributes (offsets):** values that may change over time — calibration offsets, mounting heights, anything recalibrated — live in the `imei_attributes` table as an **append-only timeline** (one row per version; the value valid at time T is the row with the greatest `effective_from <= T`). They deliberately do NOT go into `imei_map.meta` (which is current-only): a message must be corrected with the value that was in effect **when it was measured**. `GET /messages` resolves each message's attributes against the device time (`payload_time`, `received_at` as the fallback) via a lateral join and returns them as a sibling **`attributes`** field per item — not merged into `payload` — so consumers apply the corrections themselves and cannot double-apply or retro-bake them. Example: record an offset change discovered after a sensor swap by backdating it, and every message since the swap re-resolves correctly: + +```bash +curl -X PUT -H "X-App-Token: $TOKEN" -H "Content-Type: application/json" \ + -d '{"key": "temperature_offset", "value": 2.0, "effective_from": "2026-08-15T00:00:00Z", "note": "after sensor swap"}' \ + http://localhost:18080/devices/350123456789012/attributes +``` + +The listener never reads `imei_attributes`; attributes are a read-time correction, and rows may exist for IMEIs without a settings row (no FK, like `mqtt_messages`). + #### Migrating an existing deployment from the JSON files For a database that already has data (init scripts only run on an empty volume), apply the schema and migrate the JSON configs. The schema script is idempotent and **migrates the old `imei_map(device TEXT)` layout in place**: every existing row is converted to `{"device": ...}` and the legacy column is dropped: @@ -175,6 +189,9 @@ Example response: "device": "Greenhouse sensor A", "location": "Växthus 1" }, + "attributes": { + "temperature_offset": 2.0 + }, "payload_time": "2026-08-31T09:59:58+00:00" } ] diff --git a/api/app.py b/api/app.py index 4e2d5a8..6e0525a 100644 --- a/api/app.py +++ b/api/app.py @@ -7,7 +7,10 @@ PUT /configs/imei_map or — the way a management frontend should work — per device via the /devices/{imei} resource endpoints (single-row upsert, so concurrent edits of different devices never clobber each other). GET /messages merges each device's CURRENT settings document into the -payload at read time; stored payload keys win on conflicts. +payload at read time; stored payload keys win on conflicts. It also +resolves each device's TIME-VARYING attributes (offsets etc., from the +imei_attributes timeline) as of the message's device time and returns +them as a sibling "attributes" field. Authentication: app tokens stored as SHA-256 digests in the app_tokens table, passed either as the `X-App-Token` header or as an `app_token` query @@ -280,6 +283,118 @@ class ImeiMapStore: conn.close() +class ImeiAttributesStore: + """Time-varying device attributes (e.g. calibration offsets) backed by + the imei_attributes table: an append-only timeline where a new value + is a new row. The value valid at time T is the row with the greatest + effective_from <= T. + + Attributes are read-time corrections only: the listener never reads + this table, and rows may exist for IMEIs without an imei_map entry + (there is no FK, like on mqtt_messages).""" + + def __init__(self, dsn: str) -> None: + self._dsn = dsn + + def history(self, imei: str) -> dict[str, list[dict[str, Any]]]: + """Full timeline as {key: [versions]}, oldest first.""" + conn = psycopg.connect(self._dsn) + try: + with conn.cursor() as cur: + cur.execute( + "SELECT key, value, effective_from, note FROM imei_attributes " + "WHERE imei = %s ORDER BY key, effective_from", + (imei,), + ) + result: dict[str, list[dict[str, Any]]] = {} + for key, value, effective_from, note in cur.fetchall(): + result.setdefault(key, []).append( + {"value": value, "effective_from": effective_from, "note": note} + ) + return result + finally: + conn.close() + + def resolved(self, imei: str, at: datetime | None = None) -> dict[str, Any]: + """{key: value} as of `at` (None = now): for each key the version + with the greatest effective_from <= at. Keys whose first version + lies after `at` are absent (no value was in effect yet).""" + conn = psycopg.connect(self._dsn) + try: + with conn.cursor() as cur: + cur.execute( + """ + SELECT DISTINCT ON (key) key, value + FROM imei_attributes + WHERE imei = %s AND effective_from <= COALESCE(%s, now()) + ORDER BY key, effective_from DESC + """, + (imei, at), + ) + return dict(cur.fetchall()) + finally: + conn.close() + + def append( + self, + imei: str, + key: str, + value: Any, + effective_from: datetime | None = None, + note: str | None = None, + ) -> dict[str, Any]: + """Append one version (effective_from defaults to now()). + + Re-using an exact existing (imei, key, effective_from) replaces + that version in place - "edit the entry I just saved" for backdated + corrections. Returns the stored version.""" + # autocommit: a single-statement write commits immediately (a plain + # connection would roll back on close). + conn = psycopg.connect(self._dsn, autocommit=True) + try: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO imei_attributes (imei, key, value, effective_from, note) + VALUES (%s, %s, %s, COALESCE(%s, now()), %s) + ON CONFLICT (imei, key, effective_from) DO UPDATE + SET value = EXCLUDED.value, note = EXCLUDED.note, created_at = now() + RETURNING key, value, effective_from, note + """, + (imei, key, Jsonb(value), effective_from, note), + ) + row = cur.fetchone() + return { + "key": row[0], + "value": row[1], + "effective_from": row[2], + "note": row[3], + } + finally: + conn.close() + + def delete_latest(self, imei: str, key: str) -> bool: + """Delete ONLY the newest version of a key (typo/undo); the earlier + history stays immutable. Returns False when the key is unknown.""" + conn = psycopg.connect(self._dsn, autocommit=True) + try: + with conn.cursor() as cur: + cur.execute( + """ + DELETE FROM imei_attributes + WHERE imei = %s AND key = %s + AND effective_from = ( + SELECT max(effective_from) FROM imei_attributes + WHERE imei = %s AND key = %s + ) + """, + (imei, key, imei, key), + ) + return cur.rowcount > 0 + finally: + conn.close() + + def _validate_token_body(data: dict[str, Any], source: str) -> dict[str, str]: """Validate a {"tokens": {app: secret}} request body and return the inner tokens mapping. Raises ValueError with a descriptive message @@ -311,12 +426,53 @@ def _validate_single_token_body(data: dict[str, Any], source: str) -> str: return data["token"] +def _validate_attribute_body( + data: dict[str, Any], source: str +) -> tuple[str, Any, datetime | None, str | None]: + """Validate a device-attribute append body and return + (key, value, effective_from, note). effective_from None means "now". + Raises ValueError with a descriptive message otherwise.""" + if not isinstance(data, dict): + raise ValueError( + f'{source}: expected JSON on the form {{"key": ..., "value": ...}}' + ) + key = data.get("key") + if not isinstance(key, str) or not key.strip(): + raise ValueError(f'{source}: "key" must be a non-empty string') + if "value" not in data or data["value"] is None: + raise ValueError(f'{source}: "value" must carry a JSON value') + effective_from: datetime | None = None + if data.get("effective_from") is not None: + raw = data["effective_from"] + if not isinstance(raw, str): + raise ValueError(f'{source}: "effective_from" must be an ISO 8601 string') + try: + effective_from = _as_utc(datetime.fromisoformat(raw)) + except ValueError: + raise ValueError( + f'{source}: "effective_from" is not a valid ISO 8601 timestamp' + ) from None + note = data.get("note") + if note is not None and not isinstance(note, str): + raise ValueError(f'{source}: "note" must be a string when present') + return key.strip(), data["value"], effective_from, note + + def _device_repr(imei: str, doc: dict[str, Any]) -> dict[str, Any]: """API representation of a device: its settings document plus the IMEI, so a GET response can be edited and PUT straight back.""" return {"imei": imei, **doc} +def _version_repr(version: dict[str, Any]) -> dict[str, Any]: + """API representation of one device-attribute timeline version.""" + return { + "value": version["value"], + "effective_from": version["effective_from"].isoformat(), + "note": version["note"], + } + + def _env(name: str, default: str | None = None) -> str: value = os.environ.get(name) if not value: @@ -334,6 +490,7 @@ def _as_utc(value: datetime) -> datetime: def create_app( token_store: TokenStore | None = None, imei_store: ImeiMapStore | None = None, + attributes_store: ImeiAttributesStore | None = None, ) -> FastAPI: """Build the API application. @@ -358,6 +515,8 @@ def create_app( ) if imei_store is None: imei_store = ImeiMapStore(dsn) + if attributes_store is None: + attributes_store = ImeiAttributesStore(dsn) def authenticate(request: Request) -> None: token = request.headers.get("X-App-Token") @@ -510,6 +669,128 @@ def create_app( logger.info("device %s deleted", normalized) return {"status": "deleted", "imei": normalized} + # --- Time-varying device attributes (e.g. calibration offsets): an + # append-only timeline per IMEI. The value valid at T is the newest + # version with effective_from <= T; GET /messages resolves each + # message's attributes against the device time (payload_time, with + # received_at as the fallback). Attributes are API-only - the listener + # never reads them, and they may exist for IMEIs without a settings + # row (no FK). + + @app.get( + "/devices/{imei}/attributes/history", + tags=["devices"], + dependencies=[Depends(authenticate)], + ) + def get_device_attributes_history(imei: str) -> dict[str, Any]: + normalized = normalize_imei(imei) + if normalized is None: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"{imei!r} is not a 15-digit IMEI", + ) + try: + timeline = attributes_store.history(normalized) + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc + return { + "imei": normalized, + "attributes": { + key: [_version_repr(version) for version in versions] + for key, versions in timeline.items() + }, + } + + @app.get("/devices/{imei}/attributes", tags=["devices"], dependencies=[Depends(authenticate)]) + def get_device_attributes( + imei: str, + at: datetime | None = Query( + None, + description=( + "Resolve the values as of this time (ISO 8601, inclusive; naive " + "values are treated as UTC). Default: now." + ), + ), + ) -> dict[str, Any]: + normalized = normalize_imei(imei) + if normalized is None: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"{imei!r} is not a 15-digit IMEI", + ) + resolved_at = _as_utc(at) if at is not None else None + try: + values = attributes_store.resolved(normalized, resolved_at) + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc + result: dict[str, Any] = {"imei": normalized, "attributes": values} + if resolved_at is not None: + result["at"] = resolved_at.isoformat() + return result + + @app.put("/devices/{imei}/attributes", tags=["devices"], dependencies=[Depends(authenticate)]) + @app.post("/devices/{imei}/attributes", tags=["devices"], dependencies=[Depends(authenticate)]) + def add_device_attribute(imei: str, body: dict[str, Any]) -> dict[str, Any]: + normalized = normalize_imei(imei) + if normalized is None: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"{imei!r} is not a 15-digit IMEI", + ) + try: + key, value, effective_from, note = _validate_attribute_body(body, "request body") + stored = attributes_store.append(normalized, key, value, effective_from, note) + except ValueError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=str(exc), + ) from exc + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc + logger.info( + "attribute %r appended for %s (effective %s)", + stored["key"], + normalized, + stored["effective_from"].isoformat(), + ) + return { + "imei": normalized, + "key": stored["key"], + "value": stored["value"], + "effective_from": stored["effective_from"].isoformat(), + "note": stored["note"], + } + + @app.delete( + "/devices/{imei}/attributes/{key}", + tags=["devices"], + dependencies=[Depends(authenticate)], + ) + def delete_device_attribute(imei: str, key: str) -> dict[str, Any]: + normalized = normalize_imei(imei) + if normalized is None: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"{imei!r} is not a 15-digit IMEI", + ) + name = key.strip() + if not name: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="attribute key must be a non-empty string", + ) + try: + deleted = attributes_store.delete_latest(normalized, name) + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc + if not deleted: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"no versions of attribute {name!r} for device {normalized}", + ) + logger.info("attribute %r latest version deleted for %s", name, normalized) + return {"status": "deleted", "imei": normalized, "key": name} + # --- Token resources: add/rotate/remove ONE app token without # touching the other apps' rows. @@ -676,10 +957,29 @@ def create_app( # concatenation makes the RIGHT operand win on duplicate keys: the # stored payload (the device's own data plus the ingest-time # imei/device snapshot) takes precedence over the settings doc. + # + # The LATERAL join resolves the device's TIME-VARYING attributes + # (offsets etc.) as of the message's DEVICE time (payload_time, + # received_at as the fallback): for each key the newest version + # with effective_from <= that time. Returned as a sibling + # "attributes" field so consumers apply the corrections themselves; + # an absent key means no value was in effect then. query = sql.SQL( "SELECT m.id, m.received_at, m.topic, m.imei, m.device, " - "COALESCE(im.meta, '{}'::jsonb) || m.payload, m.payload_time " - "FROM mqtt_messages m LEFT JOIN imei_map im ON im.imei = m.imei" + "COALESCE(im.meta, '{}'::jsonb) || m.payload, m.payload_time, " + "attrs.resolved " + "FROM mqtt_messages m " + "LEFT JOIN imei_map im ON im.imei = m.imei " + "LEFT JOIN LATERAL (" + " SELECT COALESCE(jsonb_object_agg(latest.key, latest.value), '{}'::jsonb) AS resolved " + " FROM (" + " SELECT DISTINCT ON (a.key) a.key, a.value " + " FROM imei_attributes a " + " WHERE a.imei = m.imei " + " AND a.effective_from <= COALESCE(m.payload_time, m.received_at) " + " ORDER BY a.key, a.effective_from DESC " + " ) latest" + ") attrs ON true" ) + where + order_by try: @@ -709,6 +1009,7 @@ def create_app( "device": row[4], "payload": row[5], "payload_time": row[6].isoformat() if row[6] is not None else None, + "attributes": row[7], } for row in rows ], diff --git a/init/01_schema.sql b/init/01_schema.sql index a52ccdf..58368de 100644 --- a/init/01_schema.sql +++ b/init/01_schema.sql @@ -83,4 +83,26 @@ COMMENT ON TABLE app_tokens IS COMMENT ON TABLE imei_map IS 'Device settings (one JSONB document per IMEI), polled by the listener every 5 seconds; the API merges the current document into message payloads at read time.'; COMMENT ON COLUMN imei_map.meta IS - 'Settings document; "device" (non-empty string) is the device name injected at ingest, other keys are merged into message payloads at read time.'; \ No newline at end of file + 'Settings document; "device" (non-empty string) is the device name injected at ingest, other keys are merged into message payloads at read time.'; + +-- Time-varying device attributes (e.g. calibration offsets): an +-- append-only timeline. A new value = a new row; the value valid at time T +-- is the row with the greatest effective_from <= T. GET /messages resolves +-- each message's attributes against the device time (payload_time, +-- falling back to received_at). No FK to imei_map: like mqtt_messages, +-- attributes may exist for IMEIs without a settings row. The listener +-- never reads this table (attributes are read-time corrections). +CREATE TABLE IF NOT EXISTS imei_attributes ( + imei TEXT NOT NULL CHECK (imei ~ '^[0-9]{15}$'), + key TEXT NOT NULL CHECK (key <> ''), + value JSONB NOT NULL, + effective_from TIMESTAMPTZ NOT NULL DEFAULT now(), + note TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (imei, key, effective_from) +); + +COMMENT ON TABLE imei_attributes IS + 'Time-varying device attributes (e.g. offsets) as an append-only timeline; the value valid at T is the row with the greatest effective_from <= T.'; +COMMENT ON COLUMN imei_attributes.effective_from IS + 'Time from which the value applies; backdating allowed. Resolved against the device timestamp (payload_time, received_at fallback) in GET /messages.'; \ No newline at end of file diff --git a/tools/smoke_test_api.py b/tools/smoke_test_api.py index 5bf6954..dbfca8b 100644 --- a/tools/smoke_test_api.py +++ b/tools/smoke_test_api.py @@ -7,6 +7,7 @@ Checks: 4. Time framing: a wide from/to window is accepted, a reversed one is rejected 5. Device resources: a scratch device is upserted, read back and deleted 6. Token resources: a scratch token authenticates and is then removed + 7. Device attributes: an offset timeline is appended and resolved Usage: python tools/smoke_test_api.py --url http://localhost:18080 --token @@ -166,6 +167,52 @@ def main() -> int: else: failures.append(f"removed token auth -> expected 401, got {status}: {body}") + # 7. Device attributes: append an offset timeline, resolve it, delete it + attrs_url = f"{args.url}/devices/{scratch_imei}/attributes" + for value, effective in ((1.5, "2000-01-01T00:00:00Z"), (2.5, "2020-01-01T00:00:00Z")): + status, body = request( + attrs_url, + token=args.token, + method="PUT", + data=json.dumps({"key": "offset", "value": value, "effective_from": effective}).encode(), + ) + if status == 200: + print(f"PASS: attribute version {value} appended") + else: + failures.append(f"attribute append {value} -> expected 200, got {status}: {body}") + + status, body = request( + f"{attrs_url}?at={quote('2010-06-01T00:00:00Z')}", token=args.token + ) + if status == 200 and json.loads(body)["attributes"] == {"offset": 1.5}: + print("PASS: attribute resolved as of 2010 (first version)") + else: + failures.append(f"attribute resolve at 2010 -> expected offset 1.5, got {status}: {body}") + + status, body = request(attrs_url, token=args.token) + if status == 200 and json.loads(body)["attributes"] == {"offset": 2.5}: + print("PASS: attribute resolved as of now (latest version)") + else: + failures.append(f"attribute resolve now -> expected offset 2.5, got {status}: {body}") + + status, body = request(f"{attrs_url}/history", token=args.token) + if status == 200 and [v["value"] for v in json.loads(body)["attributes"]["offset"]] == [1.5, 2.5]: + print("PASS: attribute history lists both versions oldest first") + else: + failures.append(f"attribute history -> expected [1.5, 2.5], got {status}: {body}") + + status, body = request(f"{attrs_url}/offset", token=args.token, method="DELETE") + if status == 200: + print("PASS: latest attribute version deleted") + else: + failures.append(f"attribute delete -> expected 200, got {status}: {body}") + + status, body = request(attrs_url, token=args.token) + if status == 200 and json.loads(body)["attributes"] == {"offset": 1.5}: + print("PASS: earlier attribute version back in effect") + else: + failures.append(f"attribute after delete -> expected offset 1.5, got {status}: {body}") + if failures: print("\nFAILURES:") for failure in failures: diff --git a/tools/test_listener.py b/tools/test_listener.py index eb28d15..546d6a6 100644 --- a/tools/test_listener.py +++ b/tools/test_listener.py @@ -18,7 +18,14 @@ os.environ.setdefault( "DB_DSN", "postgresql://user:pass@localhost:5432/db?connect_timeout=2" ) -from api.app import ImeiMapStore, TokenStore, _validate_token_body, hash_token # noqa: E402 +from api.app import ( # noqa: E402 + ImeiAttributesStore, + ImeiMapStore, + TokenStore, + _validate_attribute_body, + _validate_token_body, + hash_token, +) from listener.process import ( # noqa: E402 InjectionResult, parse_payload_time, @@ -348,6 +355,53 @@ class TestTokenHelpers: _validate_token_body({"tokens": {"app": ""}}, "test") +class TestValidateAttributeBody: + """_validate_attribute_body is the schema check for attribute appends.""" + + def test_accepts_full_body(self) -> None: + key, value, effective_from, note = _validate_attribute_body( + {"key": "temperature_offset", "value": 2.5, + "effective_from": "2026-08-15T00:00:00Z", "note": "after sensor swap"}, + "test", + ) + assert key == "temperature_offset" + assert value == 2.5 + assert effective_from == datetime(2026, 8, 15, tzinfo=timezone.utc) + assert note == "after sensor swap" + + def test_effective_from_defaults_to_none_for_now(self) -> None: + key, value, effective_from, note = _validate_attribute_body( + {"key": "level_offset", "value": -1}, "test" + ) + assert effective_from is None + assert note is None + + def test_allows_any_json_value(self) -> None: + for value in (0, 2.5, "text", True, [1, 2], {"nested": True}): + _, parsed, _, _ = _validate_attribute_body({"key": "k", "value": value}, "test") + assert parsed == value + + def test_rejects_missing_key(self) -> None: + with pytest.raises(ValueError, match='"key" must be a non-empty string'): + _validate_attribute_body({"value": 1}, "test") + + def test_rejects_missing_or_null_value(self) -> None: + with pytest.raises(ValueError, match='"value" must carry a JSON value'): + _validate_attribute_body({"key": "k"}, "test") + with pytest.raises(ValueError, match='"value" must carry a JSON value'): + _validate_attribute_body({"key": "k", "value": None}, "test") + + def test_rejects_bad_effective_from(self) -> None: + with pytest.raises(ValueError, match="not a valid ISO 8601"): + _validate_attribute_body( + {"key": "k", "value": 1, "effective_from": "not-a-date"}, "test" + ) + + def test_rejects_non_string_note(self) -> None: + with pytest.raises(ValueError, match='"note" must be a string'): + _validate_attribute_body({"key": "k", "value": 1, "note": 42}, "test") + + class FakeTokenStore(TokenStore): """In-memory token store: no database, immediate verification.""" @@ -409,6 +463,71 @@ class FakeImeiMapStore(ImeiMapStore): self._mapping = {imei: dict(doc) for imei, doc in mapping.items()} +class FakeImeiAttributesStore(ImeiAttributesStore): + """In-memory device-attributes timeline store: no database.""" + + def __init__(self) -> None: + super().__init__("postgresql://unused") + # {imei: {key: [(effective_from, value, note), ...]}} newest last + self._timeline: dict[str, dict[str, list[tuple[datetime, Any, str | None]]]] = {} + self.appended: tuple[str, str, Any, datetime | None, str | None] | None = None + self.deleted: tuple[str, str] | None = None + + def history(self, imei: str) -> dict[str, list[dict[str, Any]]]: + result: dict[str, list[dict[str, Any]]] = {} + for key, versions in self._timeline.get(imei, {}).items(): + result[key] = [ + {"value": value, "effective_from": effective_from, "note": note} + for effective_from, value, note in sorted(versions) + ] + return result + + def resolved(self, imei: str, at: datetime | None = None) -> dict[str, Any]: + import datetime as dt_mod + + when = at if at is not None else dt_mod.datetime.now(dt_mod.timezone.utc) + resolved: dict[str, Any] = {} + for key, versions in self._timeline.get(imei, {}).items(): + applicable = [ + (effective_from, value) + for effective_from, value, _note in versions + if effective_from <= when + ] + if applicable: + resolved[key] = max(applicable)[1] + return resolved + + def append( + self, + imei: str, + key: str, + value: Any, + effective_from: datetime | None = None, + note: str | None = None, + ) -> dict[str, Any]: + import datetime as dt_mod + + when = effective_from if effective_from is not None else dt_mod.datetime.now(dt_mod.timezone.utc) + self.appended = (imei, key, value, effective_from, note) + versions = self._timeline.setdefault(imei, {}).setdefault(key, []) + # replace-on-conflict: same (imei, key, effective_from) edits in place + versions = [v for v in versions if v[0] != when] + versions.append((when, value, note)) + self._timeline[imei][key] = versions + return {"key": key, "value": value, "effective_from": when, "note": note} + + def delete_latest(self, imei: str, key: str) -> bool: + self.deleted = (imei, key) + versions = self._timeline.get(imei, {}).get(key) + if not versions: + return False + newest = max(effective_from for effective_from, _value, _note in versions) + self._timeline[imei][key] = [ + v for v in versions if v[0] != newest + ] + return True + + @pytest.fixture() def app_module(monkeypatch): """Import api.app with a fake environment (create_app runs at import @@ -433,13 +552,17 @@ def app_module(monkeypatch): @pytest.fixture() def api(app_module): - """ASGI app wired to in-memory fakes; yields (app, token_store, imei_store).""" + """ASGI app wired to in-memory fakes; yields + (app, token_store, imei_store, attributes_store).""" tokens = FakeTokenStore({"test-app": "secret-abc"}) imeis = FakeImeiMapStore( {"350123456789012": {"device": "Greenhouse sensor A"}} ) - app = app_module.create_app(token_store=tokens, imei_store=imeis) - return app, tokens, imeis + attributes = FakeImeiAttributesStore() + app = app_module.create_app( + token_store=tokens, imei_store=imeis, attributes_store=attributes + ) + return app, tokens, imeis, attributes class TestApiAuth: @@ -529,7 +652,7 @@ class TestConfigEndpoints: def client_and_stores(self, api): from fastapi.testclient import TestClient - app, tokens, imeis = api + app, tokens, imeis, _attributes = api return TestClient(app), tokens, imeis def test_get_imei_map(self, client_and_stores) -> None: @@ -634,7 +757,7 @@ class TestDeviceEndpoints: def client_and_store(self, api): from fastapi.testclient import TestClient - app, _, imeis = api + app, _, imeis, _attributes = api return TestClient(app), imeis def test_list_devices(self, client_and_store) -> None: @@ -744,7 +867,7 @@ class TestTokenEndpoints: def client_and_store(self, api): from fastapi.testclient import TestClient - app, tokens, _ = api + app, tokens, _, _attributes = api return TestClient(app), tokens def test_list_tokens_masks_secrets(self, client_and_store) -> None: @@ -790,4 +913,178 @@ class TestTokenEndpoints: client, _ = client_and_store assert client.get("/tokens").status_code == 401 assert client.put("/tokens/x", json={"token": "y"}).status_code == 401 - assert client.delete("/tokens/x").status_code == 401 \ No newline at end of file + assert client.delete("/tokens/x").status_code == 401 + + +class TestDeviceAttributeEndpoints: + """GET/PUT/DELETE /devices/{imei}/attributes — the time-varying + attribute timeline (append-only; resolved as of a point in time).""" + + HDRS = {"X-App-Token": "secret-abc"} + + @pytest.fixture() + def client_and_store(self, api): + from fastapi.testclient import TestClient + + app, _tokens, _imeis, attributes = api + return TestClient(app), attributes + + def test_endpoints_require_token(self, client_and_store) -> None: + client, _ = client_and_store + assert client.get("/devices/350123456789012/attributes").status_code == 401 + assert client.put( + "/devices/350123456789012/attributes", json={"key": "k", "value": 1} + ).status_code == 401 + assert client.delete("/devices/350123456789012/attributes/k").status_code == 401 + + def test_rejects_non_imei_url(self, client_and_store) -> None: + client, _ = client_and_store + assert client.get("/devices/123/attributes", headers=self.HDRS).status_code == 422 + assert client.put( + "/devices/123/attributes", json={"key": "k", "value": 1}, headers=self.HDRS + ).status_code == 422 + + def test_append_and_resolve_now(self, client_and_store) -> None: + client, attributes = client_and_store + r = client.put( + "/devices/350123456789012/attributes", + json={"key": "temperature_offset", "value": 1.5}, + headers=self.HDRS, + ) + assert r.status_code == 200 + body = r.json() + assert body["imei"] == "350123456789012" + assert body["key"] == "temperature_offset" + assert body["value"] == 1.5 + assert body["effective_from"] # server-side now() default + assert attributes.appended == ("350123456789012", "temperature_offset", 1.5, None, None) + + r = client.get("/devices/350123456789012/attributes", headers=self.HDRS) + assert r.status_code == 200 + assert r.json() == {"imei": "350123456789012", "attributes": {"temperature_offset": 1.5}} + + def test_append_backdated_and_resolve_at(self, client_and_store) -> None: + client, _ = client_and_store + # backdated calibration change discovered after a sensor swap + r = client.put( + "/devices/350123456789012/attributes", + json={ + "key": "temperature_offset", + "value": 2.0, + "effective_from": "2026-08-15T00:00:00Z", + "note": "after sensor swap", + }, + headers=self.HDRS, + ) + assert r.status_code == 200 + assert r.json()["effective_from"] == "2026-08-15T00:00:00+00:00" + assert r.json()["note"] == "after sensor swap" + + # before the backdated change: not in effect yet + r = client.get( + "/devices/350123456789012/attributes", + params={"at": "2026-08-01T00:00:00Z"}, + headers=self.HDRS, + ) + assert r.status_code == 200 + assert r.json()["attributes"] == {} + assert r.json()["at"] == "2026-08-01T00:00:00+00:00" + + # after it: the backdated value is in effect + r = client.get( + "/devices/350123456789012/attributes", + params={"at": "2026-09-01T00:00:00Z"}, + headers=self.HDRS, + ) + assert r.json()["attributes"] == {"temperature_offset": 2.0} + + def test_multiple_versions_resolve_to_latest_in_effect(self, client_and_store) -> None: + client, _ = client_and_store + for value, effective in ((1.0, "2026-08-01T00:00:00Z"), (2.0, "2026-08-15T00:00:00Z")): + r = client.put( + "/devices/350123456789012/attributes", + json={"key": "level_offset", "value": value, "effective_from": effective}, + headers=self.HDRS, + ) + assert r.status_code == 200 + + # midpoint resolves to the older version, after to the newer + r = client.get( + "/devices/350123456789012/attributes", + params={"at": "2026-08-10T00:00:00Z"}, + headers=self.HDRS, + ) + assert r.json()["attributes"] == {"level_offset": 1.0} + r = client.get( + "/devices/350123456789012/attributes", + params={"at": "2026-08-20T00:00:00Z"}, + headers=self.HDRS, + ) + assert r.json()["attributes"] == {"level_offset": 2.0} + + def test_history_lists_all_versions_oldest_first(self, client_and_store) -> None: + client, _ = client_and_store + for value, effective in ((2.0, "2026-08-15T00:00:00Z"), (1.0, "2026-08-01T00:00:00Z")): + client.put( + "/devices/350123456789012/attributes", + json={"key": "level_offset", "value": value, "effective_from": effective}, + headers=self.HDRS, + ) + r = client.get("/devices/350123456789012/attributes/history", headers=self.HDRS) + assert r.status_code == 200 + assert r.json() == { + "imei": "350123456789012", + "attributes": { + "level_offset": [ + {"value": 1.0, "effective_from": "2026-08-01T00:00:00+00:00", "note": None}, + {"value": 2.0, "effective_from": "2026-08-15T00:00:00+00:00", "note": None}, + ] + }, + } + + def test_append_rejects_bad_bodies(self, client_and_store) -> None: + client, _ = client_and_store + for body in ( + {}, + {"value": 1}, # missing key + {"key": "", "value": 1}, # empty key + {"key": "k"}, # missing value + {"key": "k", "value": None}, # null value + {"key": "k", "value": 1, "effective_from": "yesterday"}, # bad time + {"key": "k", "value": 1, "note": 5}, # note must be a string + ): + r = client.put( + "/devices/350123456789012/attributes", json=body, headers=self.HDRS + ) + assert r.status_code == 422, body + + def test_post_is_alias_for_put(self, client_and_store) -> None: + client, _ = client_and_store + r = client.post( + "/devices/350123456789012/attributes", + json={"key": "k", "value": 1}, + headers=self.HDRS, + ) + assert r.status_code == 200 + + def test_delete_removes_only_latest_version(self, client_and_store) -> None: + client, _ = client_and_store + for value, effective in ((1.0, "2026-08-01T00:00:00Z"), (2.0, "2026-08-15T00:00:00Z")): + client.put( + "/devices/350123456789012/attributes", + json={"key": "level_offset", "value": value, "effective_from": effective}, + headers=self.HDRS, + ) + r = client.delete("/devices/350123456789012/attributes/level_offset", headers=self.HDRS) + assert r.status_code == 200 + assert r.json() == {"status": "deleted", "imei": "350123456789012", "key": "level_offset"} + # the earlier version remains in effect + r = client.get("/devices/350123456789012/attributes", headers=self.HDRS) + assert r.json()["attributes"] == {"level_offset": 1.0} + + def test_delete_unknown_key_404(self, client_and_store) -> None: + client, _ = client_and_store + r = client.delete( + "/devices/350123456789012/attributes/does-not-exist", headers=self.HDRS + ) + assert r.status_code == 404 \ No newline at end of file