diff --git a/.gitignore b/.gitignore index 649f1e0..6233049 100644 --- a/.gitignore +++ b/.gitignore @@ -29,5 +29,18 @@ src/*.egg-info/ # User/Agent Documentation JOBS_DOCUMENTATION.md AGENTS.md +reports/ + +# AI assistant folders / files .antigravity/ -reports/ \ No newline at end of file +.claude/ +.cursor/ +.windsurf/ +.aider* +CLAUDE.md +GEMINI.md +.clinerules/ + +# Demo recordings (asciinema casts and rendered GIFs) +*.cast +*.gif \ No newline at end of file diff --git a/README.md b/README.md index 5c5a09f..d5c13a8 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,45 @@ Monitor the status of asynchronous jobs: dmtri track ``` +### Fix Inconsistencies +Automatically repair the inconsistencies found in an [`eida-consistency`](https://github.com/EIDA/eida-consistency) report. A report lists streams where the **availability** view and **dataselect** disagree; `dmtri fix` reads that report and, for each affected stream, re-runs the WFCatalog collector and rebuilds the availability view for the exact time window. + +Point it at a report file or URL: + +```bash +dmtri fix https://eida-oculus.orfeus-eu.org/consistency/NOA/2026/NOA_2026-06-07_140510.json +``` + +Fix only specific entries from the report (by their index): + +```bash +dmtri fix report.json --index 2 --index 16 +``` + +After a fix, the availability service keeps cached answers for ~20 minutes, so a stream may still look unfixed for a short while. Re-check once the cache expires — this only verifies, it changes nothing: + +```bash +dmtri fix --verify-only report.json +``` + +#### Prerequisite: the `eida-consistency` CLI + +`dmtri fix` reads the report through [`eida-consistency`](https://github.com/EIDA/eida-consistency) (version **0.5.1 or newer**, for its `explore --json` output). If it isn't already on the machine, install it any one of these ways: + +```bash +uv tool install eida-consistency # recommended — puts it on your PATH +pipx install eida-consistency +pip install eida-consistency +``` + +You don't strictly have to install it: if `uv` is present, `dmtri fix` will run it on demand via `uvx eida-consistency`. And if it's installed somewhere off your `PATH`, point dmtri at it: + +```bash +export DMTRI_EIDA_CONSISTENCY='/full/path/to/eida-consistency' +``` + +If it's missing (or too old), `dmtri fix` stops before touching anything and prints the exact install/upgrade command to run. + --- ## Customization diff --git a/pyproject.toml b/pyproject.toml index c845cb5..37c0ee0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] -name = "eida-dmtri" -version = "0.1.0" +name = "dmtri" +version = "0.1.1" description = "CLI tool for triggering datacenter metadata updates" authors = [{ name = "Nikos Sokos", email = "nsokos@noa.com" }] dependencies = [ diff --git a/src/dmtri/cli.py b/src/dmtri/cli.py index d4886a7..d4026a9 100644 --- a/src/dmtri/cli.py +++ b/src/dmtri/cli.py @@ -53,6 +53,14 @@ def main(): subparsers.add_parser("refresh", parents=[shared], help="Refresh data/metadata using configured playbooks.") subparsers.add_parser("clean", parents=[shared], help="Clean outdated data/metadata using configured playbooks.") + + fix_parser = subparsers.add_parser("fix", help="Automatically repair inconsistencies from an eida-consistency report.") + fix_parser.add_argument("report", help="Consistency report: URL or local JSON path") + fix_parser.add_argument("--verify-only", action="store_true", help="Only re-check status (no rebuilding)") + fix_parser.add_argument("--days", type=int, default=None, help="Max days to explore boundaries (passthrough to eida-consistency)") + fix_parser.add_argument("--index", type=int, action="append", dest="index", help="Restrict to specific report index(es); repeatable") + fix_parser.add_argument("--no-confirm", action="store_true", help="Skip confirmation prompt before executing") + fix_parser.add_argument("--debug", action="store_true", help="Show verbose output from Ansible") subparsers.add_parser("track", help="Track job status via tracking playbook.") doctor_parser = subparsers.add_parser("doctor", help="Check SSH connectivity to all inventory hosts") doctor_parser.add_argument("-v", "--verbose", action="store_true", help="Show full Ansible output per host") @@ -122,6 +130,20 @@ def main(): for pb in playbooks: run_hook(pb, vars_to_pass, inventory=inventory_path) + elif args.command == "fix": + from dmtri.fix import run_fix, run_verify_only + try: + if args.verify_only: + rc = run_verify_only(args.report, inventory=inventory_path, + days=args.days, indices=args.index) + else: + rc = run_fix(args.report, inventory=inventory_path, days=args.days, + indices=args.index, no_confirm=args.no_confirm, debug=args.debug) + except RuntimeError as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(2) + sys.exit(rc) + elif args.command == "track": playbooks = COMMAND_PLAYBOOKS.get("track", []) if not playbooks: diff --git a/src/dmtri/data/playbooks/fix/availability_rebuild.yml b/src/dmtri/data/playbooks/fix/availability_rebuild.yml new file mode 100644 index 0000000..7a755f6 --- /dev/null +++ b/src/dmtri/data/playbooks/fix/availability_rebuild.yml @@ -0,0 +1,89 @@ +- name: "Rebuild availability materialized view for a specific NSLC + date range" + hosts: ws_availability + gather_facts: false + + # Expected extra-vars (one stream, one range — dmtri fix calls this once per fix): + # net, sta, cha : exact codes + # loc : location code; pass "--" for the empty location + # start, end : YYYY-MM-DD. `end` is exclusive at the cacher (te <= end), + # so dmtri passes the day AFTER the last broken day. + # + # Unlike availability_refresh.yml (a global `docker restart` = last-24h rebuild), + # this triggers the in-container `avail-rebuild` console script, which rebuilds + # only the given stream/range from WFCatalog (idempotent $merge). + + tasks: + + # command + argv (not shell): values are passed as separate execve arguments, + # so even if a report-derived value contained shell metacharacters it cannot be + # interpreted by a shell. dmtri also validates these against strict allow-lists + # before they ever reach here (see fix._validate_fix). + - name: "Run scoped availability rebuild (docker exec avail-rebuild)" + command: + argv: + - docker + - exec + - fdsnws-availability-cacher + - avail-rebuild + - --net + - "{{ net }}" + - --sta + - "{{ sta }}" + - "--loc={{ loc }}" + - --cha + - "{{ cha }}" + - --start + - "{{ start }}" + - --end + - "{{ end }}" + register: rebuild_result + + - name: "DISPLAY REBUILD STATUS" + debug: + msg: + - "========================================" + - " Availability Rebuild — {{ net }}.{{ sta }}.{{ loc }}.{{ cha }}" + - "========================================" + - " Range : {{ start }} -> {{ end }} (end exclusive)" + - " Status : {{ 'SUCCESS' if rebuild_result.rc == 0 else 'FAILED' }}" + - " Exit Code: {{ rebuild_result.rc }}" + - "========================================" + + # --------------------------------------------------------------- + # JOB TRACKING + # --------------------------------------------------------------- + + - name: Ensure ~/.dmtri/jobs/{{ inventory_hostname }} directory exists + file: + path: "~/.dmtri/jobs/{{ inventory_hostname }}" + state: directory + mode: '0755' + + - name: "Load existing job list (if exists)" + slurp: + src: "~/.dmtri/jobs/{{ inventory_hostname }}/availability.json" + register: job_file + ignore_errors: true + + - name: "Parse job list or default to []" + set_fact: + job_list: >- + {{ (job_file.content | b64decode | from_json) if job_file.content is defined else [] }} + + - name: Define new job entry + set_fact: + new_job: { + "type": "availability_rebuild", + "job_id": "rebuild-{{ lookup('pipe', 'date +%s') }}", + "host": "{{ inventory_hostname }}", + "stream": "{{ net }}.{{ sta }}.{{ loc }}.{{ cha }}", + "start": "{{ start }}", + "end": "{{ end }}", + "rc": "{{ rebuild_result.rc | default('unknown') }}", + "created_at": "{{ lookup('pipe', 'date -u +%Y-%m-%dT%H:%M:%SZ') }}" + } + + - name: "Save updated job list" + copy: + content: "{{ (job_list + [new_job]) | to_nice_json }}" + dest: "~/.dmtri/jobs/{{ inventory_hostname }}/availability.json" diff --git a/src/dmtri/fix.py b/src/dmtri/fix.py new file mode 100644 index 0000000..2fb5e68 --- /dev/null +++ b/src/dmtri/fix.py @@ -0,0 +1,276 @@ +"""`dmtri fix` — automated repair of inconsistencies from an eida-consistency report. + +Flow: + 1. Shell out to `eida-consistency explore --json ` to get structured fixes + (NSLC + true window + direction + status). + 2. Keep the actionable, refresh-direction fixes (v1 is refresh-only; the clean + direction — phantom availability records — needs deletion logic we don't have yet). + 3. For each fix, run the repair chain on the existing playbook/SSH machinery: + wfcatalog_refresh (NSLC + range) -> availability_rebuild (docker exec avail-rebuild) + continue-on-error: a failed stream is recorded and skipped, not fatal; the + availability rebuild is skipped for a stream whose wfcatalog step failed. + 4. Report, and remind that the live /availability service caches answers for + ~20 min, so confirmation should come from `dmtri fix --verify-only` later. + +The availability rebuild reads from WFCatalog and is idempotent, so re-runs are safe. +""" +from __future__ import annotations # keep modern type hints lazy (dmtri floor is 3.8) + +import json +import os +import re +import shutil +import subprocess +import sys +from datetime import date, timedelta + +from dmtri.hooks.hooks import run_hook +from dmtri.paths import PLAYBOOK_WFCATALOG_REFRESH, PLAYBOOK_AVAILABILITY_REBUILD + +# Major schema version of `explore --json` we understand. Minor bumps are additive. +SUPPORTED_SCHEMA_MAJOR = "1" + + +# --------------------------------------------------------------------------- # +# Talking to eida-consistency +# --------------------------------------------------------------------------- # +def _explore_base_cmd() -> list[str]: + """Locate the eida-consistency CLI. + + Order: $DMTRI_EIDA_CONSISTENCY override -> `eida-consistency` on PATH -> + `uvx eida-consistency`. Raises a helpful error if none is available. + """ + override = os.environ.get("DMTRI_EIDA_CONSISTENCY") + if override: + return override.split() + if shutil.which("eida-consistency"): + return ["eida-consistency"] + if shutil.which("uvx"): + return ["uvx", "eida-consistency"] + raise RuntimeError( + "`dmtri fix` needs the eida-consistency CLI, which was not found on this machine.\n" + " Install it (pick one):\n" + " uv tool install eida-consistency # recommended — puts it on your PATH\n" + " pipx install eida-consistency\n" + " pip install eida-consistency\n" + " Or just make `uv` available and dmtri will run it on demand via `uvx` (no install).\n" + " Already installed elsewhere? Point dmtri at it:\n" + " export DMTRI_EIDA_CONSISTENCY='/full/path/to/eida-consistency'" + ) + + +def get_fixes(report: str, days: int | None = None, + indices: list[int] | None = None) -> dict: + """Run `eida-consistency explore --json` and return the parsed result dict. + + Raises RuntimeError on a non-zero exit, unparseable output, or an + incompatible schema major version. + """ + cmd = _explore_base_cmd() + ["explore", "--json", report] + if days is not None: + cmd += ["--days", str(days)] + for i in indices or []: + cmd += ["--index", str(i)] + + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode != 0: + combined = (proc.stderr or "") + (proc.stdout or "") + # An eida-consistency too old to know `--json` rejects the flag with a + # "No such option: --json" usage error. Turn that into an upgrade hint + # instead of a cryptic argparse/click tail. + if "--json" in combined and "no such option" in combined.lower(): + raise RuntimeError( + "eida-consistency is installed but too old: it does not support " + "`explore --json` (needed by `dmtri fix`, available since 0.5.1).\n" + " Upgrade it (pick one):\n" + " uv tool upgrade eida-consistency\n" + " pipx upgrade eida-consistency\n" + " pip install -U eida-consistency" + ) + tail = combined.strip().splitlines()[-5:] + raise RuntimeError( + "eida-consistency explore failed (exit " + f"{proc.returncode}):\n " + "\n ".join(tail) + ) + + out = proc.stdout + try: + # stdout is pure JSON (eida-consistency logs/progress go to stderr); fall + # back to the first '{' in case anything leaks in front of it. + result = json.loads(out[out.index("{"):]) if "{" in out else json.loads(out) + except (ValueError, json.JSONDecodeError) as e: + raise RuntimeError(f"could not parse explore --json output: {e}") + + version = str(result.get("schema_version", "")) + if version.split(".")[0] != SUPPORTED_SCHEMA_MAJOR: + raise RuntimeError( + f"explore --json schema_version {version!r} is incompatible with this " + f"dmtri (expects {SUPPORTED_SCHEMA_MAJOR}.x). Update dmtri or eida-consistency." + ) + return result + + +# --------------------------------------------------------------------------- # +# Fix -> playbook variables +# --------------------------------------------------------------------------- # +# Fix values come from a report fetched over a URL (untrusted input) and are +# forwarded into Ansible playbooks that shell out (docker exec / find). Validate +# them at the source against strict allow-lists so nothing shell-special can reach +# a command line. FDSN codes are short alphanumerics (+ wildcards/-/_); dates ISO. +_CODE_RE = re.compile(r"^[A-Za-z0-9*?_-]{1,8}$") # network / station / channel +_LOC_RE = re.compile(r"^[A-Za-z0-9*?_-]{0,8}$") # location (may be empty) +_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") + + +def _validate_fix(f: dict) -> None: + """Raise ValueError if any field would be unsafe to pass to a playbook.""" + for k in ("network", "station", "channel"): + if not _CODE_RE.match(str(f.get(k, ""))): + raise ValueError(f"unsafe {k}={f.get(k)!r}") + if not _LOC_RE.match(str(f.get("location", ""))): + raise ValueError(f"unsafe location={f.get('location')!r}") + for k in ("start", "end"): + if not _DATE_RE.match(str(f.get(k, ""))): + raise ValueError(f"unsafe {k}={f.get(k)!r}") + + +def _end_exclusive(end: str) -> str: + """avail-rebuild's --end is exclusive (te <= end). explore reports the last + *broken* day inclusively, so pass the day after it.""" + y, m, d = (int(x) for x in end.split("-")) + return (date(y, m, d) + timedelta(days=1)).isoformat() + + +def fix_to_vars(fix: dict) -> tuple[dict, dict]: + """Map one fix record to (wfcatalog_refresh vars, availability_rebuild vars).""" + net, sta, cha = fix["network"], fix["station"], fix["channel"] + loc = fix["location"] + start, end = fix["start"], fix["end"] + + wfcatalog_vars = { + "network": [net], + "station": [sta], + # wfcatalog discovery is largely location-agnostic; empty location -> '*'. + "location": [loc or "*"], + "channel": [cha], + "starttime": f"{start}T00:00:00", + "endtime": f"{end}T23:59:59", + "type": ["data"], + } + rebuild_vars = { + "net": net, + "sta": sta, + # avail-rebuild spells the empty location as '--'. + "loc": "--" if loc == "" else loc, + "cha": cha, + "start": start, + "end": _end_exclusive(end), + } + return wfcatalog_vars, rebuild_vars + + +def _dedup(fixes: list[dict]) -> list[dict]: + seen, out = set(), [] + for f in fixes: + key = (f["network"], f["station"], f["location"], f["channel"], f["start"], f["end"]) + if key not in seen: + seen.add(key) + out.append(f) + return out + + +def _label(f: dict) -> str: + return f"{f['network']}.{f['station']}.{f['location']}.{f['channel']} {f['start']} -> {f['end']}" + + +# --------------------------------------------------------------------------- # +# Commands +# --------------------------------------------------------------------------- # +def _partition(result: dict) -> dict: + fixes = result.get("fixes", []) + actionable = [f for f in fixes if f.get("status") == "actionable"] + return { + "refresh": _dedup([f for f in actionable if f.get("direction") == "refresh"]), + "clean": [f for f in actionable if f.get("direction") == "clean"], + "fixed": [f for f in fixes if f.get("status") == "fixed"], + "transient": [f for f in fixes if f.get("status") == "transient"], + "node": result.get("node", "?"), + } + + +def run_verify_only(report: str, inventory, days=None, indices=None) -> int: + """Re-probe via explore --json and report which rows are now fixed. No playbooks. + + Returns 0 if nothing remains actionable (all healed), else 1. + """ + parts = _partition(get_fixes(report, days, indices)) + still = parts["refresh"] + parts["clean"] + print(f"\nVerify ({parts['node']}): " + f"{len(parts['fixed'])} fixed, {len(still)} still inconsistent, " + f"{len(parts['transient'])} transient.") + for f in parts["fixed"]: + print(f" ✓ fixed {f['network']}.{f['station']}.{f['location']}.{f['channel']}") + for f in still: + print(f" ✗ still {_label(f)}") + return 0 if not still else 1 + + +def run_fix(report: str, inventory, days=None, indices=None, + no_confirm=False, debug=False) -> int: + """Drive the repair chain for every actionable refresh-direction fix.""" + parts = _partition(get_fixes(report, days, indices)) + targets = parts["refresh"] + + if parts["clean"]: + print(f"Note: {len(parts['clean'])} clean-direction inconsistency(ies) " + "skipped (not supported yet — refresh-only).") + if parts["fixed"]: + print(f"Note: {len(parts['fixed'])} row(s) already consistent — nothing to do.") + + if not targets: + print("No actionable refresh-direction fixes. Nothing to run.") + return 0 + + print(f"\ndmtri fix ({parts['node']}) — {len(targets)} stream(s) to repair:") + for f in targets: + print(f" - {_label(f)}") + print("\n Each: (1) wfcatalog_refresh on wf_catalogue, " + "(2) availability_rebuild on ws_availability.") + + if not no_confirm: + if input("\nProceed? (y/N): ").strip().lower() != "y": + print("Aborted by user.") + return 0 + + results = [] # (fix, ok, reason) + for f in targets: + try: + _validate_fix(f) # never forward untrusted/unsafe values to a playbook + except ValueError as e: + print(f"\n=== {_label(f)} ===\n REJECTED: {e}") + results.append((f, False, f"rejected ({e})")) + continue + wf_vars, rb_vars = fix_to_vars(f) + print(f"\n=== {_label(f)} ===") + rc1 = run_hook(PLAYBOOK_WFCATALOG_REFRESH, wf_vars, inventory=inventory, exit_on_error=False) + if rc1 != 0: + results.append((f, False, f"wfcatalog rc={rc1} (availability rebuild skipped)")) + continue + rc2 = run_hook(PLAYBOOK_AVAILABILITY_REBUILD, rb_vars, inventory=inventory, exit_on_error=False) + if rc2 != 0: + results.append((f, False, f"avail-rebuild rc={rc2}")) + else: + results.append((f, True, "")) + + ok = [r for r in results if r[1]] + bad = [r for r in results if not r[1]] + print("\n" + "=" * 50) + print(f" dmtri fix summary: {len(ok)} rebuilt, {len(bad)} failed") + print("=" * 50) + for f, _, reason in bad: + print(f" ✗ {_label(f)} -- {reason}") + if ok: + print("\nℹ The availability service caches answers for ~20 min, so these") + print(" may still show as 'not available' until the cache expires.") + print(f" Confirm later with: dmtri fix --verify-only ") + return 0 if not bad else 1 diff --git a/src/dmtri/hooks/hooks.py b/src/dmtri/hooks/hooks.py index 83fc72d..5ec9fd7 100644 --- a/src/dmtri/hooks/hooks.py +++ b/src/dmtri/hooks/hooks.py @@ -6,9 +6,15 @@ logger = logging.getLogger(__name__) -def run_hook(playbook_name: Path, vars: dict,inventory: Path = INVENTORY_FILE) -> None: +def run_hook(playbook_name: Path, vars: dict, inventory: Path = INVENTORY_FILE, + exit_on_error: bool = True) -> int: """ Run an Ansible playbook using ansible_runner.run_command, bypassing the project layout. + + Returns the playbook return code. By default (exit_on_error=True) it preserves + the historical behavior of refresh/clean — sys.exit(rc) on failure. Callers that + orchestrate a multi-step batch (e.g. `dmtri fix`) pass exit_on_error=False so one + failed stream does not abort the whole run; they inspect the returned rc instead. """ # Build the extra vars string manually @@ -40,6 +46,9 @@ def run_hook(playbook_name: Path, vars: dict,inventory: Path = INVENTORY_FILE) - if rc != 0: logger.error(f"Playbook {playbook_name} failed with return code {rc}") - sys.exit(rc) - + if exit_on_error: + sys.exit(rc) + return rc + logger.info(f"Playbook {playbook_name} completed successfully") + return rc diff --git a/src/dmtri/paths.py b/src/dmtri/paths.py index 4c15b8b..668c29d 100644 --- a/src/dmtri/paths.py +++ b/src/dmtri/paths.py @@ -86,6 +86,10 @@ def _resolve_playbook(relative_path: str) -> Path: PLAYBOOK_AVAILABILITY_REFRESH = _resolve_playbook("refresh/availability_refresh.yml") PLAYBOOK_METADATA_REFRESH = _resolve_playbook("refresh/seedpsd_metadata_refresh.yml") +# Consistency-repair (dmtri fix): reuses wfcatalog_refresh, plus a scoped +# availability rebuild (docker exec avail-rebuild) instead of the global restart. +PLAYBOOK_AVAILABILITY_REBUILD = _resolve_playbook("fix/availability_rebuild.yml") + PLAYBOOK_SEEDPSD_CLEAN = _resolve_playbook("clean/seedpsd_clean.yml") PLAYBOOK_WFCATALOG_CLEAN = _resolve_playbook("clean/wfcatalog_clean.yml") PLAYBOOK_AVAILABILITY_CLEAN = _resolve_playbook("clean/availability_clean.yml") diff --git a/src/dmtri/utils.py b/src/dmtri/utils.py index b7ddaa3..20d2d9b 100644 --- a/src/dmtri/utils.py +++ b/src/dmtri/utils.py @@ -9,7 +9,7 @@ def get_version(): try: # Try to get version from installed package metadata - return version("eida-dmtri") + return version("dmtri") except PackageNotFoundError: # Fallback for local development where the package might not be installed try: diff --git a/tests/test_fix.py b/tests/test_fix.py new file mode 100644 index 0000000..3b2ff85 --- /dev/null +++ b/tests/test_fix.py @@ -0,0 +1,232 @@ +import json +import subprocess +from types import SimpleNamespace + +import pytest + +import dmtri.fix as fix + + +# --------------------------------------------------------------------------- # +# get_fixes (subprocess + parsing) +# --------------------------------------------------------------------------- # +def _fake_proc(stdout="", stderr="", rc=0): + return SimpleNamespace(stdout=stdout, stderr=stderr, returncode=rc) + + +def _sample(node="NOA", fixes=None): + return {"schema_version": "1.0", "node": node, "report": "r", "fixes": fixes or []} + + +def test_get_fixes_parses_json(monkeypatch): + payload = _sample(fixes=[{"index": 1, "network": "XX", "station": "STA", + "location": "00", "channel": "BHZ", + "start": "2008-01-01", "end": "2008-02-01", + "direction": "refresh", "status": "actionable"}]) + monkeypatch.setattr(fix, "_explore_base_cmd", lambda: ["eida-consistency"]) + monkeypatch.setattr(subprocess, "run", + lambda *a, **k: _fake_proc(stdout=json.dumps(payload))) + result = fix.get_fixes("report.json") + assert result["node"] == "NOA" + assert result["fixes"][0]["channel"] == "BHZ" + + +def test_get_fixes_nonzero_exit_raises(monkeypatch): + monkeypatch.setattr(fix, "_explore_base_cmd", lambda: ["eida-consistency"]) + monkeypatch.setattr(subprocess, "run", + lambda *a, **k: _fake_proc(stderr="boom", rc=3)) + with pytest.raises(RuntimeError, match="explore failed"): + fix.get_fixes("report.json") + + +def test_get_fixes_too_old_gives_upgrade_hint(monkeypatch): + """An eida-consistency that rejects --json yields an upgrade message, not a raw tail.""" + monkeypatch.setattr(fix, "_explore_base_cmd", lambda: ["eida-consistency"]) + monkeypatch.setattr(subprocess, "run", + lambda *a, **k: _fake_proc(stderr="Error: No such option: --json", rc=2)) + with pytest.raises(RuntimeError, match="too old.*explore --json"): + fix.get_fixes("report.json") + + +def test_explore_base_cmd_missing_gives_install_hint(monkeypatch): + """When eida-consistency and uvx are both absent, explain how to install.""" + monkeypatch.delenv("DMTRI_EIDA_CONSISTENCY", raising=False) + monkeypatch.setattr(fix.shutil, "which", lambda name: None) + with pytest.raises(RuntimeError, match="uv tool install eida-consistency"): + fix._explore_base_cmd() + + +def test_explore_base_cmd_honors_override(monkeypatch): + monkeypatch.setenv("DMTRI_EIDA_CONSISTENCY", "/opt/ec/eida-consistency") + assert fix._explore_base_cmd() == ["/opt/ec/eida-consistency"] + + +def test_get_fixes_schema_mismatch_raises(monkeypatch): + payload = {"schema_version": "2.0", "node": "N", "fixes": []} + monkeypatch.setattr(fix, "_explore_base_cmd", lambda: ["eida-consistency"]) + monkeypatch.setattr(subprocess, "run", + lambda *a, **k: _fake_proc(stdout=json.dumps(payload))) + with pytest.raises(RuntimeError, match="schema_version"): + fix.get_fixes("report.json") + + +def test_get_fixes_tolerates_leading_noise(monkeypatch): + payload = _sample() + monkeypatch.setattr(fix, "_explore_base_cmd", lambda: ["eida-consistency"]) + monkeypatch.setattr(subprocess, "run", + lambda *a, **k: _fake_proc(stdout="warn\n" + json.dumps(payload))) + assert fix.get_fixes("report.json")["node"] == "NOA" + + +# --------------------------------------------------------------------------- # +# fix_to_vars (date boundary + location mapping) +# --------------------------------------------------------------------------- # +def test_fix_to_vars_empty_location_and_end_boundary(): + f = {"network": "IV", "station": "MILZ", "location": "", "channel": "HHZ", + "start": "2008-02-11", "end": "2008-05-02", + "direction": "refresh", "status": "actionable"} + wf, rb = fix.fix_to_vars(f) + # avail-rebuild: empty location -> '--', end is exclusive (last day + 1) + assert rb["loc"] == "--" + assert rb["start"] == "2008-02-11" + assert rb["end"] == "2008-05-03" + # wfcatalog: empty location -> '*', covers full end day, data type + assert wf["location"] == ["*"] + assert wf["starttime"] == "2008-02-11T00:00:00" + assert wf["endtime"] == "2008-05-02T23:59:59" + assert wf["type"] == ["data"] + + +def test_fix_to_vars_explicit_location(): + f = {"network": "IV", "station": "AQU", "location": "00", "channel": "BHZ", + "start": "2008-03-01", "end": "2008-03-01", + "direction": "refresh", "status": "actionable"} + wf, rb = fix.fix_to_vars(f) + assert rb["loc"] == "00" + assert rb["end"] == "2008-03-02" + assert wf["location"] == ["00"] + + +# --------------------------------------------------------------------------- # +# _partition (filtering + dedup) +# --------------------------------------------------------------------------- # +def test_partition_splits_and_dedups(): + base = {"network": "IV", "station": "MILZ", "location": "", "channel": "HHZ", + "start": "2008-01-01", "end": "2008-02-01"} + result = _result_with([ + {**base, "direction": "refresh", "status": "actionable"}, + {**base, "direction": "refresh", "status": "actionable"}, # dup + {**base, "channel": "HHN", "direction": "clean", "status": "actionable"}, + {**base, "channel": "HHE", "direction": "refresh", "status": "fixed"}, + {**base, "channel": "BHZ", "direction": "refresh", "status": "transient"}, + ]) + parts = fix._partition(result) + assert len(parts["refresh"]) == 1 # deduped + assert len(parts["clean"]) == 1 + assert len(parts["fixed"]) == 1 + assert len(parts["transient"]) == 1 + + +def _result_with(fixes): + return {"schema_version": "1.0", "node": "NOA", "fixes": fixes} + + +# --------------------------------------------------------------------------- # +# run_verify_only +# --------------------------------------------------------------------------- # +def test_verify_only_all_fixed_returns_0(monkeypatch, capsys): + monkeypatch.setattr(fix, "get_fixes", lambda *a, **k: _result_with([ + {"network": "IV", "station": "MILZ", "location": "", "channel": "HHZ", + "start": "2008-01-01", "end": "2008-02-01", + "direction": "refresh", "status": "fixed"}, + ])) + assert fix.run_verify_only("r", inventory=None) == 0 + assert "1 fixed" in capsys.readouterr().out + + +def test_verify_only_still_inconsistent_returns_1(monkeypatch): + monkeypatch.setattr(fix, "get_fixes", lambda *a, **k: _result_with([ + {"network": "IV", "station": "MILZ", "location": "", "channel": "HHZ", + "start": "2008-01-01", "end": "2008-02-01", + "direction": "refresh", "status": "actionable"}, + ])) + assert fix.run_verify_only("r", inventory=None) == 1 + + +# --------------------------------------------------------------------------- # +# run_fix (chain ordering + continue-on-error) +# --------------------------------------------------------------------------- # +def test_run_fix_runs_chain_in_order(monkeypatch): + f = {"network": "IV", "station": "MILZ", "location": "", "channel": "HHZ", + "start": "2008-01-01", "end": "2008-02-01", + "direction": "refresh", "status": "actionable"} + monkeypatch.setattr(fix, "get_fixes", lambda *a, **k: _result_with([f])) + calls = [] + + def fake_hook(pb, vars, inventory=None, exit_on_error=True): + calls.append(pb) + return 0 + monkeypatch.setattr(fix, "run_hook", fake_hook) + + rc = fix.run_fix("r", inventory=None, no_confirm=True) + assert rc == 0 + assert calls == [fix.PLAYBOOK_WFCATALOG_REFRESH, fix.PLAYBOOK_AVAILABILITY_REBUILD] + + +def test_run_fix_skips_rebuild_when_wfcatalog_fails(monkeypatch): + f = {"network": "IV", "station": "MILZ", "location": "", "channel": "HHZ", + "start": "2008-01-01", "end": "2008-02-01", + "direction": "refresh", "status": "actionable"} + monkeypatch.setattr(fix, "get_fixes", lambda *a, **k: _result_with([f])) + calls = [] + + def fake_hook(pb, vars, inventory=None, exit_on_error=True): + calls.append(pb) + return 2 if pb == fix.PLAYBOOK_WFCATALOG_REFRESH else 0 + monkeypatch.setattr(fix, "run_hook", fake_hook) + + rc = fix.run_fix("r", inventory=None, no_confirm=True) + assert rc == 1 # a failure -> nonzero + assert calls == [fix.PLAYBOOK_WFCATALOG_REFRESH] # rebuild skipped + + +def test_run_fix_no_targets_returns_0(monkeypatch): + monkeypatch.setattr(fix, "get_fixes", lambda *a, **k: _result_with([])) + assert fix.run_fix("r", inventory=None, no_confirm=True) == 0 + + +# --------------------------------------------------------------------------- # +# security: injection-laced report values must never reach a playbook +# --------------------------------------------------------------------------- # +def test_validate_fix_rejects_shell_metacharacters(): + bad = {"network": "IV", "station": "X; rm -rf /", "location": "", "channel": "HHZ", + "start": "2008-01-01", "end": "2008-02-01"} + with pytest.raises(ValueError, match="unsafe station"): + fix._validate_fix(bad) + + +@pytest.mark.parametrize("field,value", [ + ("channel", "HHZ$(touch pwned)"), + ("network", "IV`whoami`"), + ("start", "2008-01-01; echo x"), + ("location", "0123456789"), # too long +]) +def test_validate_fix_rejects_various(field, value): + f = {"network": "IV", "station": "STA", "location": "", "channel": "HHZ", + "start": "2008-01-01", "end": "2008-02-01"} + f[field] = value + with pytest.raises(ValueError): + fix._validate_fix(f) + + +def test_run_fix_rejects_malicious_row_without_running_playbooks(monkeypatch): + evil = {"network": "IV", "station": "MILZ`reboot`", "location": "", "channel": "HHZ", + "start": "2008-01-01", "end": "2008-02-01", + "direction": "refresh", "status": "actionable"} + monkeypatch.setattr(fix, "get_fixes", lambda *a, **k: _result_with([evil])) + calls = [] + monkeypatch.setattr(fix, "run_hook", + lambda pb, vars, inventory=None, exit_on_error=True: calls.append(pb) or 0) + rc = fix.run_fix("r", inventory=None, no_confirm=True) + assert rc == 1 # recorded as failed + assert calls == [] # no playbook ever ran for the unsafe row