From e408867aac3ff143e57a6ee1a16413c5a07d628d Mon Sep 17 00:00:00 2001 From: Shawn Date: Thu, 21 May 2026 18:53:11 -0400 Subject: [PATCH] =?UTF-8?q?Add=20diff=20tracking=20=E2=80=94=20compare=20e?= =?UTF-8?q?xports=20between=20runs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - New cantaloupe/diff.py: diff engine comparing xlsx by Asset ID, manifest tracking at ~/.cantaloupe-exports/manifest.json - New subcommand: python -m cantaloupe diff [--from FILE] [--to FILE] - Export --diff flag: auto-compare with previous export after download - Every export auto-records to manifest for history tracking - 44 new tests, 130/130 pass total --- cantaloupe/cli.py | 168 ++++++++++ cantaloupe/diff.py | 238 ++++++++++++++ tests/test_diff.py | 772 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 1178 insertions(+) create mode 100644 cantaloupe/diff.py create mode 100644 tests/test_diff.py diff --git a/cantaloupe/cli.py b/cantaloupe/cli.py index 78e19c3..6424591 100644 --- a/cantaloupe/cli.py +++ b/cantaloupe/cli.py @@ -3,11 +3,15 @@ Click CLI for cantaloupe export tool. Commands: export Download machine list Excel from mycantaloupe.com + diff Compare two machine list exports and show what changed configure Save credentials to ~/.cantaloupe.env Usage: python -m cantaloupe export --output ./exports/ python -m cantaloupe export --scheduled --output /data/exports/ + python -m cantaloupe export --diff # auto-compare with previous export + python -m cantaloupe diff --from old.xlsx --to new.xlsx + python -m cantaloupe diff # compare two most recent exports python -m cantaloupe configure """ @@ -16,6 +20,7 @@ import os import sys import time from pathlib import Path +from typing import Optional import click @@ -83,6 +88,54 @@ def _resolve_output(output: str | None, scheduled: bool) -> Path: return out_path +def _print_diff_result(result: dict, from_label: str, to_label: str) -> None: + """Pretty-print a diff result to the console.""" + click.echo() + click.echo("═" * 60) + click.echo(" Diff Report") + click.echo("═" * 60) + click.echo(f" From : {from_label} ({result['from_count']} rows)") + click.echo(f" To : {to_label} ({result['to_count']} rows)") + click.echo(f" Key : {result['id_column']}") + click.echo() + + added = result["added_count"] + removed = result["removed_count"] + changed = result["changed_count"] + + if added == 0 and removed == 0 and changed == 0: + click.echo(" āœ… No changes — exports are identical.") + else: + if added: + click.echo(f" 🟢 {added} machine(s) added") + for row in result["added"][:10]: + rid = row.get(result["id_column"], "?") + click.echo(f" + {rid}") + if added > 10: + click.echo(f" ... and {added - 10} more") + + if removed: + click.echo(f" šŸ”“ {removed} machine(s) removed") + for row in result["removed"][:10]: + rid = row.get(result["id_column"], "?") + click.echo(f" - {rid}") + if removed > 10: + click.echo(f" ... and {removed - 10} more") + + if changed: + click.echo(f" 🟔 {changed} machine(s) changed") + for entry in result["changed"][:5]: + rid = entry["id"] + field_names = list(entry["changes"].keys()) + click.echo(f" ~ {rid} ({', '.join(field_names[:5])})") + if len(field_names) > 5: + click.echo(f" +{len(field_names) - 5} more fields") + if changed > 5: + click.echo(f" ... and {changed - 5} more") + + click.echo("═" * 60) + + # ── CLI group ─────────────────────────────────────────────────────────────── @@ -137,6 +190,12 @@ def cli(ctx: click.Context, verbose: bool) -> None: is_flag=True, help="Non-interactive mode for cron usage — skips credential prompts, uses defaults.", ) +@click.option( + "--diff/--no-diff", + "diff_flag", + default=False, + help="Compare with the previous export and show changes after download.", +) @click.option( "--email", default=None, @@ -155,6 +214,7 @@ def export( output: str | None, headless: bool, scheduled: bool, + diff_flag: bool, email: str | None, password: str | None, ) -> None: @@ -163,6 +223,8 @@ def export( Authenticates with your Cantaloupe account and downloads the VueMachineList ExcelExport file, reporting file size, row count, and elapsed time on completion. + + Use --diff to auto-compare with the previous export after download. """ # Resolve credentials (CLI flags > env vars > config file) resolved_email = email or ctx.obj.get("email", "") @@ -238,6 +300,32 @@ def export( click.echo(f" Rows : {row_count:,}") click.echo(f" Time : {elapsed:.1f}s") + # ── record to manifest ───────────────────────────────────────────── + from cantaloupe.diff import record_export + + # Snapshot the previous export *before* recording this one, so + # we diff against the run-before-last, not this run. + previous_path: Path | None = None + if diff_flag: + from cantaloupe.diff import get_previous_export + previous_path = get_previous_export() + + record_export(out_path, row_count, file_size) + + # ── diff against previous export ──────────────────────────────────── + if diff_flag: + if previous_path is None: + click.echo("\n⚠ No previous export found — skipping diff.") + elif previous_path.resolve() == out_path.resolve(): + click.echo("\n⚠ Previous export is the same file — skipping diff.") + else: + try: + from cantaloupe.diff import diff_excels + result = diff_excels(previous_path, out_path) + _print_diff_result(result, previous_path.name, out_path.name) + except Exception as e: + click.echo(f"\n⚠ Diff failed: {e}", err=True) + # ── configure command ─────────────────────────────────────────────────────── @@ -268,6 +356,86 @@ def configure(email: str, password: str) -> None: click.echo(f" You can now run: python -m cantaloupe export") +# ── diff command ───────────────────────────────────────────────────────────── + + +@cli.command() +@click.option( + "--from", "from_path", + default=None, + help="Path to the older xlsx file (default: second-most-recent export).", +) +@click.option( + "--to", "to_path", + default=None, + help="Path to the newer xlsx file (default: most recent export).", +) +@click.option( + "--id-column", + default=None, + help="Column name to use as unique machine identifier (auto-detected if omitted).", +) +@click.option( + "--json", "json_output", + is_flag=True, + help="Output the diff result as JSON instead of pretty-printed text.", +) +def diff( + from_path: str | None, + to_path: str | None, + id_column: str | None, + json_output: bool, +) -> None: + """Compare two machine list Excel exports and show what changed. + + If --from and --to are omitted, the two most recent exports are used + (tracked via ~/.cantaloupe-exports/manifest.json). + + Reports machines added, removed, and changed — keyed on the Asset ID + column (auto-detected, or specified with --id-column). + """ + from cantaloupe.diff import diff_excels, get_latest_exports + + # Resolve file paths + if from_path and to_path: + f_path = Path(os.path.expanduser(from_path)).resolve() + t_path = Path(os.path.expanduser(to_path)).resolve() + elif from_path or to_path: + raise click.UsageError( + "Specify both --from and --to, or neither (to use latest exports)." + ) + else: + latest = get_latest_exports(2) + if len(latest) < 2: + click.echo( + "ERROR: Need at least two exports to diff. " + "Run an export first, or specify --from and --to explicitly.", + err=True, + ) + raise SystemExit(1) + f_path = latest[1] # older + t_path = latest[0] # newer + + if not f_path.exists(): + click.echo(f"ERROR: File not found: {f_path}", err=True) + raise SystemExit(1) + if not t_path.exists(): + click.echo(f"ERROR: File not found: {t_path}", err=True) + raise SystemExit(1) + + try: + result = diff_excels(f_path, t_path, id_column=id_column) + except (ValueError, FileNotFoundError) as e: + click.echo(f"ERROR: {e}", err=True) + raise SystemExit(1) + + if json_output: + import json + click.echo(json.dumps(result, indent=2, default=str)) + else: + _print_diff_result(result, f_path.name, t_path.name) + + # ── direct invocation ─────────────────────────────────────────────────────── if __name__ == "__main__": diff --git a/cantaloupe/diff.py b/cantaloupe/diff.py new file mode 100644 index 0000000..b4499c8 --- /dev/null +++ b/cantaloupe/diff.py @@ -0,0 +1,238 @@ +""" +Diff engine for comparing Cantaloupe machine list Excel exports. + +Reports: + - 🟢 New machines added (present in new, not in old) + - šŸ”“ Machines removed (present in old, not in new) + - 🟔 Machines with changed fields (same ID, different values) + +Also manages a JSON manifest at ~/.cantaloupe-exports/manifest.json +that tracks export history for auto-diff functionality. +""" + +import json +import os +from datetime import datetime +from pathlib import Path +from typing import Optional + +import openpyxl + +MANIFEST_DIR = Path.home() / ".cantaloupe-exports" +MANIFEST_FILE = MANIFEST_DIR / "manifest.json" + + +# ── xlsx reading ──────────────────────────────────────────────────────────── + + +def _read_xlsx(path: Path) -> tuple[list[str], list[dict]]: + """Read an xlsx file and return (headers, list of row dicts). + + Each row dict maps header name → cell value. + Returns empty lists if the file is empty or unreadable. + """ + wb = openpyxl.load_workbook(path, read_only=True) + try: + ws = wb.active + if ws is None: + return [], [] + rows = list(ws.iter_rows(values_only=True)) + finally: + wb.close() + + if not rows: + return [], [] + + headers = [str(h) if h is not None else "" for h in rows[0]] + data = [] + for row in rows[1:]: + row_dict = {} + for i, cell in enumerate(row): + if i < len(headers): + row_dict[headers[i]] = cell + data.append(row_dict) + + return headers, data + + +def _find_id_column(headers: list[str]) -> str: + """Find the best unique-identifier column from headers. + + Priority: + 1. Column whose name contains 'asset' AND 'id' (case-insensitive) + 2. Column whose name equals 'id' (case-insensitive) + 3. First column (fallback) + """ + for h in headers: + hl = h.lower() + if "asset" in hl and "id" in hl: + return h + for h in headers: + if h.lower() == "id": + return h + return headers[0] if headers else "" + + +# ── diff engine ───────────────────────────────────────────────────────────── + + +def _normalize(val) -> str: + """Convert a cell value to a string for comparison and keying.""" + if val is None: + return "" + return str(val).strip() + + +def diff_excels( + from_path: Path, + to_path: Path, + id_column: Optional[str] = None, +) -> dict: + """Compare two xlsx files and return a structured diff result. + + Args: + from_path: Path to the older ("from") xlsx file. + to_path: Path to the newer ("to") xlsx file. + id_column: Column name to use as the unique machine identifier. + Auto-detected if not provided (looks for 'Asset ID'). + + Returns: + dict with keys: + - id_column: str — column used as the key + - from_path, to_path: str — file paths + - from_count, to_count: int — data row counts + - added: list[dict] — rows present in 'to' but not 'from' + - removed: list[dict] — rows present in 'from' but not 'to' + - changed: list[dict] — {id, changes: {col: {old, new}}} + - added_count, removed_count, changed_count: int + + Raises: + ValueError: if either file has no data / no rows. + FileNotFoundError: if either file doesn't exist. + """ + from_headers, from_data = _read_xlsx(from_path) + to_headers, to_data = _read_xlsx(to_path) + + if not to_headers: + raise ValueError(f"No data found in {to_path}") + + # Determine id column (prefer the 'to' file's headers) + id_col = id_column or _find_id_column(to_headers) + + # Build lookup dicts keyed on the id column + from_by_id: dict[str, dict] = {} + for row in from_data: + key = _normalize(row.get(id_col, "")) + from_by_id[key] = row + + to_by_id: dict[str, dict] = {} + for row in to_data: + key = _normalize(row.get(id_col, "")) + to_by_id[key] = row + + from_ids = set(from_by_id.keys()) + to_ids = set(to_by_id.keys()) + + added_ids = to_ids - from_ids + removed_ids = from_ids - to_ids + common_ids = from_ids & to_ids + + # Collect changed rows + changed: list[dict] = [] + # Use headers from 'to' file as canonical + for cid in sorted(common_ids): + old_row = from_by_id[cid] + new_row = to_by_id[cid] + field_changes: dict[str, dict[str, object]] = {} + for col in to_headers: + old_val = _normalize(old_row.get(col)) + new_val = _normalize(new_row.get(col)) + if old_val != new_val: + field_changes[col] = {"old": old_row.get(col), "new": new_row.get(col)} + if field_changes: + changed.append({"id": cid, "changes": field_changes}) + + return { + "id_column": id_col, + "from_path": str(from_path), + "to_path": str(to_path), + "from_count": len(from_data), + "to_count": len(to_data), + "added": [to_by_id[aid] for aid in sorted(added_ids)], + "removed": [from_by_id[rid] for rid in sorted(removed_ids)], + "changed": changed, + "added_count": len(added_ids), + "removed_count": len(removed_ids), + "changed_count": len(changed), + } + + +# ── manifest management ───────────────────────────────────────────────────── + + +def load_manifest() -> dict: + """Load the export history manifest from ~/.cantaloupe-exports/manifest.json. + + Returns a dict with an 'exports' list. Creates an empty manifest + if the file doesn't exist. + """ + MANIFEST_DIR.mkdir(parents=True, exist_ok=True) + if MANIFEST_FILE.exists(): + try: + return json.loads(MANIFEST_FILE.read_text()) + except (json.JSONDecodeError, OSError): + return {"exports": []} + return {"exports": []} + + +def save_manifest(manifest: dict) -> None: + """Save the manifest to disk.""" + MANIFEST_DIR.mkdir(parents=True, exist_ok=True) + MANIFEST_FILE.write_text(json.dumps(manifest, indent=2, default=str) + "\n") + + +def record_export( + filepath: Path, + row_count: int, + file_size: int, +) -> None: + """Record a completed export in the manifest.""" + manifest = load_manifest() + manifest.setdefault("exports", []).append( + { + "path": str(filepath), + "filename": filepath.name, + "timestamp": datetime.now().isoformat(), + "row_count": row_count, + "file_size": file_size, + } + ) + save_manifest(manifest) + + +def get_previous_export() -> Optional[Path]: + """Return the path of the most recent export, or None if no history.""" + manifest = load_manifest() + exports = manifest.get("exports", []) + if not exports: + return None + previous = exports[-1]["path"] + p = Path(previous) + if p.exists(): + return p + # File was moved/deleted — return None so we don't diff against a ghost + return None + + +def get_latest_exports(n: int = 2) -> list[Path]: + """Return paths of the N most recent existing exports, newest first.""" + manifest = load_manifest() + exports = manifest.get("exports", []) + result: list[Path] = [] + for entry in reversed(exports): + p = Path(entry["path"]) + if p.exists(): + result.append(p) + if len(result) >= n: + break + return result diff --git a/tests/test_diff.py b/tests/test_diff.py new file mode 100644 index 0000000..388cfaa --- /dev/null +++ b/tests/test_diff.py @@ -0,0 +1,772 @@ +""" +Tests for cantaloupe diff engine and manifest tracking. + +Covers: + - diff.py: _read_xlsx, _find_id_column, diff_excels + - diff.py: load_manifest, save_manifest, record_export, + get_previous_export, get_latest_exports + - cli.py: diff command (--from/--to, --id-column, --json) + - cli.py: export --diff flag (auto-compare after download) +""" + +import json +import os +import tempfile +from pathlib import Path +from unittest.mock import MagicMock, patch + +import openpyxl +import pytest +from click.testing import CliRunner + +from cantaloupe.cli import cli, _print_diff_result +from cantaloupe.diff import ( + MANIFEST_DIR, + MANIFEST_FILE, + _find_id_column, + _read_xlsx, + diff_excels, + get_latest_exports, + get_previous_export, + load_manifest, + record_export, + save_manifest, +) + + +# ── helpers for building test xlsx files ────────────────────────────────── + + +def _make_xlsx(path: Path, headers: list[str], rows: list[list]) -> Path: + """Write a minimal xlsx file and return the path.""" + wb = openpyxl.Workbook() + ws = wb.active + ws.append(headers) + for row in rows: + ws.append(row) + wb.save(path) + wb.close() + return path + + +# ───────────────────────────────────────────────────────────────── +# _read_xlsx +# ───────────────────────────────────────────────────────────────── + + +def test_read_xlsx_basic(tmp_path): + """Read a simple xlsx file with headers and data rows.""" + path = _make_xlsx( + tmp_path / "test.xlsx", + headers=["Asset ID", "Name", "Location"], + rows=[ + ["A001", "Machine 1", "Warehouse A"], + ["A002", "Machine 2", "Warehouse B"], + ], + ) + headers, data = _read_xlsx(path) + assert headers == ["Asset ID", "Name", "Location"] + assert len(data) == 2 + assert data[0] == {"Asset ID": "A001", "Name": "Machine 1", "Location": "Warehouse A"} + assert data[1] == {"Asset ID": "A002", "Name": "Machine 2", "Location": "Warehouse B"} + + +def test_read_xlsx_empty(tmp_path): + """Empty xlsx (no rows) returns empty lists.""" + path = _make_xlsx( + tmp_path / "empty.xlsx", + headers=["Asset ID", "Name"], + rows=[], + ) + headers, data = _read_xlsx(path) + assert headers == ["Asset ID", "Name"] + assert data == [] + + +def test_read_xlsx_header_only(tmp_path): + """File with only a header row returns empty data list.""" + wb = openpyxl.Workbook() + ws = wb.active + ws.append(["Asset ID", "Name"]) + wb.save(tmp_path / "header_only.xlsx") + wb.close() + + headers, data = _read_xlsx(tmp_path / "header_only.xlsx") + assert headers == ["Asset ID", "Name"] + assert data == [] + + +def test_read_xlsx_none_cells(tmp_path): + """None / blank cells are handled gracefully.""" + path = _make_xlsx( + tmp_path / "blanks.xlsx", + headers=["Asset ID", "Name", "Notes"], + rows=[ + ["A001", None, "some notes"], + [None, "No-ID Machine", ""], + ], + ) + headers, data = _read_xlsx(path) + assert data[0]["Name"] is None + assert data[1]["Asset ID"] is None + + +def test_read_xlsx_file_not_found(tmp_path): + """Missing file raises FileNotFoundError.""" + with pytest.raises(FileNotFoundError): + _read_xlsx(tmp_path / "nonexistent.xlsx") + + +# ───────────────────────────────────────────────────────────────── +# _find_id_column +# ───────────────────────────────────────────────────────────────── + + +def test_find_id_column_exact_asset_id(): + """Finds column with 'asset' and 'id' in name.""" + headers = ["Name", "Location", "Asset ID", "Status"] + assert _find_id_column(headers) == "Asset ID" + + +def test_find_id_column_case_insensitive(): + """Case-insensitive match: 'asset id' matches 'Asset ID'.""" + headers = ["Name", "asset id", "Location"] + assert _find_id_column(headers) == "asset id" + + +def test_find_id_column_assetid_no_space(): + """'AssetID' (no space) still matches.""" + headers = ["Name", "AssetID", "Location"] + assert _find_id_column(headers) == "AssetID" + + +def test_find_id_column_just_id(): + """Falls back to column named 'id' (case-insensitive).""" + headers = ["Name", "Location", "id"] + assert _find_id_column(headers) == "id" + + +def test_find_id_column_fallback_first(): + """When no good match, returns the first column.""" + headers = ["Name", "Location", "Status"] + assert _find_id_column(headers) == "Name" + + +def test_find_id_column_empty_headers(): + """Empty headers returns empty string.""" + assert _find_id_column([]) == "" + + +# ───────────────────────────────────────────────────────────────── +# diff_excels +# ───────────────────────────────────────────────────────────────── + + +def test_diff_identical_files(tmp_path): + """Two identical files produce zero changes.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name", "Location"], + rows=[["A001", "Machine 1", "WH-A"], ["A002", "Machine 2", "WH-B"]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name", "Location"], + rows=[["A001", "Machine 1", "WH-A"], ["A002", "Machine 2", "WH-B"]], + ) + + result = diff_excels(old, new) + assert result["added_count"] == 0 + assert result["removed_count"] == 0 + assert result["changed_count"] == 0 + assert result["from_count"] == 2 + assert result["to_count"] == 2 + + +def test_diff_added_machines(tmp_path): + """New machines in 'to' file are reported as added.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"], ["A002", "Machine 2"], ["A003", "Machine 3"]], + ) + + result = diff_excels(old, new) + assert result["added_count"] == 2 + assert result["removed_count"] == 0 + assert result["changed_count"] == 0 + assert len(result["added"]) == 2 + assert result["added"][0]["Asset ID"] == "A002" + assert result["added"][1]["Asset ID"] == "A003" + + +def test_diff_removed_machines(tmp_path): + """Machines in 'from' but not 'to' are reported as removed.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"], ["A002", "Machine 2"]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"]], + ) + + result = diff_excels(old, new) + assert result["added_count"] == 0 + assert result["removed_count"] == 1 + assert result["changed_count"] == 0 + assert result["removed"][0]["Asset ID"] == "A002" + + +def test_diff_changed_fields(tmp_path): + """Same Asset ID with different field values is reported as changed.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name", "Location"], + rows=[["A001", "Machine 1", "WH-A"], ["A002", "Machine 2", "WH-B"]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name", "Location"], + rows=[["A001", "Machine 1", "WH-C"], ["A002", "Machine 2", "WH-B"]], + ) + + result = diff_excels(old, new) + assert result["added_count"] == 0 + assert result["removed_count"] == 0 + assert result["changed_count"] == 1 + assert result["changed"][0]["id"] == "A001" + assert "Location" in result["changed"][0]["changes"] + assert result["changed"][0]["changes"]["Location"]["old"] == "WH-A" + assert result["changed"][0]["changes"]["Location"]["new"] == "WH-C" + + +def test_diff_mixed_changes(tmp_path): + """Combination of added, removed, and changed machines.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name", "Location"], + rows=[ + ["A001", "Machine 1", "WH-A"], + ["A002", "Machine 2", "WH-B"], # will be removed + ["A003", "Machine 3", "WH-C"], # will be changed + ], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name", "Location"], + rows=[ + ["A001", "Machine 1", "WH-A"], # unchanged + ["A003", "Machine 3", "WH-D"], # changed Location + ["A004", "Machine 4", "WH-E"], # added + ], + ) + + result = diff_excels(old, new) + assert result["added_count"] == 1 # A004 + assert result["removed_count"] == 1 # A002 + assert result["changed_count"] == 1 # A003 + + assert result["added"][0]["Asset ID"] == "A004" + assert result["removed"][0]["Asset ID"] == "A002" + assert result["changed"][0]["id"] == "A003" + + +def test_diff_explicit_id_column(tmp_path): + """Specify a custom id column.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Machine Number", "Name"], + rows=[["M001", "Machine 1"]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Machine Number", "Name"], + rows=[["M001", "Machine 1"], ["M002", "Machine 2"]], + ) + + result = diff_excels(old, new, id_column="Machine Number") + assert result["id_column"] == "Machine Number" + assert result["added_count"] == 1 + + +def test_diff_empty_to_file(tmp_path): + """Empty 'to' file raises ValueError.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"]], + ) + # Empty new file — no rows at all + wb = openpyxl.Workbook() + wb.save(tmp_path / "new.xlsx") + wb.close() + + with pytest.raises(ValueError, match="No data found"): + diff_excels(old, tmp_path / "new.xlsx") + + +def test_diff_empty_from_file_allows_all_added(tmp_path): + """Empty 'from' file means all 'to' rows are reported as added.""" + wb = openpyxl.Workbook() + wb.save(tmp_path / "old.xlsx") + wb.close() + + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"], ["A002", "Machine 2"]], + ) + + result = diff_excels(tmp_path / "old.xlsx", new) + assert result["added_count"] == 2 + assert result["removed_count"] == 0 + assert result["changed_count"] == 0 + + +def test_diff_none_values_compared(tmp_path): + """None values are normalized to empty string for comparison.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Notes"], + rows=[["A001", None]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Notes"], + rows=[["A001", ""]], + ) + + result = diff_excels(old, new) + # None and "" are normalized to the same string → no change + assert result["changed_count"] == 0 + + +def test_diff_result_keys(tmp_path): + """Verify all expected keys are present in the result dict.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "M1"]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "M1"]], + ) + + result = diff_excels(old, new) + expected_keys = { + "id_column", "from_path", "to_path", + "from_count", "to_count", + "added", "removed", "changed", + "added_count", "removed_count", "changed_count", + } + assert expected_keys <= set(result.keys()) + assert result["id_column"] == "Asset ID" + + +# ───────────────────────────────────────────────────────────────── +# manifest management +# ───────────────────────────────────────────────────────────────── + + +def test_load_manifest_empty_creates_default(tmp_path): + """load_manifest returns default structure when no file exists.""" + with patch("cantaloupe.diff.MANIFEST_FILE", tmp_path / "manifest.json"): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + manifest = load_manifest() + assert manifest == {"exports": []} + + +def test_save_and_load_manifest(tmp_path): + """Round-trip: save a manifest then load it back.""" + manifest_file = tmp_path / "manifest.json" + manifest_dir = tmp_path + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", manifest_dir): + manifest = {"exports": [{"path": "/tmp/test.xlsx"}]} + save_manifest(manifest) + + loaded = load_manifest() + assert loaded == manifest + + +def test_record_export_appends(tmp_path): + """record_export appends to the exports list.""" + manifest_file = tmp_path / "manifest.json" + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + record_export(Path("/tmp/a.xlsx"), 10, 1024) + record_export(Path("/tmp/b.xlsx"), 20, 2048) + + manifest = load_manifest() + assert len(manifest["exports"]) == 2 + assert manifest["exports"][0]["filename"] == "a.xlsx" + assert manifest["exports"][0]["row_count"] == 10 + assert manifest["exports"][0]["file_size"] == 1024 + assert manifest["exports"][1]["filename"] == "b.xlsx" + assert manifest["exports"][1]["row_count"] == 20 + + +def test_get_previous_export_returns_last(tmp_path): + """get_previous_export returns the path of the most recent export.""" + manifest_file = tmp_path / "manifest.json" + a_path = tmp_path / "a.xlsx" + b_path = tmp_path / "b.xlsx" + a_path.write_text("dummy") + b_path.write_text("dummy") + + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + record_export(a_path, 10, 100) + record_export(b_path, 20, 200) + + prev = get_previous_export() + assert prev == b_path + + +def test_get_previous_export_empty_manifest(tmp_path): + """Returns None when manifest has no exports.""" + with patch("cantaloupe.diff.MANIFEST_FILE", tmp_path / "manifest.json"): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + assert get_previous_export() is None + + +def test_get_previous_export_missing_file(tmp_path): + """Returns None when the last export file no longer exists.""" + manifest_file = tmp_path / "manifest.json" + ghost = tmp_path / "ghost.xlsx" + + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + # Write manifest directly with a path that doesn't exist + save_manifest({"exports": [{"path": str(ghost), "timestamp": "2026-01-01"}]}) + assert get_previous_export() is None + + +def test_get_latest_exports_two(tmp_path): + """get_latest_exports returns two most recent existing files.""" + manifest_file = tmp_path / "manifest.json" + a_path = tmp_path / "a.xlsx" + b_path = tmp_path / "b.xlsx" + c_path = tmp_path / "c.xlsx" + a_path.write_text("dummy") + b_path.write_text("dummy") + c_path.write_text("dummy") + + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + record_export(a_path, 10, 100) + record_export(b_path, 20, 200) + record_export(c_path, 30, 300) + + latest = get_latest_exports(2) + assert len(latest) == 2 + assert latest[0] == c_path # newest + assert latest[1] == b_path # second newest + + +def test_get_latest_exports_skips_missing(tmp_path): + """get_latest_exports skips entries whose file doesn't exist.""" + manifest_file = tmp_path / "manifest.json" + a_path = tmp_path / "a.xlsx" + a_path.write_text("dummy") + ghost = tmp_path / "ghost.xlsx" + + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + record_export(ghost, 10, 100) # doesn't exist on disk + record_export(a_path, 20, 200) + + latest = get_latest_exports(3) + assert len(latest) == 1 + assert latest[0] == a_path + + +def test_load_manifest_corrupt_json(tmp_path): + """Corrupt manifest JSON returns default empty structure.""" + manifest_file = tmp_path / "manifest.json" + manifest_file.write_text("not valid json{{{") + + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + manifest = load_manifest() + assert manifest == {"exports": []} + + +# ───────────────────────────────────────────────────────────────── +# CLI: diff command +# ───────────────────────────────────────────────────────────────── + + +@pytest.fixture +def runner(): + return CliRunner() + + +def test_diff_command_basic(runner, tmp_path): + """diff command with explicit --from and --to files.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"], ["A002", "Machine 2"]], + ) + + result = runner.invoke(cli, [ + "diff", + "--from", str(old), + "--to", str(new), + ]) + assert result.exit_code == 0 + assert "Diff Report" in result.output + assert "1 machine(s) added" in result.output + assert "A002" in result.output + + +def test_diff_command_no_changes(runner, tmp_path): + """diff command reports no changes for identical files.""" + xlsx = _make_xlsx(tmp_path / "data.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"]], + ) + + result = runner.invoke(cli, [ + "diff", + "--from", str(xlsx), + "--to", str(xlsx), + ]) + assert result.exit_code == 0 + assert "No changes" in result.output + + +def test_diff_command_json_output(runner, tmp_path): + """--json flag outputs machine-readable JSON.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"], ["A002", "Machine 2"]], + ) + + result = runner.invoke(cli, [ + "diff", + "--from", str(old), + "--to", str(new), + "--json", + ]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data["added_count"] == 1 + assert data["id_column"] == "Asset ID" + + +def test_diff_command_missing_file(runner, tmp_path): + """Exits with error when --from file doesn't exist.""" + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "M1"]], + ) + + result = runner.invoke(cli, [ + "diff", + "--from", str(tmp_path / "nonexistent.xlsx"), + "--to", str(new), + ]) + assert result.exit_code == 1 + assert "File not found" in result.output + + +def test_diff_command_only_one_file_specified(runner, tmp_path): + """Specifying only --from (without --to) raises UsageError.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "M1"]], + ) + + result = runner.invoke(cli, [ + "diff", + "--from", str(old), + ]) + assert result.exit_code == 2 # Click exits with 2 for UsageError + assert "Specify both" in result.output + + +def test_diff_command_uses_manifest(runner, tmp_path): + """Without --from/--to, uses the two most recent exports.""" + manifest_file = tmp_path / "manifest.json" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"], ["A002", "Machine 2"]], + ) + + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + record_export(old, 1, 100) + record_export(new, 2, 200) + + result = runner.invoke(cli, ["diff"]) + assert result.exit_code == 0 + assert "Diff Report" in result.output + assert "1 machine(s) added" in result.output + + +def test_diff_command_manifest_not_enough(runner, tmp_path): + """Exits with error when manifest has fewer than 2 exports.""" + manifest_file = tmp_path / "manifest.json" + + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + result = runner.invoke(cli, ["diff"]) + assert result.exit_code == 1 + assert "Need at least two exports" in result.output + + +def test_diff_command_with_id_column(runner, tmp_path): + """--id-column specifies a custom key column.""" + old = _make_xlsx(tmp_path / "old.xlsx", + headers=["Machine Number", "Name"], + rows=[["M001", "Machine 1"]], + ) + new = _make_xlsx(tmp_path / "new.xlsx", + headers=["Machine Number", "Name"], + rows=[["M001", "Machine 1"], ["M002", "Machine 2"]], + ) + + result = runner.invoke(cli, [ + "diff", + "--from", str(old), + "--to", str(new), + "--id-column", "Machine Number", + ]) + assert result.exit_code == 0 + assert "Machine Number" in result.output + + +# ───────────────────────────────────────────────────────────────── +# CLI: export --diff flag +# ───────────────────────────────────────────────────────────────── + + +def test_export_diff_flag_no_previous(runner, tmp_path): + """--diff flag with no previous export shows a warning.""" + manifest_file = tmp_path / "manifest.json" + + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + with patch.dict( + os.environ, + {"CANTALOUPE_EMAIL": "u@t.com", "CANTALOUPE_PASSWORD": "p"}, + ): + with patch("src.download.download") as mock_download: + mock_path = tmp_path / "export.xlsx" + _make_xlsx(mock_path, + headers=["Asset ID", "Name"], + rows=[["A001", "M1"]], + ) + mock_download.return_value = mock_path + + result = runner.invoke(cli, [ + "export", "--diff", "-o", str(tmp_path), + ]) + assert result.exit_code == 0 + assert "Export complete" in result.output + assert "No previous export" in result.output + + +def test_export_diff_flag_with_previous(runner, tmp_path): + """--diff flag with a previous export shows the diff report.""" + manifest_file = tmp_path / "manifest.json" + prev_path = tmp_path / "previous.xlsx" + new_path = tmp_path / "new_export.xlsx" + + _make_xlsx(prev_path, + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"]], + ) + + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + # Record a previous export + record_export(prev_path, 1, 100) + + # Now the actual export would use a different timestamped name + # which we mock — but get_previous_export returns prev_path + _make_xlsx(new_path, + headers=["Asset ID", "Name"], + rows=[["A001", "Machine 1"], ["A002", "Machine 2"]], + ) + + with patch.dict( + os.environ, + {"CANTALOUPE_EMAIL": "u@t.com", "CANTALOUPE_PASSWORD": "p"}, + ): + with patch("src.download.download") as mock_download: + mock_download.return_value = new_path + + result = runner.invoke(cli, [ + "export", "--diff", "-o", str(tmp_path), + ]) + assert result.exit_code == 0 + assert "Export complete" in result.output + assert "Diff Report" in result.output + assert "1 machine(s) added" in result.output + + +def test_export_records_manifest(runner, tmp_path): + """Every export records to the manifest even without --diff.""" + manifest_file = tmp_path / "manifest.json" + + with patch("cantaloupe.diff.MANIFEST_FILE", manifest_file): + with patch("cantaloupe.diff.MANIFEST_DIR", tmp_path): + mock_path = tmp_path / "export.xlsx" + _make_xlsx(mock_path, + headers=["Asset ID", "Name"], + rows=[["A001", "M1"]], + ) + + with patch.dict( + os.environ, + {"CANTALOUPE_EMAIL": "u@t.com", "CANTALOUPE_PASSWORD": "p"}, + ): + with patch("src.download.download") as mock_download: + mock_download.return_value = mock_path + + result = runner.invoke(cli, [ + "export", "-o", str(tmp_path), + ]) + assert result.exit_code == 0 + + # Verify manifest was updated + manifest = load_manifest() + assert len(manifest["exports"]) == 1 + assert manifest["exports"][0]["filename"] == "export.xlsx" + + +# ───────────────────────────────────────────────────────────────── +# CLI: --help shows diff command +# ───────────────────────────────────────────────────────────────── + + +def test_cli_help_includes_diff(runner): + """--help now lists the diff command.""" + result = runner.invoke(cli, ["--help"]) + assert result.exit_code == 0 + assert "diff" in result.output + # The help line shows the full docstring first line + assert "Compare two machine list" in result.output + + +def test_export_help_includes_diff_flag(runner): + """export --help shows the --diff option.""" + result = runner.invoke(cli, ["export", "--help"]) + assert result.exit_code == 0 + assert "--diff" in result.output + + +def test_diff_help_shows_options(runner): + """diff --help shows all options.""" + result = runner.invoke(cli, ["diff", "--help"]) + assert result.exit_code == 0 + assert "--from" in result.output + assert "--to" in result.output + assert "--id-column" in result.output + assert "--json" in result.output