From f44349493e7d9c95434c1566b8b6e00db1f0b353 Mon Sep 17 00:00:00 2001 From: Dylan Gormley Date: Tue, 25 Aug 2026 12:17:24 -0500 Subject: [PATCH] feat: verify registered dataset files --- dtcli/cli.py | 3 +- dtcli/verify.py | 303 +++++++++++++++++++++++++++++++++++++++++++ tests/test_verify.py | 120 +++++++++++++++++ 3 files changed, 425 insertions(+), 1 deletion(-) create mode 100644 dtcli/verify.py create mode 100644 tests/test_verify.py diff --git a/dtcli/cli.py b/dtcli/cli.py index 9aaea1c..c5fea72 100644 --- a/dtcli/cli.py +++ b/dtcli/cli.py @@ -6,7 +6,7 @@ from click_aliasing import ClickAliasedGroup from rich import console, pretty -from dtcli import clear, config, ls, ps, pull, scout, unregistered +from dtcli import clear, config, ls, ps, pull, scout, unregistered, verify from dtcli.utilities import utilities pretty.install() @@ -47,6 +47,7 @@ def version(): cli.add_command(pull.pull) cli.add_command(scout.scout) cli.add_command(unregistered.unregistered) +cli.add_command(verify.verify) def check_version() -> None: diff --git a/dtcli/verify.py b/dtcli/verify.py new file mode 100644 index 0000000..e98d709 --- /dev/null +++ b/dtcli/verify.py @@ -0,0 +1,303 @@ +"""Datatrail dataset verification command.""" + +import json +from typing import Any, Dict, List, Optional, Set, Tuple + +import click + +from dtcli.src import functions +from dtcli.utilities import cadcclient + +NAMESPACE = "cadc:CHIMEFRB" +QUERY_BATCH_SIZE = 100 +RESULT_CATEGORIES = ( + "present", + "missing", + "size_mismatch", + "checksum_mismatch", + "unavailable", +) + + +def _normalise_uri(path: str) -> str: + """Return a full CADC URI.""" + path = path.replace("//", "/").lstrip("/") + prefix = NAMESPACE + "/" + if path.startswith(prefix): + return path + return prefix + path + + +def _relative_path(uri: str) -> str: + """Return the path below the CADC namespace.""" + start = len(NAMESPACE) + 1 + return _normalise_uri(uri)[start:] + + +def _checksum(value: Any) -> Optional[str]: + """Normalise an MD5 checksum.""" + if value is None: + return None + checksum = str(value).strip().lower() + if checksum.startswith("md5:"): + checksum = checksum[4:] + return checksum or None + + +def _size(value: Any) -> Optional[int]: + """Convert a size value to bytes.""" + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _minoc_metadata( + uris: List[str], +) -> Tuple[Dict[str, Dict[str, Any]], Set[str]]: + """Read object metadata from Minoc.""" + if not uris: + return {}, set() + try: + response = cadcclient.info([_relative_path(uri) for uri in uris]) + except Exception: + return {}, set(uris) + + metadata: Dict[str, Dict[str, Any]] = {} + for item in response: + item_uri = item.get("id") + if isinstance(item_uri, str): + metadata[_normalise_uri(item_uri)] = { + "size": _size(item.get("size")), + "checksum": _checksum(item.get("md5sum")), + } + return metadata, set() + + +def _inventory_metadata( + uris: List[str], +) -> Tuple[Dict[str, Dict[str, Any]], Set[str]]: + """Read object metadata from Luskan.""" + metadata: Dict[str, Dict[str, Any]] = {} + unavailable: Set[str] = set() + for offset in range(0, len(uris), QUERY_BATCH_SIZE): + end = offset + QUERY_BATCH_SIZE + batch = uris[offset:end] + quoted = ",".join("'" + uri.replace("'", "''") + "'" for uri in batch) + query = ( + "select uri,contentLength,contentChecksum " + "from inventory.Artifact where uri in (" + quoted + ")" + ) + try: + rows = cadcclient.query(query) + except Exception: + unavailable.update(batch) + continue + batch_set = set(batch) + for row in rows: + if len(row) < 3 or not row[0]: + continue + uri = _normalise_uri(str(row[0])) + if uri in batch_set: + metadata[uri] = { + "size": _size(row[1]), + "checksum": _checksum(row[2]), + } + return metadata, unavailable + + +def _empty_report(scope: str, dataset: str) -> Dict[str, Any]: + """Create an empty verification report.""" + return { + "scope": scope, + "dataset": dataset, + "registered": 0, + "ok": False, + "summary": {category: 0 for category in RESULT_CATEGORIES}, + "results": {category: [] for category in RESULT_CATEGORIES}, + } + + +def verify_dataset(scope: str, dataset: str) -> Dict[str, Any]: + """Compare registered files with Minoc and Luskan metadata.""" + report = _empty_report(scope, dataset) + results = report["results"] + try: + dataset_info = functions.get_dataset_file_info(scope, dataset) + except Exception: + dataset_info = None + if not isinstance(dataset_info, dict) or dataset_info.get("error"): + results["unavailable"].append( + { + "uri": None, + "services": ["datatrail"], + "reason": "Dataset information is unavailable.", + } + ) + _finish_report(report) + return report + + locations = dataset_info.get("file_replica_locations") + if not isinstance(locations, dict): + results["unavailable"].append( + { + "uri": None, + "services": ["datatrail"], + "reason": "Dataset file information is invalid.", + } + ) + _finish_report(report) + return report + minoc_files = locations.get("minoc", []) + if not isinstance(minoc_files, list) or not all( + isinstance(path, str) for path in minoc_files + ): + results["unavailable"].append( + { + "uri": None, + "services": ["datatrail"], + "reason": "Registered Minoc files are invalid.", + } + ) + _finish_report(report) + return report + + uris = sorted({_normalise_uri(path) for path in minoc_files}) + report["registered"] = len(uris) + minoc, minoc_unavailable = _minoc_metadata(uris) + inventory, inventory_unavailable = _inventory_metadata(uris) + for uri in uris: + unavailable_services = [] + if uri in minoc_unavailable: + unavailable_services.append("minoc") + if uri in inventory_unavailable: + unavailable_services.append("luskan") + if unavailable_services: + results["unavailable"].append( + { + "uri": uri, + "services": unavailable_services, + "reason": "Metadata service is unavailable.", + } + ) + continue + + missing_services = [] + if uri not in minoc: + missing_services.append("minoc") + if uri not in inventory: + missing_services.append("luskan") + if missing_services: + results["missing"].append({"uri": uri, "services": missing_services}) + continue + + _compare_metadata(uri, minoc[uri], inventory[uri], results) + + _finish_report(report) + return report + + +def _compare_metadata( + uri: str, + minoc: Dict[str, Any], + inventory: Dict[str, Any], + results: Dict[str, List[Dict[str, Any]]], +) -> None: + """Add one file to its result categories.""" + incomplete = [] + incomplete_services = [] + for field in ("size", "checksum"): + if minoc.get(field) is None: + if "minoc" not in incomplete_services: + incomplete_services.append("minoc") + incomplete.append(field) + if inventory.get(field) is None: + if "luskan" not in incomplete_services: + incomplete_services.append("luskan") + if field not in incomplete: + incomplete.append(field) + if incomplete: + results["unavailable"].append( + { + "uri": uri, + "services": incomplete_services, + "fields": incomplete, + "reason": "Metadata is incomplete.", + } + ) + + mismatch = False + if minoc.get("size") is not None and inventory.get("size") is not None: + if minoc["size"] != inventory["size"]: + mismatch = True + results["size_mismatch"].append( + { + "uri": uri, + "minoc": minoc["size"], + "luskan": inventory["size"], + } + ) + if minoc.get("checksum") is not None and inventory.get("checksum") is not None: + if minoc["checksum"] != inventory["checksum"]: + mismatch = True + results["checksum_mismatch"].append( + { + "uri": uri, + "minoc": minoc["checksum"], + "luskan": inventory["checksum"], + } + ) + if not incomplete and not mismatch: + results["present"].append( + { + "uri": uri, + "size": minoc["size"], + "checksum": minoc["checksum"], + } + ) + + +def _finish_report(report: Dict[str, Any]) -> None: + """Set report totals and status.""" + results = report["results"] + report["summary"] = { + category: len(results[category]) for category in RESULT_CATEGORIES + } + report["ok"] = not any( + results[category] + for category in ( + "missing", + "size_mismatch", + "checksum_mismatch", + "unavailable", + ) + ) + + +def _show_report(report: Dict[str, Any]) -> None: + """Print a concise verification report.""" + click.echo(f"{report['scope']} {report['dataset']}") + click.echo(f"registered: {report['registered']}") + for category in RESULT_CATEGORIES: + label = category.replace("_", "-") + click.echo(f"{label}: {report['summary'][category]}") + for category in RESULT_CATEGORIES[1:]: + for item in report["results"][category]: + target = item.get("uri") or ",".join(item.get("services", [])) + click.echo(f"{category.replace('_', '-')}: {target}") + + +@click.command(name="verify", help="Verify registered Minoc files.") +@click.argument("scope", required=True, type=click.STRING, nargs=1) +@click.argument("dataset", required=True, type=click.STRING, nargs=1) +@click.option("--json", "output_json", is_flag=True, help="Output as JSON.") +@click.pass_context +def verify(ctx: click.Context, scope: str, dataset: str, output_json: bool) -> None: + """Verify registered Minoc files against CADC metadata.""" + report = verify_dataset(scope, dataset) + if output_json: + click.echo(json.dumps(report, indent=2)) + else: + _show_report(report) + if not report["ok"]: + ctx.exit(2 if report["results"]["unavailable"] else 1) diff --git a/tests/test_verify.py b/tests/test_verify.py new file mode 100644 index 0000000..c9dc1be --- /dev/null +++ b/tests/test_verify.py @@ -0,0 +1,120 @@ +"""Tests for dataset verification.""" + +import json +from typing import Any, Dict + +from click.testing import CliRunner + +from dtcli import verify +from dtcli.cli import cli + + +def _metadata(size: int, checksum: str) -> Dict[str, Any]: + """Create file metadata for a test.""" + return {"size": size, "checksum": checksum} + + +def test_verify_dataset_categories(monkeypatch) -> None: + """Classify exact file and metadata outcomes.""" + names = ["a.h5", "b.h5", "c.h5", "d.h5", "e.h5"] + uris = [f"cadc:CHIMEFRB/data/{name}" for name in names] + monkeypatch.setattr( + verify.functions, + "get_dataset_file_info", + lambda scope, dataset: { + "file_replica_locations": {"minoc": [f"data/{name}" for name in names]} + }, + ) + monkeypatch.setattr( + verify, + "_minoc_metadata", + lambda files: ( + { + uris[0]: _metadata(10, "aaa"), + uris[2]: _metadata(30, "ccc"), + uris[3]: _metadata(40, "ddd"), + uris[4]: _metadata(50, "eee"), + }, + set(), + ), + ) + monkeypatch.setattr( + verify, + "_inventory_metadata", + lambda files: ( + { + uris[0]: _metadata(10, "aaa"), + uris[1]: _metadata(20, "bbb"), + uris[2]: _metadata(31, "ccc"), + uris[3]: _metadata(40, "different"), + }, + {uris[4]}, + ), + ) + + report = verify.verify_dataset("test.scope", "event") + + assert report["ok"] is False + assert report["summary"] == { + "present": 1, + "missing": 1, + "size_mismatch": 1, + "checksum_mismatch": 1, + "unavailable": 1, + } + assert report["results"]["present"][0]["uri"] == uris[0] + assert report["results"]["missing"][0]["services"] == ["minoc"] + assert report["results"]["size_mismatch"][0]["uri"] == uris[2] + assert report["results"]["checksum_mismatch"][0]["uri"] == uris[3] + assert report["results"]["unavailable"][0]["services"] == ["luskan"] + + +def test_inventory_metadata_uses_exact_uris(monkeypatch) -> None: + """Query only the registered CADC URIs.""" + calls = [] + + def fake_query(query: str): + """Return one inventory row.""" + calls.append(query) + return [["cadc:CHIMEFRB/data/a.h5", "12", "md5:ABC"], [""]] + + monkeypatch.setattr(verify.cadcclient, "query", fake_query) + + metadata, unavailable = verify._inventory_metadata(["cadc:CHIMEFRB/data/a.h5"]) + + assert unavailable == set() + assert metadata == {"cadc:CHIMEFRB/data/a.h5": {"size": 12, "checksum": "abc"}} + assert "where uri in ('cadc:CHIMEFRB/data/a.h5')" in calls[0] + + +def test_verify_json_reports_unavailable_service(monkeypatch) -> None: + """Return machine-readable failure when Datatrail is unavailable.""" + monkeypatch.setattr("dtcli.cli.check_version", lambda: None) + monkeypatch.setattr( + verify.functions, + "get_dataset_file_info", + lambda scope, dataset: {"error": "connection failed"}, + ) + + result = CliRunner().invoke(cli, ["verify", "test.scope", "event", "--json"]) + + assert result.exit_code == 2 + report = json.loads(result.output) + assert report["ok"] is False + assert report["summary"]["unavailable"] == 1 + assert report["results"]["unavailable"][0]["services"] == ["datatrail"] + assert "connection failed" not in result.output + + +def test_verify_empty_registration_is_clean(monkeypatch) -> None: + """Treat an empty registered file list as a valid result.""" + monkeypatch.setattr( + verify.functions, + "get_dataset_file_info", + lambda scope, dataset: {"file_replica_locations": {"minoc": []}}, + ) + + report = verify.verify_dataset("test.scope", "empty") + + assert report["registered"] == 0 + assert report["ok"] is True