From b4d4f7522b0e01876afcab480475fae18e7ebe14 Mon Sep 17 00:00:00 2001 From: Dylan Gormley Date: Mon, 24 Aug 2026 15:16:22 -0500 Subject: [PATCH 1/3] feat(ls): add --match and --expand for dataset discovery --- docs/list.md | 71 +++++++++++++++++++++++++++ dtcli/ls.py | 103 +++++++++++++++++++++++++++++++++++++++- dtcli/src/functions.py | 82 ++++++++++++++++++++++++++++++++ tests/test_cli.py | 94 ++++++++++++++++++++++++++++++++++++ tests/test_functions.py | 74 +++++++++++++++++++++++++++++ 5 files changed, 423 insertions(+), 1 deletion(-) diff --git a/docs/list.md b/docs/list.md index 68cfaf3..a1744ce 100644 --- a/docs/list.md +++ b/docs/list.md @@ -12,6 +12,10 @@ Options: -q, --quiet Only errors shown in logs. --write Write the events to file. --json Output as JSON. + --match TEXT Comma-separated, case-insensitive terms a larger dataset must + all contain. + --expand Open each matched larger dataset one level and list its + children. --help Show this message and exit. ``` @@ -102,6 +106,73 @@ Within Datatrail, there are two types of datasets: Please see the CLI reference page for more information on the `list` command: [datatrail list](../cli/#datatrail-list) +## Finding datasets with `--match` and `--expand` + +Navigating the hierarchy one name at a time gets slow when you do not know +where a dataset lives. `--match` filters the larger datasets of a scope, or +of **every** scope when no scope is given, by one or more comma-separated, +case-insensitive terms, which must all appear in the combined +`scope dataset` text: + +```shell +$> datatrail ls chime.acquisition.processed --match gains + Datatrail: Dataset Map ++-----------------------------+---------------+ +| Scope | Dataset | ++-----------------------------+---------------+ +| chime.acquisition.processed | complex_gains | ++-----------------------------+---------------+ +``` + +A hit may be a container whose children are the datasets you actually want. +`--expand` opens each matched larger dataset one level and lists the children +it finds, recording the parent; a match whose children cannot be listed keeps +its own row: + +```shell +$> datatrail ls --match gain --expand + Datatrail: Dataset Map ++-----------------------------+---------------+---------------+ +| Scope | Dataset | Parent | ++-----------------------------+---------------+---------------+ +| chime.acquisition.processed | complex_gains | | ++-----------------------------+---------------+---------------+ +| gbo.acquisition.processed | 20230716 | complex_gains | +| gbo.acquisition.processed | 20230715 | complex_gains | +| ... | ... | ... | ++-----------------------------+---------------+---------------+ +``` + +Rows reached through a parent resolve directly with +`datatrail ps `; a row kept for a matched dataset whose +children could not be listed (or that has none) may still be a container. + +!!! warning "Incomplete maps" + + If Datatrail does not answer for a scope or dataset during the walk, the + map is reported as **incomplete** and the unanswered queries are listed, + rather than silently showing them as empty. With `--json`, those queries + appear in the `failed` list. A partial map still exits 0; a map with **no** + rows and unanswered queries exits 1, since nothing was determined. + +With `--json`, the map is emitted as structured rows for scripting; `parent` +is `null` for rows that were not reached through `--expand`: + +```bash +$ datatrail ls --match gain --expand --json +{ + "results": [ + { + "scope": "gbo.acquisition.processed", + "dataset": "20230525", + "parent": "complex_gains" + }, + ... + ], + "failed": [] +} +``` + ## 🤖 Machine-readable JSON output The `--json` flag outputs structured JSON instead of formatted tables, making it easy to parse the output in scripts and pipelines: diff --git a/dtcli/ls.py b/dtcli/ls.py index d6335da..e30cca7 100644 --- a/dtcli/ls.py +++ b/dtcli/ls.py @@ -2,7 +2,7 @@ import json import logging -from typing import Optional +from typing import Any, Dict, Optional import click from requests.exceptions import ConnectionError @@ -35,6 +35,17 @@ @click.option("-q", "--quiet", is_flag=True, help="Only errors shown in logs.") @click.option("--write", is_flag=True, help="Write the events to file.") @click.option("--json", "output_json", is_flag=True, help="Output as JSON.") +@click.option( + "--match", + type=click.STRING, + default=None, + help="Comma-separated, case-insensitive terms a larger dataset must all contain.", +) +@click.option( + "--expand", + is_flag=True, + help="Open each matched larger dataset one level and list its children.", +) @click.pass_context def list( # noqa: C901 ctx: click.Context, @@ -44,6 +55,8 @@ def list( # noqa: C901 quiet: bool = False, write: bool = False, output_json: bool = False, + match: Optional[str] = None, + expand: bool = False, ): """List Datatrail Scopes & Datasets. @@ -55,6 +68,8 @@ def list( # noqa: C901 quiet (bool): Only errors shown in logs. write (bool): Write the events to file. output_json (bool): Output as JSON. + match (str): Comma-separated terms a larger dataset must all contain. + expand (bool): Open each matched larger dataset one level. """ # Set logging level. set_log_level(logger, verbose, quiet) @@ -74,6 +89,24 @@ def list( # noqa: C901 error_console.print(e) ctx.exit(1) return None + if match is not None or expand: + if datasets: + error_console.print( + "--match and --expand map larger datasets; " + "omit the DATASETS argument." + ) + ctx.exit(1) + return None + if expand and match is None and not scope: + error_console.print( + "--expand alone would open every dataset in the archive; " + "give a SCOPE or --match to narrow it." + ) + ctx.exit(1) + return None + discovery = functions.discover_datasets(scope, match, expand, verbose, quiet) + _display_discovery(discovery, ctx, expand, write, output_json, scope) + return None results = functions.list(scope, datasets, verbose, quiet) # Output JSON if requested. @@ -143,3 +176,71 @@ def list( # noqa: C901 if "error" in results.keys(): error_console.print(results["error"]) ctx.exit(1) + + +def _display_discovery( + results: Dict[str, Any], + ctx: click.Context, + expand: bool, + write: bool, + output_json: bool, + scope: Optional[str], +) -> None: + """Display the dataset map built by functions.discover_datasets. + + An empty map with unanswered queries exits non-zero: nothing was + determined. A partial map is shown, with the unanswered queries listed. + + Args: + results (Dict[str, Any]): Dictionary from functions.discover_datasets. + ctx (click.Context): Click context. + expand (bool): Whether children were listed, adding a parent column. + write (bool): Write the map to file. + output_json (bool): Output as JSON. + scope (Optional[str]): Scope walked, None when all were. + """ + if output_json: + print(json.dumps(results, indent=2)) + if "error" in results: + ctx.exit(1) + if not results["results"] and results["failed"]: + ctx.exit(1) + return + if "error" in results: + error_console.print(results["error"]) + ctx.exit(1) + return + rows = results["results"] + failed = results["failed"] + if write: + with open(f"./dataset_map_{scope if scope else 'all_scopes'}.json", "w") as f: + json.dump(results, f) + if rows: + table = Table( + title="Datatrail: Dataset Map", + header_style="magenta", + title_style="bold magenta", + ) + table.add_column("Scope") + table.add_column("Dataset") + if expand: + table.add_column("Parent") + previous = None + for row in rows: + if previous is not None and row["scope"] != previous: + table.add_section() + previous = row["scope"] + line = [row["scope"], row["dataset"]] + if expand: + line.append(row["parent"] if row["parent"] else "") + table.add_row(*line) + with console.pager(styles=False): + console.print(table) + elif not failed: + console.print("No datasets matched.") + if failed: + error_console.print("Map is incomplete -- Datatrail did not answer for:") + for item in failed: + error_console.print(f" {item}") + if not rows: + ctx.exit(1) diff --git a/dtcli/src/functions.py b/dtcli/src/functions.py index 2804c68..abbee2f 100644 --- a/dtcli/src/functions.py +++ b/dtcli/src/functions.py @@ -116,6 +116,88 @@ def list( # noqa: C901 return {} +def discover_datasets( + scope: Optional[str] = None, + match: Optional[str] = None, + expand: bool = False, + verbose: int = 0, + quiet: bool = False, +) -> Dict[str, Any]: + """Map larger datasets across scopes, with filtering and expansion. + + Walks one scope, or every scope when none is given, and keeps the larger + datasets whose "scope dataset" text contains every comma-separated, + case-insensitive match term. With expand, each kept dataset is opened one + level and its children become the rows, recording the opened dataset as + their parent; a dataset whose children cannot be listed keeps its own row. + A scope or dataset Datatrail does not answer for is reported in 'failed' + rather than shown as empty. + + Args: + scope (Optional[str], optional): Scope to walk. Defaults to None, + which walks every scope. + match (Optional[str], optional): Comma-separated terms a dataset must + all contain. Defaults to None. + expand (bool, optional): Open each kept dataset one level. Defaults + to False. + verbose (int, optional): Verbosity. Defaults to 0. + quiet (bool, optional): Minimal logging. Defaults to False. + + Returns: + Dict[str, Any]: Keys 'results', rows of scope, dataset and parent, + and 'failed', the queries Datatrail did not answer. Key 'error' + on a configuration or connection failure. + """ + # Set logging level. + utilities.set_log_level(logger, verbose, quiet) + terms = [t.strip().lower() for t in (match or "").split(",") if t.strip()] + if scope: + scopes = [scope] + else: + found = list(verbose=verbose, quiet=quiet) + if "error" in found: + return found + answer = found.get("scopes") + # A non-200 response body is passed through as a string; never walk + # it, or any other non-list shape, as if it were the scopes list. + # NB: isinstance against the builtin list is unavailable here, since + # this module's list() shadows it. + if isinstance(answer, str) or not isinstance(answer, Sequence): + return {"error": "Datatrail did not answer the scopes query."} + if not answer: + return { + "error": "Datatrail reports zero scopes: an account or " + "configuration problem, not an empty archive." + } + scopes = sorted(answer) + results: List[Dict[str, Optional[str]]] = [] + failed: List[str] = [] + for s in scopes: + listed = list(s, verbose=verbose, quiet=quiet) + datasets = None if "error" in listed else listed.get("larger_datasets") + if datasets is None: + failed.append(f"datasets in {s}") + continue + kept = [ + d for d in sorted(datasets) if all(t in f"{s} {d}".lower() for t in terms) + ] + for d in kept: + if not expand: + results.append({"scope": s, "dataset": d, "parent": None}) + continue + opened = list(s, d, verbose=verbose, quiet=quiet) + children = None if "error" in opened else opened.get("datasets") + if children is None: + failed.append(f"children of {s} {d}") + results.append({"scope": s, "dataset": d, "parent": None}) + elif children: + for c in sorted(children, reverse=True): + results.append({"scope": s, "dataset": c, "parent": d}) + else: + results.append({"scope": s, "dataset": d, "parent": None}) + return {"results": results, "failed": failed} + + def ps( scope: str, dataset: str, diff --git a/tests/test_cli.py b/tests/test_cli.py index 77ddee8..4f02f06 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -106,6 +106,8 @@ def test_cli_list_help(runner: CliRunner) -> None: assert "--write" in result.output assert "--json" in result.output assert "Output as JSON" in result.output + assert "--match" in result.output + assert "--expand" in result.output def test_cli_ps_help(runner: CliRunner) -> None: @@ -366,6 +368,71 @@ def test_cli_list_children(runner: CliRunner) -> None: assert "289007650" in result.output +def test_cli_list_match(runner: CliRunner) -> None: + """Test for CLI list to filter larger datasets with --match. + + Args: + runner (CliRunner): Click runner. + """ + result = runner.invoke( + datatrail, ["ls", "chime.event.baseband.raw", "--match", "classified"] + ) + assert result.exit_code == 0 + assert "classified.FRB" in result.output + + +def test_cli_list_match_expand(runner: CliRunner) -> None: + """Test for CLI list to expand matched larger datasets one level. + + Args: + runner (CliRunner): Click runner. + """ + result = runner.invoke( + datatrail, + ["ls", "chime.event.baseband.raw", "--match", "classified.FRB", "--expand"], + ) + assert result.exit_code == 0 + assert "289007650" in result.output + assert "classified.FRB" in result.output + + +def test_cli_list_match_no_hits(runner: CliRunner) -> None: + """Test for CLI list with --match matching nothing. + + Args: + runner (CliRunner): Click runner. + """ + result = runner.invoke( + datatrail, + ["ls", "chime.event.baseband.raw", "--match", "no.such.dataset.term"], + ) + assert result.exit_code == 0 + assert "No datasets matched." in result.output + + +def test_cli_list_match_with_dataset_argument(runner: CliRunner) -> None: + """Test for CLI list rejecting --match combined with a dataset argument. + + Args: + runner (CliRunner): Click runner. + """ + result = runner.invoke( + datatrail, + ["ls", "chime.event.baseband.raw", "classified.FRB", "--match", "FRB"], + ) + assert result.exit_code == 1 + + +def test_cli_list_bare_expand(runner: CliRunner) -> None: + """Test for CLI list rejecting --expand without a scope or --match. + + Args: + runner (CliRunner): Click runner. + """ + result = runner.invoke(datatrail, ["ls", "--expand"]) + assert result.exit_code == 1 + + @pytest.mark.cadc def test_cli_ps(runner: CliRunner) -> None: """Test for CLI ps command. @@ -682,6 +749,33 @@ def test_cli_list_children_json(runner: CliRunner) -> None: assert "289007650" in output_data["datasets"] +def test_cli_list_match_json(runner: CliRunner) -> None: + """Test for CLI list to output the dataset map as JSON. + + Args: + runner (CliRunner): Click runner. + """ + import json + + result = runner.invoke( + datatrail, + ["ls", "chime.event.baseband.raw", "--match", "classified", "--json"], + ) + assert result.exit_code == 0 + # Extract JSON from output (skip version check message if present) + json_start = result.output.find("{") + json_output = result.output[json_start:] + # Parse the output as JSON + output_data = json.loads(json_output) + # Should have 'results' rows and a 'failed' list + assert { + "scope": "chime.event.baseband.raw", + "dataset": "classified.FRB", + "parent": None, + } in output_data["results"] + assert output_data["failed"] == [] + + @pytest.mark.cadc def test_cli_ps_json(runner: CliRunner) -> None: """Test for CLI ps command with JSON output. diff --git a/tests/test_functions.py b/tests/test_functions.py index 59155b1..f3922c1 100644 --- a/tests/test_functions.py +++ b/tests/test_functions.py @@ -112,3 +112,77 @@ def json() -> Dict[str, Any]: ) results: Dict[str, Any] = functions.list() assert results == {"error": "Datatrail did not answer the scopes query."} + + +def _fake_list(scope=None, dataset=None, verbose=0, quiet=False): + """Stand-in for functions.list with one scope not answering.""" + if scope is None: + return {"scopes": ["b.scope", "a.scope", "c.scope"]} + if dataset is None: + if scope == "a.scope": + return {"larger_datasets": ["data.other", "data.good", "skip.me"]} + if scope == "c.scope": + return {"larger_datasets": []} + return {"error": "Datatrail Server at CHIME is not responding."} + if dataset == "data.good": + return {"datasets": ["child1", "child2"]} + return {"error": "Datatrail Server at CHIME is not responding."} + + +def test_discover_datasets(monkeypatch) -> None: + """Test discover_datasets filtering, sorting, and expansion.""" + monkeypatch.setattr(functions, "list", _fake_list) + results: Dict[str, Any] = functions.discover_datasets(match="data", expand=True) + assert results["results"] == [ + {"scope": "a.scope", "dataset": "child2", "parent": "data.good"}, + {"scope": "a.scope", "dataset": "child1", "parent": "data.good"}, + {"scope": "a.scope", "dataset": "data.other", "parent": None}, + ] + # An unanswered query is reported, never shown as empty. + assert results["failed"] == [ + "children of a.scope data.other", + "datasets in b.scope", + ] + + +def test_discover_datasets_no_expand(monkeypatch) -> None: + """Test discover_datasets without expansion, terms ANDed against scope.""" + monkeypatch.setattr(functions, "list", _fake_list) + results: Dict[str, Any] = functions.discover_datasets(match="a.scope,data") + assert results["results"] == [ + {"scope": "a.scope", "dataset": "data.good", "parent": None}, + {"scope": "a.scope", "dataset": "data.other", "parent": None}, + ] + assert results["failed"] == ["datasets in b.scope"] + + +def test_discover_datasets_single_scope(monkeypatch) -> None: + """Test discover_datasets walking one named scope only.""" + monkeypatch.setattr(functions, "list", _fake_list) + results: Dict[str, Any] = functions.discover_datasets(scope="a.scope") + assert [r["dataset"] for r in results["results"]] == [ + "data.good", + "data.other", + "skip.me", + ] + assert results["failed"] == [] + + +def test_discover_datasets_empty_scope_is_not_failure(monkeypatch) -> None: + """Test a scope that answers with no datasets is empty, not failed.""" + monkeypatch.setattr(functions, "list", _fake_list) + results: Dict[str, Any] = functions.discover_datasets(scope="c.scope") + assert results["results"] == [] + assert results["failed"] == [] + + +def test_discover_datasets_unanswered_scopes_query(monkeypatch) -> None: + """Test a non-list scopes answer is an error, never walked as text.""" + + def bad_list(scope=None, dataset=None, verbose=0, quiet=False): + return {"scopes": "Bad Gateway"} + + monkeypatch.setattr(functions, "list", bad_list) + results: Dict[str, Any] = functions.discover_datasets(match="gain") + assert "error" in results + assert "results" not in results From 1d51ae012e06d860e3fb18bf25828edb5e8b9585 Mon Sep 17 00:00:00 2001 From: Dylan Gormley Date: Tue, 25 Aug 2026 12:15:07 -0500 Subject: [PATCH 2/3] feat(ls): add recursive dataset discovery --- docs/list.md | 35 ++++++++++- dtcli/ls.py | 81 +++++++++++++++++-------- dtcli/src/functions.py | 88 +++++++++++++++++++++++++-- tests/test_cli.py | 72 ++++++++++++++++++++++ tests/test_functions.py | 128 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 371 insertions(+), 33 deletions(-) diff --git a/docs/list.md b/docs/list.md index a1744ce..11af455 100644 --- a/docs/list.md +++ b/docs/list.md @@ -16,6 +16,8 @@ Options: all contain. --expand Open each matched larger dataset one level and list its children. + --recursive Open each matched larger dataset through all descendant + levels. --help Show this message and exit. ``` @@ -155,8 +157,24 @@ children could not be listed (or that has none) may still be a container. appear in the `failed` list. A partial map still exits 0; a map with **no** rows and unanswered queries exits 1, since nothing was determined. +`--recursive` follows every descendant of each matched larger dataset instead +of stopping after one level. It emits terminal datasets and records the full +path used to reach each one: + +```shell +$> datatrail ls gbo.acquisition.processed --match gains --recursive +``` + +The walk visits each dataset once, so shared descendants are not duplicated +and hierarchy cycles cannot loop forever. The first path found in sorted order +is retained. An answered empty child list is a terminal dataset. A branch that +does not answer is retained as a partial row and also listed under `failed`. +Like `--expand`, a recursive walk across all scopes requires `--match` to keep +the request bounded. + With `--json`, the map is emitted as structured rows for scripting; `parent` -is `null` for rows that were not reached through `--expand`: +is `null` for rows that were not reached through expansion. Recursive rows +also include `path`, from the matched larger dataset through the terminal row: ```bash $ datatrail ls --match gain --expand --json @@ -173,6 +191,21 @@ $ datatrail ls --match gain --expand --json } ``` +```bash +$ datatrail ls gbo.acquisition.processed --match gains --recursive --json +{ + "results": [ + { + "scope": "gbo.acquisition.processed", + "dataset": "20230525", + "parent": "complex_gains", + "path": ["complex_gains", "20230525"] + } + ], + "failed": [] +} +``` + ## 🤖 Machine-readable JSON output The `--json` flag outputs structured JSON instead of formatted tables, making it easy to parse the output in scripts and pipelines: diff --git a/dtcli/ls.py b/dtcli/ls.py index e30cca7..babd1e1 100644 --- a/dtcli/ls.py +++ b/dtcli/ls.py @@ -2,7 +2,7 @@ import json import logging -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional import click from requests.exceptions import ConnectionError @@ -46,6 +46,11 @@ is_flag=True, help="Open each matched larger dataset one level and list its children.", ) +@click.option( + "--recursive", + is_flag=True, + help="Open each matched larger dataset through all descendant levels.", +) @click.pass_context def list( # noqa: C901 ctx: click.Context, @@ -57,6 +62,7 @@ def list( # noqa: C901 output_json: bool = False, match: Optional[str] = None, expand: bool = False, + recursive: bool = False, ): """List Datatrail Scopes & Datasets. @@ -70,6 +76,7 @@ def list( # noqa: C901 output_json (bool): Output as JSON. match (str): Comma-separated terms a larger dataset must all contain. expand (bool): Open each matched larger dataset one level. + recursive (bool): Open all descendants of each matched larger dataset. """ # Set logging level. set_log_level(logger, verbose, quiet) @@ -89,23 +96,32 @@ def list( # noqa: C901 error_console.print(e) ctx.exit(1) return None - if match is not None or expand: + if match is not None or expand or recursive: if datasets: error_console.print( - "--match and --expand map larger datasets; " + "--match, --expand, and --recursive map larger datasets; " "omit the DATASETS argument." ) ctx.exit(1) return None - if expand and match is None and not scope: + if (expand or recursive) and match is None and not scope: error_console.print( - "--expand alone would open every dataset in the archive; " + "Expansion alone would open every dataset in the archive; " "give a SCOPE or --match to narrow it." ) ctx.exit(1) return None - discovery = functions.discover_datasets(scope, match, expand, verbose, quiet) - _display_discovery(discovery, ctx, expand, write, output_json, scope) + discovery = functions.discover_datasets( + scope=scope, + match=match, + expand=expand, + verbose=verbose, + quiet=quiet, + recursive=recursive, + ) + _display_discovery( + discovery, ctx, expand or recursive, write, output_json, scope, recursive + ) return None results = functions.list(scope, datasets, verbose, quiet) @@ -185,6 +201,7 @@ def _display_discovery( write: bool, output_json: bool, scope: Optional[str], + recursive: bool = False, ) -> None: """Display the dataset map built by functions.discover_datasets. @@ -198,6 +215,7 @@ def _display_discovery( write (bool): Write the map to file. output_json (bool): Output as JSON. scope (Optional[str]): Scope walked, None when all were. + recursive (bool): Whether rows include their full hierarchy path. """ if output_json: print(json.dumps(results, indent=2)) @@ -216,26 +234,8 @@ def _display_discovery( with open(f"./dataset_map_{scope if scope else 'all_scopes'}.json", "w") as f: json.dump(results, f) if rows: - table = Table( - title="Datatrail: Dataset Map", - header_style="magenta", - title_style="bold magenta", - ) - table.add_column("Scope") - table.add_column("Dataset") - if expand: - table.add_column("Parent") - previous = None - for row in rows: - if previous is not None and row["scope"] != previous: - table.add_section() - previous = row["scope"] - line = [row["scope"], row["dataset"]] - if expand: - line.append(row["parent"] if row["parent"] else "") - table.add_row(*line) with console.pager(styles=False): - console.print(table) + console.print(_discovery_table(rows, expand, recursive)) elif not failed: console.print("No datasets matched.") if failed: @@ -244,3 +244,32 @@ def _display_discovery( error_console.print(f" {item}") if not rows: ctx.exit(1) + + +def _discovery_table( + rows: List[Dict[str, Any]], expand: bool, recursive: bool = False +) -> Table: + """Build the dataset map table.""" + table = Table( + title="Datatrail: Dataset Map", + header_style="magenta", + title_style="bold magenta", + ) + table.add_column("Scope") + table.add_column("Dataset") + if expand: + table.add_column("Parent") + if recursive: + table.add_column("Path") + previous = None + for row in rows: + if previous is not None and row["scope"] != previous: + table.add_section() + previous = row["scope"] + line = [row["scope"], row["dataset"]] + if expand: + line.append(row["parent"] if row["parent"] else "") + if recursive: + line.append(" / ".join(row["path"])) + table.add_row(*line) + return table diff --git a/dtcli/src/functions.py b/dtcli/src/functions.py index abbee2f..245457e 100644 --- a/dtcli/src/functions.py +++ b/dtcli/src/functions.py @@ -9,7 +9,7 @@ from collections import Counter from collections.abc import Sequence from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Optional, Set, Tuple import requests @@ -122,6 +122,7 @@ def discover_datasets( expand: bool = False, verbose: int = 0, quiet: bool = False, + recursive: bool = False, ) -> Dict[str, Any]: """Map larger datasets across scopes, with filtering and expansion. @@ -129,9 +130,10 @@ def discover_datasets( datasets whose "scope dataset" text contains every comma-separated, case-insensitive match term. With expand, each kept dataset is opened one level and its children become the rows, recording the opened dataset as - their parent; a dataset whose children cannot be listed keeps its own row. - A scope or dataset Datatrail does not answer for is reported in 'failed' - rather than shown as empty. + their parent. With recursive, each kept dataset is opened until terminal + datasets are reached. A dataset whose children cannot be listed keeps its + own row. A scope or dataset Datatrail does not answer for is reported in + 'failed' rather than shown as empty. Args: scope (Optional[str], optional): Scope to walk. Defaults to None, @@ -142,11 +144,14 @@ def discover_datasets( to False. verbose (int, optional): Verbosity. Defaults to 0. quiet (bool, optional): Minimal logging. Defaults to False. + recursive (bool, optional): Open all descendants of each kept dataset. + Defaults to False. Returns: Dict[str, Any]: Keys 'results', rows of scope, dataset and parent, - and 'failed', the queries Datatrail did not answer. Key 'error' - on a configuration or connection failure. + plus path for recursive rows, and 'failed', the branches Datatrail + did not answer. Key 'error' on a configuration or connection + failure. """ # Set logging level. utilities.set_log_level(logger, verbose, quiet) @@ -181,6 +186,13 @@ def discover_datasets( kept = [ d for d in sorted(datasets) if all(t in f"{s} {d}".lower() for t in terms) ] + if recursive: + rows, branch_failures = _discover_descendants( + s, kept, verbose=verbose, quiet=quiet + ) + results.extend(rows) + failed.extend(branch_failures) + continue for d in kept: if not expand: results.append({"scope": s, "dataset": d, "parent": None}) @@ -198,6 +210,70 @@ def discover_datasets( return {"results": results, "failed": failed} +def _discover_descendants( + scope: str, + roots: Sequence[str], + verbose: int = 0, + quiet: bool = False, +) -> Tuple[List[Dict[str, Any]], List[str]]: + """Return unique terminal datasets below the given roots.""" + results: List[Dict[str, Any]] = [] + failed: List[str] = [] + visited: Set[str] = set() + emitted: Set[str] = set() + stack: List[Tuple[str, Optional[str], Tuple[str, ...]]] = [ + (root, None, (root,)) for root in reversed(sorted(set(roots))) + ] + + def add_row(dataset: str, parent: Optional[str], path: Tuple[str, ...]) -> None: + if dataset in emitted: + return + results.append( + { + "scope": scope, + "dataset": dataset, + "parent": parent, + "path": [*path], + } + ) + emitted.add(dataset) + + while stack: + dataset, parent, path = stack.pop() + if dataset in visited: + continue + visited.add(dataset) + opened = list(scope, dataset, verbose=verbose, quiet=quiet) + children = None if "error" in opened else opened.get("datasets") + if ( + children is None + or isinstance(children, str) + or not isinstance(children, Sequence) + or any(not isinstance(child, str) or not child.strip() for child in children) + ): + failed.append(f"children of {scope} {' / '.join(path)}") + add_row(dataset, parent, path) + continue + + child_names = sorted(set(children)) + if not child_names: + add_row(dataset, parent, path) + continue + + cycle_found = False + for child in reversed(child_names): + if child in path: + failed.append(f"cycle in {scope}: {' / '.join(path + (child,))}") + cycle_found = True + continue + if child not in visited: + stack.append((child, dataset, path + (child,))) + if cycle_found: + add_row(dataset, parent, path) + + return results, failed + + def ps( scope: str, dataset: str, diff --git a/tests/test_cli.py b/tests/test_cli.py index 4f02f06..e91ea4b 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -108,6 +108,7 @@ def test_cli_list_help(runner: CliRunner) -> None: assert "Output as JSON" in result.output assert "--match" in result.output assert "--expand" in result.output + assert "--recursive" in result.output def test_cli_ps_help(runner: CliRunner) -> None: @@ -433,6 +434,77 @@ def test_cli_list_bare_expand(runner: CliRunner) -> None: assert result.exit_code == 1 +def test_cli_list_bare_recursive(runner: CliRunner) -> None: + """Test for CLI list rejecting an unconstrained recursive walk. + + Args: + runner (CliRunner): Click runner. + """ + result = runner.invoke(datatrail, ["ls", "--recursive"]) + assert result.exit_code == 1 + + +def test_cli_list_recursive_plain(runner: CliRunner, monkeypatch) -> None: + """Test for CLI list showing recursive paths in the table. + + Args: + runner (CliRunner): Click runner. + monkeypatch: Pytest monkeypatch fixture. + """ + + def fake_discovery(**kwargs): + assert kwargs["recursive"] is True + return { + "results": [ + { + "scope": "test.scope", + "dataset": "leaf", + "parent": "branch", + "path": ["root", "branch", "leaf"], + } + ], + "failed": [], + } + + monkeypatch.setattr("dtcli.ls.functions.discover_datasets", fake_discovery) + result = runner.invoke(datatrail, ["ls", "--match", "root", "--recursive"]) + assert result.exit_code == 0 + assert "leaf" in result.output + assert "root / branch / leaf" in result.output + + +def test_cli_list_recursive_json(runner: CliRunner, monkeypatch) -> None: + """Test for CLI list retaining recursive paths in JSON. + + Args: + runner (CliRunner): Click runner. + monkeypatch: Pytest monkeypatch fixture. + """ + import json + + expected = { + "results": [ + { + "scope": "test.scope", + "dataset": "leaf", + "parent": "branch", + "path": ["root", "branch", "leaf"], + } + ], + "failed": [], + } + + def fake_discovery(**kwargs): + assert kwargs["recursive"] is True + return expected + + monkeypatch.setattr("dtcli.ls.functions.discover_datasets", fake_discovery) + result = runner.invoke(datatrail, ["ls", "--match", "root", "--recursive", "--json"]) + assert result.exit_code == 0 + json_start = result.output.find("{") + assert json.loads(result.output[json_start:]) == expected + + @pytest.mark.cadc def test_cli_ps(runner: CliRunner) -> None: """Test for CLI ps command. diff --git a/tests/test_functions.py b/tests/test_functions.py index f3922c1..d362206 100644 --- a/tests/test_functions.py +++ b/tests/test_functions.py @@ -186,3 +186,131 @@ def bad_list(scope=None, dataset=None, verbose=0, quiet=False): results: Dict[str, Any] = functions.discover_datasets(match="gain") assert "error" in results assert "results" not in results + + +def test_discover_datasets_recursive_paths(monkeypatch) -> None: + """Test recursive discovery paths, ordering, filtering, and duplicates.""" + calls = [] + children = { + "wanted.root": ["branch.b", "branch.a", "branch.a"], + "branch.a": ["leaf.shared", "leaf.a"], + "branch.b": ["leaf.b", "leaf.shared"], + } + + def fake_list(scope=None, dataset=None, verbose=0, quiet=False): + if dataset is None: + return { + "larger_datasets": [ + "wanted.root", + "skip.root", + "wanted.empty", + "wanted.root", + ] + } + calls.append(dataset) + return {"datasets": children.get(dataset, [])} + + monkeypatch.setattr(functions, "list", fake_list) + results = functions.discover_datasets( + scope="test.scope", match="wanted", recursive=True + ) + assert results == { + "results": [ + { + "scope": "test.scope", + "dataset": "wanted.empty", + "parent": None, + "path": ["wanted.empty"], + }, + { + "scope": "test.scope", + "dataset": "leaf.a", + "parent": "branch.a", + "path": ["wanted.root", "branch.a", "leaf.a"], + }, + { + "scope": "test.scope", + "dataset": "leaf.shared", + "parent": "branch.a", + "path": ["wanted.root", "branch.a", "leaf.shared"], + }, + { + "scope": "test.scope", + "dataset": "leaf.b", + "parent": "branch.b", + "path": ["wanted.root", "branch.b", "leaf.b"], + }, + ], + "failed": [], + } + assert "skip.root" not in calls + assert calls.count("wanted.root") == 1 + assert calls.count("leaf.shared") == 1 + + +def test_discover_datasets_recursive_empty_and_failed(monkeypatch) -> None: + """Test recursive discovery keeps empty and failed branches distinct.""" + + def fake_list(scope=None, dataset=None, verbose=0, quiet=False): + if dataset is None: + return {"larger_datasets": ["root"]} + if dataset == "root": + return {"datasets": ["offline", "malformed", "empty"]} + if dataset == "offline": + return {"error": "service unavailable"} + if dataset == "malformed": + return {"datasets": [None]} + return {"datasets": []} + + monkeypatch.setattr(functions, "list", fake_list) + results = functions.discover_datasets(scope="test.scope", recursive=True) + assert results["results"] == [ + { + "scope": "test.scope", + "dataset": "empty", + "parent": "root", + "path": ["root", "empty"], + }, + { + "scope": "test.scope", + "dataset": "malformed", + "parent": "root", + "path": ["root", "malformed"], + }, + { + "scope": "test.scope", + "dataset": "offline", + "parent": "root", + "path": ["root", "offline"], + }, + ] + assert results["failed"] == [ + "children of test.scope root / malformed", + "children of test.scope root / offline", + ] + + +def test_discover_datasets_recursive_cycle(monkeypatch) -> None: + """Test recursive discovery stops and reports a hierarchy cycle.""" + calls = [] + + def fake_list(scope=None, dataset=None, verbose=0, quiet=False): + if dataset is None: + return {"larger_datasets": ["root"]} + calls.append(dataset) + return {"datasets": ["branch"] if dataset == "root" else ["root"]} + + monkeypatch.setattr(functions, "list", fake_list) + results = functions.discover_datasets(scope="test.scope", recursive=True) + assert results["results"] == [ + { + "scope": "test.scope", + "dataset": "branch", + "parent": "root", + "path": ["root", "branch"], + } + ] + assert results["failed"] == [ + "cycle in test.scope: root / branch / root", + ] + assert calls == ["root", "branch"] From 7ab966230e05995b7b3f99e928848421de798dd7 Mon Sep 17 00:00:00 2001 From: Dylan Gormley Date: Tue, 25 Aug 2026 12:22:58 -0500 Subject: [PATCH 3/3] feat(inventory): add resumable manifests --- docs/commands.md | 2 + docs/inventory.md | 71 +++++++ dtcli/cli.py | 3 +- dtcli/inventory.py | 402 ++++++++++++++++++++++++++++++++++++++++ mkdocs.yml | 1 + tests/test_inventory.py | 288 ++++++++++++++++++++++++++++ 6 files changed, 766 insertions(+), 1 deletion(-) create mode 100644 docs/inventory.md create mode 100644 dtcli/inventory.py create mode 100644 tests/test_inventory.py diff --git a/docs/commands.md b/docs/commands.md index 6448da2..3016d25 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -9,6 +9,8 @@ The commands available to you are: - `clear`: This removes all files belonging to the 'scope' and 'dataset', only available for the local and canfar sites. - `config`: Edit the `.datatrail/config.yaml` configuration file. +- `inventory`: Recursively discover datasets and write their file replica URIs + to a resumable JSON manifest. - `list`: This list either the 'scopes' available or all of the datasets belonging to the given dataset. - `ps`: This provides detailed information for the given 'scope' and 'dataset' combination. diff --git a/docs/inventory.md b/docs/inventory.md new file mode 100644 index 0000000..c24cdc4 --- /dev/null +++ b/docs/inventory.md @@ -0,0 +1,71 @@ +# Building a durable inventory + +`datatrail inventory` recursively discovers terminal datasets and records each +file replica URI in a versioned JSON manifest. The manifest can later drive a +batch download without repeating the archive crawl. + +Every run must be bounded in one of these ways: + +```shell +$> datatrail inventory chime.event.baseband.raw +$> datatrail inventory --match classified,baseband +$> datatrail inventory gbo.acquisition.processed --parent complex_gains +``` + +A scope limits traversal to that scope. `--match` may search across scopes but +only opens matching larger datasets. `--parent` requires a scope and starts at +one known dataset. It cannot be combined with `--match`. + +Use `--output` to choose the manifest path: + +```shell +$> datatrail inventory chime.event.baseband.raw --match classified \ + --output baseband-inventory.json +``` + +The command writes the manifest atomically after discovery and after every +dataset query. A rerun with the same selection reuses `ready` and `empty` +entries, then retries `pending` and `failed` entries. A different selection is +refused so unrelated inventories cannot be mixed. + +An inventory with any failed discovery branch or file query exits nonzero. +`--allow-incomplete` keeps the incomplete manifest but exits zero when a caller +wants to inspect or process the available subset. + +## Manifest format + +The first format is `datatrail.inventory/v1`: + +```json +{ + "schema": "datatrail.inventory/v1", + "selection": { + "scope": "gbo.acquisition.processed", + "match": [], + "parent": "complex_gains" + }, + "complete": true, + "discovery_failures": [], + "datasets": [ + { + "scope": "gbo.acquisition.processed", + "dataset": "20230525", + "parent": "complex_gains", + "path": ["complex_gains", "20230525"], + "status": "ready", + "replicas": [ + { + "storage_element": "minoc", + "uri": "cadc:CHIMEFRB/example/file.h5" + } + ] + } + ] +} +``` + +Replica rows contain only Datatrail information. Size and checksum fields may +be added in a later schema when they can be obtained without requiring a CADC +credential during inventory creation. A valid dataset with no replica URIs has +status `empty`; an unavailable or invalid file response has status `failed` +and an `error` field. diff --git a/dtcli/cli.py b/dtcli/cli.py index c123bfa..8ab057e 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, inventory, ls, ps, pull, scout, unregistered from dtcli.utilities import utilities pretty.install() @@ -43,6 +43,7 @@ def version(): cli.add_command(clear.clear) cli.add_command(config.config) +cli.add_command(inventory.inventory) cli.add_command(ls.list, aliases=["ls"]) cli.add_command(ps.ps) cli.add_command(pull.pull) diff --git a/dtcli/inventory.py b/dtcli/inventory.py new file mode 100644 index 0000000..41d9dec --- /dev/null +++ b/dtcli/inventory.py @@ -0,0 +1,402 @@ +"""Datatrail Inventory Command.""" + +import json +import os +import tempfile +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +import click + +from dtcli.src import functions + +SCHEMA = "datatrail.inventory/v1" +FINISHED_STATUSES = {"ready", "empty"} +ENTRY_STATUSES = FINISHED_STATUSES | {"pending", "failed"} + + +@click.command(name="inventory", help="Build a resumable dataset file inventory.") +@click.argument("scope", required=False, type=click.STRING) +@click.option( + "--match", + type=click.STRING, + default=None, + help="Comma-separated terms a larger dataset must all contain.", +) +@click.option( + "--parent", + type=click.STRING, + default=None, + help="Start at this dataset within SCOPE.", +) +@click.option( + "--output", + "-o", + type=click.Path( + file_okay=True, + dir_okay=False, + writable=True, + resolve_path=True, + path_type=Path, + ), + default=Path("datatrail-inventory.json"), + show_default=True, + help="Manifest path.", +) +@click.option( + "--allow-incomplete", + is_flag=True, + help="Exit successfully while unresolved entries remain.", +) +@click.option("-v", "--verbose", count=True, help="Verbosity: v=INFO, vv=DEBUG.") +@click.option("-q", "--quiet", is_flag=True, help="Only errors shown in logs.") +@click.pass_context +def inventory( + ctx: click.Context, + scope: Optional[str], + match: Optional[str], + parent: Optional[str], + output: Path, + allow_incomplete: bool, + verbose: int, + quiet: bool, +) -> None: + """Build a durable inventory of dataset replica URIs. + + Args: + ctx (click.Context): Click context. + scope (Optional[str]): Scope to inventory. + match (Optional[str]): Terms used to select larger datasets. + parent (Optional[str]): Dataset where traversal starts. + output (Path): Manifest path. + allow_incomplete (bool): Exit zero with unresolved entries. + verbose (int): Verbosity level. + quiet (bool): Minimal logging. + """ + try: + manifest = build_inventory( + scope=scope, + match=match, + parent=parent, + output=output, + verbose=verbose, + quiet=quiet, + ) + except (OSError, ValueError) as error: + raise click.ClickException(str(error)) from error + + ready = sum(entry["status"] == "ready" for entry in manifest["datasets"]) + empty = sum(entry["status"] == "empty" for entry in manifest["datasets"]) + files = sum(len(entry["replicas"]) for entry in manifest["datasets"]) + click.echo(f"Wrote {output}: {ready} ready, {empty} empty, {files} replica URIs.") + if not manifest["complete"]: + unresolved = sum( + entry["status"] not in FINISHED_STATUSES for entry in manifest["datasets"] + ) + discovery = len(manifest["discovery_failures"]) + click.echo( + f"Inventory incomplete: {unresolved} dataset entries and " + f"{discovery} discovery branches unresolved.", + err=True, + ) + if not allow_incomplete: + ctx.exit(1) + + +def build_inventory( + scope: Optional[str], + match: Optional[str], + parent: Optional[str], + output: Path, + verbose: int = 0, + quiet: bool = False, +) -> Dict[str, Any]: + """Build or resume an inventory manifest.""" + selection = _selection(scope, match, parent) + manifest = _load_manifest(output, selection) + rows, discovery_failures = _discover(selection, verbose, quiet) + manifest["discovery_failures"] = discovery_failures + _merge_rows(manifest, rows) + _set_complete(manifest) + _write_manifest(output, manifest) + + for entry in manifest["datasets"]: + if entry["status"] in FINISHED_STATUSES: + continue + replacement = _inspect_dataset(entry, verbose, quiet) + entry.clear() + entry.update(replacement) + _set_complete(manifest) + _write_manifest(output, manifest) + + _set_complete(manifest) + _write_manifest(output, manifest) + return manifest + + +def _selection( + scope: Optional[str], match: Optional[str], parent: Optional[str] +) -> Dict[str, Any]: + """Normalize and validate the traversal boundary.""" + clean_scope = scope.strip() if scope else None + clean_parent = parent.strip() if parent else None + terms = sorted( + {term.strip().lower() for term in (match or "").split(",") if term.strip()} + ) + if match is not None and not terms: + raise ValueError("--match must contain at least one non-empty term.") + if clean_parent and not clean_scope: + raise ValueError("--parent requires SCOPE.") + if clean_parent and terms: + raise ValueError("Use either --parent or --match, not both.") + if not clean_scope and not terms: + raise ValueError("Give SCOPE, --match, or SCOPE with --parent.") + return {"scope": clean_scope, "match": terms, "parent": clean_parent} + + +def _discover( + selection: Dict[str, Any], verbose: int, quiet: bool +) -> Tuple[List[Dict[str, Any]], List[str]]: + """Discover terminal datasets within the selected boundary.""" + scope = selection["scope"] + parent = selection["parent"] + try: + if parent: + return functions._discover_descendants( + scope, [parent], verbose=verbose, quiet=quiet + ) + result = functions.discover_datasets( + scope=scope, + match=",".join(selection["match"]) or None, + recursive=True, + verbose=verbose, + quiet=quiet, + ) + except Exception as error: + return [], [f"discovery failed: {_error_text(error)}"] + if "error" in result: + return [], [f"discovery failed: {_error_text(result['error'])}"] + rows = result.get("results") + failures = result.get("failed") + if not isinstance(rows, list) or not isinstance(failures, list): + return [], ["discovery failed: unexpected result shape"] + return rows, [str(failure) for failure in failures] + + +def _inspect_dataset(entry: Dict[str, Any], verbose: int, quiet: bool) -> Dict[str, Any]: + """Fetch and normalize one dataset's replica URIs.""" + context = {key: entry[key] for key in ("scope", "dataset", "parent", "path")} + try: + response = functions.get_dataset_file_info( + entry["scope"], entry["dataset"], verbose=verbose, quiet=quiet + ) + replicas = _replicas(response) + except Exception as error: + return { + **context, + "status": "failed", + "replicas": [], + "error": _error_text(error), + } + return { + **context, + "status": "ready" if replicas else "empty", + "replicas": replicas, + } + + +def _replicas(response: Any) -> List[Dict[str, str]]: + """Normalize a Datatrail file response.""" + if not isinstance(response, dict): + raise ValueError("Datatrail returned an unexpected file response.") + if "error" in response: + raise ValueError(_error_text(response["error"])) + locations = response.get("file_replica_locations") + if not isinstance(locations, dict): + raise ValueError("Datatrail file response has no replica locations.") + + replicas: List[Dict[str, str]] = [] + for storage_element in sorted(locations): + uris = locations[storage_element] + if not isinstance(storage_element, str) or not storage_element.strip(): + raise ValueError("Datatrail returned an invalid storage element.") + if not isinstance(uris, list) or any( + not isinstance(uri, str) or not uri.strip() for uri in uris + ): + raise ValueError( + f"Datatrail returned invalid replica URIs for {storage_element}." + ) + replicas.extend( + {"storage_element": storage_element, "uri": uri} for uri in sorted(set(uris)) + ) + return replicas + + +def _load_manifest(path: Path, selection: Dict[str, Any]) -> Dict[str, Any]: + """Load a compatible manifest or create a new one.""" + if not path.exists(): + return { + "schema": SCHEMA, + "selection": selection, + "complete": False, + "discovery_failures": [], + "datasets": [], + } + try: + manifest = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + raise ValueError(f"Cannot read inventory manifest {path}: {error}") from error + if not isinstance(manifest, dict) or manifest.get("schema") != SCHEMA: + raise ValueError(f"Unsupported inventory manifest schema in {path}.") + if manifest.get("selection") != selection: + raise ValueError(f"Inventory selection does not match {path}.") + datasets = manifest.get("datasets") + if not isinstance(datasets, list): + raise ValueError(f"Inventory manifest datasets are invalid in {path}.") + _validate_entries(datasets, path) + manifest["discovery_failures"] = [] + manifest["complete"] = False + return manifest + + +def _validate_entries(entries: List[Any], path: Path) -> None: + """Validate resumable dataset entries.""" + seen = set() + for entry in entries: + if not isinstance(entry, dict): + raise ValueError(f"Inventory manifest contains an invalid entry in {path}.") + key = (entry.get("scope"), entry.get("dataset")) + if ( + not all(isinstance(value, str) and value for value in key) + or key in seen + or not _valid_entry(entry) + ): + raise ValueError(f"Inventory manifest contains an invalid entry in {path}.") + seen.add(key) + + +def _valid_entry(entry: Dict[str, Any]) -> bool: + """Check one saved dataset entry.""" + dataset = entry["dataset"] + parent = entry.get("parent") + path = entry.get("path") + replicas = entry.get("replicas") + status = entry.get("status") + if ( + status not in ENTRY_STATUSES + or not isinstance(path, list) + or not path + or any(not isinstance(part, str) or not part for part in path) + or path[-1] != dataset + or (parent is not None and (not isinstance(parent, str) or not parent)) + or parent != (path[-2] if len(path) > 1 else None) + or not isinstance(replicas, list) + or any(not _valid_replica(replica) for replica in replicas) + ): + return False + if status == "ready": + return bool(replicas) + if replicas: + return False + return status != "failed" or ( + isinstance(entry.get("error"), str) and bool(entry["error"]) + ) + + +def _valid_replica(replica: Any) -> bool: + """Check one saved replica row.""" + return ( + isinstance(replica, dict) + and isinstance(replica.get("storage_element"), str) + and bool(replica["storage_element"]) + and isinstance(replica.get("uri"), str) + and bool(replica["uri"]) + ) + + +def _merge_rows(manifest: Dict[str, Any], rows: List[Dict[str, Any]]) -> None: + """Add newly discovered datasets without resetting finished entries.""" + by_key = { + (entry["scope"], entry["dataset"]): entry for entry in manifest["datasets"] + } + for row in rows: + context = _row_context(row) + key = (context["scope"], context["dataset"]) + if key in by_key: + by_key[key]["parent"] = context["parent"] + by_key[key]["path"] = context["path"] + continue + entry = {**context, "status": "pending", "replicas": []} + manifest["datasets"].append(entry) + by_key[key] = entry + _sort_entries(manifest) + + +def _row_context(row: Dict[str, Any]) -> Dict[str, Any]: + """Validate a recursive discovery row.""" + if not isinstance(row, dict): + raise ValueError("Recursive discovery returned an invalid dataset row.") + scope = row.get("scope") + dataset = row.get("dataset") + parent = row.get("parent") + path = row.get("path") + if ( + not isinstance(scope, str) + or not scope + or not isinstance(dataset, str) + or not dataset + or (parent is not None and (not isinstance(parent, str) or not parent)) + or not isinstance(path, list) + or not path + or any(not isinstance(part, str) or not part for part in path) + or path[-1] != dataset + or parent != (path[-2] if len(path) > 1 else None) + ): + raise ValueError("Recursive discovery returned an invalid dataset row.") + return {"scope": scope, "dataset": dataset, "parent": parent, "path": path} + + +def _set_complete(manifest: Dict[str, Any]) -> None: + """Update the manifest completion flag.""" + manifest["complete"] = not manifest["discovery_failures"] and all( + entry["status"] in FINISHED_STATUSES for entry in manifest["datasets"] + ) + + +def _sort_entries(manifest: Dict[str, Any]) -> None: + """Keep manifest output deterministic.""" + manifest["datasets"].sort( + key=lambda entry: ( + entry["scope"], + tuple(entry["path"]), + entry["dataset"], + ) + ) + + +def _write_manifest(path: Path, manifest: Dict[str, Any]) -> None: + """Atomically write the manifest beside its temporary file.""" + path.parent.mkdir(parents=True, exist_ok=True) + _sort_entries(manifest) + descriptor, temporary = tempfile.mkstemp( + dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp" + ) + try: + with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as handle: + json.dump(manifest, handle, indent=2) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + except Exception: + try: + os.unlink(temporary) + except FileNotFoundError: + pass + raise + + +def _error_text(error: Any) -> str: + """Return a useful error string.""" + text = str(error).strip() + return text if text else type(error).__name__ diff --git a/mkdocs.yml b/mkdocs.yml index 53fe8aa..e442e9f 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -10,6 +10,7 @@ nav: - Initialise: initialising.md - Commands: - clear: clear.md + - inventory: inventory.md - list: list.md - ps: ps.md - pull: pull.md diff --git a/tests/test_inventory.py b/tests/test_inventory.py new file mode 100644 index 0000000..ee36d97 --- /dev/null +++ b/tests/test_inventory.py @@ -0,0 +1,288 @@ +"""Tests for durable inventory manifests.""" + +import json +from pathlib import Path +from typing import Any, Dict, List + +import pytest +from click.testing import CliRunner + +from dtcli import inventory as inventory_command +from dtcli.cli import cli as datatrail + + +def _row(dataset: str, path: List[str]) -> Dict[str, Any]: + """Build a recursive discovery row.""" + return { + "scope": "test.scope", + "dataset": dataset, + "parent": path[-2] if len(path) > 1 else None, + "path": path, + } + + +def test_inventory_builds_atomic_manifest(tmp_path: Path, monkeypatch) -> None: + """Test deterministic replicas and per-dataset checkpoints.""" + output = tmp_path / "inventory.json" + calls = [] + writes = [] + + def discover(**kwargs): + assert kwargs["match"] == "baseband,classified" + return { + "results": [ + _row("ready", ["root", "ready"]), + _row("empty", ["root", "empty"]), + ], + "failed": [], + } + + def file_info(scope, dataset, verbose=0, quiet=False): + calls.append(dataset) + if dataset == "empty": + return {"file_replica_locations": {}} + return { + "file_replica_locations": { + "minoc": ["cadc:file-b", "cadc:file-a", "cadc:file-a"], + "archive": ["archive:file"], + } + } + + original_write = inventory_command._write_manifest + + def record_write(path, manifest): + original_write(path, manifest) + writes.append(json.loads(json.dumps(manifest))) + + monkeypatch.setattr(inventory_command.functions, "discover_datasets", discover) + monkeypatch.setattr(inventory_command.functions, "get_dataset_file_info", file_info) + monkeypatch.setattr(inventory_command, "_write_manifest", record_write) + + manifest = inventory_command.build_inventory( + scope=None, + match="classified, BASEBAND", + parent=None, + output=output, + ) + + assert manifest["schema"] == "datatrail.inventory/v1" + assert manifest["selection"] == { + "scope": None, + "match": ["baseband", "classified"], + "parent": None, + } + assert manifest["complete"] is True + assert [entry["dataset"] for entry in manifest["datasets"]] == [ + "empty", + "ready", + ] + assert manifest["datasets"][0]["status"] == "empty" + assert manifest["datasets"][1]["replicas"] == [ + {"storage_element": "archive", "uri": "archive:file"}, + {"storage_element": "minoc", "uri": "cadc:file-a"}, + {"storage_element": "minoc", "uri": "cadc:file-b"}, + ] + assert calls == ["empty", "ready"] + assert len(writes) == 4 + assert [entry["status"] for entry in writes[0]["datasets"]] == [ + "pending", + "pending", + ] + assert [entry["status"] for entry in writes[1]["datasets"]] == [ + "empty", + "pending", + ] + assert json.loads(output.read_text()) == manifest + assert list(tmp_path.glob(".inventory.json.*.tmp")) == [] + + +def test_inventory_resume_retries_only_failures(tmp_path: Path, monkeypatch) -> None: + """Test a rerun reuses finished entries and retries failures.""" + output = tmp_path / "inventory.json" + attempts = {"ready": 0, "empty": 0, "flaky": 0} + + def discover(**kwargs): + return { + "results": [ + _row("ready", ["root", "ready"]), + _row("empty", ["root", "empty"]), + _row("flaky", ["root", "flaky"]), + ], + "failed": [], + } + + def file_info(scope, dataset, verbose=0, quiet=False): + attempts[dataset] += 1 + if dataset == "empty": + return {"file_replica_locations": {}} + if dataset == "flaky" and attempts[dataset] == 1: + return {"error": "service unavailable"} + return {"file_replica_locations": {"minoc": [f"cadc:{dataset}"]}} + + monkeypatch.setattr(inventory_command.functions, "discover_datasets", discover) + monkeypatch.setattr(inventory_command.functions, "get_dataset_file_info", file_info) + + first = inventory_command.build_inventory( + scope="test.scope", match=None, parent=None, output=output + ) + assert first["complete"] is False + assert ( + next(entry for entry in first["datasets"] if entry["dataset"] == "flaky")[ + "status" + ] + == "failed" + ) + + second = inventory_command.build_inventory( + scope="test.scope", match=None, parent=None, output=output + ) + assert second["complete"] is True + assert attempts == {"ready": 1, "empty": 1, "flaky": 2} + flaky = next(entry for entry in second["datasets"] if entry["dataset"] == "flaky") + assert flaky["status"] == "ready" + assert "error" not in flaky + + +def test_inventory_parent_starts_at_subtree(tmp_path: Path, monkeypatch) -> None: + """Test a named parent starts recursive traversal directly.""" + output = tmp_path / "inventory.json" + + def descendants(scope, roots, verbose=0, quiet=False): + assert scope == "test.scope" + assert roots == ["root"] + return [_row("leaf", ["root", "leaf"])], [] + + monkeypatch.setattr( + inventory_command.functions, "_discover_descendants", descendants + ) + monkeypatch.setattr( + inventory_command.functions, + "get_dataset_file_info", + lambda *args, **kwargs: {"file_replica_locations": {}}, + ) + + manifest = inventory_command.build_inventory( + scope="test.scope", match=None, parent="root", output=output + ) + assert manifest["complete"] is True + assert manifest["selection"]["parent"] == "root" + assert manifest["datasets"][0]["path"] == ["root", "leaf"] + assert manifest["datasets"][0]["status"] == "empty" + + +def test_inventory_default_exit_fails_on_discovery_gap( + tmp_path: Path, monkeypatch +) -> None: + """Test incomplete discovery needs an explicit successful-exit option.""" + output = tmp_path / "inventory.json" + + monkeypatch.setattr( + inventory_command.functions, + "discover_datasets", + lambda **kwargs: { + "results": [_row("leaf", ["root", "leaf"])], + "failed": ["children of test.scope root / offline"], + }, + ) + monkeypatch.setattr( + inventory_command.functions, + "get_dataset_file_info", + lambda *args, **kwargs: {"file_replica_locations": {"minoc": ["cadc:leaf"]}}, + ) + runner = CliRunner() + result = runner.invoke( + datatrail, ["inventory", "test.scope", "--output", str(output)] + ) + assert result.exit_code == 1 + assert "Inventory incomplete" in result.output + assert json.loads(output.read_text())["complete"] is False + + allowed = runner.invoke( + datatrail, + [ + "inventory", + "test.scope", + "--output", + str(output), + "--allow-incomplete", + ], + ) + assert allowed.exit_code == 0 + assert "Inventory incomplete" in allowed.output + + +def test_inventory_default_exit_fails_on_file_query(tmp_path: Path, monkeypatch) -> None: + """Test a failed file query makes the command fail by default.""" + output = tmp_path / "inventory.json" + monkeypatch.setattr( + inventory_command.functions, + "discover_datasets", + lambda **kwargs: { + "results": [_row("leaf", ["root", "leaf"])], + "failed": [], + }, + ) + monkeypatch.setattr( + inventory_command.functions, + "get_dataset_file_info", + lambda *args, **kwargs: {"error": "service unavailable"}, + ) + result = CliRunner().invoke( + datatrail, ["inventory", "test.scope", "--output", str(output)] + ) + assert result.exit_code == 1 + manifest = json.loads(output.read_text()) + assert manifest["complete"] is False + assert manifest["datasets"][0]["status"] == "failed" + assert manifest["datasets"][0]["error"] == "service unavailable" + + +@pytest.mark.parametrize( + "arguments,message", + [ + ([], "Give SCOPE"), + (["--parent", "root"], "--parent requires SCOPE"), + ( + ["test.scope", "--parent", "root", "--match", "gain"], + "either --parent or --match", + ), + (["test.scope", "--match", " , "], "--match must contain"), + ], +) +def test_inventory_rejects_unbounded_or_ambiguous_selection( + tmp_path: Path, arguments: List[str], message: str +) -> None: + """Test inventory selection always has one clear boundary.""" + output = tmp_path / "inventory.json" + runner = CliRunner() + result = runner.invoke(datatrail, ["inventory", *arguments, "--output", str(output)]) + assert result.exit_code == 1 + assert message in result.output + assert not output.exists() + + +def test_inventory_refuses_different_selection(tmp_path: Path, monkeypatch) -> None: + """Test a manifest cannot be resumed with a different selection.""" + output = tmp_path / "inventory.json" + monkeypatch.setattr( + inventory_command.functions, + "discover_datasets", + lambda **kwargs: {"results": [], "failed": []}, + ) + inventory_command.build_inventory( + scope="first.scope", match=None, parent=None, output=output + ) + with pytest.raises(ValueError, match="selection does not match"): + inventory_command.build_inventory( + scope="second.scope", match=None, parent=None, output=output + ) + + +def test_inventory_help() -> None: + """Test the inventory command is registered with its main options.""" + result = CliRunner().invoke(datatrail, ["inventory", "--help"]) + assert result.exit_code == 0 + assert "--match" in result.output + assert "--parent" in result.output + assert "--output" in result.output + assert "--allow-incomplete" in result.output