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