diff --git a/dtcli/cli.py b/dtcli/cli.py index d95c158..e0797cc 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() @@ -48,6 +48,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/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..1a9de70 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,105 @@ +"""HTTP fixtures for tests that do not need the live services.""" + +import json + +import pytest +import requests + + +@pytest.fixture +def http_responses(monkeypatch): + """Serve explicit responses through Requests without bypassing JSON decoding.""" + responses = {} + unexpected = [] + + def add(method, url, data, payload=None): + key = (method, url, json.dumps(payload, sort_keys=True)) + responses[key] = data + + def send(session, prepared, **kwargs): + try: + payload = json.loads(prepared.body) if prepared.body else None + except (TypeError, ValueError) as error: + unexpected.append((prepared.method, prepared.url, "non-JSON body")) + raise AssertionError("Unexpected non-JSON request body") from error + key = (prepared.method, prepared.url, json.dumps(payload, sort_keys=True)) + if key not in responses: + unexpected.append(key) + raise AssertionError(f"Unexpected HTTP request: {key}") + response = requests.Response() + response.status_code = 200 + response.url = prepared.url + response.request = prepared + response.headers["Content-Type"] = "application/json" + response.encoding = "utf-8" + response._content = json.dumps(responses[key]).encode("utf-8") + return response + + monkeypatch.setattr(requests.Session, "send", send) + yield add + # CLI commands can catch exceptions, so also check for unexpected requests here. + assert not unexpected, f"Unexpected HTTP requests: {unexpected}" + + +@pytest.fixture +def datatrail_api(http_responses): + """Supply a small dataset hierarchy for CLI output and filtering tests.""" + server = "https://frb.chimenet.ca/datatrail" + scope = "chime.event.baseband.raw" + http_responses( + "GET", + f"{server}/query/dataset/scopes", + [scope, "chime.event.intensity.raw"], + ) + http_responses( + "GET", + f"{server}/query/dataset/larger?scope={scope}", + {"larger_datasets": ["classified.FRB", "unclassified"]}, + ) + http_responses( + "GET", + f"{server}/query/dataset/children/{scope}/classified.FRB", + {"contains": ["289007650", "289007651"]}, + ) + http_responses( + "POST", + f"{server}/query/dataset/find", + { + "file_replica_locations": { + "minoc": ["data/test-missing/222266914/file.msgpack"] + } + }, + {"scope": "chime.event.intensity.raw", "name": "222266914"}, + ) + http_responses( + "POST", + "https://frb.chimenet.ca/results/view", + [], + { + "query": { + "pipeline": "datatrail-unregistered-datasets", + "results.dataset_name": "not-an-event", + }, + "projection": {"results.files": 0}, + "limit": 100, + }, + ) + + +@pytest.fixture +def results_response(http_responses): + """Register the expected query and response for a workflow-results request.""" + + def add(pipeline, query, projection, data, limit=100): + http_responses( + "POST", + "https://frb.chimenet.ca/results/view", + data, + { + "query": {"pipeline": pipeline, **query}, + "projection": projection, + "limit": limit, + }, + ) + + return add diff --git a/tests/test_cli.py b/tests/test_cli.py index 20983d0..f38d7f3 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -8,6 +8,7 @@ from click.testing import CliRunner from dtcli.cli import cli as datatrail +from dtcli.utilities import utilities @pytest.fixture(scope="module") @@ -40,12 +41,15 @@ def list_specific_files(directory): @pytest.fixture -def runner() -> CliRunner: +def runner(request, monkeypatch) -> CliRunner: """Click CLI runner for testing. Returns: (CliRunner) -> None: """ + if request.node.get_closest_marker("cadc") is None: + request.getfixturevalue("datatrail_api") + monkeypatch.setattr(utilities, "cli_is_latest_release", lambda: True) return CliRunner() diff --git a/tests/test_functions.py b/tests/test_functions.py index 3333593..5cd9b5c 100644 --- a/tests/test_functions.py +++ b/tests/test_functions.py @@ -1,9 +1,7 @@ """Tests for Datatrail CLI.""" from datetime import datetime as dt -from typing import Any, Dict, List, Optional - -import pytest +from typing import Any, Dict from dtcli.src import functions from dtcli.src.functions import ( @@ -14,81 +12,100 @@ from dtcli.utilities.results import failure -def test_view_results() -> None: - """Test view_results.""" - pipeline: str = "datatrail-registration-last-completed-date" - query: Dict[str, Any] = {"site": "chime"} - projection: Dict[str, Any] = {"results": 1} - results: List[Dict[str, Any]] = view_results(pipeline, query, projection) +def test_view_results(results_response) -> None: + """Test view_results with a fixed pipeline record.""" + pipeline = "datatrail-registration-last-completed-date" + query = {"site": "chime"} + projection = {"results": 1} + expected = [{"results": {"last_completed_date": "2024-01-02"}}] + results_response(pipeline, query, projection, expected) + results = view_results(pipeline, query, projection) + assert results == expected assert dt.strptime( - results[0]["results"]["last_completed_date"], "%Y-%M-%d" - ) > dt.strptime("2023-12-01", "%Y-%M-%d") - - -def test_view_results_bad_pipeline() -> None: - """Test view_results with bad pipeline.""" - pipeline: str = "bad-pipeline-name" - query: Dict[str, Any] = {"site": "chime"} - projection: Dict[str, Any] = {"results": 1} - results: List[Dict[str, Any]] = view_results(pipeline, query, projection) - if results: - assert results == [] - else: - pytest.skip("No results found.") - - -def test_get_unregistered_dataset() -> None: - """Test get_unregistered_dataset.""" - pipeline: str = "datatrail-unregistered-datasets" - query: Dict[str, Any] = {} - projection: Dict[str, Any] = {"results.dataset_name": 1, "results.dataset_scope": 1} - limit: int = 1 - try: - results: Dict[str, Any] = view_results(pipeline, query, projection, limit)[0] - except IndexError: - pytest.skip("No unregistered datasets found.") - dataset_name: str = results["results"]["dataset_name"] - dataset_scope: str = results["results"]["dataset_scope"] - - unregistered_dataset: Optional[Dict[str, Any]] = get_unregistered_dataset( - dataset_name, dataset_scope + results[0]["results"]["last_completed_date"], "%Y-%m-%d" + ) > dt.strptime("2023-12-01", "%Y-%m-%d") + + +def test_view_results_bad_pipeline(results_response) -> None: + """Test an unknown pipeline returns an empty result.""" + pipeline = "bad-pipeline-name" + query = {"site": "chime"} + projection = {"results": 1} + results_response(pipeline, query, projection, []) + assert view_results(pipeline, query, projection) == [] + + +def test_get_unregistered_dataset(results_response) -> None: + """Test the event lookup returns its registration details.""" + record = { + "results": { + "dataset_name": "289007650", + "dataset_scope": "chime.event.baseband.raw", + "attach_to_dataset": "classified.FRB", + "reason": "Parent dataset is missing.", + } + } + results_response( + "datatrail-unregistered-datasets", + {"site": "chime", "results.dataset_name": "289007650"}, + {"results.files": 0}, + [record], + limit=1, + ) + result = get_unregistered_dataset("289007650", "chime.event.baseband.raw") + assert result == record + assert "attach_to_dataset" in result["results"] + assert "reason" in result["results"] + + +def test_find_unregistered_datasets(results_response) -> None: + """Test exact, scoped, and partial event queries against fixed responses.""" + pipeline = "datatrail-unregistered-datasets" + projection = {"results.files": 0} + name, scope = "289007650", "chime.event.baseband.raw" + record = { + "results": { + "dataset_name": name, + "dataset_scope": scope, + "reason": "Parent dataset is missing.", + } + } + results_response(pipeline, {"results.dataset_name": name}, projection, [record]) + results_response( + pipeline, + {"results.dataset_name": name, "results.dataset_scope": scope}, + projection, + [record], ) - if unregistered_dataset: - assert "attach_to_dataset" in unregistered_dataset["results"].keys() - assert "reason" in unregistered_dataset["results"].keys() - else: - pytest.skip("No unregistered datasets found.") - - -def test_find_unregistered_datasets() -> None: - """Test find_unregistered_datasets.""" - pipeline: str = "datatrail-unregistered-datasets" - projection: Dict[str, Any] = {"results.dataset_name": 1, "results.dataset_scope": 1} - try: - known: Dict[str, Any] = view_results(pipeline, {}, projection, 1)[0] - except IndexError: - pytest.skip("No unregistered datasets found.") - dataset_name: str = known["results"]["dataset_name"] - dataset_scope: str = known["results"]["dataset_scope"] - - results: List[Dict[str, Any]] = find_unregistered_datasets(dataset_name) - assert len(results) > 0 - assert all(r["results"]["dataset_name"] == dataset_name for r in results) - assert "reason" in results[0]["results"].keys() - - # Scope of the dataset filters nothing out, an unrelated scope filters all. - assert find_unregistered_datasets(dataset_name, scope=dataset_scope) - assert find_unregistered_datasets(dataset_name, scope="not.a.scope") == [] - - # A partial search finds at least the datasets an exact search does. - partial: List[Dict[str, Any]] = find_unregistered_datasets( - dataset_name[:-1], partial=True + results_response( + pipeline, + {"results.dataset_name": name, "results.dataset_scope": "not.a.scope"}, + projection, + [], + ) + results_response( + pipeline, + {"results.dataset_name": {"$regex": name[:-1]}}, + projection, + [record], + ) + results = find_unregistered_datasets(name) + assert results == [record] + assert all(r["results"]["dataset_name"] == name for r in results) + assert "reason" in results[0]["results"] + assert find_unregistered_datasets(name, scope=scope) == [record] + assert find_unregistered_datasets(name, scope="not.a.scope") == [] + assert find_unregistered_datasets(name[:-1], partial=True) == [record] + + +def test_find_unregistered_datasets_no_match(results_response) -> None: + """Test an event with no unregistered records.""" + results_response( + "datatrail-unregistered-datasets", + {"results.dataset_name": "not-an-event"}, + {"results.files": 0}, + [], ) - assert len(partial) >= len(results) - - -def test_find_unregistered_datasets_no_match() -> None: - """Test find_unregistered_datasets with an event that is not unregistered.""" assert find_unregistered_datasets("not-an-event") == [] 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