thresholds.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/complexity/thresholds.py
 
"""
Purpose
-------
Turn measurements into a gate-callable verdict with a real exit code.
 
Responsibilities
----------------
- Resolve thresholds from explicit flags over an ``.octrc.json`` section.
- Apply the don't-get-worse ratchet against a recorded baseline.
- Name every breaching function as ``relpath:line qualname CC=n [CRAP=m]``.
- Carry the verdict as both machine fields and compact ASCII lines.
 
Diagnostics
-----------
Domain: OCT-COMPLEXITY
Levels:
    L2 - verdict lifecycle
    L3 - threshold and breach details
    L4 - deep tracing
 
Contracts
---------
- No active threshold means no verdict, so today's output is unchanged.
- Exit 0 is clean, exit 2 is at least one breach, exit 3 is a named refusal.
- A malformed threshold is refused by name and never silently ignored.
- The module imports only the Python standard library and sibling measurement
  modules, so it stays callable from a zero-dependency context.
 
Dependencies
------------
- oct.complexity.service (result rows and relative-path display)
- oct.complexity.baseline (lazy; BaselineEntry type hints for the ratchet comparison)
"""
 
# OCT-LINT: disable-file=dbg_usage reason="The verdict layer emits only its explicit report and must not add diagnostics noise to JSON or ASCII output"
# OCT-LINT: disable-file=dbg_import reason="Diagnostics would contaminate explicit JSON and ASCII verdict output"
 
from __future__ import annotations
 
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Mapping, Sequence
 
from oct.complexity.service import (
    FunctionResult,
    function_identity,
    relative_display_path,
)
 
if TYPE_CHECKING:  # pragma: no cover - typing only, avoids an import cycle
    from oct.complexity.baseline import BaselineEntry
 
 
#: Advisory starting values quoted in the CLI help text. They are conventions
#: for operators to adopt deliberately, never defaults applied on their behalf.
ADVISORY_MAX_CC = 15
ADVISORY_MAX_CRAP = 30
 
EXIT_CLEAN = 0
EXIT_BREACH = 2
EXIT_REFUSAL = 3
 
SOURCE_FLAG = "flag"
SOURCE_CONFIG = "octrc"
 
CONFIG_SECTION = "complexity"
 
 
class GateRefusalError(ValueError):
    """Raised when a gate-mode invocation cannot be honestly evaluated."""
 
 
@dataclass(frozen=True, slots=True)
class Thresholds:
    """The effective CC and CRAP limits plus where each one came from."""
 
    max_cc: int | None = None
    max_crap: float | None = None
    max_cc_source: str | None = None
    max_crap_source: str | None = None
 
    @property
    def active(self) -> bool:
        """Return whether any threshold is in force for this invocation."""
        return self.max_cc is not None or self.max_crap is not None
 
    def summary(self) -> str:
        """Return a compact ASCII description of the active limits."""
        parts = []
        if self.max_cc is not None:
            parts.append(f"max CC {self.max_cc}")
        if self.max_crap is not None:
            parts.append(f"max CRAP {_number(self.max_crap)}")
        return ", ".join(parts) if parts else "no thresholds"
 
 
@dataclass(frozen=True, slots=True)
class Breach:
    """One function that fails the active thresholds, with its numbers."""
 
    file: str
    lineno: int
    qualname: str
    complexity: int
    crap: float | None
    metrics: tuple[str, ...]
    reason: str
    baseline_complexity: int | None = None
    baseline_crap: float | None = None
 
    def to_dict(self) -> dict[str, object]:
        """Return a JSON-ready breach record."""
        return {
            "file": self.file,
            "lineno": self.lineno,
            "qualname": self.qualname,
            "complexity": self.complexity,
            "crap": self.crap,
            "metrics": list(self.metrics),
            "reason": self.reason,
            "baseline_complexity": self.baseline_complexity,
            "baseline_crap": self.baseline_crap,
        }
 
    def format_line(self) -> str:
        """Return the pinned ``relpath:line qualname CC=n [CRAP=m]`` line."""
        line = f"{self.file}:{self.lineno} {self.qualname} CC={self.complexity}"
        if self.crap is not None:
            line = f"{line} CRAP={self.crap:.2f}"
        return f"{line} ({self.reason})"
 
 
