Leo
f7ffa0e6cc
- 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
222 lines
8.4 KiB
Python
222 lines
8.4 KiB
Python
"""
|
|
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)
|