diff --git a/cliffy/ai.py b/cliffy/ai.py index d1edcb0..84732b4 100644 --- a/cliffy/ai.py +++ b/cliffy/ai.py @@ -219,6 +219,9 @@ def validate_manifest_yaml(text: str, cli_name: str) -> str: ) raise ManifestOutputError(f"The manifest does not match the schema:\n{problems}") from e + if "commands" not in data and not manifest.callbacks: + # the schema allows a callbacks-only CLI, so a reply without `commands` passes its validation + raise ManifestOutputError("The manifest does not match the schema:\n- commands: Field required") if manifest.name != cli_name: raise ManifestOutputError(f"The manifest `name` must be `{cli_name}`, not `{manifest.name}`.") if manifest.includes: diff --git a/cliffy/blueprint.py b/cliffy/blueprint.py index e5a5c49..c3ae20f 100644 --- a/cliffy/blueprint.py +++ b/cliffy/blueprint.py @@ -1,64 +1,290 @@ -from datetime import datetime -from typing import Any, Self +"""Blueprints: bring your own YAML vocabulary. -from pydantic import BaseModel, Field +A blueprint is a typed spec. It is a pydantic model that describes a custom YAML schema, plus a +``to_manifest`` method that says how that schema maps onto a cliffy manifest. Everything downstream of the +manifest (both commanders, ``render``, ``test``, ``docs``, ``build``) works unchanged. +A YAML file selects its blueprint with a top-level ``blueprint:`` key or with ``--blueprint`` on the command +line. A blueprint reference is one of: -class Blueprint(BaseModel): - """Base class for custom manifest blueprints that define CLI generation schemas""" +* a path to a Python file, optionally with a class name: ``./tasks.py`` or ``./tasks.py:Tasks`` +* an importable class: ``package.module:Tasks`` (or ``package.module`` when it defines one blueprint) +* the name of an installed plugin, registered under the ``cliffy.blueprints`` entry-point group +""" - name: str = Field(..., description="Name of the blueprint") - version: str = Field(..., description="Version of the blueprint") - help: str = Field(default="", description="Help text for the blueprint") - loaded: datetime = Field(default_factory=datetime.now) +from __future__ import annotations - def validate_hook(self) -> bool: - """Hook for custom validation logic +import hashlib +import importlib +import importlib.util +import inspect +import os +import sys +from abc import ABC, abstractmethod +from dataclasses import dataclass +from importlib.metadata import EntryPoint, entry_points +from pathlib import Path +from types import ModuleType +from typing import Any, ClassVar - Returns: - bool: True if validation passes, False otherwise - """ - return True +from pydantic import BaseModel, ConfigDict, ValidationError - def load_hook(self) -> None: - """Hook called when blueprint is loaded""" +from cliffy.manifest import CLIManifest - def init_hook(self) -> None: - """Hook called during blueprint initialization""" +ENTRY_POINT_GROUP = "cliffy.blueprints" +BLUEPRINT_KEY = "blueprint" - def pre_process_hook(self) -> None: - """Hook called before processing blueprint""" - def post_process_hook(self) -> None: - """Hook called after processing blueprint""" +class BlueprintError(Exception): + """A blueprint could not be found, loaded or applied. The message is meant to be shown to the user.""" + + +class Blueprint(BaseModel, ABC): + """Base class for a custom YAML spec. + + Subclass it, declare the fields of your spec as pydantic fields, and implement ``to_manifest``. + Unknown keys are rejected by default so that typos in a spec are reported; set + ``model_config = ConfigDict(extra="ignore")`` in a subclass to change that. + + Hooks run in this order:: + + pre_validate(raw dict) -> validation -> to_manifest() -> post_manifest(manifest) -> post_codegen(code) + """ + + model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid") + + @classmethod + def pre_validate(cls, raw: dict[str, Any]) -> dict[str, Any]: + """Hook: adjust the raw YAML mapping before it is validated against the spec.""" + return raw + + @abstractmethod + def to_manifest(self) -> CLIManifest | dict[str, Any]: + """Map the validated spec onto a cliffy manifest (a ``CLIManifest`` or a mapping that validates as one).""" + + def post_manifest(self, manifest: CLIManifest) -> CLIManifest: + """Hook: adjust the manifest after ``to_manifest`` and before code generation.""" + return manifest + + def post_codegen(self, code: str) -> str: + """Hook: adjust the generated CLI source code.""" + return code + + @classmethod + def describe(cls) -> str: + """One-line description of the blueprint: the first line of the class docstring.""" + doc = inspect.cleandoc(cls.__doc__ or "") + return doc.splitlines()[0] if doc else "" @classmethod - def create(cls, **data: Any) -> Self: - """Factory method to create a new blueprint instance - - Args: - **data: Blueprint configuration data - - Returns: - Blueprint instance - """ - instance = cls(**data) - instance.init_hook() - return instance - - def process(self) -> None: - """Main processing pipeline for the blueprint - - Executes hooks in order: - 1. pre_process_hook - 2. validate_hook - 3. load_hook - 4. post_process_hook - """ - self.pre_process_hook() - - if not self.validate_hook(): - raise ValueError(f"Blueprint validation failed for {self.name}") - - self.load_hook() - self.post_process_hook() + def spec_json_schema(cls) -> dict[str, Any]: + """JSON schema for YAML files written in this spec, including the optional ``blueprint`` key.""" + schema = cls.model_json_schema() + schema.setdefault("properties", {}).setdefault( + BLUEPRINT_KEY, + {"type": "string", "title": "Blueprint", "description": "Blueprint that turns this file into a CLI."}, + ) + return schema + + +@dataclass(frozen=True) +class InstalledBlueprint: + """A blueprint registered by an installed package under the ``cliffy.blueprints`` entry-point group.""" + + name: str + target: str + package: str + entry_point: EntryPoint + + def load(self) -> type[Blueprint]: + try: + loaded = self.entry_point.load() + except Exception as e: + raise BlueprintError( + f"could not load blueprint `{self.name}` ({self.target}) from package `{self.package}`: " + f"{type(e).__name__}: {e}" + ) from e + if inspect.ismodule(loaded): + return _pick_blueprint_class(loaded, None, self.target) + return _ensure_blueprint_class(loaded, self.target) + + +def installed_blueprints() -> list[InstalledBlueprint]: + """All blueprints provided by installed packages, sorted by name.""" + found = [ + InstalledBlueprint(name=ep.name, target=ep.value, package=ep.dist.name if ep.dist else "", entry_point=ep) + for ep in entry_points(group=ENTRY_POINT_GROUP) + ] + return sorted(found, key=lambda bp: (bp.name, bp.package)) + + +def normalize_reference(reference: str, base_dir: str | os.PathLike[str] | None = None) -> str: + """Make a file reference absolute (relative to ``base_dir``, default: cwd). Other references are unchanged.""" + target, attr = _split_reference(reference.strip()) + if not _is_file_reference(target): + return reference.strip() + path = Path(target).expanduser() + if not path.is_absolute(): + path = Path(base_dir or Path.cwd()) / path + resolved = str(path.resolve()) + return f"{resolved}:{attr}" if attr else resolved + + +def resolve_blueprint(reference: str, base_dir: str | os.PathLike[str] | None = None) -> type[Blueprint]: + """Resolve a blueprint reference to a ``Blueprint`` subclass. + + Raises: + BlueprintError: the reference does not exist, is ambiguous, or is not a blueprint. + """ + reference = reference.strip() + if not reference: + raise BlueprintError("empty blueprint reference") + + target, attr = _split_reference(reference) + if _is_file_reference(target): + path = Path(_split_reference(normalize_reference(reference, base_dir))[0]) + if not path.is_file(): + raise BlueprintError(f"blueprint file not found: {path}") + return _pick_blueprint_class(_import_file(path), attr, str(path)) + + if attr is None: + plugins = [bp for bp in installed_blueprints() if bp.name == target] + if len(plugins) > 1: + providers = ", ".join(f"{bp.package} ({bp.target})" for bp in plugins) + raise BlueprintError( + f"blueprint `{target}` is ambiguous, it is provided by: {providers}. " + "Use a `module:Class` reference to pick one." + ) + if plugins: + return plugins[0].load() + + return _pick_blueprint_class(_import_module(target, reference), attr, target) + + +def build_manifest( + blueprint_cls: type[Blueprint], raw: dict[str, Any], source: str = "spec" +) -> tuple[Blueprint, CLIManifest]: + """Validate ``raw`` against the blueprint's spec and transform it into a cliffy manifest. + + Raises: + BlueprintError: the spec is invalid, or the blueprint produced an invalid manifest. + """ + name = blueprint_cls.__name__ + data = {key: value for key, value in raw.items() if key != BLUEPRINT_KEY} + try: + spec = blueprint_cls.model_validate(blueprint_cls.pre_validate(data)) + except ValidationError as e: + raise BlueprintError(f"{source} is not a valid {name} spec\n{format_validation_error(e)}") from e + except Exception as e: + raise BlueprintError(f"blueprint {name} failed to validate {source}: {type(e).__name__}: {e}") from e + + try: + produced = spec.to_manifest() + manifest = produced if isinstance(produced, CLIManifest) else CLIManifest.model_validate(produced) + manifest = spec.post_manifest(manifest) + except ValidationError as e: + raise BlueprintError( + f"blueprint {name} produced an invalid cliffy manifest for {source}\n{format_validation_error(e)}" + ) from e + except Exception as e: + raise BlueprintError(f"blueprint {name} failed to transform {source}: {type(e).__name__}: {e}") from e + + if not isinstance(manifest, CLIManifest): + raise BlueprintError(f"blueprint {name}.post_manifest must return a CLIManifest, got {type(manifest).__name__}") + return spec, manifest + + +def format_validation_error(error: ValidationError) -> str: + """Render a pydantic validation error as one `` path.to.key: message`` line per problem.""" + lines = [] + for problem in error.errors(): + location = "" + for part in problem["loc"]: + location += f"[{part}]" if isinstance(part, int) else f".{part}" if location else str(part) + value = problem.get("input") + got = "" if isinstance(value, dict | list) or value is None else f" (got {value!r})" + lines.append(f" {location or ''}: {problem['msg']}{got}") + return "\n".join(lines) + + +def _split_reference(reference: str) -> tuple[str, str | None]: + """``target:Class`` -> (target, Class). A Windows drive letter is not mistaken for a target.""" + target, sep, attr = reference.rpartition(":") + if sep and target and attr.isidentifier(): + return target, attr + return reference, None + + +def _is_file_reference(target: str) -> bool: + return target.endswith(".py") or "/" in target or os.sep in target + + +def _import_file(path: Path) -> ModuleType: + digest = hashlib.sha256(str(path).encode()).hexdigest()[:12] + module_name = f"cliffy_blueprint_{digest}" + spec = importlib.util.spec_from_file_location(module_name, path) + if spec is None or spec.loader is None: + raise BlueprintError(f"blueprint file is not a Python module: {path}") + + module = importlib.util.module_from_spec(spec) + # registered while executing so that pydantic can resolve the module's postponed annotations + sys.modules[module_name] = module + try: + spec.loader.exec_module(module) + except Exception as e: + sys.modules.pop(module_name, None) + raise BlueprintError(f"could not import blueprint file {path}: {type(e).__name__}: {e}") from e + return module + + +def _import_module(module_name: str, reference: str) -> ModuleType: + try: + return importlib.import_module(module_name) + except ModuleNotFoundError as e: + if e.name and (module_name == e.name or module_name.startswith(f"{e.name}.")): + known = ", ".join(bp.name for bp in installed_blueprints()) or "none" + raise BlueprintError( + f"blueprint `{reference}` not found. It is not an installed blueprint (installed: {known}), " + "not an importable module, and not a path to a .py file." + ) from e + raise BlueprintError(f"could not import blueprint `{reference}`: {type(e).__name__}: {e}") from e + except Exception as e: + raise BlueprintError(f"could not import blueprint `{reference}`: {type(e).__name__}: {e}") from e + + +def _is_concrete_blueprint(obj: Any) -> bool: + return inspect.isclass(obj) and issubclass(obj, Blueprint) and not inspect.isabstract(obj) + + +def _ensure_blueprint_class(obj: Any, reference: str) -> type[Blueprint]: + if not _is_concrete_blueprint(obj): + raise BlueprintError( + f"`{reference}` is not a blueprint. A blueprint is a subclass of cliffy.blueprint.Blueprint " + "that implements to_manifest()." + ) + blueprint_cls: type[Blueprint] = obj + return blueprint_cls + + +def _pick_blueprint_class(module: ModuleType, attr: str | None, reference: str) -> type[Blueprint]: + if attr: + if not hasattr(module, attr): + raise BlueprintError(f"`{reference}` has no attribute `{attr}`") + return _ensure_blueprint_class(getattr(module, attr), f"{reference}:{attr}") + + candidates = [obj for obj in vars(module).values() if _is_concrete_blueprint(obj)] + # prefer the blueprints defined in the module over the ones it only imports + defined = [obj for obj in candidates if obj.__module__ == module.__name__] + candidates = defined or candidates + if not candidates: + raise BlueprintError( + f"no blueprint found in `{reference}`. Define a subclass of cliffy.blueprint.Blueprint " + "that implements to_manifest()." + ) + if len(candidates) > 1: + names = ", ".join(sorted(cls.__name__ for cls in candidates)) + raise BlueprintError( + f"`{reference}` defines more than one blueprint ({names}). Pick one with `{reference}:ClassName`." + ) + blueprint_cls: type[Blueprint] = candidates[0] + return blueprint_cls diff --git a/cliffy/builder.py b/cliffy/builder.py index a8318ea..101abda 100644 --- a/cliffy/builder.py +++ b/cliffy/builder.py @@ -1,37 +1,63 @@ -import os +import contextlib +import io +import re +import subprocess import sys +from dataclasses import dataclass +from importlib import metadata from io import TextIOWrapper from pathlib import Path -from shutil import copy from tempfile import NamedTemporaryFile, TemporaryDirectory -from click.testing import CliRunner, Result +from packaging.requirements import InvalidRequirement, Requirement +from packaging.specifiers import InvalidSpecifier, SpecifierSet +from packaging.utils import canonicalize_name from shiv import cli as shiv_cli -from shiv import pip from cliffy.commander import CLI from cliffy.helper import TEMP_FILES, delete_temp_files, import_module_from_path from cliffy.transformer import Transformer +DEFAULT_INTERPRETER = "/usr/bin/env python3" +# Package that provides the framework a generated CLI imports +FRAMEWORK_PACKAGES = {"typer": "typer", "rich_click": "rich-click", "click": "click"} +GENERATED_HEADER = re.compile(r"\A## Generated (\S+) on .*\n") + + +@dataclass +class BuildResult: + """Outcome of a build. `output` holds the reason when `exit_code` is not 0.""" + + exit_code: int + output: str = "" + exception: BaseException | None = None + path: str | None = None + requires_python: str | None = None + def build_cli_from_manifest( - manifestIO: TextIOWrapper, output_dir: str | None = None, interpreter: str = "/usr/bin/env python3" -) -> tuple[CLI, Result]: - T = Transformer(manifestIO, validate_requires=False) - with NamedTemporaryFile(mode="w", prefix=f"{T.cli.name}_", suffix=".py", delete=False, encoding="utf-8") as script: - TEMP_FILES.append(script) - try: - script.write(T.cli.code) - script.flush() - result = build_cli( - T.cli.name, - script_path=script.name, - deps=T.cli.requires, - output_dir=output_dir, - interpreter=interpreter, - ) - finally: - delete_temp_files() + manifestIO: TextIOWrapper, + output_dir: str | None = None, + interpreter: str = DEFAULT_INTERPRETER, + blueprint: str | None = None, +) -> tuple[CLI, BuildResult]: + """Build a manifest, or a YAML file written in a blueprint's spec, into a single-file executable. + + The blueprint is applied here, at build time. The executable holds the generated CLI and its + dependencies and needs neither the blueprint nor the YAML file to run. + """ + T = Transformer(manifestIO, validate_requires=False, blueprint=blueprint) + # the directory is removed with the script in it, also when the build fails + with TemporaryDirectory() as script_dir: + script_path = Path(script_dir) / f"{T.cli.name}.py" + script_path.write_text(T.cli.code, encoding="utf-8") + result = build_cli( + T.cli.name, + script_path=str(script_path), + deps=T.cli.requires, + output_dir=output_dir, + interpreter=interpreter, + ) return T.cli, result @@ -41,35 +67,112 @@ def build_cli( script_path: str, deps: list[str] | None = None, output_dir: str | None = None, - interpreter: str = "/usr/bin/env python3", -) -> Result: - if deps is None: - deps = [] + interpreter: str = DEFAULT_INTERPRETER, +) -> BuildResult: + """Bundle a generated CLI script and its dependencies into one executable zipapp named `cli_name`. - if output_dir: - os.makedirs(output_dir, exist_ok=True) + Dependencies are installed with the pip of the running interpreter. The framework the CLI imports + (typer or rich-click) is pinned to the version installed next to cliffy unless `deps` names it, and + the archive is written with fixed timestamps, so equal inputs give an identical file. + """ + try: + code = Path(script_path).read_text(encoding="utf-8") + except OSError as e: + return BuildResult(1, f"could not read the generated CLI at {script_path}", e) + + module_name = cli_name.replace("-", "_") + output_path = Path(output_dir or ".") / cli_name + requirements = get_build_requirements(code, deps or []) + + with TemporaryDirectory() as staging: + pip_install = [sys.executable, "-m", "pip", "--disable-pip-version-check", "install", "--no-compile"] + pip = subprocess.run([*pip_install, "--target", staging, *requirements], capture_output=True, text=True) + if pip.returncode != 0: + reason = (pip.stderr.strip() or pip.stdout.strip()).splitlines()[-15:] + return BuildResult(1, f"could not install {', '.join(requirements)}:\n" + "\n".join(reason)) + + requires_python = get_requires_python(Path(staging)) + code = GENERATED_HEADER.sub(r"## Generated \1 with cliffy\n", code) + Path(staging, f"{module_name}.py").write_text( + get_python_guard(cli_name, requires_python) + code, encoding="utf-8" + ) - with TemporaryDirectory() as tdist: - copy(script_path, os.path.join(tdist, f"{cli_name}.py")) - pip_deps = [*get_framework_requirements(script_path), *deps] + shiv_args = ["--site-packages", staging, "--compressed", "--reproducible"] + shiv_args += ["-e", f"{module_name}.cli", "-o", str(output_path), "-p", interpreter] + shiv_output = io.StringIO() + try: + output_path.parent.mkdir(parents=True, exist_ok=True) + with contextlib.redirect_stdout(shiv_output), contextlib.redirect_stderr(shiv_output): + shiv_cli.main.main(args=shiv_args, standalone_mode=False) + except (Exception, SystemExit) as e: + return BuildResult(1, f"{shiv_output.getvalue()}could not write {output_path}", e) - pip.install(["--target", tdist, *pip_deps]) + return BuildResult(0, path=str(output_path), requires_python=requires_python) - runner = CliRunner() - output_file = os.path.join(output_dir, f"{cli_name}") if output_dir else cli_name - return runner.invoke( - shiv_cli.main, # type: ignore - ["--site-packages", tdist, "--compressed", "-e", f"{cli_name}.cli", "-o", output_file, "-p", interpreter], - ) + +def get_build_requirements(code: str, deps: list[str]) -> list[str]: + """Requirements to bundle: the manifest's `requires` plus the framework the generated code imports. + + The framework is read from the generated code, because a loaded CLI is built from its script alone. + """ + named = set() + for dep in deps: + with contextlib.suppress(InvalidRequirement): + named.add(canonicalize_name(Requirement(dep).name)) + + requirements = list(deps) + for module, package in FRAMEWORK_PACKAGES.items(): + if package in named or not re.search(rf"^\s*(import|from) {module}\b", code, flags=re.MULTILINE): + continue + try: + requirements.append(f"{package}=={metadata.version(package)}") + except metadata.PackageNotFoundError: + requirements.append(package) + return requirements -def get_framework_requirements(script_path: str) -> list[str]: - """Packages the generated CLI imports no matter what the manifest requires. +def get_requires_python(site_packages: Path) -> str: + """Python versions the bundled packages can run on, as `>=3.X` or `==3.X.*`. - Read from the generated code, because a loaded CLI is built from its script alone. + A bundle with compiled extension modules only runs on the Python minor version it was built with. + Otherwise the lowest version accepted by every bundled package is the floor. """ - code = Path(script_path).read_text(encoding="utf-8") - return ["rich-click"] if "import rich_click as click" in code else ["typer"] + compiled = [ + path + for pattern in ("*.so", "*.pyd", "*.dylib") + for path in site_packages.rglob(pattern) + if ".abi3." not in path.name + ] + if compiled: + return f"==3.{sys.version_info.minor}.*" + + specifiers = [] + for metadata_file in site_packages.glob("*.dist-info/METADATA"): + for line in metadata_file.read_text(encoding="utf-8", errors="replace").splitlines(): + if not line.strip(): + break + if line.lower().startswith("requires-python:"): + with contextlib.suppress(InvalidSpecifier): + specifiers.append(SpecifierSet(line.split(":", 1)[1].strip())) + + for minor in range(sys.version_info.minor): + if all(specifier.contains(f"3.{minor}.99") for specifier in specifiers): + return f">=3.{minor}" + return f">=3.{sys.version_info.minor}" + + +def get_python_guard(cli_name: str, requires_python: str) -> str: + """Code put at the top of the bundled CLI so that an unsupported interpreter gets a readable error.""" + exact = requires_python.startswith("==") + minor = int(requires_python.removesuffix(".*").split(".")[1]) + wanted = f"Python 3.{minor}" if exact else f"Python 3.{minor} or newer" + message = f"{cli_name} was built for {wanted}, but is running on Python " + return f"""import sys + +if not sys.version_info[:2] {"==" if exact else ">="} (3, {minor}): + raise SystemExit({message!r} + sys.version.split()[0] + ". Run it with a matching interpreter.") + +""" def run_cli(cli_name: str, script_code: str, args: tuple) -> None: diff --git a/cliffy/cli.py b/cliffy/cli.py index 2d97fe8..c3313c6 100644 --- a/cliffy/cli.py +++ b/cliffy/cli.py @@ -1,9 +1,11 @@ +import json import sys import traceback from io import TextIOWrapper from typing import Any, TextIO from cliffy.ai import ai +from cliffy.blueprint import BlueprintError, installed_blueprints, resolve_blueprint from cliffy.builder import build_cli, build_cli_from_manifest, run_cli from cliffy.doc import DocGenerator from cliffy.helper import ( @@ -32,6 +34,7 @@ "rm": "remove", "rm-all": "remove-all", "rmall": "remove-all", + "blueprints": "blueprint list", } @@ -43,6 +46,16 @@ def show_aliases_callback(ctx: Any, param: Any, val: bool) -> None: ctx.exit() +blueprint_option = click.option( + "--blueprint", + "-b", + type=str, + default=None, + help="Blueprint that maps a custom YAML spec onto a CLI: a .py file, module:Class or an installed blueprint name. " + "Overrides the `blueprint:` key of the file.", +) + + @click.group(context_settings=CONTEXT_SETTINGS) @click.version_option() @click.option("--aliases", type=bool, is_flag=True, is_eager=True, callback=show_aliases_callback) @@ -51,10 +64,11 @@ def cli(aliases: bool) -> None: @click.argument("manifests", type=click.File("rb"), nargs=-1) -def load(manifests: list[TextIO]) -> None: +@blueprint_option +def load(manifests: list[TextIO], blueprint: str | None = None) -> None: """Load CLI for given manifest(s)""" for manifest in manifests: - T = Transformer(manifest) + T = Transformer(manifest, blueprint=blueprint) Loader.load_from_cli(T.cli) save_metadata(manifest.name, T.cli) out(f"✨ Generated {T.cli.name} CLI v{T.cli.version} ✨", fg="green") @@ -69,7 +83,7 @@ def update(cli_names: list[str]) -> None: for cli_name in cli_names: if cli_metadata := get_metadata(cli_name): with open(cli_metadata.runner_path, encoding="utf-8") as manifest: - T = Transformer(manifest) + T = Transformer(manifest, blueprint=cli_metadata.blueprint) Loader.load_from_cli(T.cli) save_metadata(cli_metadata.runner_path, T.cli) out(f"✨ Reloaded {T.cli.name} CLI v{T.cli.version} ✨", fg="green") @@ -84,9 +98,10 @@ def update(cli_names: list[str]) -> None: @click.argument("manifest", type=click.File("rb")) -def render(manifest: TextIO) -> None: +@blueprint_option +def render(manifest: TextIO, blueprint: str | None = None) -> None: """Render the CLI manifest generation as code""" - T = Transformer(manifest) + T = Transformer(manifest, blueprint=blueprint) console = Console() console.print(T.cli.code, overflow="fold", emoji=False, markup=False) out(f"# Rendered {T.cli.name} CLI v{T.cli.version} ~", fg="green") @@ -94,9 +109,10 @@ def render(manifest: TextIO) -> None: @click.argument("manifest", type=click.File("rb")) @click.argument("cli_args", type=str, nargs=-1) -def cliffy_run(manifest: TextIO, cli_args: tuple[str]) -> None: +@blueprint_option +def cliffy_run(manifest: TextIO, cli_args: tuple[str], blueprint: str | None = None) -> None: """Run CLI for a manifest""" - T = Transformer(manifest) + T = Transformer(manifest, blueprint=blueprint) run_cli(T.cli.name, T.cli.code, cli_args) @@ -194,12 +210,17 @@ def remove_all() -> None: default="/usr/bin/env python3", show_default=True, ) -def build(cli_or_manifests: list[TextIOWrapper | str], output_dir: str, python: str) -> None: - """Build a CLI manifest or a loaded CLI into a zipapp""" +@blueprint_option +def build( + cli_or_manifests: list[TextIOWrapper | str], output_dir: str, python: str, blueprint: str | None = None +) -> None: + """Build a CLI manifest or a loaded CLI into a single-file executable (zipapp)""" failed = False for cli_or_manifest in cli_or_manifests: if isinstance(cli_or_manifest, TextIOWrapper): - cli, result = build_cli_from_manifest(cli_or_manifest, output_dir=output_dir, interpreter=python) + cli, result = build_cli_from_manifest( + cli_or_manifest, output_dir=output_dir, interpreter=python, blueprint=blueprint + ) cli_name = cli.name else: cli_name = cli_or_manifest @@ -225,6 +246,7 @@ def build(cli_or_manifests: list[TextIOWrapper | str], output_dir: str, python: continue out(f"+ {cli_name} built 📦", fg="green") + out(f" {result.path} (single file, needs Python {result.requires_python})") if failed: raise SystemExit(1) @@ -237,6 +259,8 @@ def info(cli_name: str) -> None: out(f"{click.style('name:', fg='blue')} {metadata.cli_name}") out(f"{click.style('version:', fg='blue')} {metadata.version}") out(f"{click.style('requires:', fg='blue')} {metadata.requires}") + if metadata.blueprint: + out(f"{click.style('blueprint:', fg='blue')} {metadata.blueprint}") out(f"{click.style('age:', fg='blue')} {age_datetime(metadata.loaded)} ({metadata.loaded.ctime()})") out(f"{click.style('manifest:', fg='blue')}\n{indent_block(metadata.manifest, spaces=2)}") @@ -249,7 +273,8 @@ def info(cli_name: str) -> None: is_flag=True, ) @click.argument("run-cli-args", type=str, nargs=-1) -def dev(manifest: str, run_cli: bool, run_cli_args: tuple[str]) -> None: +@blueprint_option +def dev(manifest: str, run_cli: bool, run_cli_args: tuple[str], blueprint: str | None = None) -> None: """Start hot-reloader for a manifest for active development. Examples: @@ -261,7 +286,10 @@ def dev(manifest: str, run_cli: bool, run_cli_args: tuple[str]) -> None: - cli dev examples/hello.yaml --run-cli hello """ out(f"🔄 Watching {manifest} for changes...\n", fg="magenta") - Reloader.watch(manifest, run_cli, run_cli_args) + if blueprint: + Reloader.watch(manifest, run_cli, run_cli_args, blueprint=blueprint) + else: + Reloader.watch(manifest, run_cli, run_cli_args) @click.argument("manifest", type=click.Path(exists=True, file_okay=True, dir_okay=False, readable=True)) @@ -273,9 +301,10 @@ def dev(manifest: str, run_cli: bool, run_cli_args: tuple[str]) -> None: default=False, help="exit instantly on first error or failed test.", ) -def test(manifest: str, exitfirst: bool) -> None: +@blueprint_option +def test(manifest: str, exitfirst: bool, blueprint: str | None = None) -> None: """Run tests defined in a manifest""" - tester = Tester(manifest) + tester = Tester(manifest, blueprint=blueprint) if not tester.test_pipeline: exit_err("Missing tests section in manifest") out("✨ Invoking tests ✨", nl=False) @@ -326,10 +355,11 @@ def test(manifest: str, exitfirst: bool) -> None: @click.argument("manifest", type=click.File("rb"), required=True) -def validate(manifest: TextIO) -> None: +@blueprint_option +def validate(manifest: TextIO, blueprint: str | None = None) -> None: """Validate the syntax and structure of a CLI manifest""" try: - Transformer(manifest) + Transformer(manifest, blueprint=blueprint) out(f"Manifest {manifest.name} is valid", fg="green") except Exception as e: exit_err(f"Manifest {manifest.name} is invalid: {e}") @@ -340,10 +370,11 @@ def validate(manifest: TextIO) -> None: @click.option( "--output-dir", "-o", type=click.Path(exists=True, dir_okay=True, file_okay=False), help="Output directory" ) -def docs(cli_or_manifest: str, format: str, output_dir: str) -> None: +@blueprint_option +def docs(cli_or_manifest: str, format: str, output_dir: str, blueprint: str | None = None) -> None: """Generate documentation for a CLI""" if isinstance(cli_or_manifest, TextIOWrapper): - T = Transformer(cli_or_manifest) + T = Transformer(cli_or_manifest, blueprint=blueprint) doc_generator = DocGenerator(T.manifest) # type: ignore doc_generator.generate(format, output_dir) out(f"+ {T.cli.name}.{format}") @@ -352,12 +383,50 @@ def docs(cli_or_manifest: str, format: str, output_dir: str) -> None: # metadata.manifest is the raw manifest text. It still has to be rendered and merged with its # includes, so it is transformed from its source like `cli update` does. with open(metadata.runner_path, encoding="utf-8") as manifest: - T = Transformer(manifest, validate_requires=False) + T = Transformer(manifest, validate_requires=False, blueprint=blueprint or metadata.blueprint) doc_generator = DocGenerator(T.manifest) # type: ignore doc_generator.generate(format, output_dir) out(f"+ {metadata.cli_name}.{format}") +@click.group("blueprint", context_settings=CONTEXT_SETTINGS) +def blueprint_group() -> None: + """Work with blueprints: custom YAML specs that map onto a CLI""" + + +def blueprint_list() -> None: + """List the blueprints provided by installed packages""" + blueprints = installed_blueprints() + if not blueprints: + out("No blueprints installed. Use a .py file or module:Class reference, or install a blueprint package.") + return + + rows = [] + for blueprint in blueprints: + try: + description = blueprint.load().describe() + except BlueprintError as e: + description = f"error: {e}" + rows.append([blueprint.name, blueprint.target, blueprint.package, description]) + print_rich_table(["Name", "Blueprint", "Package", "Description"], rows, styles=["cyan", "magenta", "green", "blue"]) + + +@click.argument("blueprint", type=str) +@click.option("--output", "-o", type=click.Path(dir_okay=False, writable=True), help="Write the schema to a file") +def blueprint_schema(blueprint: str, output: str | None) -> None: + """Print the JSON schema of a blueprint's spec, for validation and editor completion""" + try: + schema = json.dumps(resolve_blueprint(blueprint).spec_json_schema(), indent=2) + except BlueprintError as e: + exit_err(f"~ {e}") + + if output: + write_to_file(output, schema + "\n") + out(f"+ {output}", fg="green") + else: + out(schema) + + # register commands load_command = cli.command("load")(load) build_command = cli.command("build")(build) @@ -374,11 +443,15 @@ def docs(cli_or_manifest: str, format: str, output_dir: str) -> None: validate_command = cli.command("validate")(validate) docs_command = cli.command("docs")(docs) cli.add_command(ai) +cli.add_command(blueprint_group) +blueprint_list_command = blueprint_group.command("list")(blueprint_list) +blueprint_schema_command = blueprint_group.command("schema")(blueprint_schema) # register aliases cli.command("add", hidden=True, epilog="Alias for load")( - click.argument("manifests", type=click.File("rb"), nargs=-1)(load) + click.argument("manifests", type=click.File("rb"), nargs=-1)(blueprint_option(load)) ) +cli.command("blueprints", hidden=True, epilog="Alias for blueprint list")(blueprint_list) cli.command("ls", hidden=True, epilog="Alias for list")(cliffy_list) cli.command("rm", hidden=True, epilog="Alias for remove")(click.argument("cli_names", type=str, nargs=-1)(remove)) cli.command("rm-all", hidden=True, epilog="Alias for remove-all")(remove_all) diff --git a/cliffy/commander.py b/cliffy/commander.py index 6c3270a..0003746 100644 --- a/cliffy/commander.py +++ b/cliffy/commander.py @@ -9,6 +9,7 @@ from cliffy.helper import py_str, to_identifier from cliffy.manifest import ( + Callback, CLIManifest, Command, CommandConfig, @@ -29,6 +30,7 @@ class CLI(BaseModel): version: str code: str requires: list[str] = [] + blueprint: str | None = None class BaseGroup(BaseModel): @@ -97,6 +99,7 @@ class Commander(ABC): __slots__ = ( "aliases_by_commands", "base_imports", + "callbacks_by_group", "cli", "commands", "greedy", @@ -123,6 +126,10 @@ def __init__(self, manifest: CLIManifest) -> None: self.base_imports: set[str] = set() self.aliases_by_commands: dict[str, list[str]] = defaultdict(list) + # callbacks keyed by group name, "" is the CLI itself + self.callbacks_by_group: dict[str, list[Callback]] = defaultdict(list) + for callback in self.manifest.callbacks: + self.callbacks_by_group[callback.group].append(callback) self.commands: list[Command] = [ command if isinstance(command, Command) else Command(name=name, run=command) @@ -334,6 +341,10 @@ def from_greedy_make_lazy_command(self, greedy_command: Command, group: str) -> lazy_command.params = lazy_parsed_params return lazy_command + def callbacks_for(self, group: BaseGroup | None = None) -> list[Callback]: + """Callbacks of a group, or of the CLI itself when no group is given.""" + return self.callbacks_by_group.get(group.name if group else "", []) + @abstractmethod def define_groups(self) -> None: raise NotImplementedError diff --git a/cliffy/commanders/click.py b/cliffy/commanders/click.py index 03f4ec9..9aedb8d 100644 --- a/cliffy/commanders/click.py +++ b/cliffy/commanders/click.py @@ -4,7 +4,15 @@ from cliffy.commander import BaseGroup, Commander from cliffy.helper import py_str -from cliffy.manifest import CLIManifest, Command, CommandParam, GenericCommandParam, SimpleCommandParam +from cliffy.manifest import ( + Callback, + CLIManifest, + Command, + CommandParam, + GenericCommandParam, + ParamBlock, + SimpleCommandParam, +) from cliffy.parser import Parser @@ -36,8 +44,11 @@ def show_aliases(ctx, param, value): print({py_str(self.aliases_table())}) ctx.exit() """ + callbacks = self.callbacks_for() self.cli += """ @click.group(context_settings=CONTEXT_SETTINGS""" + if "invoke_without_command" not in self.manifest.cli_options: + self.cli += self.parse_callback_options(callbacks, leading=", ") if self.manifest.cli_options: self.cli += f",{self.click_parser.to_args(self.manifest.cli_options)}" self.cli += f""") @@ -47,22 +58,46 @@ def show_aliases(ctx, param, value): self.cli += """ @click.option('--aliases', is_flag=True, callback=show_aliases, expose_value=False, is_eager=True, help='Show command aliases.')""" - self.cli += f""" -def cli(): + self.cli += f"""{self.parse_callback_decorators(callbacks)} +def cli({self.parse_callback_param_names(callbacks)}): {py_str(self.manifest.help)} - pass - +{self.parse_callback_run(callbacks)} """ + def parse_callback_options(self, callbacks: list[Callback], leading: str = "") -> str: + return f"{leading}invoke_without_command=True" if any(cb.invoke_without_command for cb in callbacks) else "" + + def parse_callback_decorators(self, callbacks: list[Callback]) -> str: + """Decorators for a group function with callbacks: its params, then the context.""" + if not callbacks: + return "" + + decorators = self.click_parser.parse_param_list(self.callback_params(callbacks)) + return ("\n" + decorators if decorators else "") + "\n@click.pass_context" + + def parse_callback_param_names(self, callbacks: list[Callback]) -> str: + if not callbacks: + return "" + + names = self.click_parser.get_param_list_names(self.callback_params(callbacks)) + return f"ctx, {names}" if names else "ctx" + + @staticmethod + def callback_params(callbacks: list[Callback]) -> list[ParamBlock]: + return [param for callback in callbacks for param in callback.params] + + def parse_callback_run(self, callbacks: list[Callback]) -> str: + return "".join(self.click_parser.parse_run_block(callback.run) for callback in callbacks) or " pass\n" + def define_groups(self) -> None: for group in self.groups.values(): parsed_help = group.help.replace("\n", "") if group.help else "" + callbacks = self.callbacks_for(group) self.cli += f""" -@click.group() -def {group.var_name}(): +@click.group({self.parse_callback_options(callbacks)}){self.parse_callback_decorators(callbacks)} +def {group.var_name}({self.parse_callback_param_names(callbacks)}): {py_str(parsed_help)} - pass -""" +{self.parse_callback_run(callbacks)}""" def add_group(self, group: BaseGroup) -> None: parent_var = group.parent_group.var_name if group.parent_group else "cli" @@ -130,13 +165,17 @@ class ClickParser(Parser): GENERIC_OPTION_REGEX = re.compile(r"@click\.option\(\s*['\"]--([\w-]+)") GENERIC_ARGUMENT_REGEX = re.compile(r"@click\.argument\(\s*['\"]([\w-]+)['\"]") - def _combined_params(self, command: Command) -> list[CommandParam | SimpleCommandParam | GenericCommandParam]: + def _combined_params(self, command: Command) -> list[ParamBlock]: return self.manifest.global_params + command.params def get_param_names(self, command: Command) -> str: - """Get parameter names for function definition""" + """Get parameter names for the function definition of a command""" + return self.get_param_list_names(self._combined_params(command)) + + def get_param_list_names(self, params: list[ParamBlock]) -> str: + """Get parameter names for the function that is decorated with these params""" names: list[str] = [] - for param in self._combined_params(command): + for param in params: if isinstance(param, GenericCommandParam): # a raw decorator: the name can only be read back from its first declaration match = self.GENERIC_OPTION_REGEX.search(param.root) or self.GENERIC_ARGUMENT_REGEX.search(param.root) @@ -149,9 +188,13 @@ def get_param_names(self, command: Command) -> str: return ", ".join(names) def parse_params(self, command: Command) -> str: - """Parse parameters into Click decorators""" + """Parse the parameters of a command, global ones included, into Click decorators""" + return self.parse_param_list(self._combined_params(command)) + + def parse_param_list(self, params: list[ParamBlock]) -> str: + """Parse a list of parameters into Click decorators""" decorators = [] - for param in self._combined_params(command): + for param in params: if isinstance(param, GenericCommandParam): decorators.append(param.root.strip()) elif isinstance(param, CommandParam): diff --git a/cliffy/commanders/typer.py b/cliffy/commanders/typer.py index 75d6138..4a0ff4a 100644 --- a/cliffy/commanders/typer.py +++ b/cliffy/commanders/typer.py @@ -2,7 +2,7 @@ from cliffy.commander import BaseGroup, Commander from cliffy.helper import py_str -from cliffy.manifest import CLIManifest, Command +from cliffy.manifest import Callback, CLIManifest, Command class TyperCommander(Commander): @@ -48,9 +48,10 @@ def aliases_callback(value: bool): print({py_str(self.aliases_table())}) raise typer.Exit() """ - self.cli += """ -@cli.callback() -def main(""" + callbacks = self.callbacks_for() + self.cli += f""" +@cli.callback({self.parse_callback_options(callbacks)}) +def main({self.parse_callback_params(callbacks)}""" if self.aliases_by_commands: self.cli += """ aliases: Optional[bool] = typer.Option(None, '--aliases', callback=aliases_callback, is_eager=True),""" @@ -58,9 +59,25 @@ def main(""" self.cli += """ version: Optional[bool] = typer.Option(None, '--version', callback=version_callback, is_eager=True) ): - pass - """ + self.cli += self.parse_callback_run(callbacks) + "\n" + + def parse_callback_options(self, callbacks: list[Callback]) -> str: + return "invoke_without_command=True" if any(cb.invoke_without_command for cb in callbacks) else "" + + def parse_callback_params(self, callbacks: list[Callback]) -> str: + """Signature lines for a callback function: the context first, then the params of each callback.""" + if not callbacks: + return "" + + signature = "\n ctx: typer.Context," + for callback in callbacks: + for param in callback.params: + signature += f"\n {self.parser.parse_param(param).strip().rstrip(',')}," + return signature + + def parse_callback_run(self, callbacks: list[Callback]) -> str: + return "".join(self.parser.parse_run_block(callback.run) for callback in callbacks) or " pass\n" def add_root_command(self, command: Command) -> None: """ @@ -97,6 +114,13 @@ def define_groups(self) -> None: parsed_help = group.help.replace("\n", "") if group.help else "" empty_or_help = f"help={py_str(parsed_help)}," if parsed_help else "" self.cli += f"""{group.var_name} = typer.Typer({empty_or_help}) +""" + if callbacks := self.callbacks_for(group): + self.cli += f""" +@{group.var_name}.callback({self.parse_callback_options(callbacks)}) +def {group.var_name}_callback({self.parse_callback_params(callbacks)} +): +{self.parse_callback_run(callbacks)} """ def add_group(self, group: BaseGroup) -> None: diff --git a/cliffy/homer.py b/cliffy/homer.py index d88d41a..5b8a754 100644 --- a/cliffy/homer.py +++ b/cliffy/homer.py @@ -33,6 +33,7 @@ def save_metadata(manifest_path: str, cli: CLI) -> None: loaded=datetime.now(), manifest=manifest.read(), requires=cli.requires, + blueprint=cli.blueprint, ).model_dump(), default=str, ), diff --git a/cliffy/manifest.py b/cliffy/manifest.py index c169239..034072f 100644 --- a/cliffy/manifest.py +++ b/cliffy/manifest.py @@ -2,9 +2,9 @@ from collections.abc import ItemsView, Iterator from datetime import datetime from functools import cached_property -from typing import Any +from typing import Any, Self -from pydantic import BaseModel, Field, RootModel, ValidationInfo, field_validator +from pydantic import BaseModel, Field, RootModel, ValidationInfo, field_validator, model_validator from .helper import CLI_NAME_PATTERN, wrap_as_comment @@ -254,6 +254,28 @@ class CommandTemplate(BaseModel): ) +class Callback(BaseModel): + """ + Code that runs before the commands of the CLI, or of one command group. A callback can declare its own + params, which become options of the CLI (or group) itself, i.e. `mycli --verbose command`. + The Typer/Click context is available in the run block as `ctx`. + """ + + run: RunBlock = Field(..., description="Code to run before the invoked command. `ctx` holds the context.") + group: str = Field( + default="", + description="Command group the callback belongs to, i.e. `db` or `db.tables`. Leave empty for the CLI itself.", + ) + params: list[ParamBlock] = Field( + default=[], + description="Params of the CLI (or group) itself, given before the command name. Same syntax as command params.", + ) + invoke_without_command: bool = Field( + default=False, + description="Also run the callback when the CLI (or group) is invoked without a command, instead of showing the help.", + ) + + class CLIManifest(BaseModel): manifestVersion: str = LATEST_SCHEMA_VERSION name: str = Field(..., description="The name of the CLI, used when invoking from command line.") @@ -303,7 +325,7 @@ class CLIManifest(BaseModel): ) commands: dict[str, CommandBlock] | list[Command] = Field( - ..., + default_factory=dict, description="A mapping containing the command definitions for the CLI. " "Each command should have a unique key- which can be either a group command or nested subcommands. " "Nested subcommands are joined by '.' in between each level. " @@ -311,6 +333,13 @@ class CLIManifest(BaseModel): "A special '(*)' wildcard can be used to spread the subcommand to all group-level commands", ) + callbacks: list[Callback] = Field( + default=[], + description="Code that runs before any command of the CLI, or of one command group. " + "Callbacks for the same target run in the order given and their params are combined. " + "The built-in --version and --aliases options are kept.", + ) + cli_options: dict[str, Any] = Field(default_factory=dict, description="Additional CLI configuration options") tests: list[str | dict[str, str]] = Field(default=[], description="Test cases for commands") @@ -344,6 +373,25 @@ def is_v3(cls, value: str) -> str: raise ValueError(f"Unrecognized manifest version {value}. Latest is {LATEST_SCHEMA_VERSION}.") return value + @model_validator(mode="after") + def callback_groups_exist(self) -> Self: + if not self.callbacks: + return self + + names = self.commands if isinstance(self.commands, dict) else [command.name for command in self.commands] + groups: set[str] = set() + for name in names: + parts = name.split("|")[0].strip().split(".") + groups.update(".".join(parts[:depth]) for depth in range(1, len(parts))) + + for callback in self.callbacks: + if callback.group and callback.group not in groups: + known = ", ".join(sorted(groups)) or "none" + raise ValueError( + f"callback group `{callback.group}` does not match any command group (groups: {known})" + ) + return self + @classmethod def get_field_description(cls, field_name: str, as_comment: bool = True) -> str: field = cls.model_fields.get(field_name) @@ -558,6 +606,7 @@ class CLIMetadata(BaseModel): loaded: datetime manifest: str requires: list[str] + blueprint: str | None = None if __name__ == "__main__": diff --git a/cliffy/reloader.py b/cliffy/reloader.py index f8d0d18..931eab7 100644 --- a/cliffy/reloader.py +++ b/cliffy/reloader.py @@ -15,11 +15,12 @@ class Reloader(FileSystemEventHandler): - def __init__(self, manifest_path: str, run_cli: bool, run_cli_args: tuple) -> None: + def __init__(self, manifest_path: str, run_cli: bool, run_cli_args: tuple, blueprint: str | None = None) -> None: self.manifest_path = manifest_path self.manifest_realpath = os.path.realpath(manifest_path) self.run_cli = run_cli self.run_cli_args = run_cli_args + self.blueprint = blueprint self.last_modified = datetime.now() super().__init__() @@ -38,7 +39,10 @@ def on_modified(self, event: FileSystemEvent) -> None: self.last_modified = datetime.now() t = threading.Thread( - target=self.reload, args=(self.manifest_path, self.run_cli, self.run_cli_args), daemon=True + target=self.reload, + args=(self.manifest_path, self.run_cli, self.run_cli_args), + kwargs={"blueprint": self.blueprint}, + daemon=True, ) try: t.start() @@ -47,8 +51,8 @@ def on_modified(self, event: FileSystemEvent) -> None: out_err(f"Failed to start thread: {e}") @classmethod - def watch(cls, manifest_path: str, run_cli: bool, run_cli_args: tuple) -> None: - event_handler = cls(manifest_path, run_cli, run_cli_args) + def watch(cls, manifest_path: str, run_cli: bool, run_cli_args: tuple, blueprint: str | None = None) -> None: + event_handler = cls(manifest_path, run_cli, run_cli_args, blueprint) observer = Observer() observer.schedule(event_handler, path=str(Path(manifest_path).parent), recursive=False) observer.start() @@ -64,9 +68,9 @@ def watch(cls, manifest_path: str, run_cli: bool, run_cli_args: tuple) -> None: observer.join() @staticmethod - def reload(manifest_path: str, run_cli: bool, run_cli_args: tuple) -> None: + def reload(manifest_path: str, run_cli: bool, run_cli_args: tuple, blueprint: str | None = None) -> None: with open(manifest_path, encoding="utf-8") as manifest_io: - T = Transformer(manifest_io) + T = Transformer(manifest_io, blueprint=blueprint) if blueprint else Transformer(manifest_io) Loader.load_from_cli(T.cli) save_metadata(manifest_path, T.cli) out(f"✨ Reloaded {T.cli.name} CLI v{T.cli.version} ✨", fg="green") diff --git a/cliffy/tester.py b/cliffy/tester.py index 1a9b232..ecdffb2 100644 --- a/cliffy/tester.py +++ b/cliffy/tester.py @@ -37,9 +37,9 @@ def is_shelled(cls, value: str) -> str: class Tester: - def __init__(self, manifest_path: str) -> None: + def __init__(self, manifest_path: str, blueprint: str | None = None) -> None: with open(manifest_path, encoding="utf-8") as manifest_io: - self.T = Transformer(manifest_io) + self.T = Transformer(manifest_io, blueprint=blueprint) with NamedTemporaryFile( mode="w", prefix=f"{self.T.cli.name}_test_", suffix=".py", delete=False, encoding="utf-8" @@ -56,7 +56,8 @@ def __init__(self, manifest_path: str) -> None: self.module_funcs = inspect.getmembers(self.module, inspect.isfunction) manifest = cast(CLIManifest, self.T.manifest) - # typer vendors its own click: its runner cannot invoke a CLI that was generated for click + # typer vendors its own click and its runner only accepts typer apps: a CLI generated with + # `use_click` is a plain click group and needs click's own runner self.runner: CliRunner | ClickCliRunner = ClickCliRunner() if manifest.use_click else CliRunner() self.parser = Parser(manifest) diff --git a/cliffy/transformer.py b/cliffy/transformer.py index fb25c0c..9b1ebf2 100644 --- a/cliffy/transformer.py +++ b/cliffy/transformer.py @@ -5,6 +5,14 @@ from jinja2 import BaseLoader, Environment, FileSystemLoader from pydantic import ValidationError +from cliffy.blueprint import ( + BLUEPRINT_KEY, + Blueprint, + BlueprintError, + build_manifest, + normalize_reference, + resolve_blueprint, +) from cliffy.commander import generate_cli from cliffy.commanders.click import ClickCommander from cliffy.commanders.typer import TyperCommander @@ -17,6 +25,8 @@ class Transformer: """Loads command manifest and transforms it into a CLI""" __slots__ = ( + "blueprint", + "blueprint_ref", "cli", "command_config", "include_chain", @@ -34,6 +44,7 @@ def __init__( as_include: bool = False, validate_requires: bool = True, include_chain: tuple[str, ...] = (), + blueprint: str | None = None, ) -> None: self.manifest_io = manifest_io self.manifest_path = os.path.realpath(manifest_io.name) @@ -41,13 +52,15 @@ def __init__( self.include_chain = (*include_chain, self.manifest_path) self.command_config = self.load_manifest(manifest_io) self.manifest_version = self.command_config.pop("manifestVersion", LATEST_SCHEMA_VERSION) - if self.command_config.get("includes"): + self.blueprint: Blueprint | None = None + self.blueprint_ref = None if as_include else self.find_blueprint_reference(blueprint) + if not self.blueprint_ref and self.command_config.get("includes"): self.includes_config = self.resolve_includes() cliffy_merger.merge(self.command_config, self.includes_config) manifest_cls = IncludeManifest if as_include else CLIManifest try: - self.manifest = manifest_cls(**self.command_config) + self.manifest = self.apply_blueprint() if self.blueprint_ref else manifest_cls(**self.command_config) except ValidationError as e: out(f"{e}") exit_err(f"~ error validating {manifest_io.name}") @@ -58,6 +71,33 @@ def __init__( if isinstance(self.manifest, CLIManifest): commander_cls = ClickCommander if self.manifest.use_click else TyperCommander self.cli = generate_cli(self.manifest, commander_cls=commander_cls) + if self.blueprint: + self.cli.code = self.blueprint.post_codegen(self.cli.code) + self.cli.blueprint = self.blueprint_ref + + def find_blueprint_reference(self, blueprint: str | None) -> str | None: + """Blueprint for this file: the `--blueprint` option wins over the file's own `blueprint:` key. + + A path given on the command line is relative to the working directory, a path in the file is + relative to the file. + """ + in_file = self.command_config.pop(BLUEPRINT_KEY, None) + if blueprint: + return normalize_reference(blueprint) + if in_file is None: + return None + if not isinstance(in_file, str) or not in_file.strip(): + exit_err(f"~ error loading {self.manifest_io.name}: `{BLUEPRINT_KEY}` must be a blueprint reference") + return normalize_reference(in_file, os.path.dirname(self.manifest_path)) + + def apply_blueprint(self) -> CLIManifest: + """Validate the file against its blueprint's spec and transform it into a manifest""" + try: + blueprint_cls = resolve_blueprint(str(self.blueprint_ref)) + self.blueprint, manifest = build_manifest(blueprint_cls, self.command_config, source=self.manifest_io.name) + except BlueprintError as e: + exit_err(f"~ {e}") + return manifest def validate_cli_requires(self) -> None: if not self.manifest.requires: diff --git a/docs/blueprints.md b/docs/blueprints.md new file mode 100644 index 0000000..465379c --- /dev/null +++ b/docs/blueprints.md @@ -0,0 +1,284 @@ +# Blueprints + +A cliffy manifest describes a CLI in cliffy's own vocabulary: `commands`, `params`, `run`. A **blueprint** lets you +bring your own vocabulary instead. You define the YAML schema that fits your domain and tell cliffy how it maps +onto a CLI. Files written in that schema then work with `cli load`, `cli run`, `cli render`, `cli test`, `cli dev`, +`cli docs` and `cli build`, which turns them into a single-file executable. + +``` +your YAML spec ──▶ Blueprint (validate, to_manifest) ──▶ cliffy manifest ──▶ Typer or Click code ──▶ one executable file +``` + +## Writing a blueprint + +A blueprint is a [pydantic](https://docs.pydantic.dev) model with one required method. The fields are your schema, +and `to_manifest` returns the cliffy manifest for a validated file. + +```python title="tasks.py" +from pydantic import BaseModel, Field + +from cliffy.blueprint import Blueprint +from cliffy.manifest import CLIManifest, Command, RunBlock + + +class Task(BaseModel): + run: str | list[str] = Field(description="Shell line(s) to run, in order.") + help: str = Field(default="", description="What the task does, shown in --help.") + + +class Tasks(Blueprint): + """Task runner: every entry under `tasks` becomes a command that runs shell lines""" + + name: str = Field(description="Name of the CLI.") + version: str = Field(default="0.1.0", description="Version of the CLI.") + help: str = Field(default="", description="Description of the CLI.") + tasks: dict[str, str | list[str] | Task] = Field(min_length=1, description="Tasks by name.") + + def to_manifest(self) -> CLIManifest: + commands = {} + for name, entry in self.tasks.items(): + task = entry if isinstance(entry, Task) else Task(run=entry) + lines = [task.run] if isinstance(task.run, str) else task.run + script = "\n".join(f"subprocess.run({line!r}, shell=True, check=True)" for line in lines) + commands[name] = Command(help=task.help, run=RunBlock(script)) + + return CLIManifest(name=self.name, version=self.version, help=self.help, commands=commands) +``` + +A file written in that spec names its blueprint with a top-level `blueprint:` key: + +```yaml title="tasks.yaml" +blueprint: tasks.py + +name: tasks +help: Project tasks + +tasks: + greet: echo "hello from tasks" + check: + - echo "linting" + - echo "testing" + release: + help: Check, then tag the release + run: + - echo "linting" + - echo "tagging" +``` + +``` +$ cli run tasks.yaml -- greet +hello from tasks + +$ cli load tasks.yaml +$ tasks check +linting +testing +``` + +`to_manifest` may return a `CLIManifest` or a plain mapping that validates as one. Because the result is an +ordinary manifest, everything cliffy does with manifests applies: both the Typer and the Click generator +(`use_click`), `requires`, `tests`, command aliases, callbacks and so on. + +The whole file is rendered with Jinja2 before the blueprint sees it, exactly like a manifest. + +## Selecting a blueprint + +Either in the file, or on the command line. The command line wins. + +```yaml +blueprint: tasks.py +``` + +``` +cli run jobs.yaml --blueprint tasks.py -- greet +``` + +`--blueprint` / `-b` is available on `load`, `run`, `render`, `build`, `dev`, `test`, `validate` and `docs`. +`cli update` reuses the blueprint a CLI was loaded with, and `cli info` shows it. + +A blueprint reference is one of: + +| Reference | Meaning | +|---|---| +| `tasks.py`, `./bp/tasks.py:Tasks` | A Python file. In a YAML file the path is relative to that file; on the command line it is relative to the working directory. Add `:ClassName` when the file defines several blueprints. | +| `mypackage.blueprints:Tasks` | An importable class. `mypackage.blueprints` alone works when the module defines exactly one blueprint. | +| `tasks` | A blueprint provided by an installed package (see [Sharing blueprints](#sharing-blueprints)). | + +Cliffy reports a reference it cannot use and says why: the file or module does not exist, the name is provided by +more than one package, the module defines no blueprint or more than one, or the target is not a blueprint. + +## Validation errors + +Unknown keys are rejected by default, so typos are caught. Errors are reported against your spec, with the path +to the offending key: + +``` +$ cli validate server.yaml +~ server.yaml is not a valid MCPTools spec + tools[0].name: Value error, `serve` is a reserved name (got 'serve') + tools[1].parameters[0].type: Input should be 'string', 'integer', 'number', 'boolean' or 'array' (got 'strng') + tols: Extra inputs are not permitted 💔 +``` + +Use pydantic validators for rules that go beyond types. To accept unknown keys, set +`model_config = ConfigDict(extra="ignore")` on your blueprint. + +## Hooks + +Three optional hooks run around `to_manifest`: + +``` +pre_validate(raw) ──▶ validation ──▶ to_manifest() ──▶ post_manifest(manifest) ──▶ code generation ──▶ post_codegen(code) +``` + +| Hook | Receives | Use it to | +|---|---|---| +| `pre_validate(cls, raw)` (classmethod) | The YAML mapping, without the `blueprint` key | Accept shorthand forms, set defaults, rename legacy keys | +| `post_manifest(self, manifest)` | The `CLIManifest` | Adjust the manifest, i.e. add commands every CLI of this kind gets | +| `post_codegen(self, code)` | The generated Python source | Adjust the generated code, i.e. prepend a license header | + +```python +class Tasks(Blueprint): + ... + + @classmethod + def pre_validate(cls, raw): + # allow `jobs:` as an older name for `tasks:` + if "jobs" in raw: + raw = {**raw, "tasks": raw.pop("jobs")} + return raw + + def post_codegen(self, code): + return "# generated from a Tasks spec, do not edit\n" + code +``` + +## Editor completion + +`cli blueprint schema` prints the JSON schema of a blueprint's spec. + +``` +cli blueprint schema tasks.py -o tasks.schema.json +``` + +Reference it from your YAML files to get validation and completion from the YAML language server: + +```yaml +# yaml-language-server: $schema=tasks.schema.json +blueprint: tasks.py +``` + +Field descriptions (`Field(description=...)`) show up as hover documentation. + +## Building a single-file executable + +`cli build` applies the blueprint at build time and bundles the generated CLI with its dependencies into one +executable [zipapp](https://docs.python.org/3/library/zipapp.html). + +``` +$ cli build tasks.yaml -o dist ++ tasks built 📦 + dist/tasks (single file, needs Python >=3.10) + +$ ./dist/tasks greet +hello from tasks +``` + +The file runs on any machine with a matching Python. It does not need cliffy, the blueprint, pydantic or the YAML +file. + +* **Dependencies.** Whatever the manifest lists under `requires` is bundled. Let your blueprint set `requires` + for the packages its generated code imports. The CLI framework (typer, or rich-click with `use_click`) is added + automatically, pinned to the version installed next to cliffy unless `requires` names it. +* **Interpreter.** `--python` / `-p` sets the shebang, by default `/usr/bin/env python3`. The build reports which + Python versions the bundle supports. A bundle of pure-Python packages runs on every version its packages accept. + A bundle with compiled extensions only runs on the Python minor version it was built with, so build with the + version you deploy to. Started with an unsupported Python, the executable exits with a message that says so. +* **Reproducible.** Building the same input twice with the same package versions gives a byte-identical file. +* **Errors.** When a dependency cannot be installed, the build prints pip's reason and writes no file. + +On first run the executable unpacks its packages into `~/.shiv` (set `SHIV_ROOT` to change the location). + +## Sharing blueprints + +A package can register blueprints under the `cliffy.blueprints` entry-point group: + +```toml title="pyproject.toml" +[project.entry-points."cliffy.blueprints"] +tasks = "mypackage.blueprints:Tasks" +``` + +Once the package is installed, the blueprint is available by name: + +```yaml +blueprint: tasks +``` + +``` +$ cli blueprints +┏━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +┃ Name ┃ Blueprint ┃ Package ┃ Description ┃ +┡━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩ +│ tasks │ mypackage.blueprints:Tasks │ mypackage │ Task runner: every entry under... │ +└───────┴────────────────────────────┴───────────┴───────────────────────────────────┘ +``` + +`cli blueprints` is short for `cli blueprint list`. The description is the first line of the class docstring. + +## Examples + +Two worked examples live in [`examples/blueprints`](https://github.com/jaykv/cliffy/tree/main/examples/blueprints): + +* `tasks.py` with `tasks.yaml`: the task runner from this page. +* `mcp_tools.py` with `text-tools.yaml`: an [MCP](https://modelcontextprotocol.io) tool server. The spec lists + tools with typed parameters and a Python body. The generated CLI has one command per tool, a `tools` command + that prints the tool definitions, and a `serve` command that speaks MCP over stdio without extra dependencies. + `cli build examples/blueprints/text-tools.yaml` gives a single file you can register as an MCP server. + +## Callbacks + +Callbacks are a manifest feature that blueprints often need: code that runs before every command of the CLI, or +of one command group. They work with both Typer and Click, and the built-in `--version` and `--aliases` options +stay in place. + +```yaml +callbacks: + # no `group`: runs before every command, its params become options of the CLI itself + - params: + - name: --verbose + type: bool + default: false + short: -v + help: Print what is about to run + invoke_without_command: true + run: | + ctx.obj = {"verbose": verbose} + if ctx.invoked_subcommand is None: + print("no command given, try --help") + elif verbose: + print(f"running {ctx.invoked_subcommand}") + + # runs before every command of the `db` group + - group: db + run: print("connecting to db") + +commands: + hello: print("hello") + db.migrate: print("migrating") +``` + +``` +$ mycli --verbose db migrate +running db +connecting to db +migrating +``` + +| Key | Meaning | +|---|---| +| `run` | Code to run. The Typer or Click context is available as `ctx`. | +| `group` | Command group the callback belongs to, i.e. `db` or `db.tables`. Leave it out for the CLI itself. | +| `params` | Options given before the command name, in the same syntax as command params. | +| `invoke_without_command` | Also run when no command is given, instead of showing the help. | + +Several callbacks for the same target run in the order given. A manifest with only callbacks and no `commands` +is valid, which makes a single-purpose CLI. diff --git a/examples/blueprints/mcp_tools.py b/examples/blueprints/mcp_tools.py new file mode 100644 index 0000000..f106a20 --- /dev/null +++ b/examples/blueprints/mcp_tools.py @@ -0,0 +1,249 @@ +"""A realistic blueprint: an MCP tool server. + +The spec describes tools the way the Model Context Protocol does: a name, a description, typed parameters +and a Python body. The generated CLI has + +* one command per tool, to call it from a shell: ``text-tools format-text " hi " --trim`` +* ``tools``, which prints the tool definitions as MCP ``tools/list`` JSON +* ``serve``, a dependency-free MCP server over stdio (JSON-RPC 2.0, one message per line) + + cli run examples/blueprints/text-tools.yaml -- count-words "one two three" + cli build examples/blueprints/text-tools.yaml -o dist # single file, serves MCP with `dist/text-tools serve` +""" + +import re +import textwrap +from typing import Any, Literal + +from pydantic import BaseModel, Field, field_validator, model_validator + +from cliffy.blueprint import Blueprint +from cliffy.manifest import CLIManifest, Command, CommandParam, RunBlock + +# names the generated module already uses: built-in commands, imports and the typer app +RESERVED_NAMES = {"serve", "tools", "main", "cli", "json", "sys", "typer", "subprocess"} +PYTHON_TYPES = {"string": "str", "integer": "int", "number": "float", "boolean": "bool", "array": "list[str]"} + +# Shared by every generated server. Kept free of third-party imports so that a built CLI stays small. +RUNTIME = ''' +def _describe_tools(): + return [ + {"name": name, "description": tool["description"], "inputSchema": tool["inputSchema"]} + for name, tool in TOOLS.items() + ] + + +def _render_result(value): + return value if isinstance(value, str) else json.dumps(value, indent=2, default=str) + + +def _call_tool(name, arguments): + if name not in TOOLS: + raise KeyError(f"unknown tool: {name}") + schema = TOOLS[name]["inputSchema"] + unknown = sorted(set(arguments) - set(schema["properties"])) + missing = [key for key in schema["required"] if key not in arguments] + if unknown or missing: + raise ValueError(f"invalid arguments for {name}: missing {missing}, unknown {unknown}") + return _render_result(TOOLS[name]["function"](**arguments)) + + +def _handle_message(message): + """Answer one JSON-RPC message. Returns None for notifications, which get no reply.""" + method, params, message_id = message.get("method"), message.get("params") or {}, message.get("id") + if message_id is None: + return None + + if method == "initialize": + result = { + "protocolVersion": params.get("protocolVersion", PROTOCOL_VERSION), + "capabilities": {"tools": {}}, + "serverInfo": SERVER_INFO, + } + elif method == "ping": + result = {} + elif method == "tools/list": + result = {"tools": _describe_tools()} + elif method == "tools/call": + try: + text, failed = _call_tool(params.get("name"), params.get("arguments") or {}), False + except Exception as e: + text, failed = f"{type(e).__name__}: {e}", True + result = {"content": [{"type": "text", "text": text}], "isError": failed} + else: + return {"jsonrpc": "2.0", "id": message_id, "error": {"code": -32601, "message": f"Method not found: {method}"}} + return {"jsonrpc": "2.0", "id": message_id, "result": result} + + +def _serve(): + for line in sys.stdin: + if not line.strip(): + continue + try: + reply = _handle_message(json.loads(line)) + except ValueError: + reply = {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "Parse error"}} + if reply is not None: + sys.stdout.write(json.dumps(reply) + "\\n") + sys.stdout.flush() +''' + + +class ToolParameter(BaseModel): + name: str = Field(pattern=r"^[a-z][a-z0-9_]*$", description="Name of the parameter, a Python identifier.") + type: Literal["string", "integer", "number", "boolean", "array"] = Field( + default="string", description="JSON schema type. An `array` holds strings." + ) + description: str = Field(default="", description="What the parameter means, shown to the model and in --help.") + required: bool = Field(default=False, description="Required parameters become positional arguments.") + default: Any = Field(default=None, description="Value used when an optional parameter is not given.") + + @model_validator(mode="after") + def check_shape(self) -> "ToolParameter": + if self.required and self.default is not None: + raise ValueError("a required parameter cannot have a default") + if self.required and self.type == "boolean": + raise ValueError("a boolean parameter is a flag and cannot be required") + return self + + @property + def python_type(self) -> str: + return PYTHON_TYPES[self.type] + + @property + def python_default(self) -> str: + if self.type == "boolean": + return repr(bool(self.default)) + return repr(self.default) + + def to_json_schema(self) -> dict[str, Any]: + schema: dict[str, Any] = {"type": self.type} + if self.type == "array": + schema["items"] = {"type": "string"} + if self.description: + schema["description"] = self.description + return schema + + def to_param(self) -> CommandParam: + # the generated code puts help in double quotes + help_text = self.description.replace('"', "'") + if self.required: + return CommandParam(name=self.name, type=self.python_type, required=True, help=help_text) + + default = self.default if self.type == "string" else self.python_default + if self.default is None and self.type != "boolean": + default = None + return CommandParam( + name=f"--{self.name.replace('_', '-')}", type=self.python_type, default=default, help=help_text + ) + + +class Tool(BaseModel): + name: str = Field(pattern=r"^[a-z][a-z0-9_]*$", description="Name of the tool, as the model calls it.") + description: str = Field(description="What the tool does, shown to the model and in --help.") + parameters: list[ToolParameter] = Field(default=[], description="Inputs of the tool.") + run: str = Field(description="Python body of the tool. Parameters are local variables. Return the result.") + + @field_validator("name") + @classmethod + def not_reserved(cls, name: str) -> str: + if name in RESERVED_NAMES: + raise ValueError(f"`{name}` is a reserved name") + return name + + @field_validator("parameters") + @classmethod + def unique_parameters(cls, parameters: list[ToolParameter]) -> list[ToolParameter]: + names = [parameter.name for parameter in parameters] + if duplicates := sorted({name for name in names if names.count(name) > 1}): + raise ValueError(f"duplicate parameter names: {', '.join(duplicates)}") + return parameters + + @property + def function_name(self) -> str: + return f"_tool_{self.name}" + + def to_function(self) -> str: + ordered = sorted(self.parameters, key=lambda parameter: not parameter.required) + signature = ", ".join( + f"{p.name}: {p.python_type}" if p.required else f"{p.name}: {p.python_type} = {p.python_default}" + for p in ordered + ) + body = textwrap.indent(textwrap.dedent(self.run).strip() or "pass", " " * 4) + return f"def {self.function_name}({signature}):\n{body}\n" + + def to_definition(self) -> str: + """Python source of this tool's entry in the generated TOOLS registry.""" + input_schema = { + "type": "object", + "properties": {parameter.name: parameter.to_json_schema() for parameter in self.parameters}, + "required": [parameter.name for parameter in self.parameters if parameter.required], + } + return ( + f" {self.name!r}: {{\n" + f" 'function': {self.function_name},\n" + f" 'description': {self.description!r},\n" + f" 'inputSchema': {input_schema!r},\n" + f" }},\n" + ) + + def to_command(self) -> Command: + arguments = ", ".join(f"{parameter.name}={parameter.name}" for parameter in self.parameters) + return Command( + help=re.sub(r"\s+", " ", self.description).replace('"', "'"), + params=[parameter.to_param() for parameter in self.parameters], + run=RunBlock(f"print(_render_result({self.function_name}({arguments})))"), + ) + + +class MCPTools(Blueprint): + """MCP tool server: every tool becomes a command, `serve` speaks MCP over stdio""" + + name: str = Field(description="Name of the CLI and of the MCP server.") + version: str = Field(default="0.1.0", description="Version of the CLI and of the MCP server.") + description: str = Field(default="", description="What the server offers.") + protocol_version: str = Field(default="2025-06-18", description="MCP protocol revision the server reports.") + imports: list[str] = Field(default=[], description="Import lines the tool bodies need, i.e. `import re`.") + requires: list[str] = Field(default=[], description="Packages the tool bodies need. Bundled by `cli build`.") + tools: list[Tool] = Field(min_length=1, description="The tools, as a list or as a mapping keyed by tool name.") + + @classmethod + def pre_validate(cls, raw: dict[str, Any]) -> dict[str, Any]: + # `tools` may be written as a mapping of name -> tool, which reads better for a handful of tools + tools = raw.get("tools") + if isinstance(tools, dict): + raw = {**raw, "tools": [{"name": name, **(tool or {})} for name, tool in tools.items()]} + return raw + + @field_validator("tools") + @classmethod + def unique_tools(cls, tools: list[Tool]) -> list[Tool]: + names = [tool.name for tool in tools] + if duplicates := sorted({name for name in names if names.count(name) > 1}): + raise ValueError(f"duplicate tool names: {', '.join(duplicates)}") + return tools + + def to_manifest(self) -> CLIManifest: + functions = "\n\n".join(tool.to_function() for tool in self.tools) + functions += "\n\nTOOLS = {\n" + "".join(tool.to_definition() for tool in self.tools) + "}\n" + functions += f"SERVER_INFO = {{'name': {self.name!r}, 'version': {self.version!r}}}\n" + functions += f"PROTOCOL_VERSION = {self.protocol_version!r}\n" + functions += RUNTIME + + commands = {tool.name.replace("_", "-"): tool.to_command() for tool in self.tools} + commands["tools"] = Command( + help="Print the tool definitions as MCP tools/list JSON", + run=RunBlock("print(json.dumps(_describe_tools(), indent=2))"), + ) + commands["serve"] = Command(help="Serve the tools over MCP on stdio", run=RunBlock("_serve()")) + + return CLIManifest( + name=self.name, + version=self.version, + help=self.description, + requires=self.requires, + imports=["import json", "import sys", *self.imports], + functions=functions, + commands=commands, + tests=[{"tools": f"assert {self.tools[0].name!r} in result.output"}], + ) diff --git a/examples/blueprints/mcp_tools.schema.json b/examples/blueprints/mcp_tools.schema.json new file mode 100644 index 0000000..c42b3fe --- /dev/null +++ b/examples/blueprints/mcp_tools.schema.json @@ -0,0 +1,150 @@ +{ + "$defs": { + "Tool": { + "properties": { + "name": { + "description": "Name of the tool, as the model calls it.", + "pattern": "^[a-z][a-z0-9_]*$", + "title": "Name", + "type": "string" + }, + "description": { + "description": "What the tool does, shown to the model and in --help.", + "title": "Description", + "type": "string" + }, + "parameters": { + "default": [], + "description": "Inputs of the tool.", + "items": { + "$ref": "#/$defs/ToolParameter" + }, + "title": "Parameters", + "type": "array" + }, + "run": { + "description": "Python body of the tool. Parameters are local variables. Return the result.", + "title": "Run", + "type": "string" + } + }, + "required": [ + "name", + "description", + "run" + ], + "title": "Tool", + "type": "object" + }, + "ToolParameter": { + "properties": { + "name": { + "description": "Name of the parameter, a Python identifier.", + "pattern": "^[a-z][a-z0-9_]*$", + "title": "Name", + "type": "string" + }, + "type": { + "default": "string", + "description": "JSON schema type. An `array` holds strings.", + "enum": [ + "string", + "integer", + "number", + "boolean", + "array" + ], + "title": "Type", + "type": "string" + }, + "description": { + "default": "", + "description": "What the parameter means, shown to the model and in --help.", + "title": "Description", + "type": "string" + }, + "required": { + "default": false, + "description": "Required parameters become positional arguments.", + "title": "Required", + "type": "boolean" + }, + "default": { + "default": null, + "description": "Value used when an optional parameter is not given.", + "title": "Default" + } + }, + "required": [ + "name" + ], + "title": "ToolParameter", + "type": "object" + } + }, + "additionalProperties": false, + "description": "MCP tool server: every tool becomes a command, `serve` speaks MCP over stdio", + "properties": { + "name": { + "description": "Name of the CLI and of the MCP server.", + "title": "Name", + "type": "string" + }, + "version": { + "default": "0.1.0", + "description": "Version of the CLI and of the MCP server.", + "title": "Version", + "type": "string" + }, + "description": { + "default": "", + "description": "What the server offers.", + "title": "Description", + "type": "string" + }, + "protocol_version": { + "default": "2025-06-18", + "description": "MCP protocol revision the server reports.", + "title": "Protocol Version", + "type": "string" + }, + "imports": { + "default": [], + "description": "Import lines the tool bodies need, i.e. `import re`.", + "items": { + "type": "string" + }, + "title": "Imports", + "type": "array" + }, + "requires": { + "default": [], + "description": "Packages the tool bodies need. Bundled by `cli build`.", + "items": { + "type": "string" + }, + "title": "Requires", + "type": "array" + }, + "tools": { + "description": "The tools, as a list or as a mapping keyed by tool name.", + "items": { + "$ref": "#/$defs/Tool" + }, + "minItems": 1, + "title": "Tools", + "type": "array" + }, + "blueprint": { + "type": "string", + "title": "Blueprint", + "description": "Blueprint that turns this file into a CLI." + } + }, + "required": [ + "name", + "tools" + ], + "title": "MCPTools", + "type": "object" +} diff --git a/examples/blueprints/tasks.py b/examples/blueprints/tasks.py new file mode 100644 index 0000000..dd43587 --- /dev/null +++ b/examples/blueprints/tasks.py @@ -0,0 +1,41 @@ +"""A small blueprint: a task runner. + +The spec is a name and a mapping of tasks. Every task becomes a command that runs its shell lines in order +and stops at the first one that fails. + + cli run examples/blueprints/tasks.yaml -- greet + cli build examples/blueprints/tasks.yaml -o dist +""" + +from pydantic import BaseModel, Field + +from cliffy.blueprint import Blueprint +from cliffy.manifest import CLIManifest, Command, RunBlock + + +class Task(BaseModel): + run: str | list[str] = Field(description="Shell line(s) to run, in order.") + help: str = Field(default="", description="What the task does, shown in --help.") + + +class Tasks(Blueprint): + """Task runner: every entry under `tasks` becomes a command that runs shell lines""" + + name: str = Field(description="Name of the CLI.") + version: str = Field(default="0.1.0", description="Version of the CLI.") + help: str = Field(default="", description="Description of the CLI.") + vars: dict[str, str] = Field(default={}, description="Values for the `{{ name }}` expressions in the file.") + tasks: dict[str, str | list[str] | Task] = Field( + min_length=1, + description="Tasks by name. A task is a shell line, a list of shell lines, or a mapping with `run` and `help`.", + ) + + def to_manifest(self) -> CLIManifest: + commands = {} + for name, entry in self.tasks.items(): + task = entry if isinstance(entry, Task) else Task(run=entry) + lines = [task.run] if isinstance(task.run, str) else task.run + script = "\n".join(f"subprocess.run({line!r}, shell=True, check=True)" for line in lines) + commands[name] = Command(help=task.help, run=RunBlock(script)) + + return CLIManifest(name=self.name, version=self.version, help=self.help, commands=commands) diff --git a/examples/blueprints/tasks.schema.json b/examples/blueprints/tasks.schema.json new file mode 100644 index 0000000..e9cc464 --- /dev/null +++ b/examples/blueprints/tasks.schema.json @@ -0,0 +1,97 @@ +{ + "$defs": { + "Task": { + "properties": { + "run": { + "anyOf": [ + { + "type": "string" + }, + { + "items": { + "type": "string" + }, + "type": "array" + } + ], + "description": "Shell line(s) to run, in order.", + "title": "Run" + }, + "help": { + "default": "", + "description": "What the task does, shown in --help.", + "title": "Help", + "type": "string" + } + }, + "required": [ + "run" + ], + "title": "Task", + "type": "object" + } + }, + "additionalProperties": false, + "description": "Task runner: every entry under `tasks` becomes a command that runs shell lines", + "properties": { + "name": { + "description": "Name of the CLI.", + "title": "Name", + "type": "string" + }, + "version": { + "default": "0.1.0", + "description": "Version of the CLI.", + "title": "Version", + "type": "string" + }, + "help": { + "default": "", + "description": "Description of the CLI.", + "title": "Help", + "type": "string" + }, + "vars": { + "additionalProperties": { + "type": "string" + }, + "default": {}, + "description": "Values for the `{{ name }}` expressions in the file.", + "title": "Vars", + "type": "object" + }, + "tasks": { + "additionalProperties": { + "anyOf": [ + { + "type": "string" + }, + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "$ref": "#/$defs/Task" + } + ] + }, + "description": "Tasks by name. A task is a shell line, a list of shell lines, or a mapping with `run` and `help`.", + "minProperties": 1, + "title": "Tasks", + "type": "object" + }, + "blueprint": { + "type": "string", + "title": "Blueprint", + "description": "Blueprint that turns this file into a CLI." + } + }, + "required": [ + "name", + "tasks" + ], + "title": "Tasks", + "type": "object" +} diff --git a/examples/blueprints/tasks.yaml b/examples/blueprints/tasks.yaml new file mode 100644 index 0000000..23b38d0 --- /dev/null +++ b/examples/blueprints/tasks.yaml @@ -0,0 +1,22 @@ +# yaml-language-server: $schema=tasks.schema.json +# Written in the spec defined by tasks.py, not in the cliffy manifest schema. +blueprint: tasks.py + +name: tasks +version: 0.1.0 +help: Project tasks + +# jinja2 variables: the whole file is rendered before the blueprint sees it +vars: + tag: v0.1.0 + +tasks: + greet: echo "hello from tasks" + check: + - echo "linting" + - echo "testing" + release: + help: Check, then tag the release + run: + - echo "linting" + - echo "tagging {{ tag }}" diff --git a/examples/blueprints/text-tools.yaml b/examples/blueprints/text-tools.yaml new file mode 100644 index 0000000..0edb696 --- /dev/null +++ b/examples/blueprints/text-tools.yaml @@ -0,0 +1,69 @@ +# yaml-language-server: $schema=mcp_tools.schema.json +# Written in the spec defined by mcp_tools.py, not in the cliffy manifest schema. +blueprint: mcp_tools.py + +name: text-tools +version: 0.1.0 +description: Text tools, as a CLI and as an MCP server + +imports: + - import re + +tools: + format_text: + description: Change the case of a text and trim it + parameters: + - name: text + description: The text to format + required: true + - name: uppercase + type: boolean + description: Convert the text to uppercase + - name: trim + type: boolean + description: Remove leading and trailing whitespace + run: | + if uppercase: + text = text.upper() + return text.strip() if trim else text + + count_words: + description: Count the words in a text + parameters: + - name: text + description: The text to count words in + required: true + run: | + return {"words": len(text.split())} + + extract_entities: + description: Find dates and email addresses in a text + parameters: + - name: text + description: The text to search + required: true + - name: entity_types + type: array + description: Only return these kinds of entities (dates, emails) + run: | + found = { + "dates": re.findall(r"\d{1,2}[/-]\d{1,2}[/-]\d{2,4}", text), + "emails": re.findall(r"[\w.%+-]+@[\w.-]+\.[a-zA-Z]{2,}", text), + } + return {kind: values for kind, values in found.items() if not entity_types or kind in entity_types} + + repeat: + description: Repeat a text + parameters: + - name: text + description: The text to repeat + required: true + - name: times + type: integer + default: 2 + description: How many times + - name: separator + default: " " + description: Text put between the repetitions + run: | + return separator.join([text] * times) diff --git a/examples/callbacks.yaml b/examples/callbacks.yaml new file mode 100644 index 0000000..a61bd0b --- /dev/null +++ b/examples/callbacks.yaml @@ -0,0 +1,38 @@ +# yaml-language-server: $schema=cliffy_schema.json + +name: callbacks +version: 0.1.0 +help: Callbacks run before the commands of a CLI or of a command group + +callbacks: + # no `group`: runs before every command, and its params become options of the CLI itself + - params: + - name: --verbose + type: bool + default: false + short: -v + help: Print what is about to run + invoke_without_command: true + run: | + ctx.obj = {"verbose": verbose} + if ctx.invoked_subcommand is None: + print("no command given, try --help") + elif verbose: + print(f"running {ctx.invoked_subcommand}") + + # runs before every command of the `db` group + - group: db + run: print("connecting to db") + +commands: + hello: print("hello") + db.migrate: print("migrating") + db.status: print("up to date") + +tests: + - hello: assert result.output == "hello\n" + - --verbose hello: assert result.output == "running hello\nhello\n" + - -v db migrate: assert result.output == "running db\nconnecting to db\nmigrating\n" + - db status: assert result.output == "connecting to db\nup to date\n" + - "": assert result.output == "no command given, try --help\n" + - --version: assert result.output == "callbacks, 0.1.0\n" diff --git a/examples/cliffy_schema.json b/examples/cliffy_schema.json index d3f7f42..9f3a10c 100644 --- a/examples/cliffy_schema.json +++ b/examples/cliffy_schema.json @@ -1,5 +1,50 @@ { "$defs": { + "Callback": { + "description": "Code that runs before the commands of the CLI, or of one command group. A callback can declare its own\nparams, which become options of the CLI (or group) itself, i.e. `mycli --verbose command`.\nThe Typer/Click context is available in the run block as `ctx`.", + "properties": { + "run": { + "$ref": "#/$defs/RunBlock", + "description": "Code to run before the invoked command. `ctx` holds the context." + }, + "group": { + "default": "", + "description": "Command group the callback belongs to, i.e. `db` or `db.tables`. Leave empty for the CLI itself.", + "title": "Group", + "type": "string" + }, + "params": { + "default": [], + "description": "Params of the CLI (or group) itself, given before the command name. Same syntax as command params.", + "items": { + "anyOf": [ + { + "$ref": "#/$defs/CommandParam" + }, + { + "$ref": "#/$defs/SimpleCommandParam" + }, + { + "$ref": "#/$defs/GenericCommandParam" + } + ] + }, + "title": "Params", + "type": "array" + }, + "invoke_without_command": { + "default": false, + "description": "Also run the callback when the CLI (or group) is invoked without a command, instead of showing the help.", + "title": "Invoke Without Command", + "type": "boolean" + } + }, + "required": [ + "run" + ], + "title": "Callback", + "type": "object" + }, "Command": { "description": "Defines a single command within the CLI. It specifies the command's execution logic,\nparameters, and configuration.", "properties": { @@ -459,6 +504,15 @@ "description": "A mapping containing the command definitions for the CLI. Each command should have a unique key- which can be either a group command or nested subcommands. Nested subcommands are joined by '.' in between each level. Aliases for commands can be separated in the key by '|'. A special '(*)' wildcard can be used to spread the subcommand to all group-level commands", "title": "Commands" }, + "callbacks": { + "default": [], + "description": "Code that runs before any command of the CLI, or of one command group. Callbacks for the same target run in the order given and their params are combined. The built-in --version and --aliases options are kept.", + "items": { + "$ref": "#/$defs/Callback" + }, + "title": "Callbacks", + "type": "array" + }, "cli_options": { "additionalProperties": true, "description": "Additional CLI configuration options", @@ -503,8 +557,7 @@ }, "required": [ "name", - "version", - "commands" + "version" ], "title": "CLIManifest", "type": "object" diff --git a/mkdocs.yaml b/mkdocs.yaml index 6a5981e..e31a538 100644 --- a/mkdocs.yaml +++ b/mkdocs.yaml @@ -3,6 +3,7 @@ nav: - Cliffy: index.md - Getting Started: getting-started.md - Features: features.md + - Blueprints: blueprints.md - Schema: schema.md theme: name: material diff --git a/tests/test_blueprint.py b/tests/test_blueprint.py new file mode 100644 index 0000000..1a44dd9 --- /dev/null +++ b/tests/test_blueprint.py @@ -0,0 +1,505 @@ +import json +import textwrap +from pathlib import Path +from typing import Any + +import pytest +from click.testing import CliRunner +from pydantic import Field + +from cliffy.blueprint import ( + Blueprint, + BlueprintError, + build_manifest, + installed_blueprints, + normalize_reference, + resolve_blueprint, +) +from cliffy.cli import cli +from cliffy.homer import get_metadata +from cliffy.manifest import CLIManifest +from cliffy.reloader import Reloader +from cliffy.transformer import Transformer + +EXAMPLES = Path(__file__).parent.parent / "examples" / "blueprints" +# pytest imports this file as a top-level module, so the blueprints below are importable by reference +SHOUT = f"{__name__}:Shout" + +GREETER_BLUEPRINT = ''' +from pydantic import Field + +from cliffy.blueprint import Blueprint + + +class Greeter(Blueprint): + """Greeter: one command per greeting + + More details that are not part of the description. + """ + + name: str + greetings: dict[str, str] = Field(min_length=1) + + def to_manifest(self): + commands = {key: f"print({text!r})" for key, text in self.greetings.items()} + return {"name": self.name, "version": "1.0.0", "commands": commands} +''' + + +class Shout(Blueprint): + """Shout: prints every word in uppercase""" + + name: str + words: list[str] = Field(min_length=1) + + @classmethod + def pre_validate(cls, raw: dict[str, Any]) -> dict[str, Any]: + if isinstance(raw.get("words"), str): + raw = {**raw, "words": raw["words"].split()} + return raw + + def to_manifest(self) -> CLIManifest: + commands = {word: f"print({word.upper()!r})" for word in self.words} + return CLIManifest(name=self.name, version="1.0.0", commands=commands) + + def post_manifest(self, manifest: CLIManifest) -> CLIManifest: + manifest.help = f"{len(self.words)} words" + return manifest + + def post_codegen(self, code: str) -> str: + return code + "\n# shouted\n" + + +class Broken(Blueprint): + name: str + + def to_manifest(self) -> dict[str, Any]: + return {"name": self.name} + + +class Crashing(Blueprint): + name: str + + def to_manifest(self) -> CLIManifest: + raise RuntimeError("boom") + + +@pytest.fixture +def greeter(tmp_path): + """A blueprint file and a spec that selects it with a relative `blueprint:` key.""" + (tmp_path / "greeter.py").write_text(GREETER_BLUEPRINT) + spec = tmp_path / "specs" / "hi.yaml" + spec.parent.mkdir() + spec.write_text("blueprint: ../greeter.py\nname: bp-hi\ngreetings:\n hello: hello there\n bye: see you\n") + return spec + + +def add_plugin(directory: Path, dist: str, entry_points: dict[str, str]) -> None: + """Install a fake distribution that registers blueprints, the way a real package would.""" + # like a wheel does, the directory uses the normalized name (importlib.metadata splits it on the first "-") + dist_info = directory / f"{dist.replace('-', '_')}-1.0.dist-info" + dist_info.mkdir(parents=True) + (dist_info / "METADATA").write_text(f"Metadata-Version: 2.1\nName: {dist}\nVersion: 1.0\n") + lines = "".join(f"{name} = {target}\n" for name, target in entry_points.items()) + (dist_info / "entry_points.txt").write_text(f"[cliffy.blueprints]\n{lines}") + + +@pytest.fixture +def plugin(tmp_path, monkeypatch): + plugin_dir = tmp_path / "plugins" + plugin_dir.mkdir() + (plugin_dir / "bp_greeter_plugin.py").write_text(GREETER_BLUEPRINT) + add_plugin(plugin_dir, "bp-greeter-plugin", {"greeter": "bp_greeter_plugin:Greeter"}) + monkeypatch.syspath_prepend(str(plugin_dir)) + return plugin_dir + + +def test_blueprint_base_is_abstract(): + with pytest.raises(TypeError): + Blueprint() # type: ignore[abstract] + + +def test_build_manifest_runs_hooks_in_order(): + spec, manifest = build_manifest(Shout, {"blueprint": "ignored", "name": "shout", "words": "a b"}) + + assert isinstance(spec, Shout) + assert spec.words == ["a", "b"] + assert manifest.name == "shout" + assert manifest.help == "2 words" + assert set(manifest.commands) == {"a", "b"} + + +def test_build_manifest_reports_spec_errors_against_the_spec(): + with pytest.raises(BlueprintError) as error: + build_manifest(Shout, {"name": "shout", "words": [], "wrods": ["a"]}, source="shout.yaml") + + message = str(error.value) + assert "shout.yaml is not a valid Shout spec" in message + assert "words: List should have at least 1 item" in message + assert "wrods: Extra inputs are not permitted" in message + + +def test_build_manifest_reports_nested_locations(): + blueprint_cls = resolve_blueprint(str(EXAMPLES / "mcp_tools.py")) + raw = {"name": "x", "tools": [{"name": "ok", "description": "d", "run": "return 1", "parameters": [{}]}]} + with pytest.raises(BlueprintError, match=r"tools\[0\]\.parameters\[0\]\.name: Field required"): + build_manifest(blueprint_cls, raw) + + +def test_build_manifest_reports_invalid_manifest_from_blueprint(): + with pytest.raises(BlueprintError) as error: + build_manifest(Broken, {"name": "broken"}) + + assert "blueprint Broken produced an invalid cliffy manifest" in str(error.value) + assert "version: Field required" in str(error.value) + + +def test_build_manifest_reports_crashing_blueprint(): + with pytest.raises(BlueprintError, match="blueprint Crashing failed to transform spec: RuntimeError: boom"): + build_manifest(Crashing, {"name": "crash"}) + + +def test_describe_and_json_schema(): + assert Shout.describe() == "Shout: prints every word in uppercase" + assert Broken.describe() == "" + + schema = Shout.spec_json_schema() + assert schema["required"] == ["name", "words"] + assert schema["additionalProperties"] is False + assert schema["properties"]["blueprint"]["type"] == "string" + assert "blueprint" not in Shout.model_json_schema()["properties"] + + +def test_resolve_file(greeter): + blueprint_cls = resolve_blueprint("greeter.py", base_dir=greeter.parent.parent) + assert blueprint_cls.__name__ == "Greeter" + assert blueprint_cls.describe() == "Greeter: one command per greeting" + + assert resolve_blueprint(f"{greeter.parent.parent / 'greeter.py'}:Greeter").__name__ == "Greeter" + + +def test_resolve_file_errors(tmp_path): + with pytest.raises(BlueprintError, match="blueprint file not found"): + resolve_blueprint("missing.py", base_dir=tmp_path) + + (tmp_path / "empty.py").write_text("x = 1\n") + with pytest.raises(BlueprintError, match="no blueprint found"): + resolve_blueprint(str(tmp_path / "empty.py")) + + (tmp_path / "broken.py").write_text("import module_that_does_not_exist_xyz\n") + with pytest.raises(BlueprintError, match=r"could not import blueprint file .*ModuleNotFoundError"): + resolve_blueprint(str(tmp_path / "broken.py")) + + (tmp_path / "two.py").write_text(GREETER_BLUEPRINT + "\n\nclass Second(Greeter):\n pass\n") + with pytest.raises(BlueprintError, match=r"defines more than one blueprint \(Greeter, Second\)"): + resolve_blueprint(str(tmp_path / "two.py")) + assert resolve_blueprint(f"{tmp_path / 'two.py'}:Second").__name__ == "Second" + + with pytest.raises(BlueprintError, match="has no attribute `Missing`"): + resolve_blueprint(f"{tmp_path / 'two.py'}:Missing") + + +def test_resolve_module_reference(): + assert resolve_blueprint(SHOUT) is Shout + + with pytest.raises(BlueprintError, match="defines more than one blueprint"): + resolve_blueprint(__name__) + + with pytest.raises(BlueprintError, match="is not a blueprint"): + resolve_blueprint(f"{__name__}:EXAMPLES") + + with pytest.raises(BlueprintError, match="is not a blueprint"): + resolve_blueprint("cliffy.blueprint:Blueprint") + + with pytest.raises(BlueprintError, match=r"no blueprint found in `cliffy\.manifest`"): + resolve_blueprint("cliffy.manifest") + + +def test_resolve_not_found_lists_installed(plugin): + with pytest.raises(BlueprintError) as error: + resolve_blueprint("no_such_blueprint_xyz") + + assert "blueprint `no_such_blueprint_xyz` not found" in str(error.value) + assert "installed: greeter" in str(error.value) + + with pytest.raises(BlueprintError, match="empty blueprint reference"): + resolve_blueprint(" ") + + +def test_resolve_installed_plugin(plugin): + installed = {blueprint.name: blueprint for blueprint in installed_blueprints()} + assert installed["greeter"].target == "bp_greeter_plugin:Greeter" + assert installed["greeter"].package == "bp-greeter-plugin" + + assert resolve_blueprint("greeter").__name__ == "Greeter" + + +def test_resolve_ambiguous_plugin(plugin): + add_plugin(plugin, "bp-other-plugin", {"greeter": "bp_greeter_plugin"}) + + with pytest.raises(BlueprintError) as error: + resolve_blueprint("greeter") + + assert "blueprint `greeter` is ambiguous" in str(error.value) + assert "bp-greeter-plugin (bp_greeter_plugin:Greeter)" in str(error.value) + assert "bp-other-plugin (bp_greeter_plugin)" in str(error.value) + # a full reference still works + assert resolve_blueprint("bp_greeter_plugin:Greeter").__name__ == "Greeter" + + +def test_resolve_plugin_that_fails_to_load(plugin): + add_plugin(plugin, "bp-stale-plugin", {"stale": "bp_missing_module_xyz:Nope"}) + + with pytest.raises(BlueprintError, match="could not load blueprint `stale`"): + resolve_blueprint("stale") + + +def test_normalize_reference(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + resolved = tmp_path.resolve() + + assert normalize_reference("bp.py") == str(resolved / "bp.py") + assert normalize_reference("sub/bp.py:Cls") == f"{resolved / 'sub' / 'bp.py'}:Cls" + assert normalize_reference("bp.py", base_dir=tmp_path / "specs") == str(resolved / "specs" / "bp.py") + assert normalize_reference("package.module:Cls") == "package.module:Cls" + assert normalize_reference(" plugin-name ") == "plugin-name" + + +def test_transformer_applies_blueprint_from_file_key(greeter): + with open(greeter) as spec: + T = Transformer(spec) + + assert T.cli.name == "bp-hi" + assert T.cli.blueprint == str((greeter.parent.parent / "greeter.py").resolve()) + assert isinstance(T.manifest, CLIManifest) + assert set(T.manifest.commands) == {"hello", "bye"} + assert "print('hello there')" in T.cli.code + + +def test_transformer_option_overrides_file_key_and_runs_post_codegen(tmp_path): + spec_path = tmp_path / "shout.yaml" + spec_path.write_text("blueprint: does-not-exist\nname: bp-shout\nwords: [hey]\n") + + with open(spec_path) as spec: + T = Transformer(spec, blueprint=SHOUT) + + assert T.cli.blueprint == SHOUT + assert T.cli.code.endswith("\n# shouted\n") + assert "print('HEY')" in T.cli.code + + +def test_transformer_renders_jinja_before_the_blueprint(tmp_path): + spec_path = tmp_path / "shout.yaml" + spec_path.write_text("name: bp-shout\nwords: [\"{{ 'template' ~ 'd' }}\"]\n") + + with open(spec_path) as spec: + T = Transformer(spec, blueprint=SHOUT) + + assert "print('TEMPLATED')" in T.cli.code + + +@pytest.mark.parametrize( + ("content", "blueprint", "expected"), + [ + ("blueprint: nope_xyz\nname: x\n", None, "blueprint `nope_xyz` not found"), + ("blueprint: 12\nname: x\n", None, "`blueprint` must be a blueprint reference"), + ("name: x\nwords: []\n", SHOUT, "is not a valid Shout spec"), + ("name: x\n", f"{__name__}:Broken", "produced an invalid cliffy manifest"), + ], +) +def test_transformer_blueprint_errors_exit_with_a_message(tmp_path, capsys, content, blueprint, expected): + spec_path = tmp_path / "spec.yaml" + spec_path.write_text(content) + + with open(spec_path) as spec, pytest.raises(SystemExit) as exit_info: + Transformer(spec, blueprint=blueprint) + + assert exit_info.value.code == 1 + assert expected in capsys.readouterr().err + + +def test_plain_manifest_is_untouched_without_blueprint(): + with open("examples/hello.yaml") as manifest: + T = Transformer(manifest) + + assert T.blueprint is None + assert T.cli.blueprint is None + + +def test_cli_render_run_and_validate_with_blueprint(greeter): + runner = CliRunner() + + rendered = runner.invoke(cli, ["render", str(greeter)]) + assert rendered.exit_code == 0 + assert "print('see you')" in rendered.output + + ran = runner.invoke(cli, ["run", str(greeter), "--", "hello"]) + assert ran.exit_code == 0 + assert "hello there" in ran.output + + assert runner.invoke(cli, ["validate", str(greeter)]).exit_code == 0 + + invalid = runner.invoke(cli, ["validate", str(greeter), "--blueprint", SHOUT]) + assert invalid.exit_code == 1 + assert "is not a valid Shout spec" in invalid.output + + +def test_cli_load_remembers_blueprint_for_update_and_info(greeter): + runner = CliRunner() + spec = greeter.with_name("loaded.yaml") + spec.write_text("name: bp-loaded\nwords: [one]\n") + + loaded = runner.invoke(cli, ["load", str(spec), "--blueprint", SHOUT]) + assert loaded.exit_code == 0, loaded.output + metadata = get_metadata("bp-loaded") + assert metadata is not None + assert metadata.blueprint == SHOUT + + info = runner.invoke(cli, ["info", "bp-loaded"]) + assert f"blueprint: {SHOUT}" in info.output + + # `update` has no --blueprint option: it has to reuse the one given to `load` + spec.write_text("name: bp-loaded\nwords: [one, two]\n") + updated = runner.invoke(cli, ["update", "bp-loaded"]) + assert updated.exit_code == 0, updated.output + assert "Reloaded bp-loaded" in updated.output + + assert runner.invoke(cli, ["remove", "bp-loaded"]).exit_code == 0 + + +def test_reloader_reloads_with_blueprint(tmp_path): + spec = tmp_path / "reloaded.yaml" + spec.write_text("name: bp-reloaded\nwords: [one]\n") + + Reloader.reload(str(spec), False, (), blueprint=SHOUT) + + metadata = get_metadata("bp-reloaded") + assert metadata is not None + assert metadata.blueprint == SHOUT + assert CliRunner().invoke(cli, ["remove", "bp-reloaded"]).exit_code == 0 + + +def test_cli_test_with_blueprint(tmp_path): + (tmp_path / "tested.py").write_text( + textwrap.dedent( + """ + from cliffy.blueprint import Blueprint + + + class Tested(Blueprint): + name: str + + def to_manifest(self): + return { + "name": self.name, + "version": "1.0.0", + "commands": {"ping": "print('pong')"}, + "tests": [{"ping": "assert result.output == 'pong\\\\n'"}], + } + """ + ) + ) + spec = tmp_path / "tested.yaml" + spec.write_text("name: bp-tested\n") + + result = CliRunner().invoke(cli, ["test", str(spec), "-b", str(tmp_path / "tested.py")]) + assert "All tests passed" in result.output + assert result.exit_code == 0 + + +def test_cli_docs_with_blueprint(greeter): + result = CliRunner().invoke(cli, ["docs", str(greeter), "-o", str(greeter.parent)]) + assert result.exit_code == 0, result.output + assert "see you" in (greeter.parent / "bp-hi.md").read_text() or "bye" in (greeter.parent / "bp-hi.md").read_text() + + +def test_cli_blueprint_list(plugin): + runner = CliRunner() + for command in (["blueprint", "list"], ["blueprints"]): + result = runner.invoke(cli, command) + assert result.exit_code == 0 + assert "greeter" in result.output + assert "bp-greeter-plugin" in result.output + assert "per greeting" in result.output + + +def test_cli_blueprint_list_shows_broken_plugins(plugin): + add_plugin(plugin, "bp-stale-plugin", {"stale": "bp_missing_module_xyz:Nope"}) + + result = CliRunner().invoke(cli, ["blueprint", "list"]) + assert result.exit_code == 0 + assert "stale" in result.output + assert "error" in result.output + + +def test_cli_blueprint_schema(tmp_path): + runner = CliRunner() + + printed = runner.invoke(cli, ["blueprint", "schema", SHOUT]) + assert printed.exit_code == 0 + assert json.loads(printed.output)["required"] == ["name", "words"] + + output = tmp_path / "shout.schema.json" + written = runner.invoke(cli, ["blueprint", "schema", SHOUT, "-o", str(output)]) + assert written.exit_code == 0 + assert json.loads(output.read_text())["properties"]["blueprint"]["type"] == "string" + + missing = runner.invoke(cli, ["blueprint", "schema", "nope_xyz"]) + assert missing.exit_code == 1 + assert "not found" in missing.output + + +@pytest.mark.parametrize("example", ["tasks", "mcp_tools"]) +def test_example_schema_is_up_to_date(example): + """The schema files next to the examples give editor completion; they must match the blueprints.""" + blueprint_cls = resolve_blueprint(str(EXAMPLES / f"{example}.py")) + committed = json.loads((EXAMPLES / f"{example}.schema.json").read_text()) + assert committed == blueprint_cls.spec_json_schema() + + +@pytest.mark.parametrize( + ("spec", "args", "expected"), + [ + ("tasks", ["greet"], "hello from tasks"), + ("tasks", ["check"], "linting\ntesting"), + ("tasks", ["--version"], "tasks, 0.1.0"), + ("text-tools", ["format-text", " hi there ", "--uppercase", "--trim"], "HI THERE\n"), + ("text-tools", ["count-words", "one two three"], '"words": 3'), + ("text-tools", ["extract-entities", "a@b.io on 1/2/2026", "--entity-types", "emails"], "a@b.io"), + ("text-tools", ["repeat", "ab", "--times", "3", "--separator", "-"], "ab-ab-ab"), + ("text-tools", ["tools"], '"name": "extract_entities"'), + ], +) +def test_example_specs_run(capfd, spec, args, expected): + result = CliRunner().invoke(cli, ["run", str(EXAMPLES / f"{spec}.yaml"), "--", *args]) + + assert result.exit_code == 0, result.output + # task lines run in subprocesses, their output is only visible on the real stdout + assert expected in result.output + capfd.readouterr().out + + +@pytest.mark.parametrize("spec", ["tasks", "text-tools"]) +def test_example_specs_validate(spec): + result = CliRunner().invoke(cli, ["validate", str(EXAMPLES / f"{spec}.yaml")]) + assert result.exit_code == 0, result.output + + +def test_example_spec_tests_pass(): + result = CliRunner().invoke(cli, ["test", str(EXAMPLES / "text-tools.yaml")]) + assert "All tests passed" in result.output + assert result.exit_code == 0 + + +def test_example_mcp_blueprint_rejects_bad_specs(): + blueprint_cls = resolve_blueprint(str(EXAMPLES / "mcp_tools.py")) + tool = {"description": "d", "run": "return 1"} + + with pytest.raises(BlueprintError, match="`serve` is a reserved name"): + build_manifest(blueprint_cls, {"name": "x", "tools": {"serve": tool}}) + + with pytest.raises(BlueprintError, match="duplicate tool names: same"): + build_manifest(blueprint_cls, {"name": "x", "tools": [{"name": "same", **tool}, {"name": "same", **tool}]}) + + required_flag = {**tool, "parameters": [{"name": "flag", "type": "boolean", "required": True}]} + with pytest.raises(BlueprintError, match="a boolean parameter is a flag and cannot be required"): + build_manifest(blueprint_cls, {"name": "x", "tools": {"ok": required_flag}}) diff --git a/tests/test_builder.py b/tests/test_builder.py index 35990c4..795ef69 100644 --- a/tests/test_builder.py +++ b/tests/test_builder.py @@ -1,13 +1,31 @@ import glob +import json import os +import subprocess +import sys import tempfile -from unittest.mock import patch +import zipfile +from importlib import metadata +from pathlib import Path +from unittest.mock import MagicMock, patch import pytest +from click.testing import CliRunner -from cliffy.builder import build_cli, build_cli_from_manifest, run_cli +from cliffy.builder import ( + build_cli, + build_cli_from_manifest, + get_build_requirements, + get_python_guard, + get_requires_python, + run_cli, +) +from cliffy.cli import cli from cliffy.helper import TEMP_FILES +EXAMPLES = Path(__file__).parent.parent / "examples" / "blueprints" +TYPER_CODE = "## Generated demo on 2026-01-01 00:00:00\nfrom typing import Optional, Any\nimport typer\n" +CLICK_CODE = "## Generated demo on 2026-01-01 00:00:00\nimport rich_click as click\n" TYPER_MANIFEST = "name: buildme\nversion: '1'\ncommands:\n a: print(1)\n" CLICK_MANIFEST = "name: buildme\nversion: '1'\nuse_click: true\ncommands:\n a: print(1)\n" @@ -21,6 +39,229 @@ def temp_dir(monkeypatch, tmp_path): return temp_dir +def skip_pip(): + """Patches the pip run of a build: the zipapp is then built from the script alone""" + pip_run = subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="") + return patch("cliffy.builder.subprocess.run", return_value=pip_run) + + +def pip_requirements(mock_pip_run: MagicMock) -> list[str]: + """Requirements a patched pip run was asked to install into the staging directory""" + pip_args = mock_pip_run.call_args.args[0] + assert pip_args[:3] == [sys.executable, "-m", "pip"] + assert "install" in pip_args + return pip_args[pip_args.index("--target") + 2 :] + + +def run_artifact( + artifact: Path, args: list[str], workdir: Path, stdin: str | None = None +) -> subprocess.CompletedProcess: + """Run a built CLI the way a user would on another machine: other directory, no cliffy, no blueprint.""" + environment = { + "PATH": os.environ.get("PATH", ""), + # where the zipapp unpacks its bundled packages on first run + "SHIV_ROOT": str(workdir / "shiv-root"), + "PYTHONNOUSERSITE": "1", + } + return subprocess.run( + # -I: isolated mode, so nothing from the test environment can leak into the artifact + [sys.executable, "-I", str(artifact), *args], + capture_output=True, + encoding="utf-8", + cwd=workdir, + env=environment, + input=stdin, + ) + + +def test_build_requirements_pin_the_framework(): + requirements = get_build_requirements(TYPER_CODE, ["requests>=2", "six"]) + + assert requirements[:2] == ["requests>=2", "six"] + assert len(requirements) == 3 + assert requirements[2].startswith("typer==") + + click_requirements = get_build_requirements(CLICK_CODE, []) + assert len(click_requirements) == 1 + assert click_requirements[0].startswith("rich-click==") + + +def test_build_requirements_keep_the_users_framework_pin(): + assert get_build_requirements(TYPER_CODE, ["Typer>=0.24"]) == ["Typer>=0.24"] + # an unparsable requirement is left for pip to report + assert get_build_requirements("print(1)\n", ["not a requirement!"]) == ["not a requirement!"] + + +def write_dist(site_packages: Path, name: str, requires_python: str | None) -> None: + dist_info = site_packages / f"{name}-1.0.dist-info" + dist_info.mkdir(parents=True) + line = f"Requires-Python: {requires_python}\n" if requires_python else "" + (dist_info / "METADATA").write_text( + f"Metadata-Version: 2.1\nName: {name}\nVersion: 1.0\n{line}\nRequires-Python: >=9\n" + ) + + +def test_requires_python_is_the_highest_floor_of_pure_packages(tmp_path): + write_dist(tmp_path, "old", ">=3.6") + write_dist(tmp_path, "new", ">=3.9,<4") + write_dist(tmp_path, "any", None) + + assert get_requires_python(tmp_path) == ">=3.9" + + +def test_requires_python_is_exact_with_compiled_extensions(tmp_path): + write_dist(tmp_path, "fast", ">=3.8") + (tmp_path / "fast.cpython-311-darwin.so").write_bytes(b"") + + assert get_requires_python(tmp_path) == f"==3.{sys.version_info.minor}.*" + + # abi3 extensions work across Python versions + (tmp_path / "fast.cpython-311-darwin.so").rename(tmp_path / "fast.abi3.so") + assert get_requires_python(tmp_path) == ">=3.8" + + +@pytest.mark.parametrize( + ("requires_python", "passes", "expected"), + [ + (">=3.8", True, ""), + (f"==3.{sys.version_info.minor}.*", True, ""), + (">=3.99", False, "demo was built for Python 3.99 or newer, but is running on Python 3."), + ("==3.1.*", False, "demo was built for Python 3.1, but is running on Python 3."), + ], +) +def test_python_guard(requires_python, passes, expected): + guard = get_python_guard("demo", requires_python) + + if passes: + exec(guard, {}) + return + + with pytest.raises(SystemExit) as exit_info: + exec(guard, {}) + assert expected in str(exit_info.value.code) + + +def test_build_reports_missing_script(tmp_path): + result = build_cli("demo", script_path=str(tmp_path / "missing.py"), output_dir=str(tmp_path)) + + assert result.exit_code == 1 + assert "could not read the generated CLI" in result.output + assert result.path is None + + +def test_build_reports_dependency_that_cannot_be_installed(tmp_path): + script = tmp_path / "demo.py" + script.write_text("print(1)\n") + + result = build_cli("demo", script_path=str(script), deps=["not a requirement!"], output_dir=str(tmp_path / "out")) + + assert result.exit_code == 1 + assert "could not install not a requirement!" in result.output + assert not (tmp_path / "out" / "demo").exists() + + +def test_build_command_shows_the_reason_of_a_failed_build(tmp_path): + manifest = tmp_path / "broken.yaml" + manifest.write_text( + "name: bp-broken-build\nversion: 1.0.0\nrequires: ['not a requirement!']\ncommands:\n a: print(1)\n" + ) + + result = CliRunner().invoke(cli, ["build", str(manifest), "-o", str(tmp_path / "out")]) + + # a failed build is reported with the exit code as well + assert result.exit_code == 1 + assert "could not install" in result.output + assert "bp-broken-build build failed" in result.output + + +def test_build_blueprint_spec_into_a_single_self_contained_file(tmp_path): + """End to end: a YAML file in a custom spec becomes one executable that needs neither cliffy nor the blueprint.""" + output_dir = tmp_path / "dist" + result = CliRunner().invoke(cli, ["build", str(EXAMPLES / "text-tools.yaml"), "-o", str(output_dir)]) + + assert result.exit_code == 0, result.output + assert "text-tools built" in result.output + assert "single file, needs Python >=3." in result.output + + assert [path.name for path in output_dir.iterdir()] == ["text-tools"] + artifact = output_dir / "text-tools" + assert os.access(artifact, os.X_OK) + assert artifact.read_bytes().startswith(b"#!/usr/bin/env python3\n") + + with zipfile.ZipFile(artifact) as archive: + names = archive.namelist() + assert "site-packages/text_tools.py" in names + assert any(name.startswith("site-packages/typer/") for name in names) + # the blueprint is applied at build time, neither it nor cliffy is shipped + assert not any("cliffy" in name or "mcp_tools" in name or "pydantic" in name for name in names) + + workdir = tmp_path / "elsewhere" + workdir.mkdir() + + formatted = run_artifact(artifact, ["format-text", " hello there ", "--uppercase", "--trim"], workdir) + assert formatted.returncode == 0, formatted.stderr + assert formatted.stdout == "HELLO THERE\n" + + version = run_artifact(artifact, ["--version"], workdir) + assert version.stdout == "text-tools, 0.1.0\n" + + requests = [ + {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18"}}, + {"jsonrpc": "2.0", "method": "notifications/initialized"}, + {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}, + { + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": {"name": "count_words", "arguments": {"text": "a b"}}, + }, + {"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "nope", "arguments": {}}}, + ] + served = run_artifact( + artifact, ["serve"], workdir, stdin="".join(json.dumps(request) + "\n" for request in requests) + ) + assert served.returncode == 0, served.stderr + replies = [json.loads(line) for line in served.stdout.splitlines()] + + assert [reply["id"] for reply in replies] == [1, 2, 3, 4] + assert replies[0]["result"]["serverInfo"] == {"name": "text-tools", "version": "0.1.0"} + assert [tool["name"] for tool in replies[1]["result"]["tools"]] == [ + "format_text", + "count_words", + "extract_entities", + "repeat", + ] + assert json.loads(replies[2]["result"]["content"][0]["text"]) == {"words": 2} + assert replies[2]["result"]["isError"] is False + assert replies[3]["result"]["isError"] is True + + +def test_build_is_reproducible_and_takes_blueprint_and_interpreter_options(tmp_path): + spec = tmp_path / "jobs.yaml" + spec.write_text("name: bp-jobs\ntasks:\n greet: echo hello from a build\n") + blueprint = str(EXAMPLES / "tasks.py") + interpreter = "/usr/bin/env -S python3 -I" + + builds = [] + for output_dir in (tmp_path / "one", tmp_path / "two"): + with open(spec) as spec_io: + built_cli, result = build_cli_from_manifest( + spec_io, output_dir=str(output_dir), interpreter=interpreter, blueprint=blueprint + ) + assert result.exit_code == 0, result.output + assert built_cli.name == "bp-jobs" + assert result.path == str(output_dir / "bp-jobs") + assert result.requires_python is not None and result.requires_python.startswith(">=3.") + builds.append((output_dir / "bp-jobs").read_bytes()) + + assert builds[0] == builds[1] + assert builds[0].startswith(f"#!{interpreter}\n".encode()) + + ran = run_artifact(tmp_path / "one" / "bp-jobs", ["greet"], tmp_path) + assert ran.returncode == 0, ran.stderr + assert "hello from a build" in ran.stdout + + def test_run_cli_removes_temp_file_when_import_fails(temp_dir): with pytest.raises(ImportError): run_cli("broken", "import no_such_module_for_cliffy", ()) @@ -41,38 +282,41 @@ def test_build_removes_temp_file_when_build_fails(temp_dir, tmp_path): build_cli_from_manifest(manifest) # type: ignore[arg-type] assert glob.glob(f"{temp_dir}/buildme_*") == [] + # the script is written into a temporary directory: neither it nor the directory is left behind + assert list(temp_dir.iterdir()) == [] assert TEMP_FILES == [] @pytest.mark.parametrize( - "manifest, expected_requirements", - [(TYPER_MANIFEST, ["typer"]), (CLICK_MANIFEST, ["rich-click"])], + "manifest, framework", + [(TYPER_MANIFEST, "typer"), (CLICK_MANIFEST, "rich-click")], ids=["typer", "click"], ) -def test_build_installs_the_framework_the_cli_imports(tmp_path, manifest, expected_requirements): +def test_build_installs_the_framework_the_cli_imports(tmp_path, manifest, framework): manifest_path = tmp_path / "buildme.yaml" manifest_path.write_text(manifest) output_dir = tmp_path / "out" # pip is not run: the zipapp is built from the script alone - with patch("cliffy.builder.pip.install") as mock_pip_install, open(manifest_path) as manifest_io: + with skip_pip() as mock_pip_run, open(manifest_path) as manifest_io: cli, result = build_cli_from_manifest(manifest_io, output_dir=str(output_dir)) # type: ignore[arg-type] assert result.exit_code == 0, result.output assert os.path.exists(output_dir / cli.name) - pip_args = mock_pip_install.call_args.args[0] - assert pip_args[0] == "--target" - assert pip_args[2:] == expected_requirements + pip_args = mock_pip_run.call_args.args[0] + assert pip_args[pip_args.index("--target") + 1] != str(output_dir) + # only the framework of the generated code, pinned to the version installed next to cliffy + assert pip_requirements(mock_pip_run) == [f"{framework}=={metadata.version(framework)}"] def test_framework_requirements_and_deps_are_combined(tmp_path): script_path = tmp_path / "script.py" script_path.write_text("import rich_click as click\n") - with patch("cliffy.builder.pip.install") as mock_pip_install: + with skip_pip() as mock_pip_run: build_cli("script", str(script_path), deps=["requests>=2,<3"], output_dir=str(tmp_path / "out")) - assert mock_pip_install.call_args.args[0][2:] == ["rich-click", "requests>=2,<3"] + assert pip_requirements(mock_pip_run) == ["requests>=2,<3", f"rich-click=={metadata.version('rich-click')}"] def test_build_creates_nested_output_dir(tmp_path): @@ -80,7 +324,7 @@ def test_build_creates_nested_output_dir(tmp_path): script_path.write_text("import typer\n") output_dir = tmp_path / "dist" / "bin" - with patch("cliffy.builder.pip.install"): + with skip_pip(): result = build_cli("script", str(script_path), output_dir=str(output_dir)) assert result.exit_code == 0, result.output diff --git a/tests/test_callbacks.py b/tests/test_callbacks.py new file mode 100644 index 0000000..caf09d8 --- /dev/null +++ b/tests/test_callbacks.py @@ -0,0 +1,195 @@ +import click +import pytest +from click.testing import CliRunner +from pydantic import ValidationError + +from cliffy.cli import cli +from cliffy.cli import test_command as manifest_test_command +from cliffy.commanders.click import ClickCommander +from cliffy.commanders.typer import TyperCommander +from cliffy.manifest import Callback, CLIManifest, CommandParam, RunBlock, SimpleCommandParam +from cliffy.tester import Tester as ManifestTester + +CALLBACKS_MANIFEST = """ +name: {name} +version: 1.2.3 +use_click: {use_click} + +callbacks: + - params: + - name: --verbose + type: bool + default: false + short: -v + help: Print what is about to run + - --profile: str = "dev" + invoke_without_command: true + run: | + ctx.obj = profile + if ctx.invoked_subcommand is None: + print("no command") + elif verbose: + print(f"running {{ctx.invoked_subcommand}} on {{profile}}") + - run: print("second root callback") + - group: db + run: print("connecting") + - group: db.tables + params: + - --schema: str = "public" + run: print(f"schema {{schema}}") + +commands: + hello | hi: print("hello") + db.migrate: print("migrating") + db.tables.list: print("tables") + +tests: + - hello: assert result.output == "second root callback\\nhello\\n" + - -v hello: assert result.output == "running hello on dev\\nsecond root callback\\nhello\\n" + - --verbose --profile prod hello: assert result.output.startswith("running hello on prod\\n") + - db migrate: assert result.output == "second root callback\\nconnecting\\nmigrating\\n" + - db tables --schema app list: assert result.output.endswith("connecting\\nschema app\\ntables\\n") + - "": assert result.output == "no command\\nsecond root callback\\n" + - --version: assert "1.2.3" in result.output and "callback" not in result.output + - --aliases: assert "hi" in result.output and "callback" not in result.output +""" + + +@pytest.fixture(params=[False, True], ids=["typer", "click"]) +def callbacks_manifest(request, tmp_path): + framework = "click" if request.param else "typer" + path = tmp_path / f"callbacks_{framework}.yaml" + path.write_text(CALLBACKS_MANIFEST.format(name=f"cb-{framework}", use_click=str(request.param).lower())) + return str(path) + + +def test_callbacks_run_before_commands_and_keep_builtin_options(callbacks_manifest): + tester = ManifestTester(callbacks_manifest) + assert tester.total_cases == 8 + + for case in tester.test_pipeline: + runs = tester.invoke_test(case.command, case.assert_script) # type: ignore[union-attr] + result = next(runs) + assert result.exception is None or isinstance(result.exception, SystemExit), result.output + assert result.exit_code == 0, result.output + # runs the assert of the manifest test + assert next(runs, None) is None + + +def test_callbacks_manifest_passes_cli_test(callbacks_manifest): + result = CliRunner().invoke(manifest_test_command, [callbacks_manifest]) + assert "All tests passed" in result.output + assert result.exit_code == 0 + + +def test_callbacks_example_passes_cli_test(): + result = CliRunner().invoke(manifest_test_command, ["examples/callbacks.yaml"]) + assert "All tests passed" in result.output + assert result.exit_code == 0 + + +def test_callback_params_are_options_of_the_cli(callbacks_manifest): + result = CliRunner().invoke(cli, ["run", callbacks_manifest, "--", "--help"]) + assert result.exit_code == 0 + # rich colours the help when it thinks it is on a terminal (it does on GitHub Actions) + output = click.unstyle(result.output) + for option in ("--verbose", "-v", "--profile", "--version", "--aliases", "Print what is about to run"): + assert option in output + + +def test_manifest_without_callbacks_generates_the_same_entry_point(): + manifest = CLIManifest(name="plain", version="1.0.0", commands={"a.b": "print(1)"}) + + typer_code = TyperCommander(manifest) + typer_code.generate_cli() + assert "@cli.callback()\ndef main(\n version: Optional[bool]" in typer_code.cli + assert "ctx" not in typer_code.cli + assert "a_app = typer.Typer()\n" in typer_code.cli + + click_code = ClickCommander(manifest) + click_code.generate_cli() + assert "def cli():\n" in click_code.cli + assert "@click.group()\ndef a_app():\n" in click_code.cli + assert "pass_context" not in click_code.cli + + +def test_typer_callback_codegen(): + manifest = CLIManifest( + name="cb", + version="1.0.0", + commands={"db.up": "print(1)"}, + callbacks=[ + Callback(run=RunBlock("print('root')"), params=[SimpleCommandParam({"--debug": "bool = False"})]), + Callback(group="db", run=RunBlock("$ echo db"), invoke_without_command=True), + ], + ) + commander = TyperCommander(manifest) + commander.generate_cli() + + assert ( + "@cli.callback()\ndef main(\n ctx: typer.Context,\n debug: bool = typer.Option(False)," in commander.cli + ) + assert "version: Optional[bool] = typer.Option(None, '--version'" in commander.cli + assert "@db_app.callback(invoke_without_command=True)\ndef db_app_callback(\n ctx: typer.Context,\n):" in ( + commander.cli + ) + # shell lines work in callbacks like in commands + assert "subprocess.run" in commander.cli + + +def test_click_callback_codegen(): + manifest = CLIManifest( + name="cb", + version="1.0.0", + use_click=True, + commands={"db.up": "print(1)"}, + callbacks=[ + Callback( + run=RunBlock("print('root')"), + params=[CommandParam(name="--debug", type="bool", short="-d", help="Debug output")], + invoke_without_command=True, + ), + Callback(group="db", run=RunBlock("print('db')")), + ], + ) + commander = ClickCommander(manifest) + commander.generate_cli() + + assert "@click.group(context_settings=CONTEXT_SETTINGS, invoke_without_command=True)" in commander.cli + assert "@click.version_option('1.0.0')" in commander.cli + assert ( + "@click.option('--debug/--no-debug', '-d', help='Debug output')\n@click.pass_context\ndef cli(ctx, debug):" + in (commander.cli) + ) + assert "@click.group()\n@click.pass_context\ndef db_app(ctx):" in commander.cli + + +def test_click_cli_options_take_precedence_over_callback_invoke_without_command(): + manifest = CLIManifest( + name="cb", + version="1.0.0", + use_click=True, + commands={"up": "print(1)"}, + cli_options={"invoke_without_command": False}, + callbacks=[Callback(run=RunBlock("print('root')"), invoke_without_command=True)], + ) + commander = ClickCommander(manifest) + commander.generate_cli() + + assert commander.cli.count("invoke_without_command") == 1 + assert "invoke_without_command=False" in commander.cli + + +def test_callback_group_must_exist(): + with pytest.raises(ValidationError, match=r"callback group `nope` does not match any command group \(groups: db\)"): + CLIManifest( + name="cb", + version="1.0.0", + commands={"db.up | u": "print(1)", "hello": "print(2)"}, + callbacks=[Callback(group="nope", run=RunBlock("pass"))], + ) + + +def test_callback_requires_run(): + with pytest.raises(ValidationError): + Callback() # type: ignore[call-arg]