From f7ffa0e6ccf9cec237d7a92988820200120bb798 Mon Sep 17 00:00:00 2001 From: Leo Date: Thu, 21 May 2026 19:16:53 -0400 Subject: [PATCH] cantaloupe_sync: fix column mapping heuristic, add comprehensive tests - Fix greedy matching where 'name' keyword matched 'Location Name' header - Use word-subset matching instead of substring matching - Normalize keywords (replace _ and # with spaces) to match header normalization - Reorder heuristics: multi-word specific patterns before single-word generics - Add 10 unit tests for mapping, diff, and Excel parsing - Add 23 integration tests for all sync API endpoints - All 33 tests pass --- cantaloupe_sync.py | 79 +++++++++----- test_cantaloupe_sync.py | 221 ++++++++++++++++++++++++++++++++++++++++ test_sync_api.py | 212 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 487 insertions(+), 25 deletions(-) create mode 100644 test_cantaloupe_sync.py create mode 100644 test_sync_api.py diff --git a/cantaloupe_sync.py b/cantaloupe_sync.py index 1cf5e34..4a9d0fd 100644 --- a/cantaloupe_sync.py +++ b/cantaloupe_sync.py @@ -98,37 +98,59 @@ ASSET_COLUMNS = [ COLUMN_HEURISTICS = [ # machine_id is the primary key for upserts (["asset id", "assetid", "machine id", "machineid", "machine_id", "equipment id", "unit id", "asset number"], "machine_id"), - (["serial", "serial number", "serialnumber", "serial_no", "s/n"], "serial_number"), - (["name", "asset name", "equipment name", "description"], "name"), - (["description", "notes", "detail", "comments"], "description"), + # Multi-word / specific heuristics first to avoid greedy single-word matches + (["location name", "location_name", "site name"], "_location_name"), + (["building name", "building_name", "bldg name"], "building_name"), + (["building number", "building_number", "building_no", "bldg number", "bldg #", "building #"], "building_number"), + (["trailer number", "trailer_number", "trailer_no", "trailer #", "mobile unit"], "trailer_number"), + (["walking directions", "walking_directions", "how to get there"], "walking_directions"), + (["map link", "map_link", "google maps", "map url", "location url"], "map_link"), + (["parking location", "parking_location", "parking spot"], "parking_location"), + (["street address", "location address", "site address"], "address"), + (["asset name", "equipment name"], "name"), + (["model number", "model_no", "model #"], "model"), + (["serial number", "serialnumber", "serial_no", "s/n"], "serial_number"), + (["room number", "room #", "suite"], "room"), + (["customer name", "client name", "account name", "company"], "_customer_name"), + (["photo path", "photo_path", "image path"], "photo_path"), + (["route number", "route #", "delivery route"], "_route"), + (["sales rep", "salesperson", "account manager"], "_sales_rep"), + (["qr code", "tag number", "tag #"], "_barcode"), + (["geofence radius", "geofence_radius", "geo radius"], "geofence_radius_meters"), + (["gps lat", "lat"], "latitude"), + (["gps lng", "gps long", "long", "lng", "lon"], "longitude"), + # Single-word/generic heuristics (last, after specific multi-word ones) + (["serial", "s/n"], "serial_number"), (["category", "type", "asset type", "equipment type", "class"], "category"), (["status", "state", "condition"], "status"), (["make", "manufacturer", "brand", "vendor"], "make"), - (["model", "model number", "model_no", "model #"], "model"), - (["address", "street", "street address", "location address", "site address"], "address"), - (["building name", "building_name", "bldg name", "site name"], "building_name"), - (["building number", "building_number", "building_no", "bldg number", "bldg #", "building #"], "building_number"), + (["address", "street"], "address"), (["floor", "level", "story"], "floor"), - (["room", "room number", "room #", "suite"], "room"), - (["trailer", "trailer number", "trailer_no", "trailer #", "mobile unit"], "trailer_number"), - (["walking directions", "walking_directions", "directions", "how to get there"], "walking_directions"), - (["map link", "map_link", "google maps", "map url", "location url"], "map_link"), - (["parking", "parking location", "parking_location", "parking spot"], "parking_location"), - (["latitude", "lat", "gps lat"], "latitude"), - (["longitude", "long", "lng", "lon", "gps lng", "gps long"], "longitude"), - (["geofence", "geo radius", "radius", "geofence radius", "geofence_radius"], "geofence_radius_meters"), - (["customer", "client", "account", "company", "customer name"], "_customer_name"), - (["location name", "location_name", "site", "site name", "location"], "_location_name"), - (["photo", "photo path", "photo_path", "image", "picture"], "photo_path"), - (["route", "route number", "route #", "delivery route"], "_route"), - (["sales", "sales rep", "salesperson", "account manager"], "_sales_rep"), - (["barcode", "qr code", "tag", "tag number", "tag #"], "_barcode"), + (["latitude", "gps"], "latitude"), + (["longitude", "geo"], "longitude"), + (["geofence", "radius"], "geofence_radius_meters"), + (["customer", "client", "account"], "_customer_name"), + (["location", "site"], "_location_name"), + (["name"], "name"), + (["description", "notes", "detail", "comments"], "description"), + (["model"], "model"), + (["building"], "building_name"), + (["room"], "room"), + (["trailer"], "trailer_number"), + (["walking", "directions"], "walking_directions"), + (["map"], "map_link"), + (["parking"], "parking_location"), + (["photo", "image", "picture"], "photo_path"), + (["route"], "_route"), + (["sales", "rep"], "_sales_rep"), + (["barcode", "tag", "qr"], "_barcode"), ] def _normalize(s: str) -> str: - """Lowercase, strip, collapse whitespace.""" - return " ".join(str(s).lower().strip().split()) + """Lowercase, strip, collapse whitespace. Replace underscores with spaces + so 'machine_id' matches 'machine id'.""" + return " ".join(str(s).lower().strip().replace("_", " ").replace("#", " ").split()) def map_columns(excel_headers: list[str]) -> dict[str, str]: @@ -146,10 +168,17 @@ def map_columns(excel_headers: list[str]) -> dict[str, str]: if not norm: continue matched = False + norm_words = set(norm.split()) for keywords, canonical in COLUMN_HEURISTICS: for kw in keywords: - # Match if the header contains the keyword or vice versa - if kw in norm or norm in kw: + kw_norm = _normalize(kw) + if not kw_norm: + continue + kw_words = set(kw_norm.split()) + # Match: all keyword words must be present in the header. + # This prevents short headers like "name" from matching + # multi-word keywords like "location name". + if kw_words.issubset(norm_words): mapping[header] = canonical discovered.append(f"{header} → {canonical}") matched = True diff --git a/test_cantaloupe_sync.py b/test_cantaloupe_sync.py new file mode 100644 index 0000000..49d5287 --- /dev/null +++ b/test_cantaloupe_sync.py @@ -0,0 +1,221 @@ +""" +Unit tests for cantaloupe_sync column mapping and diff computation. +""" +import json +import sqlite3 +from pathlib import Path + +from cantaloupe_sync import ( + _normalize, map_columns, apply_mapping, parse_excel, + compute_diff, _fetch_live_data, COLUMN_HEURISTICS, +) + +DB_PATH = str(Path(__file__).parent / "assets.db") + + +def test_normalize(): + assert _normalize("Asset ID") == "asset id" + assert _normalize(" Location Name ") == "location name" + assert _normalize("machine_id") == "machine id" + assert _normalize("Building #") == "building" + assert _normalize("S/N") == "s/n" # slash is not stripped, just whitespace/underscore/hash + assert _normalize("") == "" + + +def test_map_columns_basic(): + """Test that standard Cantaloupe headers map correctly.""" + headers = [ + "Asset ID", "Name", "Serial Number", "Customer", + "Location Name", "Make", "Model", "Status", "Category", "Address" + ] + mapping = map_columns(headers) + assert mapping["Asset ID"] == "machine_id" + assert mapping["Name"] == "name" + assert mapping["Serial Number"] == "serial_number" + assert mapping["Customer"] == "_customer_name" + assert mapping["Location Name"] == "_location_name" + assert mapping["Address"] == "address" + + +def test_map_columns_no_greedy_match(): + """Name should not match location_name heuristic.""" + headers = ["Name", "Location Name"] + mapping = map_columns(headers) + assert mapping["Name"] == "name", f"Expected name, got {mapping.get('Name')}" + assert mapping["Location Name"] == "_location_name" + + +def test_map_columns_variants(): + """Test various header variants.""" + test_cases = [ + (["Asset ID"], {"Asset ID": "machine_id"}), + (["machine_id"], {"machine_id": "machine_id"}), + (["Machine ID"], {"Machine ID": "machine_id"}), + (["Equipment ID"], {"Equipment ID": "machine_id"}), + (["Serial"], {"Serial": "serial_number"}), + (["S/N"], {"S/N": "serial_number"}), + (["Make"], {"Make": "make"}), + (["Manufacturer"], {"Manufacturer": "make"}), + (["Model #"], {"Model #": "model"}), + (["Category"], {"Category": "category"}), + (["Status"], {"Status": "status"}), + (["Customer Name"], {"Customer Name": "_customer_name"}), + (["Client"], {"Client": "_customer_name"}), + (["Building Name"], {"Building Name": "building_name"}), + (["Bldg #"], {"Bldg #": "building_number"}), + ] + for headers, expected in test_cases: + mapping = map_columns(headers) + assert mapping == expected, f"Failed for {headers}: got {mapping}, expected {expected}" + + +def test_apply_mapping(): + """Test that Excel rows are correctly mapped to canteen columns.""" + rows = [ + {"Asset ID": "M001", "Name": "Test", "Customer": "Acme", "UnknownCol": "extra"}, + ] + col_map = {"Asset ID": "machine_id", "Name": "name", "Customer": "_customer_name"} + result = apply_mapping(rows, col_map) + assert result[0]["machine_id"] == "M001" + assert result[0]["name"] == "Test" + assert result[0]["_customer_name"] == "Acme" + assert result[0]["_raw_UnknownCol"] == "extra" + + +def test_compute_diff_no_live_data(): + """Diff against empty DB should show all as new.""" + imported = [ + {"machine_id": "M001", "name": "Test Asset", "_customer_name": "Acme"}, + {"machine_id": "M002", "name": "Test 2", "_customer_name": "Beta"}, + ] + live = {"assets": [], "customers": [], "locations": []} + diff = compute_diff(imported, live) + assert diff["import_count"] == 2 + assert len(diff["new_assets"]) == 2 + assert len(diff["removed_assets"]) == 0 + assert len(diff["changed_assets"]) == 0 + assert "Acme" in diff["new_customers"] + assert "Beta" in diff["new_customers"] + + +def test_compute_diff_with_existing(): + """Diff should detect new, removed, and changed assets.""" + imported = [ + {"machine_id": "M001", "name": "Updated Name", "status": "active"}, + ] + live = { + "assets": [ + {"machine_id": "M001", "name": "Old Name", "status": "maintenance", + "serial_number": "", "description": "", "category": "", "make": "", + "model": "", "address": "", "building_name": "", "building_number": "", + "floor": "", "room": "", "trailer_number": "", "walking_directions": "", + "map_link": "", "parking_location": "", "photo_path": "", + "latitude": None, "longitude": None, "geofence_radius_meters": 50}, + {"machine_id": "M002", "name": "Removed Asset", "status": "active", + "serial_number": "", "description": "", "category": "", "make": "", + "model": "", "address": "", "building_name": "", "building_number": "", + "floor": "", "room": "", "trailer_number": "", "walking_directions": "", + "map_link": "", "parking_location": "", "photo_path": "", + "latitude": None, "longitude": None, "geofence_radius_meters": 50}, + ], + "customers": [], + "locations": [], + } + diff = compute_diff(imported, live) + assert len(diff["new_assets"]) == 0 + assert len(diff["removed_assets"]) == 1 + assert diff["removed_assets"][0]["machine_id"] == "M002" + assert len(diff["changed_assets"]) == 1 + assert diff["changed_assets"][0]["machine_id"] == "M001" + changes = {c["field"]: c for c in diff["changed_assets"][0]["changes"]} + assert changes["name"]["old_value"] == "Old Name" + assert changes["name"]["new_value"] == "Updated Name" + assert changes["status"]["old_value"] == "maintenance" + assert changes["status"]["new_value"] == "active" + + +def test_parse_excel(): + """Test parsing a simple Excel file.""" + import openpyxl + import tempfile + wb = openpyxl.Workbook() + ws = wb.active + ws.append(["Asset ID", "Name", "Status"]) + ws.append(["M001", "Test Machine", "active"]) + ws.append(["M002", "Fridge", "maintenance"]) + + with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as f: + wb.save(f.name) + tmp_path = f.name + + try: + headers, rows = parse_excel(tmp_path) + assert headers == ["Asset ID", "Name", "Status"] + assert len(rows) == 2 + assert rows[0]["Asset ID"] == "M001" + assert rows[1]["Name"] == "Fridge" + finally: + Path(tmp_path).unlink(missing_ok=True) + + +def test_empty_rows_skipped(): + """Empty rows in Excel should be skipped.""" + import openpyxl + import tempfile + wb = openpyxl.Workbook() + ws = wb.active + ws.append(["Asset ID", "Name"]) + ws.append(["M001", "Test"]) + ws.append([None, None]) # empty row + ws.append(["M002", "Test2"]) + + with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as f: + wb.save(f.name) + tmp_path = f.name + + try: + headers, rows = parse_excel(tmp_path) + assert len(rows) == 2 # empty row skipped + finally: + Path(tmp_path).unlink(missing_ok=True) + + +def test_first_match_wins(): + """When multiple heuristics could match, first wins due to ordering.""" + # Verify that heuristics are ordered: specific multi-word before generic + # The "location name" entry should come before "name" entry + location_idx = None + name_single_idx = None + for i, (keywords, canonical) in enumerate(COLUMN_HEURISTICS): + if canonical == "_location_name" and "location name" in keywords: + location_idx = i + if canonical == "name" and keywords == ["name"]: + name_single_idx = i + assert location_idx is not None, "location name heuristic not found" + assert name_single_idx is not None, "single-word name heuristic not found" + assert location_idx < name_single_idx, ( + f"location_name heuristic ({location_idx}) must come before " + f"single-word name heuristic ({name_single_idx})" + ) + + +if __name__ == "__main__": + import sys + tests = [ + test_normalize, test_map_columns_basic, test_map_columns_no_greedy_match, + test_map_columns_variants, test_apply_mapping, + test_compute_diff_no_live_data, test_compute_diff_with_existing, + test_parse_excel, test_empty_rows_skipped, test_first_match_wins, + ] + passed = 0 + failed = 0 + for t in tests: + try: + t() + print(f" ✓ {t.__name__}") + passed += 1 + except Exception as e: + print(f" ✗ {t.__name__}: {e}") + failed += 1 + print(f"\nUnit tests: {passed} passed, {failed} failed") + sys.exit(0 if failed == 0 else 1) diff --git a/test_sync_api.py b/test_sync_api.py new file mode 100644 index 0000000..6aa0f2e --- /dev/null +++ b/test_sync_api.py @@ -0,0 +1,212 @@ +""" +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/ returns 200", resp.status_code == 200, str(resp.status_code)) + detail = resp.json() + check("GET /batches/ has raw_data", "raw_data" in detail, str(detail.keys())) + check("GET /batches/ 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()