@dataclass(frozen=True, slots=True)
class Verdict:
    """Everything a gate or an agent needs to branch on one measurement run."""
 
    status: str
    exit_code: int
    thresholds: Thresholds
    breaches: tuple[Breach, ...]
    baseline: str | None = None
    baseline_functions: int | None = None
    baseline_written: str | None = None
    baseline_written_functions: int | None = None
 
    def to_dict(self) -> dict[str, object]:
        """Return the machine-readable verdict envelope."""
        return {
            "status": self.status,
            "exit_code": self.exit_code,
            "max_cc": self.thresholds.max_cc,
            "max_crap": self.thresholds.max_crap,
            "max_cc_source": self.thresholds.max_cc_source,
            "max_crap_source": self.thresholds.max_crap_source,
            "baseline": self.baseline,
            "baseline_functions": self.baseline_functions,
            "baseline_written": self.baseline_written,
            "baseline_written_functions": self.baseline_written_functions,
            "breach_count": len(self.breaches),
            "breaches": [breach.to_dict() for breach in self.breaches],
        }
 
 
def resolve_thresholds(
    flag_max_cc: int | None,
    flag_max_crap: float | None,
    config: dict | None,
) -> Thresholds:
    """Resolve the effective thresholds, with flags overriding the config.
 
    ``config`` is the whole ``.octrc.json`` document; only its optional
    ``complexity`` object is read. A malformed value raises
    :class:`GateRefusalError` rather than degrading to "no threshold",
    because a silently inactive gate reads exactly like a passing one.
    """
    section = config.get(CONFIG_SECTION) if isinstance(config, dict) else None
    if section is not None and not isinstance(section, dict):
        raise GateRefusalError(
            f"octrc '{CONFIG_SECTION}' must be a JSON object, got "
            f"{type(section).__name__}"
        )
    settings = section or {}
    config_cc = _coerce_limit(settings.get("max_cc"), "max_cc", as_int=True)
    config_crap = _coerce_limit(settings.get("max_crap"), "max_crap", as_int=False)
    max_cc = flag_max_cc if flag_max_cc is not None else config_cc
    max_crap = flag_max_crap if flag_max_crap is not None else config_crap
    return Thresholds(
        max_cc=max_cc,
        max_crap=max_crap,
        max_cc_source=_source(flag_max_cc, config_cc),
        max_crap_source=_source(flag_max_crap, config_crap),
    )
 
 
def _source(flag_value: object, config_value: object) -> str | None:
    """Return which input supplied the effective limit, if any did."""
    if flag_value is not None:
        return SOURCE_FLAG
    if config_value is not None:
        return SOURCE_CONFIG
    return None
 
 
def _coerce_limit(
    value: object,
    key: str,
    as_int: bool,
) -> int | float | None:
    """Validate one configured limit or refuse it by name."""
    if value is None:
        return None
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise GateRefusalError(
            f"octrc '{CONFIG_SECTION}.{key}' must be a positive number, got "
            f"{type(value).__name__}"
        )
    if as_int:
        if not isinstance(value, int) or value < 1:
            raise GateRefusalError(
                f"octrc '{CONFIG_SECTION}.{key}' must be an integer of at "
                f"least 1, got {value!r}"
            )
        return value
    if value <= 0:
        raise GateRefusalError(
            f"octrc '{CONFIG_SECTION}.{key}' must be greater than 0, got {value!r}"
        )
    return float(value)
 
 
def evaluate(
    results: Sequence[FunctionResult],
    thresholds: Thresholds,
    root: str | Path,
    baseline: Mapping[str, "BaselineEntry"] | None = None,
) -> tuple[Breach, ...]:
    """Return every function that fails the active thresholds, worst first.
 
    Without a baseline, failing means exceeding a threshold. With one, the
    ratchet applies: a function the baseline never saw fails on the threshold
    alone, and a recorded function fails only when it is BOTH over the
    threshold and strictly worse than its recorded value. Existing debt held
    at or below its recorded value never fails - this is hygiene, not
    rigidity.
 
    ``results`` arrives already ordered worst-first by the service, so the
    breach list names the most severe function first.
    """
    if not thresholds.active:
        return ()
    breaches = []
    for item in results:
        metrics = _breached_metrics(item, thresholds)
        if not metrics:
            continue
        recorded = None
        if baseline is not None:
            recorded = baseline.get(
                function_identity(item.file, item.qualname, root)
            )
            if recorded is not None:
                metrics = _worsened_metrics(item, metrics, recorded)
                if not metrics:
                    continue
        breaches.append(
            Breach(
                file=relative_display_path(item.file, root),
                lineno=item.lineno,
                qualname=item.qualname,
                complexity=item.complexity,
                crap=item.crap,
                metrics=metrics,
                reason=_reason(baseline, recorded, metrics),
                baseline_complexity=None if recorded is None else recorded.complexity,
                baseline_crap=None if recorded is None else recorded.crap,
            )
        )
    return tuple(breaches)
 
 
