canteen-admin-server/test_sync_api.py

"""
Test script for Cantaloupe sync API.
Runs the admin server on a random port, creates test Excel data,
and tests all endpoints.
"""
import json
import os
import subprocess
import sys
import tempfile
import threading
import time
from pathlib import Path

import openpyxl
import httpx

DB_PATH = str(Path(__file__).parent / "assets.db")


def create_test_excel(path: str):
    """Create a test Cantaloupe Excel export file."""
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.append([
        "Asset ID", "Name", "Serial Number", "Customer", "Location Name",
        "Make", "Model", "Status", "Category", "Address"
    ])
    ws.append([
        "MACH001", "Coffee Machine 1", "SN001", "Acme Corp", "Building A",
        "Canteen", "CM2000", "active", "Appliances", "123 Main St"
    ])
    ws.append([
        "MACH002", "Fridge Unit 2", "SN002", "Acme Corp", "Building B",
        "Hobart", "HF500", "active", "Appliances", "456 Oak Ave"
    ])
    wb.save(path)
    print(f"  Created test Excel with {ws.max_row - 1} data rows")


def run_server(port: int, ready_event: threading.Event):
    """Run the admin server in a subprocess."""
    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__),
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    # Wait for server to be ready
    for _ in range(30):
        time.sleep(0.2)
        try:
            resp = httpx.get(f"http://127.0.0.1:{port}/health", timeout=2)
            if resp.status_code == 200:
                ready_event.set()
                return proc, port
        except Exception:
            pass
    proc.kill()
    raise RuntimeError("Server did not start")


