lint_baseline.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/linter/lint_baseline.py
 
"""
Purpose
-------
FS-508 - lint baseline: snapshot current violations so subsequent runs
can report only new vs resolved issues.
 
Responsibilities
----------------
- Save a baseline from lint results to ``.option_c/cache/lint-baseline.json``.
- Load an existing baseline from disk.
- Diff current results against a baseline to find new and resolved violations.
- Pick the blocking ones out of the new violations; only they fail
  ``--against-baseline`` (operator ruling 2026-09-28).
 
Diagnostics
-----------
Domain: OCT-LINTER
Levels:
    L2 - (none) the save, load and diff are reported at L3
    L3 - baseline save/load/diff
    L4 - (none)
 
Contracts
---------
- Baseline file is JSON with ``_schema_version``, ``entries``, metadata.
- Corrupt or missing file returns ``None`` (no crash).
- Violations are keyed by ``(path, rule)`` for diff stability.
 
Dependencies
------------
- oct (__version__, recorded in the baseline)
- oct.core.diagnostics (_dbg structured logger)
- oct.linter.rule_registry (SEVERITY_BLOCKING)
"""
 
from __future__ import annotations
 
import json
from dataclasses import dataclass, field, asdict
from datetime import datetime
from pathlib import Path
 
from oct import __version__
from oct.core.diagnostics import _dbg
from oct.linter.rule_registry import SEVERITY_BLOCKING
 
 
_SCHEMA_VERSION = "1.0"
 
 
@dataclass
class BaselineEntry:
    """One baselined violation."""
 
    path: str
    rule: str
    message: str
 
 
@dataclass
class LintBaseline:
    """A full baseline snapshot."""
 
    entries: list[BaselineEntry]
    oct_version: str
    timestamp: str
    profile: str
 
 
def save_baseline(
    results: list[dict],
    baseline_path: Path,
    profile: str,
) -> int:
    """Extract violations from lint *results* and write to *baseline_path*.
 
    Returns the number of violations saved.
    """
    func = "save_baseline"
    _dbg("OCT-LINTER", func, f"saving baseline for profile {profile} to {baseline_path}", 3)
    entries: list[BaselineEntry] = []
    for r in results:
        detail = r.get("json_detail", {})
        rel_path = detail.get("path", r.get("rel", ""))
        checks = detail.get("checks", {})
        for rule_name, check in checks.items():
            if not check.get("pass", True):
                entries.append(BaselineEntry(
                    path=rel_path,
                    rule=rule_name,
                    message=check.get("message", ""),
                ))
 
    baseline = LintBaseline(
        entries=entries,
        oct_version=__version__,
        timestamp=datetime.now().isoformat(),
        profile=profile,
    )
 
    baseline_path.parent.mkdir(parents=True, exist_ok=True)
    baseline_path.write_text(
        json.dumps(
            {
                "_schema_version": _SCHEMA_VERSION,
                **asdict(baseline),
            },
            indent=2,
        ),
        encoding="utf-8",
    )
    return len(entries)
 
 
def load_baseline(baseline_path: Path) -> LintBaseline | None:
    """Load a baseline from disk. Returns ``None`` on missing/corrupt file."""
    func = "load_baseline"
    _dbg("OCT-LINTER", func, f"loading baseline from {baseline_path}", 3)
    if not baseline_path.is_file():
        return None
    try:
        raw = json.loads(baseline_path.read_text(encoding="utf-8"))
        if not isinstance(raw, dict):
            return None
        if raw.get("_schema_version") != _SCHEMA_VERSION:
            return None
        entries_raw = raw.get("entries", [])
        if not isinstance(entries_raw, list):
            return None
        entries = [
            BaselineEntry(path=e["path"], rule=e["rule"], message=e["message"])
            for e in entries_raw
            if isinstance(e, dict) and "path" in e and "rule" in e
        ]
        return LintBaseline(
            entries=entries,
            oct_version=raw.get("oct_version", ""),
            timestamp=raw.get("timestamp", ""),
            profile=raw.get("profile", ""),
        )
    except (json.JSONDecodeError, OSError, KeyError):
        return None
 
 
def diff_against_baseline(
    results: list[dict],
    baseline: LintBaseline,
) -> tuple[list[tuple[str, str, str]], list[tuple[str, str, str]]]:
    """Compare current *results* against *baseline*.
 
    Returns ``(new_violations, resolved_violations)`` where each item is
    ``(path, rule, message)``.
 
    A violation is identified by its ``(path, rule)`` pair. If a
    ``(path, rule)`` appears in current results but not in the baseline,
    it is "new". If it appears in the baseline but not in current results,
    it is "resolved".
    """
    func = "diff_against_baseline"
    _dbg("OCT-LINTER", func, "diffing results against the baseline", 3)
    baseline_set: set[tuple[str, str]] = set()
    baseline_msgs: dict[tuple[str, str], str] = {}
    for entry in baseline.entries:
        key = (entry.path, entry.rule)
        baseline_set.add(key)
        baseline_msgs[key] = entry.message
 
    current_set: set[tuple[str, str]] = set()
    current_msgs: dict[tuple[str, str], str] = {}
    for r in results:
        detail = r.get("json_detail", {})
        rel_path = detail.get("path", r.get("rel", ""))
        checks = detail.get("checks", {})
        for rule_name, check in checks.items():
            if not check.get("pass", True):
                key = (rel_path, rule_name)
                current_set.add(key)
                current_msgs[key] = check.get("message", "")
 
    new_keys = current_set - baseline_set
    resolved_keys = baseline_set - current_set
 
    new_violations = sorted(
        [(p, r, current_msgs[(p, r)]) for p, r in new_keys],
    )
    resolved_violations = sorted(
        [(p, r, baseline_msgs[(p, r)]) for p, r in resolved_keys],
    )
    return new_violations, resolved_violations
 
 
def blocking_violations(
    violations: list[tuple[str, str, str]],
    results: list[dict],
) -> list[tuple[str, str, str]]:
    """The *violations* whose check is blocking in *results*.
 
    ``oct lint --against-baseline`` exits 1 only for these (operator ruling
    2026-09-28); a new advisory failure is listed but does not fail the
    command. A check recording no severity counts as blocking, as the
    compliance rule (FS-C4) treats it.
    """
    severity: dict[tuple[str, str], str] = {}
    for r in results:
        detail = r.get("json_detail", {})
        rel_path = detail.get("path", r.get("rel", ""))
        for rule_name, check in detail.get("checks", {}).items():
            severity[(rel_path, rule_name)] = check.get("severity", SEVERITY_BLOCKING)
    return [
        v for v in violations
        if severity.get((v[0], v[1]), SEVERITY_BLOCKING) == SEVERITY_BLOCKING
    ]