def _worsened_metrics(
    item: FunctionResult,
    metrics: tuple[str, ...],
    recorded: "BaselineEntry",
) -> tuple[str, ...]:
    """Keep only the over-threshold metrics that are worse than recorded.
 
    A recorded ``crap`` of ``None`` means the baseline was written without
    coverage data, so there is no accepted value to sit at or below. That
    fails closed: an unrecorded metric is never grandfathered.
    """
    kept = []
    for metric in metrics:
        if metric == "cc" and item.complexity > recorded.complexity:
            kept.append(metric)
        elif metric == "crap" and (
            recorded.crap is None
            or (item.crap is not None and item.crap > recorded.crap)
        ):
            kept.append(metric)
    return tuple(kept)
 
 
def _reason(
    baseline: Mapping[str, "BaselineEntry"] | None,
    recorded: "BaselineEntry | None",
    metrics: tuple[str, ...],
) -> str:
    """Return the human-readable cause of one breach."""
    if baseline is None:
        return "over-threshold"
    if recorded is None:
        return "not in baseline"
    parts = []
    if "cc" in metrics:
        parts.append(f"worse than baseline CC={recorded.complexity}")
    if "crap" in metrics:
        parts.append(
            "no baseline CRAP recorded"
            if recorded.crap is None
            else f"worse than baseline CRAP={recorded.crap:.2f}"
        )
    return "; ".join(parts)
 
 
def _breached_metrics(
    item: FunctionResult,
    thresholds: Thresholds,
) -> tuple[str, ...]:
    """Return which active metrics this one function exceeds."""
    metrics = []
    if thresholds.max_cc is not None and item.complexity > thresholds.max_cc:
        metrics.append("cc")
    if (
        thresholds.max_crap is not None
        and item.crap is not None
        and item.crap > thresholds.max_crap
    ):
        metrics.append("crap")
    return tuple(metrics)
 
 
def build_verdict(
    thresholds: Thresholds,
    breaches: Sequence[Breach],
    baseline: str | None = None,
    baseline_functions: int | None = None,
    baseline_written: str | None = None,
    baseline_written_functions: int | None = None,
) -> Verdict:
    """Return the verdict for one gate-mode invocation."""
    breached = bool(breaches)
    return Verdict(
        status="breached" if breached else "clean",
        exit_code=EXIT_BREACH if breached else EXIT_CLEAN,
        thresholds=thresholds,
        breaches=tuple(breaches),
        baseline=baseline,
        baseline_functions=baseline_functions,
        baseline_written=baseline_written,
        baseline_written_functions=baseline_written_functions,
    )
 
 
def render_verdict(verdict: Verdict) -> str:
    """Render the verdict as compact ASCII lines for terminal users."""
    lines = []
    if verdict.baseline_written is not None:
        lines.append(
            f"Baseline written: {verdict.baseline_written} "
            f"({verdict.baseline_written_functions} function(s))."
        )
    if verdict.baseline is not None:
        lines.append(
            f"Baseline: {verdict.baseline} "
            f"({verdict.baseline_functions} function(s) recorded)."
        )
    lines.extend(_render_outcome(verdict))
    return "\n".join(lines)
 
 
def _render_outcome(verdict: Verdict) -> list[str]:
    """Return the verdict's outcome lines, breaches included."""
    if not verdict.thresholds.active:
        return ["No thresholds active; measurement only."]
    limits = verdict.thresholds.summary()
    if not verdict.breaches:
        return [f"No threshold breaches ({limits})."]
    lines = [f"Threshold breaches ({limits}):"]
    lines.extend(f"  {breach.format_line()}" for breach in verdict.breaches)
    lines.append(
        f"{len(verdict.breaches)} function(s) over the active thresholds."
    )
    return lines
 
 
def _number(value: float) -> str:
    """Render a limit without a trailing ``.0`` on whole numbers."""
    return str(int(value)) if float(value).is_integer() else str(value)