From e19d0618445d3d0e05fc2e8fe7de54828225f0de Mon Sep 17 00:00:00 2001 From: Leo Date: Fri, 22 May 2026 15:15:03 -0400 Subject: [PATCH] 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. --- test_admin_crud.py | 290 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 290 insertions(+) create mode 100644 test_admin_crud.py diff --git a/test_admin_crud.py b/test_admin_crud.py new file mode 100644 index 0000000..c837720 --- /dev/null +++ b/test_admin_crud.py @@ -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