def run_tests(base_url: str, excel_path: str):
    """Test all Cantaloupe sync API endpoints."""
    passed = 0
    failed = 0

    def check(name: str, condition: bool, detail: str = ""):
        nonlocal passed, failed
        if condition:
            print(f"  ✓ {name}")
            passed += 1
        else:
            print(f"  ✗ {name} — {detail}")
            failed += 1

    # Clean up any leftover test data from previous runs
    import sqlite3
    conn = sqlite3.connect(DB_PATH)
    conn.execute("DELETE FROM cantaloupe_sync_batches")
    conn.execute("DELETE FROM assets WHERE machine_id IN ('MACH001', 'MACH002')")
    conn.execute("DELETE FROM customers WHERE name = 'Acme Corp'")
    conn.execute("DELETE FROM locations")
    conn.commit()
    conn.close()

    # 1. List batches (should be empty now)
    resp = httpx.get(f"{base_url}/api/admin/cantaloupe/batches")
    check("GET /batches returns 200", resp.status_code == 200, str(resp.status_code))
    batches = resp.json()
    check("GET /batches returns list", isinstance(batches, list), str(type(batches)))
    check("GET /batches is empty initially", len(batches) == 0, f"count={len(batches)}")

    # 2. Run sync
    resp = httpx.post(
        f"{base_url}/api/admin/cantaloupe/sync",
        params={"file_path": excel_path},
    )
    check("POST /sync returns 200", resp.status_code == 200, f"{resp.status_code}: {resp.text[:200]}")
    sync_result = resp.json()
    check("POST /sync has id", "id" in sync_result, str(sync_result.keys()))
    check("POST /sync status=pending", sync_result.get("status") == "pending", sync_result.get("status"))
    check("POST /sync has diff_summary", "diff_summary" in sync_result, str(sync_result.keys()))
    
    diff = sync_result.get("diff_summary", {})
    print(f"    Diff summary: {json.dumps({k: v for k, v in diff.items() if k != 'raw_data'}, default=str)[:200]}")
    
    batch_id = sync_result.get("id")
    assert batch_id, "No batch id in sync result"

    # 3. Get batch detail
    resp = httpx.get(f"{base_url}/api/admin/cantaloupe/batches/{batch_id}")
    check("GET /batches/<id> returns 200", resp.status_code == 200, str(resp.status_code))
    detail = resp.json()
    check("GET /batches/<id> has raw_data", "raw_data" in detail, str(detail.keys()))
    check("GET /batches/<id> has diff_rows", "diff_rows" in detail, str(detail.keys()))

    # 4. List batches again (should have 1)
    resp = httpx.get(f"{base_url}/api/admin/cantaloupe/batches")
    check("GET /batches has 1 item after sync", len(resp.json()) == 1, f"count={len(resp.json())}")

    # 5. Approve batch
    resp = httpx.post(f"{base_url}/api/admin/cantaloupe/batches/{batch_id}/approve")
    check("POST /approve returns 200", resp.status_code == 200, f"{resp.status_code}: {resp.text[:200]}")
    approve_result = resp.json()
    check("POST /approve status=approved", approve_result.get("status") == "approved", approve_result.get("status"))
    check("POST /approve has approved_at", bool(approve_result.get("approved_at")), "missing approved_at")
    check("POST /approve has applied_stats", "applied_stats" in approve_result, str(approve_result.keys()))
    if "applied_stats" in approve_result:
        stats = approve_result["applied_stats"]
        print(f"    Applied: {stats}")

    # 6. Verify customers were created (via API)
    resp = httpx.get(f"{base_url}/api/customers")
    if resp.status_code == 200:
        customers = resp.json()
        names = [c.get("name") for c in customers]
        check("Customer 'Acme Corp' created", any("Acme Corp" in str(n) for n in names), str(names))

    # 7. Verify locations were created (via API)
    resp = httpx.get(f"{base_url}/api/locations")
    if resp.status_code == 200:
        locations = resp.json()
        loc_names = [l.get("name") for l in locations]
        check("Location created from sync", len(locations) >= 2, f"count={len(locations)}")

    # 9. Re-approve should fail (already approved)
    resp = httpx.post(f"{base_url}/api/admin/cantaloupe/batches/{batch_id}/approve")
    check("Re-approve rejected (status != pending)", resp.status_code == 400, f"got {resp.status_code}")

    # 10. Create another batch and reject it
    resp = httpx.post(
        f"{base_url}/api/admin/cantaloupe/sync",
        params={"file_path": excel_path},
    )
    batch2_id = resp.json().get("id")
    check("Second sync batch created", batch2_id is not None, str(resp.json())[:100])

    if batch2_id:
        resp = httpx.post(f"{base_url}/api/admin/cantaloupe/batches/{batch2_id}/reject")
        check("POST /reject returns 200", resp.status_code == 200, f"{resp.status_code}: {resp.text[:200]}")
        reject_result = resp.json()
        check("Rejected status=rejected", reject_result.get("status") == "rejected", reject_result.get("status"))
        check("Rejected raw_data cleared", reject_result.get("raw_data") is None, str(reject_result.get("raw_data"))[:50])

    # 11. Test 404 for non-existent batch
    resp = httpx.get(f"{base_url}/api/admin/cantaloupe/batches/99999")
    check("GET /batches/99999 returns 404", resp.status_code == 404, str(resp.status_code))

    print(f"\nResults: {passed} passed, {failed} failed")
    return failed == 0


def main():
    print("=== Cantaloupe Sync API Tests ===\n")

    excel_path = "/tmp/test_cantaloupe_api.xlsx"
    create_test_excel(excel_path)

    # Find a free port
    import socket
    sock = socket.socket()
    sock.bind(('', 0))
    port = sock.getsockname()[1]
    sock.close()
    
    base_url = f"http://127.0.0.1:{port}"
    print(f"Starting server on port {port}...")

    ready = threading.Event()
    proc, _ = run_server(port, ready)
    
    try:
        ready.wait(timeout=10)
        print(f"Server ready at {base_url}\n")
        success = run_tests(base_url, excel_path)
    finally:
        proc.terminate()
        proc.wait(timeout=5)
        print("\nServer stopped.")

    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()