command.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/complexity/command.py
 
"""
Purpose
-------
Expose the on-demand complexity instrument through Click.
 
Responsibilities
----------------
- Parse paths, coverage data, JSON mode, and human truncation options.
- Parse the plan-226 gate options: thresholds and their exit-code contract.
- Scope a call to the git-modified file set under ``--changed``.
- Record and compare the don't-get-worse baseline.
- Default analysis to the discovered Option C project root.
- Convert dependency and path failures into clear CLI errors.
 
Diagnostics
-----------
Domain: OCT-COMPLEXITY
Levels:
    L2 - command lifecycle
    L3 - option and output details
    L4 - deep tracing
 
Contracts
---------
- ``--top`` affects human output only.
- With no threshold active the command is informational and exits 0, byte for
  byte as it did before plan 226.
- With a threshold active the command exits 0 when clean, 2 when any function
  breaches, and 3 when the call cannot be honestly evaluated.
- Machine-readable stdout contains JSON only.
 
Dependencies
------------
- click (CLI adapter)
- oct.complexity.baseline (ratchet document I/O)
- oct.complexity.service (analysis and rendering)
- oct.complexity.thresholds (verdict and exit-code contract)
- oct.core.octrc (optional complexity configuration section)
- oct.core.project_root (default target discovery)
- oct.complexity.coverage (missing coverage.py error, raised as a CLI error)
- oct.core.git (lazy; git-modified Python files for --changed)
"""
 
# OCT-LINT: disable-file=dbg_usage reason="Diagnostics would contaminate the command's JSON-only stdout contract"
# OCT-LINT: disable-file=dbg_import reason="Diagnostics would contaminate the command's JSON-only stdout contract"
 
from __future__ import annotations
 
import json
from pathlib import Path
 
import click
 
from oct.complexity.baseline import load_baseline, write_baseline
from oct.complexity.coverage import CoverageUnavailableError
from oct.complexity.service import (
    AnalysisReport,
    analyze_paths,
    is_excluded_path,
    render_human,
)
from oct.complexity.thresholds import (
    ADVISORY_MAX_CC,
    ADVISORY_MAX_CRAP,
    EXIT_REFUSAL,
    GateRefusalError,
    Thresholds,
    Verdict,
    build_verdict,
    evaluate,
    render_verdict,
    resolve_thresholds,
)
from oct.core.octrc import load_octrc_safe
from oct.core.project_root import get_project_root
 
 
class GateRefusal(click.ClickException):
    """A named refusal: the call could not be evaluated, so nothing is judged."""
 
    exit_code = EXIT_REFUSAL
 
 
@click.command(name="complexity")
@click.argument(
    "paths",
    nargs=-1,
    type=click.Path(exists=True, path_type=Path),
)
@click.option(
    "--coverage-data",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
    default=None,
    help="Existing coverage.py data file to map offline.",
)
@click.option(
    "--json",
    "json_mode",
    is_flag=True,
    help="Emit the full machine-readable report.",
)
@click.option(
    "--top",
    type=click.IntRange(min=1),
    default=None,
    help="Limit human output to the worst N functions.",
)
@click.option(
    "--max-cc",
    type=click.IntRange(min=1),
    default=None,
    help=(
        "Fail with exit 2 when a function exceeds this cyclomatic complexity "
        f"(advisory convention: {ADVISORY_MAX_CC})."
    ),
)
@click.option(
    "--max-crap",
    type=click.FloatRange(min=0.0, min_open=True),
    default=None,
    help=(
        "Fail with exit 2 when a function exceeds this CRAP score (advisory "
        f"convention: {ADVISORY_MAX_CRAP}); requires --coverage-data."
    ),
)
@click.option(
    "--changed",
    is_flag=True,
    help="Measure only git-modified Python files (as `oct lint --changed`).",
)
@click.option(
    "--baseline",
    "baseline_path",
    type=click.Path(dir_okay=False, path_type=Path),
    default=None,
    help=(
        "Compare against a recorded baseline: debt at or below its recorded "
        "value passes, anything new or worsened fails. Needs a threshold."
    ),
)
@click.option(
    "--write-baseline",
    "write_baseline_path",
    type=click.Path(dir_okay=False, path_type=Path),
    default=None,
    help="Record this invocation's measurements as a new baseline file.",
)
@click.pass_context
def complexity(
    ctx: click.Context,
    paths: tuple[Path, ...],
    coverage_data: Path | None,
    json_mode: bool,
    top: int | None,
    max_cc: int | None,
    max_crap: float | None,
    changed: bool,
    baseline_path: Path | None,
    write_baseline_path: Path | None,
) -> None:
    """Measure per-function cyclomatic complexity and optional CRAP scores."""
    root, root_error = _resolve_root(ctx)
    thresholds = _resolve_gate_thresholds(root, max_cc, max_crap, coverage_data)
    _check_baseline_options(thresholds, baseline_path, write_baseline_path)
    if changed:
        targets = _changed_targets(root, paths, json_mode)
    else:
        targets = _resolve_targets(paths, root, root_error)
    try:
        report = analyze_paths(targets, coverage_data=coverage_data)
    except (CoverageUnavailableError, FileNotFoundError) as exc:
        raise click.ClickException(str(exc)) from exc
    if not (thresholds.active or baseline_path or write_baseline_path):
        _emit_report(report, json_mode, top)
        return
    verdict = _build_gate_verdict(
        report,
        thresholds,
        root,
        baseline_path,
        write_baseline_path,
    )
    _emit_report(report, json_mode, top, verdict=verdict)
    if verdict.exit_code:
        ctx.exit(verdict.exit_code)
 
 
