e408867aac
- 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
773 lines
27 KiB
Python
773 lines
27 KiB
Python
"""
|
|
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
|