331 lines
14 KiB
Python
331 lines
14 KiB
Python
"""
|
|
Focused tests for uncovered API areas in Canteen Asset Tracker.
|
|
|
|
Covers: geofence point check, proximity search, service-summary export,
|
|
settings models CRUD, and auth-aware smoke test helpers.
|
|
"""
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
# ─── Test DB setup ────────────────────────────────────────────────────────
|
|
|
|
TEST_DB = Path(__file__).parent / "test_gap_coverage.db"
|
|
os.environ["CANTEEN_DB_PATH"] = str(TEST_DB)
|
|
os.environ["CANTEEN_SKIP_AUTH"] = "1"
|
|
|
|
|
|
def _clean_db():
|
|
for suffix in ("", "-shm", "-wal", "-journal"):
|
|
p = TEST_DB.with_suffix(TEST_DB.suffix + suffix)
|
|
if p.exists():
|
|
p.unlink()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_db():
|
|
_clean_db()
|
|
yield
|
|
_clean_db()
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
_clean_db()
|
|
for mod in list(sys.modules.keys()):
|
|
if mod == "server" or mod.startswith("server."):
|
|
del sys.modules[mod]
|
|
import server
|
|
import importlib
|
|
importlib.invalidate_caches()
|
|
with TestClient(server.app) as tc:
|
|
yield tc
|
|
|
|
|
|
# ─── Auth helpers ──────────────────────────────────────────────────────────
|
|
|
|
def login(client, username="admin", password="changeme"):
|
|
"""Login and return the auth header dict."""
|
|
r = client.post("/api/auth/login", json={"username": username, "password": password})
|
|
if r.status_code != 200:
|
|
pytest.skip(f"Login failed ({r.status_code}): {r.text}")
|
|
token = r.json()["token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 1. Geofence point check
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestGeofencePointCheck:
|
|
"""/api/geofences/check — test if a point is inside a geofence polygon."""
|
|
|
|
def _create_square_geofence(self, client, name, lat=0, lng=0, size=1):
|
|
"""Helper: create a square geofence centered at (lat, lng)."""
|
|
return client.post("/api/geofences", json={
|
|
"name": name,
|
|
"points": [
|
|
{"lat": lat - size, "lng": lng - size},
|
|
{"lat": lat - size, "lng": lng + size},
|
|
{"lat": lat + size, "lng": lng + size},
|
|
{"lat": lat + size, "lng": lng - size},
|
|
],
|
|
"color": "#ff0000",
|
|
})
|
|
|
|
def test_check_inside_polygon(self, client):
|
|
"""Point clearly inside a simple square geofence."""
|
|
self._create_square_geofence(client, "Square", lat=40, lng=-74, size=1)
|
|
r = client.post("/api/geofences/check", json={"lat": 40, "lng": -74})
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert isinstance(data, list)
|
|
names = [g["name"] for g in data]
|
|
assert "Square" in names, f"Expected Square in results, got: {names}"
|
|
|
|
def test_check_outside_polygon(self, client):
|
|
"""Point clearly outside the geofence — returns empty list."""
|
|
self._create_square_geofence(client, "Tiny Box", lat=0, lng=0, size=1)
|
|
r = client.post("/api/geofences/check", json={"lat": 50, "lng": 50})
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data == [], f"Expected empty list, got: {data}"
|
|
|
|
def test_check_no_geofences(self, client):
|
|
"""No geofences exist — empty array."""
|
|
r = client.post("/api/geofences/check", json={"lat": 0, "lng": 0})
|
|
assert r.status_code == 200
|
|
assert r.json() == []
|
|
|
|
def test_check_invalid_input(self, client):
|
|
"""Missing lat/lng returns 422."""
|
|
r = client.post("/api/geofences/check", json={})
|
|
assert r.status_code == 422
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 2. Proximity search
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestProximitySearch:
|
|
"""/api/proximity — find assets near a GPS point."""
|
|
|
|
def test_no_assets_nearby(self, client):
|
|
"""No assets exist — empty list."""
|
|
r = client.get("/api/proximity?lat=0&lng=0&radius_meters=1000")
|
|
assert r.status_code == 200
|
|
assert r.json() == []
|
|
|
|
def test_asset_within_radius(self, client):
|
|
"""Asset with lat/lng near the query point."""
|
|
aid = client.post("/api/assets", json={
|
|
"machine_id": "PROX-001",
|
|
"name": "Nearby Asset",
|
|
"latitude": 40.7128,
|
|
"longitude": -74.006,
|
|
}).json()["id"]
|
|
|
|
r = client.get("/api/proximity?lat=40.713&lng=-74.007&radius_meters=1000")
|
|
assert r.status_code == 200
|
|
ids = [a["id"] for a in r.json()]
|
|
assert aid in ids, f"Asset {aid} not in proximity results: {r.json()}"
|
|
|
|
def test_asset_outside_radius(self, client):
|
|
"""Asset far from query point — use NYC vs Tokyo."""
|
|
client.post("/api/assets", json={
|
|
"machine_id": "PROX-FAR",
|
|
"name": "Far Asset",
|
|
"latitude": 40.7128,
|
|
"longitude": -74.006,
|
|
})
|
|
|
|
r = client.get("/api/proximity?lat=35.6762&lng=139.6503&radius_meters=10000")
|
|
assert r.status_code == 200
|
|
machines = [a["machine_id"] for a in r.json()]
|
|
assert "PROX-FAR" not in machines, f"Far asset unexpectedly in Tokyo proximity: {r.json()}"
|
|
|
|
def test_asset_no_coords(self, client):
|
|
"""Asset without lat/lng should not appear in proximity results."""
|
|
client.post("/api/assets", json={
|
|
"machine_id": "PROX-NOCOORD",
|
|
"name": "No Coord Asset",
|
|
})
|
|
|
|
r = client.get("/api/proximity?lat=0&lng=0&radius_meters=50000")
|
|
assert r.status_code == 200
|
|
machines = [a["machine_id"] for a in r.json()]
|
|
assert "PROX-NOCOORD" not in machines, f"Asset without coords unexpectedly in results: {r.json()}"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 3. Service Summary Export
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestServiceSummaryExport:
|
|
"""/api/export/service-summary — CSV export with visit data."""
|
|
|
|
def test_export_returns_csv_content_type(self, client):
|
|
"""CSV export sets proper content type."""
|
|
r = client.get("/api/export/service-summary")
|
|
assert r.status_code == 200
|
|
assert "text/csv" in r.headers.get("content-type", "").lower()
|
|
|
|
def test_export_has_expected_headers(self, client):
|
|
"""CSV has expected columns."""
|
|
r = client.get("/api/export/service-summary")
|
|
assert r.status_code == 200
|
|
text = r.text
|
|
assert "customer_name" in text or "asset" in text, f"Unexpected headers: {text[:200]}"
|
|
|
|
def test_export_with_data_includes_rows(self, client):
|
|
"""CSV has data rows when assets exist with visits/customers."""
|
|
# Create a customer
|
|
cust = client.post("/api/customers", json={"name": "Test Customer"}).json()
|
|
# Create a location for the customer
|
|
loc = client.post("/api/locations", json={
|
|
"customer_id": cust["id"],
|
|
"name": "Test Location",
|
|
}).json()
|
|
# Create an asset at that location
|
|
aid = client.post("/api/assets", json={
|
|
"machine_id": "SRV-003",
|
|
"name": "Service Asset",
|
|
"customer_id": cust["id"],
|
|
"location_id": loc["id"],
|
|
}).json()["id"]
|
|
# Create a checkin
|
|
client.post("/api/checkins", json={"asset_id": aid, "notes": "Service visit"})
|
|
|
|
r = client.get("/api/export/service-summary")
|
|
assert r.status_code == 200
|
|
lines = r.text.strip().split("\n")
|
|
assert len(lines) >= 2, f"Expected header + data rows, got {len(lines)} lines: {r.text[:200]}"
|
|
# CSV aggregates by customer/location; check the data row has our test data
|
|
assert "Test Customer" in r.text
|
|
assert "Test Location" in r.text
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 4. Settings Models CRUD
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestSettingsModelsCRUD:
|
|
"""Models have make_id dependency — test full lifecycle."""
|
|
|
|
def test_create_model_with_make(self, client):
|
|
"""Create a make, then a model referencing it."""
|
|
make = client.post("/api/settings/makes", json={"name": "TestMake"}).json()
|
|
make_id = make["id"]
|
|
|
|
r = client.post("/api/settings/models", json={
|
|
"make_id": make_id,
|
|
"name": "TestModel",
|
|
})
|
|
assert r.status_code == 201
|
|
data = r.json()
|
|
assert data["name"] == "TestModel"
|
|
assert data["make_id"] == make_id
|
|
|
|
def test_create_model_without_make_fails(self, client):
|
|
"""Missing make_id returns 422."""
|
|
r = client.post("/api/settings/models", json={"name": "Orphan Model"})
|
|
assert r.status_code == 422
|
|
|
|
def test_list_models(self, client):
|
|
"""List models."""
|
|
r = client.get("/api/settings/models")
|
|
assert r.status_code == 200
|
|
assert isinstance(r.json(), list)
|
|
|
|
def test_update_model(self, client):
|
|
"""Update a model's name."""
|
|
make = client.post("/api/settings/makes", json={"name": "MakeForUpdate"}).json()
|
|
model = client.post("/api/settings/models", json={
|
|
"make_id": make["id"],
|
|
"name": "OldName",
|
|
}).json()
|
|
|
|
r = client.put(f"/api/settings/models/{model['id']}", json={"name": "NewName"})
|
|
assert r.status_code == 200
|
|
assert r.json()["name"] == "NewName"
|
|
|
|
def test_get_single_model(self, client):
|
|
"""Get a single model by id."""
|
|
make = client.post("/api/settings/makes", json={"name": "MakeForGet"}).json()
|
|
model = client.post("/api/settings/models", json={
|
|
"make_id": make["id"],
|
|
"name": "GetMe",
|
|
}).json()
|
|
|
|
r = client.get(f"/api/settings/models/{model['id']}")
|
|
assert r.status_code == 200
|
|
assert r.json()["name"] == "GetMe"
|
|
|
|
def test_delete_model(self, client):
|
|
"""Delete a model."""
|
|
make = client.post("/api/settings/makes", json={"name": "MakeForDel"}).json()
|
|
model = client.post("/api/settings/models", json={
|
|
"make_id": make["id"],
|
|
"name": "DeleteMe",
|
|
}).json()
|
|
|
|
r = client.delete(f"/api/settings/models/{model['id']}")
|
|
assert r.status_code == 204
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 5. Auth-aware smoke test (smoke_test.sh replacement)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestAuthSmokeWorkflow:
|
|
"""Full E2E workflow with auth: login → CRUD → checkin → verify."""
|
|
|
|
def test_full_workflow_with_auth(self, client):
|
|
auth = login(client)
|
|
|
|
# Create asset
|
|
r = client.post("/api/assets", json={
|
|
"machine_id": "E2E-001",
|
|
"name": "E2E Test Asset",
|
|
"category": "Equipment",
|
|
}, headers=auth)
|
|
assert r.status_code == 201
|
|
aid = r.json()["id"]
|
|
|
|
# Search by machine_id
|
|
r = client.get("/api/assets/search?machine_id=E2E-001", headers=auth)
|
|
assert r.status_code == 200
|
|
assert len(r.json()) > 0
|
|
|
|
# Create checkin
|
|
r = client.post("/api/checkins", json={
|
|
"asset_id": aid,
|
|
"latitude": 40.7128,
|
|
"longitude": -74.006,
|
|
"notes": "Found on site",
|
|
}, headers=auth)
|
|
assert r.status_code == 201
|
|
|
|
# Verify stats
|
|
r = client.get("/api/stats", headers=auth)
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["total_assets"] >= 1
|
|
assert data["total_checkins"] >= 1
|
|
|
|
# CSV export
|
|
r = client.get("/api/export/assets", headers=auth)
|
|
assert r.status_code == 200
|
|
assert "E2E-001" in r.text
|
|
|
|
# Delete
|
|
r = client.delete(f"/api/assets/{aid}", headers=auth)
|
|
assert r.status_code in (200, 204), f"Expected 200 or 204, got {r.status_code}: {r.text}"
|
|
|
|
# Verify deleted
|
|
r = client.get(f"/api/assets/{aid}", headers=auth)
|
|
assert r.status_code == 404
|