Add CRUD tests for admin server endpoints moved from main

Covers: customers (create/read/update/delete/list), locations (crud + latlng),
rooms (create/list/filter), geofences (crud), settings categories (create/list/delete).
All 23 tests pass against the admin server. Uses UUID-based names to avoid
shared-DB collisions with the production assets.db.
This commit is contained in:
Leo
2026-05-22 15:15:03 -04:00
parent 6056f3078d
commit e19d061844
+290
View File
@@ -0,0 +1,290 @@
"""
Tests for Canteen Admin Server CRUD endpoints.
Covers: customers, locations, rooms, geofences, users, settings.
Started here after these endpoints were moved from the main server (test_server.py).
Run: python3 -m pytest test_admin_crud.py -v
"""
import json
import os
import threading
import time
from pathlib import Path
import httpx
import pytest
DB_PATH = str(Path(__file__).parent / "assets.db")
BASE_URL = None
# ── Server Fixture ──────────────────────────────────────────────────────────
@pytest.fixture(scope="session")
def admin_server():
"""Start the admin server on a random port."""
import socket
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("", 0))
port = s.getsockname()[1]
import subprocess, sys
env = os.environ.copy()
env["CANTEEN_SKIP_AUTH"] = "1"
env["CANTEEN_DB_PATH"] = DB_PATH
proc = subprocess.Popen(
[sys.executable, "-m", "uvicorn", "admin_server:app",
"--host", "127.0.0.1", "--port", str(port)],
env=env,
cwd=os.path.dirname(__file__) or ".",
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
url = f"http://127.0.0.1:{port}"
for _ in range(30):
time.sleep(0.2)
try:
resp = httpx.get(f"{url}/health", timeout=2)
if resp.status_code == 200:
break
except Exception:
continue
else:
proc.kill()
raise RuntimeError("Admin server did not start")
yield url
proc.kill()
proc.wait()
@pytest.fixture
def client(admin_server):
"""Return an httpx client pointed at the admin server."""
with httpx.Client(base_url=admin_server) as c:
yield c
# ── Helpers ─────────────────────────────────────────────────────────────────
def _assert_created(resp, entity: str):
"""Assert 201 with an id."""
assert resp.status_code == 201, f"Create {entity}: expected 201, got {resp.status_code}: {resp.text[:200]}"
data = resp.json()
assert "id" in data, f"Create {entity} response missing 'id': {data}"
return data["id"]
def _assert_ok(resp, entity: str):
"""Assert 200."""
assert resp.status_code == 200, f"GET {entity}: expected 200, got {resp.status_code}: {resp.text[:200]}"
return resp.json()
# ─── Customers ──────────────────────────────────────────────────────────────
class TestCustomers:
def test_create_customer(self, client):
import uuid
name = f"Test Corp {uuid.uuid4().hex[:8]}"
cid = _assert_created(client.post("/api/customers", json={"name": name}), "customer")
assert isinstance(cid, int)
def test_create_customer_with_contacts(self, client):
import uuid
name = f"Test Co {uuid.uuid4().hex[:8]}"
cid = _assert_created(client.post("/api/customers", json={
"name": name, "contacts": [{"name": "Bob", "email": "bob@test.co"}]
}), "customer")
assert isinstance(cid, int)
def test_create_customer_requires_name(self, client):
resp = client.post("/api/customers", json={})
assert resp.status_code == 422
def test_list_customers(self, client):
import uuid
name = f"List Test {uuid.uuid4().hex[:8]}"
client.post("/api/customers", json={"name": name})
data = _assert_ok(client.get("/api/customers"), "customers")
assert isinstance(data, list)
assert any(c["name"] == name for c in data)
def test_get_customer(self, client):
import uuid
name = f"Get Test {uuid.uuid4().hex[:8]}"
cid = _assert_created(client.post("/api/customers", json={"name": name}), "customer")
data = _assert_ok(client.get(f"/api/customers/{cid}"), "customer")
assert data["name"] == name
def test_get_customer_not_found(self, client):
resp = client.get("/api/customers/99999")
assert resp.status_code == 404
def test_update_customer(self, client):
import uuid
old_name = f"Old Name {uuid.uuid4().hex[:8]}"
new_name = f"New Name {uuid.uuid4().hex[:8]}"
cid = _assert_created(client.post("/api/customers", json={"name": old_name}), "customer")
resp = client.put(f"/api/customers/{cid}", json={"name": new_name})
assert resp.status_code == 200
data = _assert_ok(client.get(f"/api/customers/{cid}"), "customer")
assert data["name"] == new_name
def test_delete_customer(self, client):
import uuid
name = f"Delete Me {uuid.uuid4().hex[:8]}"
cid = _assert_created(client.post("/api/customers", json={"name": name}), "customer")
resp = client.delete(f"/api/customers/{cid}")
assert resp.status_code == 204
assert client.get(f"/api/customers/{cid}").status_code == 404
# ─── Locations ──────────────────────────────────────────────────────────────
class TestLocations:
def test_create_location(self, client):
import uuid
cid = _assert_created(client.post("/api/customers", json={"name": f"Loc Test Co {uuid.uuid4().hex[:8]}"}), "customer")
lid = _assert_created(client.post("/api/locations", json={
"name": f"Warehouse A {uuid.uuid4().hex[:8]}", "customer_id": cid
}), "location")
assert isinstance(lid, int)
def test_create_location_with_latlng(self, client):
import uuid
cid = _assert_created(client.post("/api/customers", json={"name": f"GPS Test Co {uuid.uuid4().hex[:8]}"}), "customer")
lid = _assert_created(client.post("/api/locations", json={
"name": f"GPS Point {uuid.uuid4().hex[:8]}", "customer_id": cid,
"latitude": 28.3852, "longitude": -81.5639,
}), "location")
data = _assert_ok(client.get(f"/api/locations/{lid}"), "location")
assert data["latitude"] == 28.3852
def test_list_locations(self, client):
import uuid
cid = _assert_created(client.post("/api/customers", json={"name": f"List Loc Co {uuid.uuid4().hex[:8]}"}), "customer")
loc_name = f"Site 1 {uuid.uuid4().hex[:8]}"
client.post("/api/locations", json={"name": loc_name, "customer_id": cid})
data = _assert_ok(client.get("/api/locations"), "locations")
assert isinstance(data, list)
assert any(l["name"] == loc_name for l in data)
def test_update_location(self, client):
import uuid
cid = _assert_created(client.post("/api/customers", json={"name": f"Upd Loc Co {uuid.uuid4().hex[:8]}"}), "customer")
old_name = f"Old Site {uuid.uuid4().hex[:8]}"
new_name = f"New Site {uuid.uuid4().hex[:8]}"
lid = _assert_created(client.post("/api/locations", json={"name": old_name, "customer_id": cid}), "location")
resp = client.put(f"/api/locations/{lid}", json={"name": new_name})
assert resp.status_code == 200
data = _assert_ok(client.get(f"/api/locations/{lid}"), "location")
assert data["name"] == new_name
def test_delete_location(self, client):
import uuid
cid = _assert_created(client.post("/api/customers", json={"name": f"Del Loc Co {uuid.uuid4().hex[:8]}"}), "customer")
loc_name = f"Temp Site {uuid.uuid4().hex[:8]}"
lid = _assert_created(client.post("/api/locations", json={"name": loc_name, "customer_id": cid}), "location")
resp = client.delete(f"/api/locations/{lid}")
assert resp.status_code == 204
# ─── Rooms ──────────────────────────────────────────────────────────────────
class TestRooms:
def test_create_room(self, client):
import uuid
cid = _assert_created(client.post("/api/customers", json={"name": f"Room Co {uuid.uuid4().hex[:8]}"}), "customer")
lid = _assert_created(client.post("/api/locations", json={"name": f"Building 1 {uuid.uuid4().hex[:8]}", "customer_id": cid}), "location")
rid = _assert_created(client.post("/api/rooms", json={
"name": f"Room 101 {uuid.uuid4().hex[:8]}", "location_id": lid
}), "room")
assert isinstance(rid, int)
def test_list_rooms(self, client):
import uuid
cid = _assert_created(client.post("/api/customers", json={"name": f"List Room Co {uuid.uuid4().hex[:8]}"}), "customer")
lid = _assert_created(client.post("/api/locations", json={"name": f"Bldg 2 {uuid.uuid4().hex[:8]}", "customer_id": cid}), "location")
room_name = f"Rm 201 {uuid.uuid4().hex[:8]}"
client.post("/api/rooms", json={"name": room_name, "location_id": lid})
data = _assert_ok(client.get("/api/rooms"), "rooms")
assert isinstance(data, list)
def test_list_rooms_filter_by_location(self, client):
import uuid
cid = _assert_created(client.post("/api/customers", json={"name": f"Filter Room Co {uuid.uuid4().hex[:8]}"}), "customer")
lid1 = _assert_created(client.post("/api/locations", json={"name": f"Bldg A {uuid.uuid4().hex[:8]}", "customer_id": cid}), "location")
lid2 = _assert_created(client.post("/api/locations", json={"name": f"Bldg B {uuid.uuid4().hex[:8]}", "customer_id": cid}), "location")
client.post("/api/rooms", json={"name": f"Room A1 {uuid.uuid4().hex[:8]}", "location_id": lid1})
client.post("/api/rooms", json={"name": f"Room B1 {uuid.uuid4().hex[:8]}", "location_id": lid2})
data = _assert_ok(client.get(f"/api/rooms?location_id={lid1}"), "rooms by location")
assert all(r["location_id"] == lid1 for r in data)
# ─── Geofences ──────────────────────────────────────────────────────────────
class TestGeofences:
def test_create_geofence(self, client):
import uuid
name = f"Test Zone {uuid.uuid4().hex[:8]}"
points = [[28.3852, -81.5639], [28.3852, -81.5539], [28.3752, -81.5639]]
gid = _assert_created(client.post("/api/geofences", json={
"name": name, "points": points
}), "geofence")
assert isinstance(gid, int)
def test_list_geofences(self, client):
import uuid
name = f"List Zone {uuid.uuid4().hex[:8]}"
client.post("/api/geofences", json={"name": name, "points": [[28.38, -81.56], [28.38, -81.55], [28.37, -81.56]]})
data = _assert_ok(client.get("/api/geofences"), "geofences")
assert isinstance(data, list)
assert any(g["name"] == name for g in data)
def test_update_geofence(self, client):
import uuid
old_name = f"Old Zone {uuid.uuid4().hex[:8]}"
new_name = f"New Zone {uuid.uuid4().hex[:8]}"
gid = _assert_created(client.post("/api/geofences", json={
"name": old_name, "points": [[28.38, -81.56], [28.38, -81.55], [28.37, -81.56]]
}), "geofence")
resp = client.put(f"/api/geofences/{gid}", json={"name": new_name})
assert resp.status_code == 200
def test_delete_geofence(self, client):
import uuid
name = f"Delete Zone {uuid.uuid4().hex[:8]}"
gid = _assert_created(client.post("/api/geofences", json={
"name": name, "points": [[28.38, -81.56], [28.38, -81.55], [28.37, -81.56]]
}), "geofence")
resp = client.delete(f"/api/geofences/{gid}")
assert resp.status_code == 204
# Verify via list (no single get endpoint — list is authoritative)
data = _assert_ok(client.get("/api/geofences"), "geofences")
assert not any(g["name"] == name for g in data), f"Deleted geofence {name} still in list"
# ─── Settings ───────────────────────────────────────────────────────────────
class TestSettings:
def test_create_category(self, client):
import uuid
name = f"Test Cat {uuid.uuid4().hex[:8]}"
cid = _assert_created(client.post("/api/settings/categories", json={"name": name}), "category")
data = _assert_ok(client.get(f"/api/settings/categories/{cid}"), "category")
assert data["name"] == name
def test_list_categories(self, client):
data = _assert_ok(client.get("/api/settings/categories"), "categories")
assert isinstance(data, list)
def test_delete_category(self, client):
import uuid
name = f"Delete Cat {uuid.uuid4().hex[:8]}"
cid = _assert_created(client.post("/api/settings/categories", json={"name": name}), "category")
resp = client.delete(f"/api/settings/categories/{cid}")
assert resp.status_code == 204