def _check_baseline_options(
    thresholds: Thresholds,
    baseline_path: Path | None,
    write_baseline_path: Path | None,
) -> None:
    """Refuse baseline combinations that could not produce an honest verdict."""
    if baseline_path is not None and write_baseline_path is not None:
        raise GateRefusal(
            "--baseline and --write-baseline are mutually exclusive: recording "
            "and ratcheting in one call would judge against a baseline that "
            "the same call is replacing"
        )
    if baseline_path is not None and not thresholds.active:
        raise GateRefusal(
            "--baseline requires an active threshold (--max-cc / --max-crap, "
            "or the octrc 'complexity' section): with none, the ratchet has "
            "nothing to compare and would pass silently"
        )
 
 
def _build_gate_verdict(
    report: AnalysisReport,
    thresholds: Thresholds,
    root: Path,
    baseline_path: Path | None,
    write_baseline_path: Path | None,
) -> Verdict:
    """Load or write a baseline as asked, then judge the measurements."""
    try:
        recorded = load_baseline(baseline_path) if baseline_path else None
        written = (
            write_baseline(write_baseline_path, report.functions, root)
            if write_baseline_path
            else None
        )
    except GateRefusalError as exc:
        raise GateRefusal(str(exc)) from exc
    return build_verdict(
        thresholds,
        evaluate(report.functions, thresholds, root, baseline=recorded),
        baseline=None if baseline_path is None else str(baseline_path),
        baseline_functions=None if recorded is None else len(recorded),
        baseline_written=(
            None if write_baseline_path is None else str(write_baseline_path)
        ),
        baseline_written_functions=written,
    )
 
 
def _resolve_root(ctx: click.Context) -> tuple[Path, click.ClickException | None]:
    """Return the project root, tolerating undiscoverable roots.
 
    Configuration lookup and relative-path display need a root even when the
    caller measures an explicit path outside any project, so a failed
    discovery falls back to the working directory and the original error is
    handed back for the one caller that must still fail on it. An explicit
    ``--root-dir`` is never softened: that error is the operator's own.
    """
    try:
        return get_project_root(ctx), None
    except click.ClickException as exc:
        if ctx.obj and ctx.obj.get("root_dir"):
            raise
        return Path.cwd().resolve(), exc
 
 
def _resolve_gate_thresholds(
    root: Path,
    max_cc: int | None,
    max_crap: float | None,
    coverage_data: Path | None,
) -> Thresholds:
    """Resolve thresholds and refuse a CRAP limit that cannot be measured."""
    try:
        thresholds = resolve_thresholds(max_cc, max_crap, load_octrc_safe(root))
    except GateRefusalError as exc:
        raise GateRefusal(str(exc)) from exc
    if thresholds.max_crap is not None and coverage_data is None:
        raise GateRefusal(
            "--max-crap requires --coverage-data: CRAP cannot be computed "
            "without coverage measurements, and a CC-only downgrade would "
            f"pass silently (--max-crap came from the {thresholds.max_crap_source})"
        )
    return thresholds
 
 
def _changed_targets(
    root: Path,
    paths: tuple[Path, ...],
    json_mode: bool,
) -> tuple[Path, ...]:
    """Return the git-modified Python files, scoped exactly as the linter is.
 
    Discovery is :func:`oct.core.git.git_changed_files`, the same helper
    ``oct lint --changed`` uses, and the result passes through the service's
    own exclusion set so ``--changed`` narrows the measured universe without
    widening it. A git failure is refused rather than collapsed to an empty
    list: an empty measurement reads as a pass, and a gate that passes
    because git broke is worse than one that stops.
    """
    from oct.core import git as core_git
 
    if paths and not json_mode:
        click.echo("Note: --changed overrides explicit PATH targets.", err=True)
    try:
        discovered = core_git.git_changed_files(root, extensions={".py"})
    except core_git.GitError as exc:
        raise GateRefusal(
            f"--changed could not determine the modified file set: {exc}"
        ) from exc
    return tuple(
        path
        for path in discovered
        if path.is_file() and not is_excluded_path(path, root)
    )
 
 
def _resolve_targets(
    paths: tuple[Path, ...],
    root: Path,
    root_error: click.ClickException | None,
) -> tuple[Path, ...]:
    """Return the paths to measure, defaulting to the project root."""
    if paths:
        return paths
    if root_error is not None:
        raise root_error
    return (root,)
 
 
def _emit_report(
    report: AnalysisReport,
    json_mode: bool,
    top: int | None,
    verdict: Verdict | None = None,
) -> None:
    """Write the report, adding the verdict only when a gate mode is active."""
    if json_mode:
        payload = report.to_dict()
        if verdict is not None:
            payload["verdict"] = verdict.to_dict()
        click.echo(json.dumps(payload, indent=2, sort_keys=True))
        return
    text = render_human(report, top=top)
    if verdict is not None:
        text = f"{text}\n{render_verdict(verdict)}"
    click.echo(text)