diff --git a/AGENTS.md b/AGENTS.md index 7a36d14..817514b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -34,6 +34,7 @@ When you change something in milo-cli, the blast radius is: - **Terminal rendering bugs** → Alternate screen buffer not restored, cursor left off, raw mode leaked. Harm: the user's terminal is broken after our CLI exits. Happens silently unless you actually run the app to completion. - **Pipeline orchestration bugs** → Phase retries, dependency cycles, output capture. These power real deployment tooling for downstream consumers. Harm: a phase appears to succeed but didn't; or a deploy hangs forever. - **Startup-cost regressions** → The lazy-import contract is load-bearing. A stray top-level import in `__init__.py` adds latency to every CLI invocation in every downstream project. +- **Scaffold or `milo verify` regressions** → `milo new` and `milo verify` are the front door and self-diagnosis tool agents use to bootstrap and debug. A broken scaffold produces CLIs that fail their own tests; a broken verify tells agents a correctly-built CLI is wrong, or a broken one is fine. Harm: agent onboarding stalls or agents ship bad CLIs with a green check. Same severity class as schema-generation bugs — both corrupt the agent-facing contract. milo-cli is 0.2.x / alpha but has real consumers (Pounce, DORI evaluating). Calibrate accordingly — the API can still move, but not carelessly. diff --git a/README.md b/README.md index 084d235..8a758d1 100644 --- a/README.md +++ b/README.md @@ -64,14 +64,14 @@ Milo is a Python framework where every CLI is simultaneously a terminal app, a c ## Installation +**Requires Python 3.14+.** If you don't have it: `uv python install 3.14`. + ```bash pip install milo-cli ``` The PyPI package is **milo-cli**; import the **`milo`** namespace in Python. The `milo` console command is installed with the package. -Requires Python 3.14+ - --- ## Quick Start @@ -144,6 +144,52 @@ Requires Python 3.14+ --- +## Examples Index + +Pick the example closest to your use case, copy its `app.py`, and adapt. Every directory below contains a runnable `app.py` plus a focused README. + +**CLIs (typed function → CLI + MCP + llms.txt)** + +| What you want to build | Example | Key APIs | +|---|---|---| +| The simplest possible CLI | [examples/greet](examples/greet) | `CLI`, `@cli.command` | +| Dual-mode CLI ↔ MCP server (flagship) | [examples/deploy](examples/deploy) | `Annotated`, `MinLen`, `Context`, `Progress`, `--mcp` | +| Context injection, logging, progress, confirms | [examples/ctxdemo](examples/ctxdemo) | `Context`, `ctx.info`, `ctx.progress`, `ctx.confirm` | +| Nested command groups (`app repo list`) | [examples/groups](examples/groups) | `cli.group()`, `walk_commands` | +| Fast startup via deferred imports | [examples/lazyapp](examples/lazyapp) | `cli.lazy_command()` | +| Production CLI with hooks, completions, doctor | [examples/devtool](examples/devtool) | `run_doctor`, `before_run`/`after_run`, did-you-mean, completions | +| AI-native CLI surfacing tools + resources | [examples/taskman](examples/taskman) | `@command`, `@resource`, `--format`, `--llms-txt`, `--mcp` | + +**Configuration, plugins, pipelines** + +| What you want to build | Example | Key APIs | +|---|---|---| +| TOML config with profiles + overlays | [examples/configapp](examples/configapp) | `Config`, `ConfigSpec`, `Config.load`, `Config.validate` | +| Plugin system with hooks + listeners | [examples/pluggable](examples/pluggable) | `HookRegistry`, `define`, `on`, `invoke` | +| Multi-phase pipeline with deps + retries | [examples/buildpipe](examples/buildpipe) | `Pipeline`, `Phase`, `PhasePolicy`, `>>` | + +**Interactive TUIs (`App` + reducer)** + +| What you want to build | Example | Key APIs | +|---|---|---| +| The simplest TUI | [examples/counter](examples/counter) | `App.from_dir`, reducer combinators | +| Modal input with derived filtering | [examples/todo](examples/todo) | tuple state, `quit_on`, derived views | +| Tick-driven animation | [examples/stopwatch](examples/stopwatch) | `tick_rate`, `@@TICK`, `quit_on` | +| Scrollable viewport with saga I/O | [examples/filepicker](examples/filepicker) | viewport, sagas, frozen tuples | +| Multi-screen flow with forms | [examples/wizard](examples/wizard) | `Flow`, `FlowScreen`, `make_form_reducer`, `FieldSpec` | + +**Async work (sagas + Cmd pattern)** + +| What you want to build | Example | Key APIs | +|---|---|---| +| Sagas for async side effects | [examples/fetcher](examples/fetcher) | `Call`, `Put`, `Select`, `Retry` | +| Parallel concurrent work | [examples/downloader](examples/downloader) | `Fork`, `Call`, `Delay`, `Timeout` | +| Bubbletea-style Cmd thunks | [examples/spinner](examples/spinner) | `Cmd`, `Batch`, `TickCmd`, `ViewState` | + +> Don't see your use case? Run `milo new ` to scaffold a fresh CLI with tests, then `milo verify app.py` to confirm it works. + +--- + ## Usage
diff --git a/changelog.d/+agent-affordances.added.md b/changelog.d/+agent-affordances.added.md new file mode 100644 index 0000000..80ef54b --- /dev/null +++ b/changelog.d/+agent-affordances.added.md @@ -0,0 +1 @@ +Agent-native affordances: `milo new ` scaffold (app.py, tests, conftest, README), `milo verify ` six-check self-diagnosis (imports, CLI located, commands registered, schemas generate, in-process MCP list, subprocess MCP transport), `function_to_schema(..., warn_missing_docs=True)` surfacing undocumented typed params, README examples index with drift lint, and a Python 3.14+ preflight on `milo` with an actionable install hint instead of ImportError. diff --git a/docs/agent-quickstart.md b/docs/agent-quickstart.md index f0ff769..e4b65fa 100644 --- a/docs/agent-quickstart.md +++ b/docs/agent-quickstart.md @@ -12,6 +12,18 @@ If something in this doc no longer works, that's the bug — open an issue. - `uv` installed (`curl -LsSf https://astral.sh/uv/install.sh | sh`). - This repo cloned or `milo-cli` installed (`uv add milo-cli`). +## Step 0 — Scaffold (optional; skip if writing manually) + +```bash +uv run milo new my_cli +cd my_cli +``` + +Produces `app.py`, `tests/test_app.py`, `conftest.py`, and a `README.md` — the +same shape this doc walks through. Scaffold names must be lowercase with +underscores (`my_cli`, not `My-CLI`). If the directory exists, the command +refuses to overwrite; pick another name or delete the old one. + ## Step 1 — Write the function ```python @@ -113,6 +125,31 @@ You should see `my_cli` listed with `greet` as a tool. Call it: Expected result: Claude calls the tool, the tool returns `"Hello, Bob!"`, Claude echoes it back. +## Step 6 — Self-diagnose with `milo verify` + +Before registering with Claude (or any time you break something), run: + +```bash +uv run milo verify my_cli/app.py +``` + +All six checks should pass: + +``` +✓ imports: loaded app.py +✓ cli_located: found CLI instance (name='my_cli') +✓ commands_registered: 1 command(s) registered +✓ schemas_generate: 1 schema(s) generated; all params documented +✓ mcp_list_tools: 1 tool(s) listed with valid inputSchema +✓ mcp_transport: subprocess handshake succeeded; 1 tool(s) over JSON-RPC +``` + +A `⚠ schemas_generate` row listing `parameter 'X' has no description` means a +typed parameter is missing an `Args:` entry (or `Annotated[..., Description(...)]`). +A `✗` row is a failure — read the details and fix before continuing. + +`milo verify` exits 0 on warnings, nonzero on failures. Wire it into CI. + ## When things go wrong | Symptom | Likely cause | Fix | diff --git a/pyproject.toml b/pyproject.toml index 227bb70..639ce65 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,7 +40,13 @@ Issues = "https://github.com/lbliii/milo-cli/issues" where = ["src"] [tool.setuptools.package-data] -milo = ["templates/*.kida", "templates/components/*.kida", "py.typed"] +milo = [ + "templates/*.kida", + "templates/components/*.kida", + "_scaffold/default/*", + "_scaffold/default/**/*", + "py.typed", +] [tool.pytest.ini_options] testpaths = ["tests"] @@ -125,6 +131,7 @@ ignore = [ "__init__.py" = ["F401"] "tests/**/*.py" = ["S101", "S108", "S110", "S604", "SIM117", "N806", "ARG", "T201"] "examples/**/*.py" = ["S101", "S108", "S110", "S604", "SIM117", "N806", "ARG", "T201", "E402"] +"src/milo/_scaffold/default/**/*.py" = ["S101", "I001", "F401", "ARG", "T201", "E402"] "src/milo/_child.py" = ["S101"] # asserts for type narrowing on subprocess pipes "src/milo/testing/*.py" = ["S101", "ARG"] # test helpers use assert and may have unused args "src/milo/version_check.py" = ["S110", "S310"] # best-effort cache reads + known PyPI URL diff --git a/src/milo/_scaffold/__init__.py b/src/milo/_scaffold/__init__.py new file mode 100644 index 0000000..30da966 --- /dev/null +++ b/src/milo/_scaffold/__init__.py @@ -0,0 +1,55 @@ +"""Project scaffolding for `milo new`. + +Templates live under ``default/`` and are copied with ``{{name}}`` substitution. +No template engine — a plain ``str.replace`` keeps scaffolding zero-dep and +auditable. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +_TEMPLATE_DIR = Path(__file__).parent / "default" +_NAME_RE = re.compile(r"^[a-z][a-z0-9_]*$") + + +class ScaffoldError(Exception): + """Raised when a scaffold cannot be created.""" + + +def scaffold(name: str, target_dir: Path) -> Path: + """Create a new milo CLI project at ``target_dir / name``. + + Args: + name: Project name. Must match ``^[a-z][a-z0-9_]*$`` so it works + both as a directory name and as a Python identifier. + target_dir: Parent directory in which the project dir is created. + + Returns: + The created project directory path. + + Raises: + ScaffoldError: If ``name`` is invalid or the target path exists. + """ + if not _NAME_RE.match(name): + raise ScaffoldError( + f"Invalid project name '{name}'. " + f"Use lowercase letters, digits, and underscores; start with a letter." + ) + + project_dir = target_dir / name + if project_dir.exists(): + raise ScaffoldError(f"Refusing to overwrite existing path: {project_dir}") + + for src in _TEMPLATE_DIR.rglob("*"): + rel = src.relative_to(_TEMPLATE_DIR) + dst = project_dir / rel + if src.is_dir(): + dst.mkdir(parents=True, exist_ok=True) + continue + dst.parent.mkdir(parents=True, exist_ok=True) + content = src.read_text(encoding="utf-8") + dst.write_text(content.replace("{{name}}", name), encoding="utf-8") + + return project_dir diff --git a/src/milo/_scaffold/default/README.md b/src/milo/_scaffold/default/README.md new file mode 100644 index 0000000..1ddcbc7 --- /dev/null +++ b/src/milo/_scaffold/default/README.md @@ -0,0 +1,35 @@ +# {{name}} + +A milo CLI: one typed function, three protocols (CLI, MCP server, llms.txt). + +## Run + +```bash +# CLI +uv run python app.py greet --name Alice +uv run python app.py greet --name Alice --loud + +# llms.txt (agent-readable catalog) +uv run python app.py --llms-txt + +# MCP server (stdin/stdout JSON-RPC, for Claude et al.) +uv run python app.py --mcp +``` + +## Test + +```bash +uv run pytest tests/ -v +``` + +The test file covers three layers — schema, direct dispatch, MCP dispatch. +Add commands by adding `@cli.command(...)` functions to `app.py`, then add +matching tests in `tests/test_app.py`. + +## Register with Claude + +```bash +claude mcp add {{name}} -- uv run python /absolute/path/to/app.py --mcp +``` + +See the milo agent quickstart for the full walkthrough. diff --git a/src/milo/_scaffold/default/app.py b/src/milo/_scaffold/default/app.py new file mode 100644 index 0000000..ef5eed1 --- /dev/null +++ b/src/milo/_scaffold/default/app.py @@ -0,0 +1,29 @@ +"""{{name}} — a milo CLI scaffolded by `milo new`. + +Run modes: + uv run python app.py greet --name Alice + uv run python app.py --llms-txt + uv run python app.py --mcp +""" + +from __future__ import annotations + +from milo import CLI + +cli = CLI(name="{{name}}", description="What it does", version="0.1") + + +@cli.command("greet", description="Return a greeting") +def greet(name: str, loud: bool = False) -> str: + """Greet someone by name. + + Args: + name: The person to greet. + loud: If true, SHOUT. + """ + message = f"Hello, {name}!" + return message.upper() if loud else message + + +if __name__ == "__main__": + cli.run() diff --git a/src/milo/_scaffold/default/conftest.py b/src/milo/_scaffold/default/conftest.py new file mode 100644 index 0000000..f851986 --- /dev/null +++ b/src/milo/_scaffold/default/conftest.py @@ -0,0 +1,8 @@ +"""pytest configuration: make app.py importable from tests/.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) diff --git a/src/milo/_scaffold/default/tests/__init__.py b/src/milo/_scaffold/default/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/milo/_scaffold/default/tests/test_app.py b/src/milo/_scaffold/default/tests/test_app.py new file mode 100644 index 0000000..19a8559 --- /dev/null +++ b/src/milo/_scaffold/default/tests/test_app.py @@ -0,0 +1,65 @@ +"""Tests for {{name}} — schema, direct dispatch, MCP dispatch. + +Three layers cover the common regression surface: + 1. Schema — `function_to_schema(greet)` matches the function signature. + 2. Direct — `cli.invoke([...])` returns the expected output. + 3. MCP — `_call_tool(cli, {...})` returns the expected response and, + on error, structured `errorData` with `argument` context. +""" + +from __future__ import annotations + +import sys + +import pytest + +from app import cli, greet + +from milo.mcp import _call_tool, _list_tools +from milo.schema import function_to_schema + + +class TestSchema: + def test_generated_schema_matches_signature(self): + schema = function_to_schema(greet) + assert schema["type"] == "object" + assert schema["properties"]["name"]["type"] == "string" + assert schema["properties"]["loud"]["type"] == "boolean" + assert schema["required"] == ["name"] + assert schema["properties"]["loud"]["default"] is False + + def test_tool_appears_in_list_tools(self): + tools = _list_tools(cli) + assert "greet" in [t["name"] for t in tools] + + +class TestDirectDispatch: + def test_invoke_returns_string(self): + result = cli.invoke(["greet", "--name", "Alice"]) + assert result.exit_code == 0 + assert "Hello, Alice!" in result.output + + def test_invoke_with_loud_flag(self): + result = cli.invoke(["greet", "--name", "Alice", "--loud"]) + assert result.exit_code == 0 + assert "HELLO, ALICE!" in result.output + + def test_call_raw_returns_plain_value(self): + assert cli.call_raw("greet", name="Bob") == "Hello, Bob!" + + +class TestMCPDispatch: + def test_call_tool_returns_content(self): + result = _call_tool(cli, {"name": "greet", "arguments": {"name": "Agent"}}) + assert result["content"][0]["text"] == "Hello, Agent!" + assert "isError" not in result + + def test_call_tool_missing_required_arg_returns_argument_context(self): + result = _call_tool(cli, {"name": "greet", "arguments": {}}) + assert result["isError"] is True + assert result["errorData"]["argument"] == "name" + assert result["errorData"]["reason"] == "missing_required_argument" + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v"])) diff --git a/src/milo/cli.py b/src/milo/cli.py index e63daf2..9be3843 100644 --- a/src/milo/cli.py +++ b/src/milo/cli.py @@ -46,6 +46,35 @@ def _cmd_dev(args: argparse.Namespace) -> None: server.run() +def _cmd_verify(args: argparse.Namespace) -> None: + """Run diagnostic checks against an agent-built milo CLI.""" + from milo.verify import verify + + report = verify(args.target, timeout=args.timeout) + sys.stdout.write(report.format() + "\n") + sys.exit(report.exit_code) + + +def _cmd_new(args: argparse.Namespace) -> None: + """Scaffold a new milo CLI project.""" + from milo._scaffold import ScaffoldError, scaffold + + try: + project_dir = scaffold(args.name, Path(args.dir)) + except ScaffoldError as e: + sys.stderr.write(f"Error: {e}\n") + sys.exit(1) + + sys.stdout.write( + f"Created {project_dir}\n" + f"\n" + f"Next steps:\n" + f" cd {project_dir}\n" + f" uv run python app.py greet --name Alice\n" + f" uv run pytest tests/\n" + ) + + def _cmd_replay(args: argparse.Namespace) -> None: """Replay a recorded session.""" from milo.testing._record import load_recording @@ -81,8 +110,29 @@ def on_state(state, action): sys.stdout.write(f"Replay complete. Final state: {final!r}\n") +_MIN_PYTHON = (3, 14) + + +def _preflight_python_version() -> None: + """Exit with an actionable message when running under an unsupported Python. + + Runs before any milo module import that relies on 3.14+ features, so the + user sees a fix-it message instead of a `SyntaxError` or `ImportError`. + """ + if sys.version_info < _MIN_PYTHON: + have = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}" + want = f"{_MIN_PYTHON[0]}.{_MIN_PYTHON[1]}" + sys.stderr.write( + f"milo requires Python {want}+ (you have {have}).\n" + f"Install with: uv python install {want}\n" + ) + sys.exit(2) + + def main(argv: list[str] | None = None) -> None: """Main CLI entry point.""" + _preflight_python_version() + parser = argparse.ArgumentParser( prog="milo", description="Template-driven CLI applications for free-threaded Python", @@ -92,6 +142,25 @@ def main(argv: list[str] | None = None) -> None: parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") subparsers = parser.add_subparsers(dest="command") + # milo new + new_parser = subparsers.add_parser("new", help="Scaffold a new milo CLI project") + new_parser.add_argument("name", help="Project name (lowercase, underscores)") + new_parser.add_argument( + "--dir", "-d", default=".", help="Parent directory for the project (default: cwd)" + ) + + # milo verify + verify_parser = subparsers.add_parser( + "verify", help="Self-diagnose a milo CLI (schema, dispatch, MCP)" + ) + verify_parser.add_argument("target", help="Path to app.py or 'module:attr' reference") + verify_parser.add_argument( + "--timeout", + type=float, + default=5.0, + help="Seconds to wait for the subprocess MCP handshake (default: 5.0)", + ) + # milo dev dev_parser = subparsers.add_parser("dev", help="Run app with hot-reload") dev_parser.add_argument("app", help="App path as 'module:attribute'") @@ -124,7 +193,11 @@ def main(argv: list[str] | None = None) -> None: args = parser.parse_args(argv) - if args.command == "dev": + if args.command == "new": + _cmd_new(args) + elif args.command == "verify": + _cmd_verify(args) + elif args.command == "dev": _cmd_dev(args) elif args.command == "replay": _cmd_replay(args) diff --git a/src/milo/schema.py b/src/milo/schema.py index aa755ca..f06a350 100644 --- a/src/milo/schema.py +++ b/src/milo/schema.py @@ -95,8 +95,12 @@ class Description: } -@functools.lru_cache(maxsize=256) -def function_to_schema(func: Callable[..., Any], *, strict: bool = False) -> dict[str, Any]: +def function_to_schema( + func: Callable[..., Any], + *, + strict: bool = False, + warn_missing_docs: bool = False, +) -> dict[str, Any]: """Generate MCP-compatible JSON Schema from function type annotations. Parameters with defaults are optional (not in required). @@ -116,7 +120,30 @@ def function_to_schema(func: Callable[..., Any], *, strict: bool = False) -> dic When *strict* is True, unrecognized type annotations raise :class:`TypeError` instead of silently falling back to ``"string"``. + + When *warn_missing_docs* is True, every schema parameter without a + description (no ``Args:`` entry and no ``Annotated[..., Description(...)]``) + emits a :class:`UserWarning`. Default ``False`` so production schema + generation stays silent; ``milo verify`` opts in. """ + schema, undocumented = _function_to_schema_cached(func, strict=strict) + if warn_missing_docs: + for name in undocumented: + warnings.warn( + f"Parameter {name!r} has no description; " + f"add an 'Args:' entry to the docstring or " + f"Annotated[..., Description(...)] to the type", + UserWarning, + stacklevel=2, + ) + return schema + + +@functools.lru_cache(maxsize=256) +def _function_to_schema_cached( + func: Callable[..., Any], *, strict: bool = False +) -> tuple[dict[str, Any], tuple[str, ...]]: + """Compute (schema, undocumented_param_names). Cached; warnings live in the wrapper.""" sig = inspect.signature(func) # Resolve string annotations (from __future__ import annotations) # include_extras=True preserves Annotated metadata for constraint extraction @@ -144,6 +171,7 @@ def function_to_schema(func: Callable[..., Any], *, strict: bool = False) -> dic properties: dict[str, Any] = {} required: list[str] = [] defs: dict[str, dict[str, Any]] = {} + undocumented: list[str] = [] for name, param in sig.parameters.items(): annotation = hints.get(name, param.annotation) @@ -181,6 +209,12 @@ def function_to_schema(func: Callable[..., Any], *, strict: bool = False) -> dic if name in param_docs: prop["description"] = param_docs[name] + # Track params that ended up with no description from any source + # (no Args entry and no Annotated[..., Description(...)]). The wrapper + # turns this into UserWarnings when warn_missing_docs=True. + if "description" not in prop: + undocumented.append(name) + properties[name] = prop has_default = param.default is not inspect.Parameter.empty @@ -201,7 +235,7 @@ def function_to_schema(func: Callable[..., Any], *, strict: bool = False) -> dic result["required"] = required if defs: result["$defs"] = defs - return result + return result, tuple(undocumented) def _type_to_schema( diff --git a/src/milo/verify.py b/src/milo/verify.py new file mode 100644 index 0000000..dd11c5e --- /dev/null +++ b/src/milo/verify.py @@ -0,0 +1,465 @@ +"""Self-diagnosis for agent-built milo CLIs (`milo verify`). + +Answers "is this CLI correctly built?" via six checks: + +1. **Imports** — the file (or module) loads without error. +2. **CLI located** — a ``milo.CLI`` instance is reachable in the module. +3. **Commands registered** — at least one ``@cli.command`` has been attached. +4. **Schemas generate** — ``function_to_schema`` succeeds for every command; + missing docstring ``Args:`` sections surface as warnings. +5. **In-process MCP list** — ``_list_tools(cli)`` returns a well-formed list + with one entry per command. +6. **Subprocess MCP transport** — running ``python --mcp`` responds to + ``initialize`` and ``tools/list`` over JSON-RPC. (Skipped for module:attr + inputs since there's no standalone entry point.) + +The report distinguishes pass/warn/fail; `milo verify` exits non-zero only on +failures, not warnings. +""" + +from __future__ import annotations + +import contextlib +import importlib +import importlib.util +import json +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from types import ModuleType + + from milo.commands import CLI + +_MCP_PROTOCOL_VERSION = "2025-06-18" +_ICONS = {"ok": "✓", "warn": "⚠", "fail": "✗", "skip": "∙"} + + +@dataclass(frozen=True, slots=True) +class VerifyCheck: + """A single diagnostic check result.""" + + name: str + status: str # "ok" | "warn" | "fail" | "skip" + message: str + details: str = "" + + +@dataclass(frozen=True, slots=True) +class VerifyReport: + """Aggregated verify report.""" + + target: str + checks: tuple[VerifyCheck, ...] + + @property + def passed(self) -> int: + return sum(1 for c in self.checks if c.status == "ok") + + @property + def warnings(self) -> int: + return sum(1 for c in self.checks if c.status == "warn") + + @property + def failures(self) -> int: + return sum(1 for c in self.checks if c.status == "fail") + + @property + def skipped(self) -> int: + return sum(1 for c in self.checks if c.status == "skip") + + @property + def exit_code(self) -> int: + return 1 if self.failures else 0 + + def format(self) -> str: + """Render the report for terminal output.""" + lines = [f"milo verify {self.target}", ""] + for c in self.checks: + icon = _ICONS.get(c.status, "?") + lines.append(f" {icon} {c.name}: {c.message}") + if c.details: + lines.extend(f" {detail}" for detail in c.details.splitlines()) + lines.append("") + summary = ( + f"{self.passed} passed, {self.warnings} warning(s), " + f"{self.failures} failure(s), {self.skipped} skipped" + ) + lines.append(summary) + return "\n".join(lines) + + +def verify(target: str, *, timeout: float = 5.0) -> VerifyReport: + """Run all verify checks against ``target``. + + Args: + target: Either a filesystem path ending in ``.py`` or a ``module:attr`` + reference. File paths are imported via ``importlib.util`` so no + ``sys.path`` pollution is required; module:attr is resolved via + the standard import machinery with the cwd added to ``sys.path``. + timeout: Seconds to wait for the subprocess MCP handshake. + + Returns: + A :class:`VerifyReport` with every check attached. The report's + ``exit_code`` is 1 iff any check failed (warnings do not fail the + report). + """ + checks: list[VerifyCheck] = [] + + # --- Check 1: imports --- + module, file_path, import_check = _load_target(target) + checks.append(import_check) + if import_check.status == "fail" or module is None: + return VerifyReport(target=target, checks=tuple(checks)) + + # --- Check 2: locate CLI instance --- + cli = _find_cli_instance(module, target) + if isinstance(cli, VerifyCheck): + checks.append(cli) + return VerifyReport(target=target, checks=tuple(checks)) + checks.append( + VerifyCheck( + name="cli_located", + status="ok", + message=f"found CLI instance (name={cli.name!r})", + ) + ) + + # --- Check 3: commands registered --- + command_list = list(cli.walk_commands()) + if not command_list: + checks.append( + VerifyCheck( + name="commands_registered", + status="fail", + message="no commands registered", + details="Add at least one @cli.command(...) function.", + ) + ) + return VerifyReport(target=target, checks=tuple(checks)) + checks.append( + VerifyCheck( + name="commands_registered", + status="ok", + message=f"{len(command_list)} command(s) registered", + details=", ".join(path for path, _ in command_list), + ) + ) + + # --- Check 4: schemas generate --- + checks.append(_check_schemas(command_list)) + + # --- Check 5: in-process MCP list --- + expected_visible = sum(1 for _, cmd in command_list if not getattr(cmd, "hidden", False)) + checks.append(_check_in_process_mcp(cli, expected_visible)) + + # --- Check 6: subprocess MCP transport --- + if file_path is None: + checks.append( + VerifyCheck( + name="mcp_transport", + status="skip", + message="subprocess transport check skipped for module:attr input", + ) + ) + else: + checks.append(_check_subprocess_mcp(file_path, timeout=timeout)) + + return VerifyReport(target=target, checks=tuple(checks)) + + +def _load_target(target: str) -> tuple[ModuleType | None, Path | None, VerifyCheck]: + """Import the target and return ``(module, file_path, import_check)``. + + ``file_path`` is ``None`` for module:attr inputs. On failure the first + element is ``None`` and ``import_check`` carries the diagnosis. + """ + if target.endswith(".py") and Path(target).is_file(): + path = Path(target).resolve() + mod_name = f"_verify_{path.stem}" + spec = importlib.util.spec_from_file_location(mod_name, path) + if spec is None or spec.loader is None: + return ( + None, + None, + VerifyCheck( + name="imports", + status="fail", + message=f"could not create import spec for {path}", + ), + ) + module = importlib.util.module_from_spec(spec) + # Register before exec_module so @dataclass etc. can resolve + # ``sys.modules[cls.__module__]`` during class construction (Py 3.14). + sys.modules[mod_name] = module + try: + spec.loader.exec_module(module) + except Exception as e: + sys.modules.pop(mod_name, None) + return ( + None, + None, + VerifyCheck( + name="imports", + status="fail", + message=f"import failed: {type(e).__name__}: {e}", + ), + ) + return module, path, VerifyCheck(name="imports", status="ok", message=f"loaded {path.name}") + + if ":" in target: + module_path, _, _ = target.partition(":") + cwd = str(Path.cwd()) + if cwd not in sys.path: + sys.path.insert(0, cwd) + try: + module = importlib.import_module(module_path) + except Exception as e: + return ( + None, + None, + VerifyCheck( + name="imports", + status="fail", + message=f"import failed: {type(e).__name__}: {e}", + ), + ) + return ( + module, + None, + VerifyCheck(name="imports", status="ok", message=f"loaded module {module_path!r}"), + ) + + return ( + None, + None, + VerifyCheck( + name="imports", + status="fail", + message=(f"target {target!r} is neither a .py file path nor a module:attr reference"), + ), + ) + + +def _find_cli_instance(module: ModuleType, target: str) -> CLI | VerifyCheck: + """Find the CLI instance in ``module``. + + For ``module:attr`` targets, look up the named attribute. For file-path + targets, scan the module for exactly one ``CLI`` instance. + """ + from milo.commands import CLI + + if ":" in target and not target.endswith(".py"): + _, _, attr = target.partition(":") + obj = getattr(module, attr, None) + if obj is None: + return VerifyCheck( + name="cli_located", + status="fail", + message=f"attribute {attr!r} not found on module", + ) + if not isinstance(obj, CLI): + return VerifyCheck( + name="cli_located", + status="fail", + message=f"{attr!r} is {type(obj).__name__}, not milo.CLI", + ) + return obj + + instances = [ + (name, obj) + for name, obj in vars(module).items() + if isinstance(obj, CLI) and not name.startswith("_") + ] + if not instances: + return VerifyCheck( + name="cli_located", + status="fail", + message="no milo.CLI instance found at module top level", + details="Assign one: `cli = CLI(name=..., ...)`", + ) + if len(instances) > 1: + names = ", ".join(n for n, _ in instances) + return VerifyCheck( + name="cli_located", + status="fail", + message=f"multiple CLI instances found: {names}", + details="Use the module:attr form to disambiguate.", + ) + return instances[0][1] + + +def _check_schemas(command_list: list[tuple[str, Any]]) -> VerifyCheck: + """Generate schemas for every command; surface docstring coverage gaps. + + Coverage gaps come from ``function_to_schema(..., warn_missing_docs=True)`` + so verify sees the same undocumented-param judgement as production schema + generation would, were it opted in. + """ + import warnings as _warnings + + from milo.schema import function_to_schema + + failures: list[str] = [] + doc_warnings: list[str] = [] + + for path, cmd in command_list: + handler = getattr(cmd, "handler", None) + if handler is None: + failures.append(f"{path}: no handler") + continue + with _warnings.catch_warnings(record=True) as caught: + _warnings.simplefilter("always") + try: + function_to_schema(handler, warn_missing_docs=True) + except Exception as e: + failures.append(f"{path}: {type(e).__name__}: {e}") + continue + for w in caught: + if not issubclass(w.category, UserWarning): + continue + msg = str(w.message) + # Only surface missing-docs warnings here. Other UserWarnings + # (unrecognized type fallbacks, unresolved forward refs) belong + # to schema generation itself, not docstring coverage. + if "no description" in msg: + doc_warnings.append(f"{path}: {msg}") + + if failures: + return VerifyCheck( + name="schemas_generate", + status="fail", + message=f"{len(failures)} schema generation failure(s)", + details="\n".join(failures), + ) + if doc_warnings: + return VerifyCheck( + name="schemas_generate", + status="warn", + message=( + f"schemas generate ({len(command_list)}), " + f"but {len(doc_warnings)} parameter(s) lack descriptions" + ), + details="\n".join(doc_warnings), + ) + return VerifyCheck( + name="schemas_generate", + status="ok", + message=f"{len(command_list)} schema(s) generated; all params documented", + ) + + +def _check_in_process_mcp(cli: CLI, expected_count: int) -> VerifyCheck: + """Call ``_list_tools(cli)`` and validate shape.""" + from milo.mcp import _list_tools + + try: + tools = _list_tools(cli) + except Exception as e: + return VerifyCheck( + name="mcp_list_tools", + status="fail", + message=f"_list_tools raised: {type(e).__name__}: {e}", + ) + + if len(tools) != expected_count: + return VerifyCheck( + name="mcp_list_tools", + status="fail", + message=( + f"expected {expected_count} tool(s), got {len(tools)} — " + f"some commands did not reach the MCP surface" + ), + ) + for tool in tools: + if "name" not in tool or "inputSchema" not in tool: + return VerifyCheck( + name="mcp_list_tools", + status="fail", + message=f"malformed tool entry: {tool!r}", + ) + return VerifyCheck( + name="mcp_list_tools", + status="ok", + message=f"{len(tools)} tool(s) listed with valid inputSchema", + ) + + +def _check_subprocess_mcp(path: Path, *, timeout: float) -> VerifyCheck: + """Start `python --mcp`, handshake, verify tools/list response.""" + proc = subprocess.Popen( + [sys.executable, str(path), "--mcp"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + ) + try: + requests = [ + { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": {"protocolVersion": _MCP_PROTOCOL_VERSION, "capabilities": {}}, + }, + {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}, + ] + payload = "\n".join(json.dumps(r) for r in requests) + "\n" + try: + stdout, stderr = proc.communicate(input=payload, timeout=timeout) + except subprocess.TimeoutExpired: + proc.kill() + proc.communicate() + return VerifyCheck( + name="mcp_transport", + status="fail", + message=f"subprocess did not respond within {timeout}s", + details='Check that the file ends with `if __name__ == "__main__": cli.run()`.', + ) + + responses: list[dict[str, Any]] = [] + for line in stdout.splitlines(): + line = line.strip() + if not line: + continue + with contextlib.suppress(json.JSONDecodeError): + responses.append(json.loads(line)) + + if len(responses) < 2: + return VerifyCheck( + name="mcp_transport", + status="fail", + message=f"expected 2 JSON-RPC responses, got {len(responses)}", + details=(stderr or "no stderr").strip()[:500], + ) + + init_resp, tools_resp = responses[0], responses[1] + if "result" not in init_resp or "protocolVersion" not in init_resp.get("result", {}): + return VerifyCheck( + name="mcp_transport", + status="fail", + message="initialize response missing protocolVersion", + details=json.dumps(init_resp)[:300], + ) + if "result" not in tools_resp or "tools" not in tools_resp.get("result", {}): + return VerifyCheck( + name="mcp_transport", + status="fail", + message="tools/list response missing tools list", + details=json.dumps(tools_resp)[:300], + ) + + tool_count = len(tools_resp["result"]["tools"]) + return VerifyCheck( + name="mcp_transport", + status="ok", + message=f"subprocess handshake succeeded; {tool_count} tool(s) over JSON-RPC", + ) + finally: + if proc.poll() is None: + proc.kill() + proc.wait() diff --git a/tests/test_cli.py b/tests/test_cli.py index 03097cb..a3b6de0 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -37,6 +37,29 @@ def test_replay_missing_session(self): main(["replay"]) +class TestPythonPreflight: + """main() exits 2 with an actionable message on pre-3.14 Pythons.""" + + def test_rejects_old_python_with_exit_code_2(self, monkeypatch, capsys): + # Can't construct sys.version_info directly — substitute a plain + # namedtuple that exposes the same attributes we access. + from collections import namedtuple + + VInfo = namedtuple("VInfo", "major minor micro releaselevel serial") + monkeypatch.setattr(sys, "version_info", VInfo(3, 12, 7, "final", 0)) + with pytest.raises(SystemExit) as exc: + main([]) + assert exc.value.code == 2 + err = capsys.readouterr().err + assert "Python 3.14+" in err + assert "3.12.7" in err + assert "uv python install 3.14" in err + + def test_allows_current_python(self): + # Sanity: with the real interpreter (>=3.14 in this project), no exit. + main([]) + + class TestLoadApp: def test_missing_colon_exits(self, capsys): with pytest.raises(SystemExit) as exc_info: diff --git a/tests/test_readme_example_index.py b/tests/test_readme_example_index.py new file mode 100644 index 0000000..7029266 --- /dev/null +++ b/tests/test_readme_example_index.py @@ -0,0 +1,26 @@ +"""Drift gate: every examples/*/ dir must be referenced in the README index. + +When you add a new example, add a row to the "Examples Index" section in +README.md. This test fails CI otherwise — keeps the index honest. +""" + +from __future__ import annotations + +from pathlib import Path + +_REPO_ROOT = Path(__file__).resolve().parents[1] +_README = _REPO_ROOT / "README.md" +_EXAMPLES_DIR = _REPO_ROOT / "examples" + + +def _example_dirs() -> list[str]: + return sorted(p.parent.name for p in _EXAMPLES_DIR.glob("*/app.py")) + + +def test_every_example_is_referenced_in_readme(): + readme = _README.read_text(encoding="utf-8") + missing = [name for name in _example_dirs() if f"examples/{name}" not in readme] + assert not missing, ( + f"README.md is missing rows for {len(missing)} example(s): {missing}. " + f"Add them to the 'Examples Index' section." + ) diff --git a/tests/test_scaffold.py b/tests/test_scaffold.py new file mode 100644 index 0000000..5eef240 --- /dev/null +++ b/tests/test_scaffold.py @@ -0,0 +1,143 @@ +"""Tests for the project scaffold (`milo new`). + +Three layers: + 1. Pure scaffold function — file layout, substitution, error handling. + 2. End-to-end roundtrip — scaffold + import + run + run scaffolded tests. + 3. CLI entry point — `milo new` invokes the scaffold and prints next steps. +""" + +from __future__ import annotations + +import importlib.util +import subprocess +import sys + +import pytest + +from milo._scaffold import ScaffoldError, scaffold + + +class TestScaffoldFunction: + def test_creates_expected_layout(self, tmp_path): + project = scaffold("my_cli", tmp_path) + assert project == tmp_path / "my_cli" + assert (project / "app.py").is_file() + assert (project / "conftest.py").is_file() + assert (project / "tests" / "__init__.py").is_file() + assert (project / "tests" / "test_app.py").is_file() + assert (project / "README.md").is_file() + + def test_substitutes_name_placeholder(self, tmp_path): + project = scaffold("my_cli", tmp_path) + app = (project / "app.py").read_text() + readme = (project / "README.md").read_text() + assert 'CLI(name="my_cli"' in app + assert "my_cli" in readme + assert "{{name}}" not in app + assert "{{name}}" not in readme + + @pytest.mark.parametrize( + "bad_name", + ["My-CLI", "1_cli", "foo bar", "", "FOO", "_leading_underscore"], + ) + def test_rejects_invalid_names(self, bad_name, tmp_path): + with pytest.raises(ScaffoldError, match="Invalid project name"): + scaffold(bad_name, tmp_path) + + def test_refuses_to_overwrite_existing_dir(self, tmp_path): + scaffold("my_cli", tmp_path) + with pytest.raises(ScaffoldError, match="Refusing to overwrite"): + scaffold("my_cli", tmp_path) + + def test_no_template_placeholders_in_default_tree(self): + from milo._scaffold import _TEMPLATE_DIR + + for path in _TEMPLATE_DIR.rglob("*.py"): + content = path.read_text() + assert "TODO" not in content, f"{path} contains TODO" + assert "FIXME" not in content, f"{path} contains FIXME" + assert "XXX" not in content, f"{path} contains XXX" + + +class TestScaffoldRoundtrip: + """Scaffold → import → dispatch → run scaffolded tests.""" + + def test_scaffolded_app_imports_and_dispatches(self, tmp_path): + project = scaffold("roundtrip_cli", tmp_path) + + spec = importlib.util.spec_from_file_location("roundtrip_app", project / "app.py") + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + assert module.cli.name == "roundtrip_cli" + result = module.cli.invoke(["greet", "--name", "Alice"]) + assert result.exit_code == 0 + assert "Hello, Alice!" in result.output + + def test_scaffolded_llms_txt_lists_greet(self, tmp_path): + project = scaffold("llms_cli", tmp_path) + result = subprocess.run( + [sys.executable, str(project / "app.py"), "--llms-txt"], + capture_output=True, + text=True, + timeout=10, + check=True, + ) + assert "**greet**" in result.stdout + assert "**required**" in result.stdout + + def test_scaffolded_test_suite_passes(self, tmp_path): + project = scaffold("test_run_cli", tmp_path) + result = subprocess.run( + [sys.executable, "-m", "pytest", str(project / "tests"), "-v"], + capture_output=True, + text=True, + timeout=30, + check=False, + ) + assert result.returncode == 0, ( + f"Scaffolded tests failed.\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}" + ) + assert "7 passed" in result.stdout + + +class TestMiloNewCommand: + def test_milo_new_creates_project_and_prints_next_steps(self, tmp_path): + result = subprocess.run( + [sys.executable, "-m", "milo.cli", "new", "cli_cmd_test", "--dir", str(tmp_path)], + capture_output=True, + text=True, + timeout=10, + check=True, + ) + assert (tmp_path / "cli_cmd_test").is_dir() + assert (tmp_path / "cli_cmd_test" / "app.py").is_file() + assert "Next steps:" in result.stdout + assert "cd " in result.stdout + assert "uv run python app.py greet --name Alice" in result.stdout + + def test_milo_new_invalid_name_exits_nonzero_with_error(self, tmp_path): + result = subprocess.run( + [sys.executable, "-m", "milo.cli", "new", "Bad-Name", "--dir", str(tmp_path)], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + assert result.returncode == 1 + assert "Invalid project name" in result.stderr + assert not (tmp_path / "Bad-Name").exists() + + def test_milo_new_refuses_existing_target(self, tmp_path): + (tmp_path / "exists").mkdir() + result = subprocess.run( + [sys.executable, "-m", "milo.cli", "new", "exists", "--dir", str(tmp_path)], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + assert result.returncode == 1 + assert "Refusing to overwrite" in result.stderr diff --git a/tests/test_schema_v2.py b/tests/test_schema_v2.py index 6b77209..5d1b7e1 100644 --- a/tests/test_schema_v2.py +++ b/tests/test_schema_v2.py @@ -329,6 +329,76 @@ def test_parse_param_docs_sphinx(self): assert result["count"] == "How many times." +class TestWarnMissingDocs: + """`function_to_schema(..., warn_missing_docs=True)` surfaces undocumented params.""" + + def test_default_is_silent(self): + def f(name: str, count: int) -> str: + """Bare docstring.""" + return "" + + with warnings.catch_warnings(record=True) as ws: + warnings.simplefilter("always") + function_to_schema(f) + assert len(ws) == 0 + + def test_warn_missing_docs_emits_one_warning_per_undocumented_param(self): + def f(name: str, count: int) -> str: + """Bare docstring.""" + return "" + + with warnings.catch_warnings(record=True) as ws: + warnings.simplefilter("always") + function_to_schema(f, warn_missing_docs=True) + msgs = [str(w.message) for w in ws if issubclass(w.category, UserWarning)] + assert any("'name'" in m for m in msgs) + assert any("'count'" in m for m in msgs) + assert len(msgs) == 2 + + def test_warn_missing_docs_silent_when_all_documented(self): + def f(name: str, count: int) -> str: + """Greet. + + Args: + name: Who to greet. + count: How many times. + """ + return "" + + with warnings.catch_warnings(record=True) as ws: + warnings.simplefilter("always") + function_to_schema(f, warn_missing_docs=True) + msgs = [str(w.message) for w in ws if issubclass(w.category, UserWarning)] + assert msgs == [] + + def test_description_constraint_suppresses_warning(self): + """Annotated[..., Description("...")] counts as documentation.""" + + def f(name: Annotated[str, Description("via constraint")]) -> str: + """Bare docstring.""" + return "" + + with warnings.catch_warnings(record=True) as ws: + warnings.simplefilter("always") + function_to_schema(f, warn_missing_docs=True) + msgs = [str(w.message) for w in ws if issubclass(w.category, UserWarning)] + assert msgs == [] + + def test_warning_fires_on_repeated_calls(self): + """The cache must not swallow the warning on second invocation.""" + + def f(name: str) -> str: + return "" + + for _ in range(2): + with warnings.catch_warnings(record=True) as ws: + warnings.simplefilter("always") + function_to_schema(f, warn_missing_docs=True) + assert any( + "'name'" in str(w.message) for w in ws if issubclass(w.category, UserWarning) + ) + + # --------------------------------------------------------------------------- # Annotated constraint tests # --------------------------------------------------------------------------- diff --git a/tests/test_verify.py b/tests/test_verify.py new file mode 100644 index 0000000..4a38c53 --- /dev/null +++ b/tests/test_verify.py @@ -0,0 +1,241 @@ +"""Tests for the `milo verify` self-diagnosis pipeline.""" + +from __future__ import annotations + +import subprocess +import sys +import textwrap +from pathlib import Path + +import pytest + +from milo._scaffold import scaffold +from milo.verify import VerifyCheck, VerifyReport, verify + +_REPO_ROOT = Path(__file__).resolve().parents[1] +_EXAMPLES_DIR = _REPO_ROOT / "examples" + + +# Examples that legitimately cannot pass `milo verify` as-is — not bugs in +# verify, just incompatibility with the one-CLI-instance model. Document *why* +# so future contributors don't silently grow the list. +_SKIP_EXAMPLES: dict[str, str] = { + # Add entries here with a reason. Empty by default — every example should + # be verifiable, so new exceptions are a regression signal. +} + + +class TestVerifyReport: + def test_empty_report_has_zero_counts(self): + r = VerifyReport(target="x", checks=()) + assert r.passed == 0 + assert r.failures == 0 + assert r.warnings == 0 + assert r.exit_code == 0 + + def test_failure_sets_exit_code_one(self): + r = VerifyReport( + target="x", + checks=(VerifyCheck(name="k", status="fail", message="m"),), + ) + assert r.exit_code == 1 + + def test_warning_does_not_set_exit_code(self): + r = VerifyReport( + target="x", + checks=(VerifyCheck(name="k", status="warn", message="m"),), + ) + assert r.exit_code == 0 + + def test_format_contains_target_and_all_checks(self): + r = VerifyReport( + target="my.py", + checks=( + VerifyCheck(name="a", status="ok", message="good"), + VerifyCheck(name="b", status="fail", message="bad", details="line1\nline2"), + ), + ) + out = r.format() + assert "milo verify my.py" in out + assert "a: good" in out + assert "b: bad" in out + assert "line1" in out + assert "line2" in out + + +class TestVerifyScaffold: + """End-to-end: scaffold a project, then verify it passes.""" + + def test_freshly_scaffolded_project_passes(self, tmp_path): + project = scaffold("verify_scaffold", tmp_path) + report = verify(str(project / "app.py")) + assert report.exit_code == 0, report.format() + assert report.failures == 0 + assert report.passed >= 6 + # Confirm every expected check is present and passed + check_names = {c.name for c in report.checks} + for expected in { + "imports", + "cli_located", + "commands_registered", + "schemas_generate", + "mcp_list_tools", + "mcp_transport", + }: + assert expected in check_names + + +class TestVerifyFailurePaths: + def test_missing_file_fails_imports(self, tmp_path): + report = verify(str(tmp_path / "does_not_exist.py")) + assert report.exit_code == 1 + assert report.checks[0].name == "imports" + assert report.checks[0].status == "fail" + + def test_import_error_fails_imports(self, tmp_path): + broken = tmp_path / "syntax_error.py" + broken.write_text("this is not valid python :(") + report = verify(str(broken)) + assert report.exit_code == 1 + assert report.checks[0].status == "fail" + assert "SyntaxError" in report.checks[0].message + + def test_module_without_cli_instance_fails(self, tmp_path): + no_cli = tmp_path / "no_cli.py" + no_cli.write_text("def hello(): return 'hi'\n") + report = verify(str(no_cli)) + assert report.exit_code == 1 + names_and_statuses = [(c.name, c.status) for c in report.checks] + assert ("cli_located", "fail") in names_and_statuses + + def test_cli_with_no_commands_fails(self, tmp_path): + empty = tmp_path / "empty_cli.py" + empty.write_text( + textwrap.dedent( + """ + from milo import CLI + cli = CLI(name="empty", version="0.1") + """ + ) + ) + report = verify(str(empty)) + assert report.exit_code == 1 + names_and_statuses = [(c.name, c.status) for c in report.checks] + assert ("commands_registered", "fail") in names_and_statuses + + def test_undocumented_param_produces_warning(self, tmp_path): + undoc = tmp_path / "undoc.py" + undoc.write_text( + textwrap.dedent( + """ + from milo import CLI + cli = CLI(name="undoc", version="0.1") + + @cli.command("hello", description="Say hi") + def hello(name: str, extra: int = 0) -> str: + \"\"\"Say hi. + + Args: + name: Who to greet. + \"\"\" + return f"hi {name} ({extra})" + + if __name__ == "__main__": + cli.run() + """ + ) + ) + report = verify(str(undoc)) + assert report.exit_code == 0 # warning only, not failure + assert report.warnings >= 1 + schemas_check = next(c for c in report.checks if c.name == "schemas_generate") + assert schemas_check.status == "warn" + assert "extra" in schemas_check.details + + def test_unknown_target_format_fails_imports(self): + report = verify("not-a-path-not-a-module") + assert report.exit_code == 1 + assert report.checks[0].name == "imports" + assert report.checks[0].status == "fail" + + +class TestVerifyModuleAttrForm: + def test_module_attr_form_skips_subprocess_check(self, tmp_path, monkeypatch): + # Create a package-style module and import by module:attr + pkg_dir = tmp_path / "mypkg" + pkg_dir.mkdir() + (pkg_dir / "__init__.py").write_text( + textwrap.dedent( + """ + from milo import CLI + cli = CLI(name="mypkg", version="0.1") + + @cli.command("hello", description="Say hi") + def hello(name: str) -> str: + \"\"\"Say hi. + + Args: + name: Who to greet. + \"\"\" + return f"hi {name}" + """ + ) + ) + monkeypatch.chdir(tmp_path) + report = verify("mypkg:cli") + transport_check = next(c for c in report.checks if c.name == "mcp_transport") + assert transport_check.status == "skip" + assert report.exit_code == 0 + + +# --------------------------------------------------------------------------- +# Regression gate: every CLI example in examples/ must pass `milo verify`. +# `milo verify` targets the CLI/MCP protocol, so App-based TUI examples +# (which don't construct a `CLI(...)` at module scope) are out of scope. +# --------------------------------------------------------------------------- + + +def _example_ids() -> list[str]: + """Return example names whose app.py constructs a top-level `CLI(...)`.""" + ids: list[str] = [] + for app_py in _EXAMPLES_DIR.glob("*/app.py"): + text = app_py.read_text(encoding="utf-8") + if "CLI(" in text: + ids.append(app_py.parent.name) + return sorted(ids) + + +@pytest.mark.parametrize("example_name", _example_ids()) +def test_example_passes_verify(example_name): + if example_name in _SKIP_EXAMPLES: + pytest.skip(_SKIP_EXAMPLES[example_name]) + app_path = _EXAMPLES_DIR / example_name / "app.py" + report = verify(str(app_path), timeout=10.0) + if report.exit_code != 0: + pytest.fail(f"Example {example_name!r} fails `milo verify`:\n\n{report.format()}") + + +class TestMiloVerifyCommand: + def test_milo_verify_exits_zero_on_valid_cli(self, tmp_path): + project = scaffold("cmd_ok", tmp_path) + result = subprocess.run( + [sys.executable, "-m", "milo.cli", "verify", str(project / "app.py")], + capture_output=True, + text=True, + timeout=15, + check=False, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "6 passed" in result.stdout + + def test_milo_verify_exits_nonzero_on_missing_file(self, tmp_path): + result = subprocess.run( + [sys.executable, "-m", "milo.cli", "verify", str(tmp_path / "nope.py")], + capture_output=True, + text=True, + timeout=15, + check=False, + ) + assert result.returncode == 1 + assert "imports" in result.stdout + assert "fail" in result.stdout or "✗" in result.stdout