Add diff tracking — compare exports between runs

- 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
This commit is contained in:
2026-05-21 18:53:11 -04:00
parent 9c3d3ef7ed
commit e408867aac
3 changed files with 1178 additions and 0 deletions
+168
View File
@@ -3,11 +3,15 @@ Click CLI for cantaloupe export tool.
Commands: Commands:
export Download machine list Excel from mycantaloupe.com export Download machine list Excel from mycantaloupe.com
diff Compare two machine list exports and show what changed
configure Save credentials to ~/.cantaloupe.env configure Save credentials to ~/.cantaloupe.env
Usage: Usage:
python -m cantaloupe export --output ./exports/ python -m cantaloupe export --output ./exports/
python -m cantaloupe export --scheduled --output /data/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 python -m cantaloupe configure
""" """
@@ -16,6 +20,7 @@ import os
import sys import sys
import time import time
from pathlib import Path from pathlib import Path
from typing import Optional
import click import click
@@ -83,6 +88,54 @@ def _resolve_output(output: str | None, scheduled: bool) -> Path:
return out_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 ─────────────────────────────────────────────────────────────── # ── CLI group ───────────────────────────────────────────────────────────────
@@ -137,6 +190,12 @@ def cli(ctx: click.Context, verbose: bool) -> None:
is_flag=True, is_flag=True,
help="Non-interactive mode for cron usage — skips credential prompts, uses defaults.", 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( @click.option(
"--email", "--email",
default=None, default=None,
@@ -155,6 +214,7 @@ def export(
output: str | None, output: str | None,
headless: bool, headless: bool,
scheduled: bool, scheduled: bool,
diff_flag: bool,
email: str | None, email: str | None,
password: str | None, password: str | None,
) -> None: ) -> None:
@@ -163,6 +223,8 @@ def export(
Authenticates with your Cantaloupe account and downloads the Authenticates with your Cantaloupe account and downloads the
VueMachineList ExcelExport file, reporting file size, row count, VueMachineList ExcelExport file, reporting file size, row count,
and elapsed time on completion. and elapsed time on completion.
Use --diff to auto-compare with the previous export after download.
""" """
# Resolve credentials (CLI flags > env vars > config file) # Resolve credentials (CLI flags > env vars > config file)
resolved_email = email or ctx.obj.get("email", "") resolved_email = email or ctx.obj.get("email", "")
@@ -238,6 +300,32 @@ def export(
click.echo(f" Rows : {row_count:,}") click.echo(f" Rows : {row_count:,}")
click.echo(f" Time : {elapsed:.1f}s") 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 ─────────────────────────────────────────────────────── # ── configure command ───────────────────────────────────────────────────────
@@ -268,6 +356,86 @@ def configure(email: str, password: str) -> None:
click.echo(f" You can now run: python -m cantaloupe export") 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 ─────────────────────────────────────────────────────── # ── direct invocation ───────────────────────────────────────────────────────
if __name__ == "__main__": if __name__ == "__main__":
+238
View File
@@ -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
+772
View File
@@ -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