tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 8 Wiki Activity
Files
da236643f2240b1c68ac1f5a28e7f41404705605
home-automation/tests/test_api_data.py
T
tliu93 bd09523e94 M2-T13: docs wrap-up + retire frontend constraints + dependency cleanup 
- README: add 前端 v2 (React SPA) section (dev/build/codegen/hosting/gates),
  update directory listing, drop stale Jinja descriptions
- architecture-overview: retire '不引入前后端分离' constraint; reflect SPA + JSON API
- roadmap: mark M2 done
- remove orphaned jinja2 dependency (recompile requirements*.txt; no other churn)
- delete empty tests/test_auth.py stub; drop dead _extract_csrf_token in test_api_data
- verified image still builds and app imports with the slimmer deps
2026-06-13 15:20:50 +02:00

605 lines
17 KiB
Python
Raw Blame History

"""Tests for M2-T03: GET /api/locations, GET /api/poo, GET /api/public-ip."""
from __future__ import annotations
from datetime import UTC, datetime
from fastapi.testclient import TestClient
from sqlalchemy import insert
from sqlalchemy.engine import Engine
from app.models.location import Location
from app.models.poo import PooRecord
from app.models.public_ip import PublicIPHistory, PublicIPState
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _api_login(client: TestClient) -> None:
"""Log in via POST /api/auth/login so the TestClient has a session cookie."""
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 _seed_public_ip(engine: Engine) -> None:
now = datetime.now(UTC)
with engine.begin() as conn:
conn.execute(
insert(PublicIPState),
[
{
"id": 1,
"current_ipv4": "1.2.3.4",
"previous_ipv4": "1.2.3.3",
"first_seen_at": now,
"last_checked_at": now,
"last_changed_at": now,
"last_check_status": "changed",
"last_check_error": None,
"last_provider": "ipify",
}
],
)
conn.execute(
insert(PublicIPHistory),
[
{
"ipv4": "1.2.3.3",
"observed_at": datetime(2026, 1, 1, tzinfo=UTC),
"change_type": "first_seen",
"provider": "ipify",
},
{
"ipv4": "1.2.3.4",
"observed_at": now,
"change_type": "changed",
"provider": "ipify",
},
],
)
# ---------------------------------------------------------------------------
# Unauthenticated → 401
# ---------------------------------------------------------------------------
def test_locations_unauthenticated_returns_401(client: TestClient) -> None:
response = client.get("/api/locations")
assert response.status_code == 401
def test_poo_unauthenticated_returns_401(client: TestClient) -> None:
response = client.get("/api/poo")
assert response.status_code == 401
def test_public_ip_unauthenticated_returns_401(client: TestClient) -> None:
response = client.get("/api/public-ip")
assert response.status_code == 401
# ---------------------------------------------------------------------------
# GET /api/locations — basic
# ---------------------------------------------------------------------------
def test_locations_empty_returns_empty_list(location_client) -> None:
client, _engine = location_client
_api_login(client)
resp = client.get("/api/locations")
assert resp.status_code == 200
body = resp.json()
assert body["items"] == []
assert body["limit"] == 1000
assert body["offset"] == 0
def test_locations_returns_seeded_rows(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[
{
"person": "alice",
"datetime": "2026-06-01T10:00:00Z",
"latitude": 51.5,
"longitude": -0.1,
"altitude": None,
},
{
"person": "bob",
"datetime": "2026-06-02T12:00:00Z",
"latitude": 48.8,
"longitude": 2.3,
"altitude": 35.0,
},
],
)
_api_login(client)
resp = client.get("/api/locations")
assert resp.status_code == 200
items = resp.json()["items"]
assert len(items) == 2
# ordered by datetime ascending
assert items[0]["datetime"] == "2026-06-01T10:00:00Z"
assert items[1]["datetime"] == "2026-06-02T12:00:00Z"
# altitude nullable
assert items[0]["altitude"] is None
assert items[1]["altitude"] == 35.0
def test_locations_returns_all_fields(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[
{
"person": "alice",
"datetime": "2026-06-01T10:00:00Z",
"latitude": 51.5,
"longitude": -0.1,
"altitude": 10.0,
}
],
)
_api_login(client)
resp = client.get("/api/locations")
item = resp.json()["items"][0]
assert set(item.keys()) == {"person", "datetime", "latitude", "longitude", "altitude"}
assert item["person"] == "alice"
assert item["latitude"] == 51.5
assert item["longitude"] == -0.1
assert item["altitude"] == 10.0
# ---------------------------------------------------------------------------
# GET /api/locations — pagination
# ---------------------------------------------------------------------------
def test_locations_limit_and_offset(location_client) -> None:
client, engine = location_client
_seed_locations(
engine,
[
{
"person": "alice",
"datetime": f"2026-06-{i:02d}T10:00:00Z",
"latitude": 51.0 + i,
"longitude": -0.1,
"altitude": None,
}
for i in range(1, 6)
],
)
_api_login(client)
resp = client.get("/api/locations", params={"limit": 2, "offset": 1})
assert resp.status_code == 200
body = resp.json()
assert body["limit"] == 2
assert body["offset"] == 1
items = body["items"]
assert len(items) == 2
# offset=1 skips the first row (2026-06-01), so we start at 2026-06-02
assert items[0]["datetime"] == "2026-06-02T10:00:00Z"
def test_locations_limit_at_cap_returns_200(location_client) -> None:
client, _engine = location_client
_api_login(client)
resp = client.get("/api/locations", params={"limit": 5000})
assert resp.status_code == 200
assert resp.json()["limit"] == 5000
def test_locations_limit_exceeds_cap_returns_422(location_client) -> None:
client, _engine = location_client
_api_login(client)
resp = client.get("/api/locations", params={"limit": 5001})
assert resp.status_code == 422
def test_locations_limit_zero_returns_422(location_client) -> None:
client, _engine = location_client
_api_login(client)
resp = client.get("/api/locations", params={"limit": 0})
assert resp.status_code == 422
def test_locations_negative_offset_returns_422(location_client) -> None:
client, _engine = location_client
_api_login(client)
resp = client.get("/api/locations", params={"offset": -1})
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# GET /api/locations — time-window filtering (inclusive bounds)
# ---------------------------------------------------------------------------
def test_locations_start_filter_inclusive(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.1,
"altitude": None,
},
{
"person": "alice",
"datetime": "2026-06-03T10:00:00Z",
"latitude": 53.0,
"longitude": -0.1,
"altitude": None,
},
],
)
_api_login(client)
# start is inclusive: 2026-06-02 should be included
resp = client.get("/api/locations", params={"start": "2026-06-02T10:00:00Z"})
assert resp.status_code == 200
items = resp.json()["items"]
datetimes = [it["datetime"] for it in items]
assert "2026-06-02T10:00:00Z" in datetimes # inclusive
assert "2026-06-03T10:00:00Z" in datetimes
assert "2026-06-01T10:00:00Z" not in datetimes
def test_locations_end_filter_inclusive(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.1,
"altitude": None,
},
{
"person": "alice",
"datetime": "2026-06-03T10:00:00Z",
"latitude": 53.0,
"longitude": -0.1,
"altitude": None,
},
],
)
_api_login(client)
# end is inclusive: 2026-06-02 should be included
resp = client.get("/api/locations", params={"end": "2026-06-02T10:00:00Z"})
assert resp.status_code == 200
items = resp.json()["items"]
datetimes = [it["datetime"] for it in items]
assert "2026-06-01T10:00:00Z" in datetimes
assert "2026-06-02T10:00:00Z" in datetimes # inclusive
assert "2026-06-03T10:00:00Z" not in datetimes
def test_locations_start_and_end_filter(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.1,
"altitude": None,
},
{
"person": "alice",
"datetime": "2026-06-03T10:00:00Z",
"latitude": 53.0,
"longitude": -0.1,
"altitude": None,
},
],
)
_api_login(client)
resp = client.get(
"/api/locations",
params={"start": "2026-06-02T10:00:00Z", "end": "2026-06-02T10:00:00Z"},
)
assert resp.status_code == 200
items = resp.json()["items"]
assert len(items) == 1
assert items[0]["datetime"] == "2026-06-02T10:00:00Z"
# ---------------------------------------------------------------------------
# GET /api/poo — basic
# ---------------------------------------------------------------------------
def test_poo_empty_returns_empty_list(poo_client) -> None:
client, _engine = poo_client
_api_login(client)
resp = client.get("/api/poo")
assert resp.status_code == 200
body = resp.json()
assert body["items"] == []
assert body["limit"] == 100
assert body["offset"] == 0
def test_poo_returns_seeded_rows_desc(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-03T10:00Z",
"status": "fail",
"latitude": 52.0,
"longitude": -0.2,
},
{
"timestamp": "2026-06-02T10:00Z",
"status": "success",
"latitude": 53.0,
"longitude": -0.3,
},
],
)
_api_login(client)
resp = client.get("/api/poo")
assert resp.status_code == 200
items = resp.json()["items"]
assert len(items) == 3
# ordered by timestamp desc
assert items[0]["timestamp"] == "2026-06-03T10:00Z"
assert items[1]["timestamp"] == "2026-06-02T10:00Z"
assert items[2]["timestamp"] == "2026-06-01T10:00Z"
def test_poo_returns_all_fields(poo_client) -> None:
client, engine = poo_client
_seed_poo(
engine,
[
{
"timestamp": "2026-06-01T10:00Z",
"status": "success",
"latitude": 51.5,
"longitude": -0.1,
}
],
)
_api_login(client)
resp = client.get("/api/poo")
item = resp.json()["items"][0]
assert set(item.keys()) == {"timestamp", "status", "latitude", "longitude"}
assert item["status"] == "success"
# ---------------------------------------------------------------------------
# GET /api/poo — pagination
# ---------------------------------------------------------------------------
def test_poo_limit_and_offset(poo_client) -> None:
client, engine = poo_client
_seed_poo(
engine,
[
{
"timestamp": f"2026-06-{i:02d}T10:00Z",
"status": "success",
"latitude": 51.0,
"longitude": -0.1,
}
for i in range(1, 6)
],
)
_api_login(client)
resp = client.get("/api/poo", params={"limit": 2, "offset": 1})
assert resp.status_code == 200
body = resp.json()
assert body["limit"] == 2
assert body["offset"] == 1
items = body["items"]
assert len(items) == 2
# desc order: rows are 06-05, 06-04, 06-03, 06-02, 06-01
# offset=1 skips 06-05, so first item should be 06-04
assert items[0]["timestamp"] == "2026-06-04T10:00Z"
def test_poo_limit_at_cap_returns_200(poo_client) -> None:
client, _engine = poo_client
_api_login(client)
resp = client.get("/api/poo", params={"limit": 1000})
assert resp.status_code == 200
assert resp.json()["limit"] == 1000
def test_poo_limit_exceeds_cap_returns_422(poo_client) -> None:
client, _engine = poo_client
_api_login(client)
resp = client.get("/api/poo", params={"limit": 1001})
assert resp.status_code == 422
def test_poo_limit_zero_returns_422(poo_client) -> None:
client, _engine = poo_client
_api_login(client)
resp = client.get("/api/poo", params={"limit": 0})
assert resp.status_code == 422
def test_poo_negative_offset_returns_422(poo_client) -> None:
client, _engine = poo_client
_api_login(client)
resp = client.get("/api/poo", params={"offset": -1})
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# GET /api/public-ip
# ---------------------------------------------------------------------------
def test_public_ip_empty_returns_null_state_and_empty_history(client: TestClient) -> None:
_api_login(client)
resp = client.get("/api/public-ip")
assert resp.status_code == 200
body = resp.json()
assert body["state"] is None
assert body["history"] == []
def test_public_ip_returns_state_and_history(location_client) -> None:
client, engine = location_client
_seed_public_ip(engine)
_api_login(client)
resp = client.get("/api/public-ip")
assert resp.status_code == 200
body = resp.json()
state = body["state"]
assert state is not None
assert state["current_ipv4"] == "1.2.3.4"
assert state["previous_ipv4"] == "1.2.3.3"
assert state["last_check_status"] == "changed"
history = body["history"]
assert len(history) == 2
# ordered by observed_at desc — more recent item first
assert history[0]["ipv4"] == "1.2.3.4"
assert history[1]["ipv4"] == "1.2.3.3"
def test_public_ip_history_limit_at_cap_returns_200(client: TestClient) -> None:
_api_login(client)
resp = client.get("/api/public-ip", params={"limit": 1000})
assert resp.status_code == 200
def test_public_ip_history_limit_exceeds_cap_returns_422(client: TestClient) -> None:
_api_login(client)
resp = client.get("/api/public-ip", params={"limit": 1001})
assert resp.status_code == 422
def test_public_ip_history_limit_zero_returns_422(client: TestClient) -> None:
_api_login(client)
resp = client.get("/api/public-ip", params={"limit": 0})
assert resp.status_code == 422
def test_public_ip_state_has_expected_fields(location_client) -> None:
client, engine = location_client
_seed_public_ip(engine)
_api_login(client)
resp = client.get("/api/public-ip")
state = resp.json()["state"]
expected_keys = {
"id",
"current_ipv4",
"previous_ipv4",
"first_seen_at",
"last_checked_at",
"last_changed_at",
"last_check_status",
"last_check_error",
"last_provider",
}
assert set(state.keys()) == expected_keys
def test_public_ip_history_has_expected_fields(location_client) -> None:
client, engine = location_client
_seed_public_ip(engine)
_api_login(client)
resp = client.get("/api/public-ip")
h = resp.json()["history"][0]
assert set(h.keys()) == {"id", "ipv4", "observed_at", "change_type", "provider"}
Reference in New Issue View Git Blame Copy Permalink