tliu93
048414c5cb
- 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
546 lines
18 KiB
Python
546 lines
18 KiB
Python
"""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
|