canteen-admin-server/test_admin_crud.py

"""
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


# ─── Asset GPS ──────────────────────────────────────────────────────────────

class TestClearAssetGps:
    def test_clear_gps_returns_204(self, client):
        """Clear GPS returns 204 and sets lat/lng to null."""
        # Create an asset with GPS coords via direct DB insert
        import sqlite3
        conn = sqlite3.connect(DB_PATH)
        conn.execute(
            "INSERT INTO assets (machine_id, name, latitude, longitude) VALUES (?, ?, ?, ?)",
            ("GPS-ADMIN-TEST", "Test GPS Asset", 28.5, -81.3),
        )
        asset_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
        conn.commit()
        conn.close()

        resp = client.delete(f"/api/assets/{asset_id}/gps")
        assert resp.status_code == 204

        # Verify
        data = _assert_ok(client.get(f"/api/assets/{asset_id}"), "asset")
        assert data["latitude"] is None
        assert data["longitude"] is None

    def test_clear_gps_not_found_returns_404(self, client):
        resp = client.delete("/api/assets/99999/gps")
        assert resp.status_code == 404