tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 8 Wiki Activity

M2-T04: add single-row record CRUD API (patch/delete)

Browse Source 
- PATCH/DELETE /api/locations/{person}/{datetime} and /api/poo/{timestamp}
- update only non-PK fields (PK immutable); 404 on missing PK
- delete scoped to exact full PK with rowcount guard (0->404, 1->ok);
  no batch/truncate/drop path
- session + CSRF protected; bare ingestion endpoints untouched
- service helpers in app/services/location.py and poo.py; regenerate openapi/
- tests/test_api_record_crud.py
This commit is contained in:
Tianyu Liu
2026-06-12 23:33:08 +02:00
parent 9ce3f2a0b8
commit 048414c5cb
7 changed files with 1377 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/routes/api/data.py
+152 -2
View File
@@ -1,24 +1,28 @@
from __future__ import annotations
from fastapi import APIRouter, Depends, Query
from fastapi import APIRouter, Body, Depends, HTTPException, Query, status
from sqlalchemy import desc, select
from sqlalchemy.orm import Session
from app.api.routes.api.deps import require_session
from app.api.routes.api.deps import require_csrf, require_session
from app.dependencies import get_db
from app.models.location import Location
from app.models.poo import PooRecord
from app.models.public_ip import PublicIPHistory, PublicIPState
from app.schemas.data import (
LocationRecord,
LocationUpdateRequest,
LocationsResponse,
PooRecord as PooRecordSchema,
PooResponse,
PooUpdateRequest,
PublicIPHistorySchema,
PublicIPResponse,
PublicIPStateSchema,
)
from app.services.auth import AuthenticatedSession
from app.services.location import delete_location, update_location
from app.services.poo import delete_poo_record, update_poo_record
router = APIRouter(prefix="/api", tags=["api-data"])
@@ -123,3 +127,149 @@ def get_public_ip(
history = [PublicIPHistorySchema.model_validate(row) for row in history_rows]
return PublicIPResponse(state=state, history=history)
# ---------------------------------------------------------------------------
# PATCH /api/locations/{person}/{datetime}
# ---------------------------------------------------------------------------
@router.patch("/locations/{person}/{datetime}", response_model=LocationRecord)
def patch_location(
person: str,
datetime: str,
body: LocationUpdateRequest = Body(default=LocationUpdateRequest()),
db: Session = Depends(get_db),
_auth: AuthenticatedSession = Depends(require_session),
_csrf: None = Depends(require_csrf),
) -> LocationRecord:
"""
Update the non-PK fields of a single location record.
- ``person`` and ``datetime`` identify the row (composite PK) and are immutable.
- Only ``latitude``, ``longitude``, and ``altitude`` may be updated.
- Omitted body fields are left unchanged.
- Returns **404** if the PK does not exist.
"""
row = update_location(
db,
person,
datetime,
latitude=body.latitude,
longitude=body.longitude,
altitude=body.altitude,
)
if row is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="location record not found",
)
return LocationRecord(
person=row.person,
datetime=row.datetime,
latitude=row.latitude,
longitude=row.longitude,
altitude=row.altitude,
)
# ---------------------------------------------------------------------------
# DELETE /api/locations/{person}/{datetime}
# ---------------------------------------------------------------------------
@router.delete(
"/locations/{person}/{datetime}",
status_code=status.HTTP_204_NO_CONTENT,
response_model=None,
)
def delete_location_record(
person: str,
datetime: str,
db: Session = Depends(get_db),
_auth: AuthenticatedSession = Depends(require_session),
_csrf: None = Depends(require_csrf),
) -> None:
"""
Delete the single location record identified by its composite PK.
- Exactly one row is deleted; **404** if the PK does not exist.
- No batch delete / truncate path is available.
"""
deleted = delete_location(db, person, datetime)
if not deleted:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="location record not found",
)
# ---------------------------------------------------------------------------
# PATCH /api/poo/{timestamp}
# ---------------------------------------------------------------------------
@router.patch("/poo/{timestamp}", response_model=PooRecordSchema)
def patch_poo(
timestamp: str,
body: PooUpdateRequest = Body(default=PooUpdateRequest()),
db: Session = Depends(get_db),
_auth: AuthenticatedSession = Depends(require_session),
_csrf: None = Depends(require_csrf),
) -> PooRecordSchema:
"""
Update the non-PK fields of a single poo record.
- ``timestamp`` is the PK and is immutable.
- Only ``status``, ``latitude``, and ``longitude`` may be updated.
- Omitted body fields are left unchanged.
- Returns **404** if the PK does not exist.
"""
row = update_poo_record(
db,
timestamp,
status=body.status,
latitude=body.latitude,
longitude=body.longitude,
)
if row is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="poo record not found",
)
return PooRecordSchema(
timestamp=row.timestamp,
status=row.status,
latitude=row.latitude,
longitude=row.longitude,
)
# ---------------------------------------------------------------------------
# DELETE /api/poo/{timestamp}
# ---------------------------------------------------------------------------
@router.delete(
"/poo/{timestamp}",
status_code=status.HTTP_204_NO_CONTENT,
response_model=None,
)
def delete_poo(
timestamp: str,
db: Session = Depends(get_db),
_auth: AuthenticatedSession = Depends(require_session),
_csrf: None = Depends(require_csrf),
) -> None:
"""
Delete the single poo record identified by its PK.
- Exactly one row is deleted; **404** if the PK does not exist.
- No batch delete / truncate path is available.
"""
deleted = delete_poo_record(db, timestamp)
if not deleted:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="poo record not found",
)
app/schemas/data.py
+16
View File
@@ -24,6 +24,14 @@ class LocationsResponse(BaseModel):
offset: int
class LocationUpdateRequest(BaseModel):
"""PATCH body for a location record — all fields optional; PK fields excluded."""
latitude: float | None = None
longitude: float | None = None
altitude: float | None = None
# ---------------------------------------------------------------------------
# Poo
# ---------------------------------------------------------------------------
@@ -42,6 +50,14 @@ class PooResponse(BaseModel):
offset: int
class PooUpdateRequest(BaseModel):
"""PATCH body for a poo record — all fields optional; PK field excluded."""
status: str | None = None
latitude: float | None = None
longitude: float | None = None
# ---------------------------------------------------------------------------
# Public IP
# ---------------------------------------------------------------------------
app/services/location.py
+56 -1
View File
@@ -1,6 +1,6 @@
from datetime import datetime, timezone
from sqlalchemy import insert
from sqlalchemy import delete, insert, select
from sqlalchemy.orm import Session
from app.models.location import Location
@@ -40,3 +40,58 @@ def record_location(session: Session, payload: LocationRecordRequest) -> None:
)
session.execute(stmt)
session.commit()
def update_location(
session: Session,
person: str,
datetime_pk: str,
*,
latitude: float | None,
longitude: float | None,
altitude: float | None,
) -> Location | None:
"""Update non-PK fields of a single location row.
Returns the updated ORM object, or ``None`` if the PK does not exist.
The caller must not pass PK fields — they are immutable.
Only fields with a non-``None`` value are written; ``altitude`` being
``None`` in the request means "leave unchanged", not "clear to NULL".
"""
row = session.execute(
select(Location).where(
Location.person == person,
Location.datetime == datetime_pk,
)
).scalar_one_or_none()
if row is None:
return None
if latitude is not None:
row.latitude = latitude
if longitude is not None:
row.longitude = longitude
if altitude is not None:
row.altitude = altitude
session.commit()
session.refresh(row)
return row
def delete_location(session: Session, person: str, datetime_pk: str) -> bool:
"""Delete the single location row identified by its full composite PK.
Returns ``True`` if exactly one row was deleted, ``False`` if the PK did
not exist (caller should raise 404). The DELETE is scoped to the exact PK
— no batch/truncate path exists.
"""
result = session.execute(
delete(Location).where(
Location.person == person,
Location.datetime == datetime_pk,
)
)
session.commit()
return result.rowcount == 1
app/services/poo.py
+48 -1
View File
@@ -4,7 +4,7 @@ from dataclasses import dataclass
from datetime import datetime, timezone
import logging
from sqlalchemy import desc, insert, select
from sqlalchemy import delete, desc, insert, select
from sqlalchemy.orm import Session
from app.config import Settings
@@ -74,6 +74,53 @@ def record_poo(
logger.warning("Failed to trigger poo webhook on Home Assistant: %s", exc)
def update_poo_record(
session: Session,
timestamp_pk: str,
*,
status: str | None,
latitude: float | None,
longitude: float | None,
) -> PooRecord | None:
"""Update non-PK fields of a single poo record row.
Returns the updated ORM object, or ``None`` if the PK does not exist.
The ``timestamp`` PK is immutable and must not be passed as an update field.
Only fields with a non-``None`` value are written.
"""
row = session.execute(
select(PooRecord).where(PooRecord.timestamp == timestamp_pk)
).scalar_one_or_none()
if row is None:
return None
if status is not None:
row.status = status
if latitude is not None:
row.latitude = latitude
if longitude is not None:
row.longitude = longitude
session.commit()
session.refresh(row)
return row
def delete_poo_record(session: Session, timestamp_pk: str) -> bool:
"""Delete the single poo record row identified by its PK.
Returns ``True`` if exactly one row was deleted, ``False`` if the PK did
not exist (caller should raise 404). The DELETE is scoped to the exact PK
— no batch/truncate path exists.
"""
result = session.execute(
delete(PooRecord).where(PooRecord.timestamp == timestamp_pk)
)
session.commit()
return result.rowcount == 1
def get_latest_poo_record(session: Session) -> LatestPooRecord | None:
stmt = select(PooRecord).order_by(desc(PooRecord.timestamp)).limit(1)
record = session.execute(stmt).scalar_one_or_none()
openapi/openapi.json
+336
View File
@@ -542,6 +542,262 @@
}
}
},
"/api/locations/{person}/{datetime}": {
"patch": {
"tags": [
"api-data"
],
"summary": "Patch Location",
"description": "Update the non-PK fields of a single location record.\n\n- ``person`` and ``datetime`` identify the row (composite PK) and are immutable.\n- Only ``latitude``, ``longitude``, and ``altitude`` may be updated.\n- Omitted body fields are left unchanged.\n- Returns **404** if the PK does not exist.",
"operationId": "patch_location_api_locations__person___datetime__patch",
"parameters": [
{
"name": "person",
"in": "path",
"required": true,
"schema": {
"type": "string",
"title": "Person"
}
},
{
"name": "datetime",
"in": "path",
"required": true,
"schema": {
"type": "string",
"title": "Datetime"
}
},
{
"name": "X-CSRF-Token",
"in": "header",
"required": false,
"schema": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "X-Csrf-Token"
}
}
],
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/LocationUpdateRequest",
"default": {}
}
}
}
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/LocationRecord"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
},
"delete": {
"tags": [
"api-data"
],
"summary": "Delete Location Record",
"description": "Delete the single location record identified by its composite PK.\n\n- Exactly one row is deleted; **404** if the PK does not exist.\n- No batch delete / truncate path is available.",
"operationId": "delete_location_record_api_locations__person___datetime__delete",
"parameters": [
{
"name": "person",
"in": "path",
"required": true,
"schema": {
"type": "string",
"title": "Person"
}
},
{
"name": "datetime",
"in": "path",
"required": true,
"schema": {
"type": "string",
"title": "Datetime"
}
},
{
"name": "X-CSRF-Token",
"in": "header",
"required": false,
"schema": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "X-Csrf-Token"
}
}
],
"responses": {
"204": {
"description": "Successful Response"
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/api/poo/{timestamp}": {
"patch": {
"tags": [
"api-data"
],
"summary": "Patch Poo",
"description": "Update the non-PK fields of a single poo record.\n\n- ``timestamp`` is the PK and is immutable.\n- Only ``status``, ``latitude``, and ``longitude`` may be updated.\n- Omitted body fields are left unchanged.\n- Returns **404** if the PK does not exist.",
"operationId": "patch_poo_api_poo__timestamp__patch",
"parameters": [
{
"name": "timestamp",
"in": "path",
"required": true,
"schema": {
"type": "string",
"title": "Timestamp"
}
},
{
"name": "X-CSRF-Token",
"in": "header",
"required": false,
"schema": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "X-Csrf-Token"
}
}
],
"requestBody": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/PooUpdateRequest",
"default": {}
}
}
}
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/PooRecord"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
},
"delete": {
"tags": [
"api-data"
],
"summary": "Delete Poo",
"description": "Delete the single poo record identified by its PK.\n\n- Exactly one row is deleted; **404** if the PK does not exist.\n- No batch delete / truncate path is available.",
"operationId": "delete_poo_api_poo__timestamp__delete",
"parameters": [
{
"name": "timestamp",
"in": "path",
"required": true,
"schema": {
"type": "string",
"title": "Timestamp"
}
},
{
"name": "X-CSRF-Token",
"in": "header",
"required": false,
"schema": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "X-Csrf-Token"
}
}
],
"responses": {
"204": {
"description": "Successful Response"
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/api/session": {
"get": {
"tags": [
@@ -1075,6 +1331,46 @@
],
"title": "LocationRecord"
},
"LocationUpdateRequest": {
"properties": {
"latitude": {
"anyOf": [
{
"type": "number"
},
{
"type": "null"
}
],
"title": "Latitude"
},
"longitude": {
"anyOf": [
{
"type": "number"
},
{
"type": "null"
}
],
"title": "Longitude"
},
"altitude": {
"anyOf": [
{
"type": "number"
},
{
"type": "null"
}
],
"title": "Altitude"
}
},
"type": "object",
"title": "LocationUpdateRequest",
"description": "PATCH body for a location record — all fields optional; PK fields excluded."
},
"LocationsResponse": {
"properties": {
"items": {
@@ -1196,6 +1492,46 @@
],
"title": "PooResponse"
},
"PooUpdateRequest": {
"properties": {
"status": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"title": "Status"
},
"latitude": {
"anyOf": [
{
"type": "number"
},
{
"type": "null"
}
],
"title": "Latitude"
},
"longitude": {
"anyOf": [
{
"type": "number"
},
{
"type": "null"
}
],
"title": "Longitude"
}
},
"type": "object",
"title": "PooUpdateRequest",
"description": "PATCH body for a poo record — all fields optional; PK field excluded."
},
"PublicIPCheckResponse": {
"properties": {
"status": {
openapi/openapi.yaml
+224
View File
@@ -364,6 +364,189 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
/api/locations/{person}/{datetime}:
patch:
tags:
- api-data
summary: Patch Location
description: 'Update the non-PK fields of a single location record.
- ``person`` and ``datetime`` identify the row (composite PK) and are immutable.
- Only ``latitude``, ``longitude``, and ``altitude`` may be updated.
- Omitted body fields are left unchanged.
- Returns **404** if the PK does not exist.'
operationId: patch_location_api_locations__person___datetime__patch
parameters:
- name: person
in: path
required: true
schema:
type: string
title: Person
- name: datetime
in: path
required: true
schema:
type: string
title: Datetime
- name: X-CSRF-Token
in: header
required: false
schema:
anyOf:
- type: string
- type: 'null'
title: X-Csrf-Token
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/LocationUpdateRequest'
default: {}
responses:
'200':
description: Successful Response
content:
application/json:
schema:
$ref: '#/components/schemas/LocationRecord'
'422':
description: Validation Error
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
delete:
tags:
- api-data
summary: Delete Location Record
description: 'Delete the single location record identified by its composite
PK.
- Exactly one row is deleted; **404** if the PK does not exist.
- No batch delete / truncate path is available.'
operationId: delete_location_record_api_locations__person___datetime__delete
parameters:
- name: person
in: path
required: true
schema:
type: string
title: Person
- name: datetime
in: path
required: true
schema:
type: string
title: Datetime
- name: X-CSRF-Token
in: header
required: false
schema:
anyOf:
- type: string
- type: 'null'
title: X-Csrf-Token
responses:
'204':
description: Successful Response
'422':
description: Validation Error
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
/api/poo/{timestamp}:
patch:
tags:
- api-data
summary: Patch Poo
description: 'Update the non-PK fields of a single poo record.
- ``timestamp`` is the PK and is immutable.
- Only ``status``, ``latitude``, and ``longitude`` may be updated.
- Omitted body fields are left unchanged.
- Returns **404** if the PK does not exist.'
operationId: patch_poo_api_poo__timestamp__patch
parameters:
- name: timestamp
in: path
required: true
schema:
type: string
title: Timestamp
- name: X-CSRF-Token
in: header
required: false
schema:
anyOf:
- type: string
- type: 'null'
title: X-Csrf-Token
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/PooUpdateRequest'
default: {}
responses:
'200':
description: Successful Response
content:
application/json:
schema:
$ref: '#/components/schemas/PooRecord'
'422':
description: Validation Error
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
delete:
tags:
- api-data
summary: Delete Poo
description: 'Delete the single poo record identified by its PK.
- Exactly one row is deleted; **404** if the PK does not exist.
- No batch delete / truncate path is available.'
operationId: delete_poo_api_poo__timestamp__delete
parameters:
- name: timestamp
in: path
required: true
schema:
type: string
title: Timestamp
- name: X-CSRF-Token
in: header
required: false
schema:
anyOf:
- type: string
- type: 'null'
title: X-Csrf-Token
responses:
'204':
description: Successful Response
'422':
description: Validation Error
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
/api/session:
get:
tags:
@@ -735,6 +918,27 @@ components:
- longitude
- altitude
title: LocationRecord
LocationUpdateRequest:
properties:
latitude:
anyOf:
- type: number
- type: 'null'
title: Latitude
longitude:
anyOf:
- type: number
- type: 'null'
title: Longitude
altitude:
anyOf:
- type: number
- type: 'null'
title: Altitude
type: object
title: LocationUpdateRequest
description: PATCH body for a location record — all fields optional; PK fields
excluded.
LocationsResponse:
properties:
items:
@@ -824,6 +1028,26 @@ components:
- limit
- offset
title: PooResponse
PooUpdateRequest:
properties:
status:
anyOf:
- type: string
- type: 'null'
title: Status
latitude:
anyOf:
- type: number
- type: 'null'
title: Latitude
longitude:
anyOf:
- type: number
- type: 'null'
title: Longitude
type: object
title: PooUpdateRequest
description: PATCH body for a poo record — all fields optional; PK field excluded.
PublicIPCheckResponse:
properties:
status:
tests/test_api_record_crud.py
+545
View File
@@ -0,0 +1,545 @@
"""Tests for M2-T04: PATCH/DELETE /api/locations and /api/poo."""
from __future__ import annotations
from fastapi.testclient import TestClient
from sqlalchemy import insert, select
from sqlalchemy.engine import Engine
from app.models.location import Location
from app.models.poo import PooRecord
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
CSRF_HEADER = {"X-CSRF-Token": "any-value"}
def _api_login(client: TestClient) -> None:
resp = client.post(
"/api/auth/login",
json={"username": "admin", "password": "test-password"},
)
assert resp.status_code == 200, f"Login failed: {resp.status_code}"
def _seed_locations(engine: Engine, rows: list[dict]) -> None:
with engine.begin() as conn:
conn.execute(insert(Location), rows)
def _seed_poo(engine: Engine, rows: list[dict]) -> None:
with engine.begin() as conn:
conn.execute(insert(PooRecord), rows)
def _count_locations(engine: Engine) -> int:
with engine.connect() as conn:
return conn.execute(select(Location)).rowcount or len(conn.execute(select(Location)).all())
def _fetch_location(engine: Engine, person: str, dt: str) -> dict | None:
with engine.connect() as conn:
row = conn.execute(
select(Location).where(Location.person == person, Location.datetime == dt)
).one_or_none()
if row is None:
return None
return dict(row._mapping)
def _fetch_poo(engine: Engine, timestamp: str) -> dict | None:
with engine.connect() as conn:
row = conn.execute(
select(PooRecord).where(PooRecord.timestamp == timestamp)
).one_or_none()
if row is None:
return None
return dict(row._mapping)
def _all_location_count(engine: Engine) -> int:
with engine.connect() as conn:
return len(conn.execute(select(Location)).all())
def _all_poo_count(engine: Engine) -> int:
with engine.connect() as conn:
return len(conn.execute(select(PooRecord)).all())
# ---------------------------------------------------------------------------
# PATCH /api/locations/{person}/{datetime} — authentication / CSRF guards
# ---------------------------------------------------------------------------
def test_patch_location_unauthenticated_returns_401(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[{"person": "alice", "datetime": "2026-06-01T10:00:00Z", "latitude": 1.0, "longitude": 2.0, "altitude": None}],
)
resp = client.patch(
"/api/locations/alice/2026-06-01T10:00:00Z",
json={"latitude": 9.9},
headers=CSRF_HEADER,
)
assert resp.status_code == 401
def test_patch_location_missing_csrf_returns_403(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[{"person": "alice", "datetime": "2026-06-01T10:00:00Z", "latitude": 1.0, "longitude": 2.0, "altitude": None}],
)
_api_login(client)
resp = client.patch(
"/api/locations/alice/2026-06-01T10:00:00Z",
json={"latitude": 9.9},
)
assert resp.status_code == 403
# ---------------------------------------------------------------------------
# PATCH /api/locations/{person}/{datetime} — 404 for nonexistent PK
# ---------------------------------------------------------------------------
def test_patch_location_nonexistent_pk_returns_404(location_client) -> None:
client, _engine = location_client
_api_login(client)
resp = client.patch(
"/api/locations/nobody/2099-01-01T00:00:00Z",
json={"latitude": 1.0},
headers=CSRF_HEADER,
)
assert resp.status_code == 404
# ---------------------------------------------------------------------------
# PATCH /api/locations/{person}/{datetime} — updates exactly one row's fields
# ---------------------------------------------------------------------------
def test_patch_location_updates_single_row_fields(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[
{
"person": "alice",
"datetime": "2026-06-01T10:00:00Z",
"latitude": 51.0,
"longitude": -0.1,
"altitude": None,
},
{
"person": "alice",
"datetime": "2026-06-02T10:00:00Z",
"latitude": 52.0,
"longitude": -0.2,
"altitude": None,
},
],
)
_api_login(client)
resp = client.patch(
"/api/locations/alice/2026-06-01T10:00:00Z",
json={"latitude": 99.0, "longitude": 88.0},
headers=CSRF_HEADER,
)
assert resp.status_code == 200
body = resp.json()
assert body["latitude"] == 99.0
assert body["longitude"] == 88.0
assert body["person"] == "alice"
assert body["datetime"] == "2026-06-01T10:00:00Z"
# Confirm DB state — row 1 changed, row 2 unchanged
row1 = _fetch_location(engine, "alice", "2026-06-01T10:00:00Z")
assert row1 is not None
assert row1["latitude"] == 99.0
assert row1["longitude"] == 88.0
row2 = _fetch_location(engine, "alice", "2026-06-02T10:00:00Z")
assert row2 is not None
assert row2["latitude"] == 52.0 # unchanged
assert row2["longitude"] == -0.2 # unchanged
# Row count unchanged — no spurious rows added/removed
assert _all_location_count(engine) == 2
def test_patch_location_partial_update_leaves_other_fields_unchanged(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[
{
"person": "bob",
"datetime": "2026-06-10T08:00:00Z",
"latitude": 48.8,
"longitude": 2.3,
"altitude": 100.0,
}
],
)
_api_login(client)
# Only update altitude
resp = client.patch(
"/api/locations/bob/2026-06-10T08:00:00Z",
json={"altitude": 200.0},
headers=CSRF_HEADER,
)
assert resp.status_code == 200
body = resp.json()
assert body["altitude"] == 200.0
assert body["latitude"] == 48.8 # unchanged
assert body["longitude"] == 2.3 # unchanged
def test_patch_location_empty_body_is_noop(location_client) -> None:
"""Sending an empty body should not change the record but still return 200."""
client, engine = location_client
_seed_locations(
engine,
[{"person": "carol", "datetime": "2026-06-05T12:00:00Z", "latitude": 10.0, "longitude": 20.0, "altitude": None}],
)
_api_login(client)
resp = client.patch(
"/api/locations/carol/2026-06-05T12:00:00Z",
json={},
headers=CSRF_HEADER,
)
assert resp.status_code == 200
row = _fetch_location(engine, "carol", "2026-06-05T12:00:00Z")
assert row["latitude"] == 10.0
assert row["longitude"] == 20.0
def test_patch_location_response_has_correct_schema(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[{"person": "alice", "datetime": "2026-06-01T10:00:00Z", "latitude": 1.0, "longitude": 2.0, "altitude": 5.0}],
)
_api_login(client)
resp = client.patch(
"/api/locations/alice/2026-06-01T10:00:00Z",
json={"latitude": 3.0},
headers=CSRF_HEADER,
)
assert resp.status_code == 200
keys = set(resp.json().keys())
assert keys == {"person", "datetime", "latitude", "longitude", "altitude"}
# ---------------------------------------------------------------------------
# DELETE /api/locations/{person}/{datetime} — guards
# ---------------------------------------------------------------------------
def test_delete_location_unauthenticated_returns_401(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[{"person": "alice", "datetime": "2026-06-01T10:00:00Z", "latitude": 1.0, "longitude": 2.0, "altitude": None}],
)
resp = client.delete("/api/locations/alice/2026-06-01T10:00:00Z", headers=CSRF_HEADER)
assert resp.status_code == 401
def test_delete_location_missing_csrf_returns_403(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[{"person": "alice", "datetime": "2026-06-01T10:00:00Z", "latitude": 1.0, "longitude": 2.0, "altitude": None}],
)
_api_login(client)
resp = client.delete("/api/locations/alice/2026-06-01T10:00:00Z")
assert resp.status_code == 403
# ---------------------------------------------------------------------------
# DELETE /api/locations/{person}/{datetime} — 404 for nonexistent PK
# ---------------------------------------------------------------------------
def test_delete_location_nonexistent_pk_returns_404(location_client) -> None:
client, _engine = location_client
_api_login(client)
resp = client.delete("/api/locations/nobody/2099-01-01T00:00:00Z", headers=CSRF_HEADER)
assert resp.status_code == 404
# ---------------------------------------------------------------------------
# DELETE /api/locations/{person}/{datetime} — deletes exactly one row
# ---------------------------------------------------------------------------
def test_delete_location_removes_exactly_one_row(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[
{
"person": "alice",
"datetime": "2026-06-01T10:00:00Z",
"latitude": 51.0,
"longitude": -0.1,
"altitude": None,
},
{
"person": "alice",
"datetime": "2026-06-02T10:00:00Z",
"latitude": 52.0,
"longitude": -0.2,
"altitude": None,
},
],
)
_api_login(client)
before = _all_location_count(engine)
assert before == 2
resp = client.delete("/api/locations/alice/2026-06-01T10:00:00Z", headers=CSRF_HEADER)
assert resp.status_code == 204
after = _all_location_count(engine)
assert after == 1 # exactly one row removed
# The deleted row is gone
assert _fetch_location(engine, "alice", "2026-06-01T10:00:00Z") is None
# The other row still exists
assert _fetch_location(engine, "alice", "2026-06-02T10:00:00Z") is not None
def test_delete_location_second_delete_returns_404(location_client) -> None:
"""Deleting the same PK twice must return 404 on the second attempt."""
client, engine = location_client
_seed_locations(
engine,
[{"person": "alice", "datetime": "2026-06-01T10:00:00Z", "latitude": 1.0, "longitude": 2.0, "altitude": None}],
)
_api_login(client)
resp1 = client.delete("/api/locations/alice/2026-06-01T10:00:00Z", headers=CSRF_HEADER)
assert resp1.status_code == 204
resp2 = client.delete("/api/locations/alice/2026-06-01T10:00:00Z", headers=CSRF_HEADER)
assert resp2.status_code == 404
# ---------------------------------------------------------------------------
# PATCH /api/poo/{timestamp} — guards
# ---------------------------------------------------------------------------
def test_patch_poo_unauthenticated_returns_401(poo_client) -> None:
client, engine = poo_client
_seed_poo(
engine,
[{"timestamp": "2026-06-01T10:00Z", "status": "success", "latitude": 1.0, "longitude": 2.0}],
)
resp = client.patch(
"/api/poo/2026-06-01T10:00Z",
json={"status": "fail"},
headers=CSRF_HEADER,
)
assert resp.status_code == 401
def test_patch_poo_missing_csrf_returns_403(poo_client) -> None:
client, engine = poo_client
_seed_poo(
engine,
[{"timestamp": "2026-06-01T10:00Z", "status": "success", "latitude": 1.0, "longitude": 2.0}],
)
_api_login(client)
resp = client.patch(
"/api/poo/2026-06-01T10:00Z",
json={"status": "fail"},
)
assert resp.status_code == 403
# ---------------------------------------------------------------------------
# PATCH /api/poo/{timestamp} — 404
# ---------------------------------------------------------------------------
def test_patch_poo_nonexistent_pk_returns_404(poo_client) -> None:
client, _engine = poo_client
_api_login(client)
resp = client.patch(
"/api/poo/2099-01-01T00:00Z",
json={"status": "fail"},
headers=CSRF_HEADER,
)
assert resp.status_code == 404
# ---------------------------------------------------------------------------
# PATCH /api/poo/{timestamp} — updates single row
# ---------------------------------------------------------------------------
def test_patch_poo_updates_single_row_fields(poo_client) -> None:
client, engine = poo_client
_seed_poo(
engine,
[
{"timestamp": "2026-06-01T10:00Z", "status": "success", "latitude": 51.0, "longitude": -0.1},
{"timestamp": "2026-06-02T10:00Z", "status": "success", "latitude": 52.0, "longitude": -0.2},
],
)
_api_login(client)
resp = client.patch(
"/api/poo/2026-06-01T10:00Z",
json={"status": "fail", "latitude": 99.0},
headers=CSRF_HEADER,
)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "fail"
assert body["latitude"] == 99.0
assert body["timestamp"] == "2026-06-01T10:00Z"
# Other row unchanged
row2 = _fetch_poo(engine, "2026-06-02T10:00Z")
assert row2 is not None
assert row2["status"] == "success"
assert row2["latitude"] == 52.0
# Row count unchanged
assert _all_poo_count(engine) == 2
def test_patch_poo_partial_update_leaves_other_fields_unchanged(poo_client) -> None:
client, engine = poo_client
_seed_poo(
engine,
[{"timestamp": "2026-06-01T10:00Z", "status": "success", "latitude": 51.0, "longitude": -0.1}],
)
_api_login(client)
resp = client.patch(
"/api/poo/2026-06-01T10:00Z",
json={"longitude": 99.9},
headers=CSRF_HEADER,
)
assert resp.status_code == 200
body = resp.json()
assert body["longitude"] == 99.9
assert body["latitude"] == 51.0 # unchanged
assert body["status"] == "success" # unchanged
def test_patch_poo_response_has_correct_schema(poo_client) -> None:
client, engine = poo_client
_seed_poo(
engine,
[{"timestamp": "2026-06-01T10:00Z", "status": "success", "latitude": 1.0, "longitude": 2.0}],
)
_api_login(client)
resp = client.patch(
"/api/poo/2026-06-01T10:00Z",
json={"status": "fail"},
headers=CSRF_HEADER,
)
assert resp.status_code == 200
assert set(resp.json().keys()) == {"timestamp", "status", "latitude", "longitude"}
# ---------------------------------------------------------------------------
# DELETE /api/poo/{timestamp} — guards
# ---------------------------------------------------------------------------
def test_delete_poo_unauthenticated_returns_401(poo_client) -> None:
client, engine = poo_client
_seed_poo(
engine,
[{"timestamp": "2026-06-01T10:00Z", "status": "success", "latitude": 1.0, "longitude": 2.0}],
)
resp = client.delete("/api/poo/2026-06-01T10:00Z", headers=CSRF_HEADER)
assert resp.status_code == 401
def test_delete_poo_missing_csrf_returns_403(poo_client) -> None:
client, engine = poo_client
_seed_poo(
engine,
[{"timestamp": "2026-06-01T10:00Z", "status": "success", "latitude": 1.0, "longitude": 2.0}],
)
_api_login(client)
resp = client.delete("/api/poo/2026-06-01T10:00Z")
assert resp.status_code == 403
# ---------------------------------------------------------------------------
# DELETE /api/poo/{timestamp} — 404
# ---------------------------------------------------------------------------
def test_delete_poo_nonexistent_pk_returns_404(poo_client) -> None:
client, _engine = poo_client
_api_login(client)
resp = client.delete("/api/poo/2099-01-01T00:00Z", headers=CSRF_HEADER)
assert resp.status_code == 404
# ---------------------------------------------------------------------------
# DELETE /api/poo/{timestamp} — deletes exactly one row
# ---------------------------------------------------------------------------
def test_delete_poo_removes_exactly_one_row(poo_client) -> None:
client, engine = poo_client
_seed_poo(
engine,
[
{"timestamp": "2026-06-01T10:00Z", "status": "success", "latitude": 51.0, "longitude": -0.1},
{"timestamp": "2026-06-02T10:00Z", "status": "fail", "latitude": 52.0, "longitude": -0.2},
],
)
_api_login(client)
before = _all_poo_count(engine)
assert before == 2
resp = client.delete("/api/poo/2026-06-01T10:00Z", headers=CSRF_HEADER)
assert resp.status_code == 204
after = _all_poo_count(engine)
assert after == 1 # exactly one row removed
# Deleted row is gone
assert _fetch_poo(engine, "2026-06-01T10:00Z") is None
# Other row still exists
assert _fetch_poo(engine, "2026-06-02T10:00Z") is not None
def test_delete_poo_second_delete_returns_404(poo_client) -> None:
"""Deleting the same PK twice must return 404 on the second attempt."""
client, engine = poo_client
_seed_poo(
engine,
[{"timestamp": "2026-06-01T10:00Z", "status": "success", "latitude": 1.0, "longitude": 2.0}],
)
_api_login(client)
resp1 = client.delete("/api/poo/2026-06-01T10:00Z", headers=CSRF_HEADER)
assert resp1.status_code == 204
resp2 = client.delete("/api/poo/2026-06-01T10:00Z", headers=CSRF_HEADER)
assert resp2.status_code == 404