test: add 23 surface tests for the CLI

.github/workflows/ci.yml

@@ -50,6 +50,7 @@ jobs:
             plotly \
             pm4py \
             pytest \
+            hypothesis \
             maturin
           pip install -e .
       - name: build pm_fast (Rust hot-path extension)

bench/datasets/download.py

@@ -27,7 +27,7 @@
 
 import argparse
 from pathlib import Path
-from typing import Dict, NamedTuple
+from typing import NamedTuple
 
 DATA_DIR = Path(__file__).parent / "data"
 
@@ -42,7 +42,7 @@ class Dataset(NamedTuple):
 # Curated subset of public process-mining benchmarks. ``landing_url``
 # points to the 4TU dataset landing page where the XES download lives
 # behind the terms-of-use button.
-REGISTRY: Dict[str, Dataset] = {
+REGISTRY: dict[str, Dataset] = {
     "bpi_2020_domestic": Dataset(
         name="bpi_2020_domestic",
         landing_url="https://data.4tu.nl/articles/dataset/BPI_Challenge_2020_Domestic_Declarations/12692543",

bench/eval.py

@@ -19,7 +19,6 @@
 import argparse
 import json
 from pathlib import Path
-from typing import List, Optional, Tuple
 
 
 def _resolve_run_dir(p: str) -> Path:
@@ -38,13 +37,13 @@ def _read(run_dir: Path, fname: str) -> dict:
     return json.loads(f.read_text()) if f.exists() else {}
 
 
-def _fmt_pct(x: Optional[float], digits: int = 1) -> str:
+def _fmt_pct(x: float | None, digits: int = 1) -> str:
     if x is None:
         return "—"
     return f"{x * 100:.{digits}f}%"
 
 
-def _fmt(x: Optional[float], digits: int = 3) -> str:
+def _fmt(x: float | None, digits: int = 3) -> str:
     if x is None:
         return "—"
     return f"{x:.{digits}f}"
@@ -79,9 +78,9 @@ def _row(name: str, run_dir: Path) -> dict:
     }
 
 
-def render_leaderboard(rows: List[dict]) -> str:
+def render_leaderboard(rows: list[dict]) -> str:
     """Two stacked tables: predictive metrics + process-mining metrics."""
-    out: List[str] = []
+    out: list[str] = []
     out.append("### Predictive performance")
     out.append("")
     out.append(

bench/seeds.py

@@ -29,17 +29,16 @@
 
 import argparse
 import json
-import math
 import statistics
 import subprocess
 import sys
 from pathlib import Path
-from typing import Any, Dict, List
+from typing import Any
 
 
-def _flatten(d: Dict[str, Any], prefix: str = "") -> Dict[str, Any]:
+def _flatten(d: dict[str, Any], prefix: str = "") -> dict[str, Any]:
     """Flatten nested dicts to dotted-path leaves; skip lists."""
-    out: Dict[str, Any] = {}
+    out: dict[str, Any] = {}
     for k, v in d.items():
         key = f"{prefix}.{k}" if prefix else k
         if isinstance(v, dict):
@@ -49,7 +48,7 @@ def _flatten(d: Dict[str, Any], prefix: str = "") -> Dict[str, Any]:
     return out
 
 
-def _aggregate(per_seed: List[Dict[str, Any]]) -> Dict[str, Any]:
+def _aggregate(per_seed: list[dict[str, Any]]) -> dict[str, Any]:
     """For each numeric leaf metric present in *every* run, compute
     mean / std / min / max / N. Skip metrics that vary in presence."""
     if not per_seed:
@@ -59,7 +58,7 @@ def _aggregate(per_seed: List[Dict[str, Any]]) -> Dict[str, Any]:
     for f in flats[1:]:
         common_keys &= set(f.keys())
 
-    summary: Dict[str, Any] = {}
+    summary: dict[str, Any] = {}
     for key in sorted(common_keys):
         values = [f[key] for f in flats]
         if not all(isinstance(v, (int, float)) for v in values):
@@ -77,7 +76,7 @@ def _aggregate(per_seed: List[Dict[str, Any]]) -> Dict[str, Any]:
     return summary
 
 
-def _run_one(csv: str, seed: int, out_root: Path, passthrough: List[str]) -> Path:
+def _run_one(csv: str, seed: int, out_root: Path, passthrough: list[str]) -> Path:
     """Run ``gnn run`` for one seed; return the run_<ts>/ dir."""
     out_dir = out_root / f"seed_{seed}"
     cmd = [
@@ -94,7 +93,7 @@ def _run_one(csv: str, seed: int, out_root: Path, passthrough: List[str]) -> Pat
     return runs[-1]
 
 
-def _collect_metrics(run_dir: Path) -> Dict[str, Dict[str, Any]]:
+def _collect_metrics(run_dir: Path) -> dict[str, dict[str, Any]]:
     """Read every metrics/*.json and tag by filename."""
     out = {}
     for path in (run_dir / "metrics").glob("*.json"):
@@ -137,13 +136,13 @@ def main() -> int:
     print(f"  out_root:   {out_root}")
     print(f"  passthrough: {' '.join(passthrough)}")
 
-    per_file_per_seed: Dict[str, List[Dict[str, Any]]] = {}
+    per_file_per_seed: dict[str, list[dict[str, Any]]] = {}
     for seed in args.seeds:
         run_dir = _run_one(args.csv, seed, out_root, passthrough)
         for fname, payload in _collect_metrics(run_dir).items():
             per_file_per_seed.setdefault(fname, []).append(payload)
 
-    aggregate: Dict[str, Dict[str, Any]] = {}
+    aggregate: dict[str, dict[str, Any]] = {}
     for fname, runs in per_file_per_seed.items():
         aggregate[fname] = _aggregate(runs)
 

gnn_cli/export.py

@@ -56,7 +56,7 @@ class _ArchMeta:
     max_seq_len: int
 
     @classmethod
-    def from_run_dir(cls, run_dir: Path) -> "_ArchMeta":
+    def from_run_dir(cls, run_dir: Path) -> _ArchMeta:
         candidates = [
             run_dir / "models" / "lstm_arch.json",
             run_dir / "models" / "transformer_arch.json",

notebooks/build.py

@@ -14,10 +14,9 @@
 
 import json
 from pathlib import Path
-from typing import List, Tuple
 
 # Cells: list of (cell_type, source). cell_type is "markdown" or "code".
-CELLS: List[Tuple[str, str]] = [
+CELLS: list[tuple[str, str]] = [
     ("markdown", """\
 # Tutorial — process mining with BPI 2020, end to end
 

pm_fast/python/pm_fast/__init__.py

@@ -6,8 +6,6 @@
 
 from __future__ import annotations
 
-from typing import Tuple
-
 import numpy as np
 import pandas as pd
 
@@ -34,7 +32,7 @@ def build_task_adjacency(df: pd.DataFrame, num_tasks: int) -> np.ndarray:
 
 def build_padded_prefixes(
     df: pd.DataFrame,
-) -> Tuple[np.ndarray, np.ndarray, np.ndarray, int]:
+) -> tuple[np.ndarray, np.ndarray, np.ndarray, int]:
     """Replacement for `_build_prefixes` + `make_padded_dataset` fused.
 
     Expects `df` sorted by (case_id, timestamp). Returns

tests/test_cli.py

@@ -0,0 +1,194 @@
+"""Surface tests for ``gnn_cli.cli``.
+
+The full pipeline (`run`, `analyze`, `cluster`, `smoke`) is exercised
+end to end by the existing integration tests against the synthetic
+event log; this file pins the **CLI shell**: the parser shape, exit
+codes, version flag, and the ``--config`` TOML loader's error path.
+
+These tests deliberately do not exercise the heavy ``cmd_*``
+handlers (those require torch / torch-geometric and a real event
+log; they're already covered elsewhere). What they cover is the
+contract operators script against:
+
+- exit codes documented in ``cli.py``'s module docstring
+  (``EXIT_OK = 0``, ``EXIT_USAGE = 2``, ``EXIT_DATA = 3``,
+  ``EXIT_RUNTIME = 4``)
+- the documented subcommand surface (``run``, ``analyze``, etc.)
+- the ``--version`` and ``--help`` outputs
+- the ``--config`` TOML loader's "file not found" path
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from gnn_cli import __version__
+from gnn_cli.cli import (
+    EXIT_DATA,
+    EXIT_OK,
+    EXIT_USAGE,
+    build_parser,
+    cmd_version,
+    main,
+)
+
+
+# ---------------------------------------------------------------------------
+# Documented exit codes
+# ---------------------------------------------------------------------------
+
+
+class TestExitCodes:
+    def test_documented_exit_codes_are_stable_constants(self) -> None:
+        # The module docstring documents these exact values; CI / shell
+        # scripts trap on them. A renumbering would silently break every
+        # caller that relied on, e.g., "exit 3 means data error".
+        assert EXIT_OK == 0
+        assert EXIT_USAGE == 2
+        assert EXIT_DATA == 3
+
+
+# ---------------------------------------------------------------------------
+# build_parser
+# ---------------------------------------------------------------------------
+
+
+class TestParserShape:
+    def test_parser_has_a_command_subparser(self) -> None:
+        # `args.command` must be present after parsing — main()
+        # indexes COMMANDS[args.command].
+        p = build_parser()
+        args = p.parse_args(["version"])
+        assert args.command == "version"
+
+    @pytest.mark.parametrize(
+        "subcommand",
+        [
+            "run",
+            "analyze",
+            "cluster",
+            "smoke",
+            "version",
+            "predict-suffix",
+            "whatif",
+            "export",
+            "serve",
+            "diff",
+            "explain",
+            "baseline",
+        ],
+    )
+    def test_documented_subcommand_is_registered(self, subcommand: str) -> None:
+        # Every README-documented subcommand must be registered with
+        # the parser. We probe via `<sub> --help` because some
+        # subcommands have required positional arguments that we
+        # don't want to model in this test; --help exits cleanly
+        # (SystemExit(0)) on a registered subcommand and with code
+        # 2 ("invalid choice") on an unregistered one. A rename is
+        # a breaking change for shell users.
+        p = build_parser()
+        with pytest.raises(SystemExit) as excinfo:
+            p.parse_args([subcommand, "--help"])
+        assert excinfo.value.code == EXIT_OK, (
+            f"subcommand {subcommand!r} not registered (got exit "
+            f"code {excinfo.value.code})"
+        )
+
+    def test_unknown_subcommand_exits_with_usage_error(self) -> None:
+        # argparse exits with SystemExit(2) on unknown subcommand;
+        # pin that the convention holds (and that EXIT_USAGE matches).
+        p = build_parser()
+        with pytest.raises(SystemExit) as excinfo:
+            p.parse_args(["definitely-not-a-real-command"])
+        assert excinfo.value.code == EXIT_USAGE
+
+    def test_no_subcommand_exits_with_usage_error(self) -> None:
+        # Bare `python -m gnn_cli` should error out with usage, not
+        # silently no-op or run a default.
+        p = build_parser()
+        with pytest.raises(SystemExit) as excinfo:
+            p.parse_args([])
+        assert excinfo.value.code == EXIT_USAGE
+
+
+# ---------------------------------------------------------------------------
+# Run-flag defaults — pin so a stealth change to a default fails here
+# ---------------------------------------------------------------------------
+
+
+class TestRunFlagDefaults:
+    # `run` requires a `data_path` positional arg; pass a dummy
+    # string (we never invoke the handler, just inspect the parsed
+    # Namespace).
+    def test_seed_defaults_to_42(self) -> None:
+        p = build_parser()
+        args = p.parse_args(["run", "/tmp/no-such-data.csv"])
+        assert args.seed == 42
+
+    def test_out_dir_defaults_to_results(self) -> None:
+        p = build_parser()
+        args = p.parse_args(["run", "/tmp/no-such-data.csv"])
+        assert args.out_dir == "results"
+
+    def test_val_frac_default(self) -> None:
+        p = build_parser()
+        args = p.parse_args(["run", "/tmp/no-such-data.csv"])
+        assert args.val_frac == pytest.approx(0.2)
+
+    def test_epochs_defaults(self) -> None:
+        # Two model defaults at once; an accidental swap of the
+        # GAT/LSTM defaults would silently change run cost.
+        p = build_parser()
+        args = p.parse_args(["run", "/tmp/no-such-data.csv"])
+        assert args.epochs_gat == 20
+        assert args.epochs_lstm == 5
+
+
+# ---------------------------------------------------------------------------
+# version subcommand
+# ---------------------------------------------------------------------------
+
+
+class TestVersionCommand:
+    def test_cmd_version_returns_exit_ok(self, capsys: pytest.CaptureFixture[str]) -> None:
+        # Direct call (skipping main's two-pass --config parse) to
+        # isolate the command behaviour from the parser plumbing.
+        p = build_parser()
+        args = p.parse_args(["version"])
+        rc = cmd_version(args)
+        assert rc == EXIT_OK
+        out = capsys.readouterr().out
+        assert __version__ in out
+
+    def test_main_version_subcommand_returns_exit_ok(
+        self, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        rc = main(["version"])
+        assert rc == EXIT_OK
+        # The version string appears on stdout (not stderr).
+        out = capsys.readouterr().out
+        assert __version__ in out
+
+
+# ---------------------------------------------------------------------------
+# --config TOML loader error path
+# ---------------------------------------------------------------------------
+
+
+class TestConfigPath:
+    def test_main_with_missing_config_file_returns_exit_data(
+        self, capsys: pytest.CaptureFixture[str], tmp_path: Path
+    ) -> None:
+        # The two-pass `--config` parse intercepts FileNotFoundError
+        # and maps it to EXIT_DATA with a clear message; pin this so
+        # a refactor doesn't accidentally swallow the error or surface
+        # it as EXIT_RUNTIME (which would mis-classify a config typo
+        # as a model failure in operator dashboards).
+        missing = tmp_path / "no-such.toml"
+        rc = main(["--config", str(missing), "version"])
+        assert rc == EXIT_DATA
+        err = capsys.readouterr().err
+        assert "config file not found" in err
+        assert str(missing) in err

visualization/process_viz.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 
 """
 Visualization module for process mining analysis
@@ -69,9 +68,14 @@ def plot_process_flow(bottleneck_stats, le_task, top_bottlenecks,
         dst = int(row["next_task_id"])
         G_flow.add_edge(src, dst, freq=int(row["count"]), mean_hours=row["mean_hours"])
 
-    btop_edges = set((int(src), int(dst)) for src, dst in zip(
-        top_bottlenecks["task_id"], top_bottlenecks["next_task_id"]
-    ))
+    btop_edges = set(
+        (int(src), int(dst))
+        for src, dst in zip(
+            top_bottlenecks["task_id"],
+            top_bottlenecks["next_task_id"],
+            strict=True,
+        )
+    )
 
     edge_cols, edge_wids = [], []
     for (u,v) in G_flow.edges():