diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ea6c9ab..8fc259e 100644 --- a/.github/workflows/ci.yml +++ b/.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) diff --git a/bench/datasets/download.py b/bench/datasets/download.py index d9d0241..d38eb49 100644 --- a/bench/datasets/download.py +++ b/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", diff --git a/bench/eval.py b/bench/eval.py index 5e5c5bd..8f87f4d 100644 --- a/bench/eval.py +++ b/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( diff --git a/bench/seeds.py b/bench/seeds.py index d97b604..5845821 100644 --- a/bench/seeds.py +++ b/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_/ 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) diff --git a/gnn_cli/export.py b/gnn_cli/export.py index b28fef2..6e988a4 100644 --- a/gnn_cli/export.py +++ b/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", diff --git a/notebooks/build.py b/notebooks/build.py index 574c860..19925f7 100644 --- a/notebooks/build.py +++ b/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 diff --git a/pm_fast/python/pm_fast/__init__.py b/pm_fast/python/pm_fast/__init__.py index 6558bcc..99fc9a1 100644 --- a/pm_fast/python/pm_fast/__init__.py +++ b/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 diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..bf74d03 --- /dev/null +++ b/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 ` --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 diff --git a/visualization/process_viz.py b/visualization/process_viz.py index a41abea..150e926 100644 --- a/visualization/process_viz.py +++ b/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():