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
This commit is contained in:
Leo
@@ -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)
|
||||
Reference in New Issue
Block a user