dependency_check.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/deps/dependency_check.py
 
"""
Purpose
-------
Compare a module's ``Dependencies`` docstring section with what the module
actually imports. The advisory lint rule ``dependencies_match`` runs this
on every source module. The section parsing lives in
:mod:`oct.deps.oct_deps`, whose cycle detection leaves lazy and
not-imported entries out, so the rule and ``oct deps`` read an entry alike.
 
The operator ruled on 2026-09-28 (docs/CHART_C_TRACE_FOLLOWUPS_2026-09-28.md,
section 8, in the workspace):
 
1. Every import is declared. An import made only inside a function or
   under ``TYPE_CHECKING`` is declared with a purpose starting ``lazy``,
   and ``oct deps`` leaves lazy entries out of its cycle detection.
2. First-party packages (any package in the workspace) and third-party
   packages are both checked; standard-library modules are optional
   (declaring one is allowed; it is stale only if nothing of it is
   imported).
3. The rule runs at every profile, is advisory, and exempts test modules.
 
One addition, for the operator to veto: a dependency used without an
import (a CLI, a ``python -m`` subprocess, a pytest plugin, a module
loaded by name) is declared with a purpose starting ``not imported``; it
is never reported stale and never enters cycle detection.
 
Responsibilities
----------------
- :func:`parse_entries` - the ``Dependencies`` entries of a docstring, with
  their markers, and the lines that are not entries.
- :func:`collect_imports` - every dependency a module imports, each marked
  eager or lazy, standard library kept apart.
- :func:`check_dependencies_match` - the comparison: missing, stale,
  wrongly marked and unrecognised entries, as one verdict and message.
 
Diagnostics
-----------
Domain: OCT-DEPS
Levels:
    L2 - (none) called once per file by the linter; the linter logs verdicts
    L3 - workspace roots resolved for a project
    L4 - per-file comparison totals
 
Contracts
---------
- Never raises: any internal error becomes a passing verdict whose message
  says the check did not run. A raising rule would make the quality gate
  skip every rule for the file, the secret check included.
- A module's dotted name is taken from the workspace subproject root whose
  first path component is a package, so a file in a namespace directory of
  a package (``oct/oct/core``) is ``oct.core.git``; outside any package,
  from Python's own rule.
- First-party names are checked against the filesystem with exact case
  (directory listings memoised for one lint run), never by walking the
  project, so ``oct git status`` stays fast and Windows and Linux agree.
- An import is recorded as written: ``import a.b.c`` is ``a.b.c``; ``from X
  import y`` is ``X.y`` when that is a module, else ``X``. Nothing is
  shortened to whatever prefix exists on disk, so an absent generated
  module (a setuptools-scm version file) does not change the verdict.
- A first-party entry must name a module exactly, or exactly what an
  import names; an attribute (``pkg.mod.Class``) is reported as naming no
  module. A third-party or standard-library entry may name the package,
  which covers its submodules.
- An ``if TYPE_CHECKING:`` body is lazy and its ``else`` eager; ``if not
  TYPE_CHECKING:`` is the reverse; any other test is eager.
- An ``oc_diagnostics`` entry is satisfied by an import of oct's shim
  ``oct.core.diagnostics`` (Core spec: oc_diagnostics "should always be
  listed if the module uses _dbg").
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
- oct.core.workspace (lazy; the workspace manifest's subproject roots)
- oct.deps.import_graph (_is_test_path, _resolve_relative - shared rules)
- oct.deps.oct_deps (module_name_for, dependency_items, entry_marker - the shared parsing)
"""
 
from __future__ import annotations
 
import ast
import os
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
 
from oct.core.diagnostics import _dbg
from oct.deps.import_graph import _is_test_path, _resolve_relative
from oct.deps.oct_deps import dependency_items, entry_marker, module_name_for
 
#: ``name (purpose)``; the name may be wrapped in backticks.
_ENTRY_RE = re.compile(r"^`?([A-Za-z_][\w.]*)`?\s*\‍((.*)\‍)\s*$")
 
STDLIB: frozenset[str] = frozenset(sys.stdlib_module_names) | {"__future__"}
 
#: The shim through which oct modules reach oc_diagnostics.
_SHIM = "oct.core.diagnostics"
 
#: Directories that never hold an importable namespace portion.
_NOT_NAMESPACE = frozenset({"tests", "test", "docs", "__pycache__", "node_modules",
                            ".venv", "build", "dist"})
 
 
@dataclass(frozen=True)
class Entry:
    """One ``- name (purpose)`` line of a Dependencies section."""
 
    name: str
    purpose: str
    lazy: bool
    not_imported: bool
 
 
@dataclass
class Findings:
    """What :func:`check_dependencies_match` found for one module."""
 
    missing: list[str] = field(default_factory=list)
    missing_lazy: list[str] = field(default_factory=list)
    unmarked_lazy: list[str] = field(default_factory=list)
    marked_lazy_but_eager: list[str] = field(default_factory=list)
    marked_not_imported_but_imported: list[str] = field(default_factory=list)
    stale: list[str] = field(default_factory=list)
    unrecognised: list[str] = field(default_factory=list)
    unresolvable: list[str] = field(default_factory=list)
 
    @property
    def ok(self) -> bool:
        return not any(vars(self).values())
 
    def message(self) -> str:
        parts = []
        for label, items in (
            ("missing", self.missing),
            ("missing, declare as lazy", self.missing_lazy),
            ("imported only lazily, mark the entry 'lazy'", self.unmarked_lazy),
            ("marked lazy but imported at top level", self.marked_lazy_but_eager),
            ("marked 'not imported' but imported", self.marked_not_imported_but_imported),
            ("stale (not imported)", self.stale),
            ("names no module", self.unresolvable),
            ("not a '- name (purpose)' entry", self.unrecognised),
        ):
            if items:
                parts.append(f"{label}: {', '.join(items)}")
        return "; ".join(parts) or "OK"
 
 
def parse_entries(docstring: str) -> tuple[list[Entry], list[str], bool]:
    """Return ``(entries, unrecognised lines, section present)`` for *docstring*.
 
    The items come from :func:`oct.deps.oct_deps.dependency_items`: a bullet
    whose text continues on the next line is joined first, so a wrapped
    purpose still parses and its marker is read from the joined text, as
    ``oct deps`` reads it.
    """
    items = dependency_items(docstring)
    if items is None:
        return [], [], bool(re.search(r"Dependencies\s*\n\s*-+", docstring or ""))
    entries, unrecognised = [], []
    for item in items:
        e = _ENTRY_RE.match(item.text)
        if not e:
            text = item.text
            unrecognised.append(text if len(text) <= 60 else text[:57] + "...")
            continue
        marker = entry_marker(e.group(2))
        entries.append(Entry(e.group(1), e.group(2), marker == "lazy", marker == "not imported"))
    return entries, unrecognised, True
 
 
# -- the filesystem ------------------------------------------------------------------------
 
 
class _Probe:
    """Exact-case filesystem probes, memoised in a dict the caller owns.
 
    The linter passes ``LinterContext.dependency_memo``, which lives exactly
    as long as one lint run, so a long-lived process (the MCP server) sees
    files created between runs, and nothing needs resetting. The memo is
    shared by the linter's worker threads; a race only repeats a probe.
    """
 
    def __init__(self, memo: dict) -> None:
        self.memo = memo
        self._listings: dict[Path, frozenset[str]] = memo.setdefault("listings", {})
        self._modules: dict[tuple[Path, str, bool], bool] = memo.setdefault("modules", {})
 
    def listing(self, directory: Path) -> frozenset[str]:
        got = self._listings.get(directory)
        if got is None:
            try:
                got = frozenset(os.listdir(directory))
            except OSError:
                got = frozenset()
            self._listings[directory] = got
        return got
 
    def is_dir(self, parent: Path, name: str) -> bool:
        return name in self.listing(parent) and (parent / name).is_dir()
 
    def is_package(self, directory: Path) -> bool:
        return "__init__.py" in self.listing(directory)
 
    def module_exists(self, root: Path, dotted: str, namespace_top: bool = False) -> bool:
        """Whether *dotted* is a module under *root*, with exact case.
 
        A directory counts as a (namespace) package below the top level; at
        the top level only with an ``__init__.py``, unless *namespace_top*.
        """
        key = (root, dotted, namespace_top)
        got = self._modules.get(key)
        if got is None:
            got = self._module_exists(root, dotted, namespace_top)
            self._modules[key] = got
        return got
 
    def _module_exists(self, root: Path, dotted: str, namespace_top: bool) -> bool:
        parts = dotted.split(".")
        d = root
        for part in parts[:-1]:
            if not self.is_dir(d, part):
                return False
            d = d / part
        leaf = parts[-1]
        if leaf + ".py" in self.listing(d):
            return True
        if self.is_dir(d, leaf):
            return self.is_package(d / leaf) or len(parts) > 1 or namespace_top
        return False
 
 
def _workspace_roots(project_root: Path, probe: _Probe) -> tuple[Path, ...]:
    """Every subproject root of the workspace holding *project_root*, then it."""
    func = "_workspace_roots"
    cached = probe.memo.setdefault("roots", {}).get(project_root)
    if cached is not None:
        return cached
    roots: list[Path] = [project_root]
    try:
        from oct.core.workspace import find_workspace_root, load_workspace
        ws_root = find_workspace_root(project_root)
        if ws_root is not None:
            roots = [sp.path.resolve() for sp in load_workspace(ws_root).subprojects] + roots
    except Exception as exc:  # a broken manifest must not stop the lint
        _dbg("OCT-DEPS", func, f"workspace roots unavailable ({exc!r}); project only", 3)
    ordered = tuple(sorted(dict.fromkeys(roots), key=lambda r: len(r.parts), reverse=True))
    _dbg("OCT-DEPS", func, f"{project_root}: {len(ordered)} roots", 3)
    probe.memo["roots"][project_root] = ordered
    return ordered
 
 
def _naming_root(py_file: Path, roots: tuple[Path, ...], probe: _Probe) -> tuple[Path, bool]:
    """The directory *py_file*'s dotted name is relative to, and whether it is packaged.
 
    Preferred: the most specific workspace root under which the file's first
    path component is a package. Otherwise Python's rule from the file's
    own directory; ``False`` then means a script outside any package.
    """
    for r in roots:
        try:
            rel = py_file.relative_to(r)
        except ValueError:
            continue
        if len(rel.parts) > 1 and probe.is_package(r / rel.parts[0]):
            return r, True
    d = py_file.parent
    if not probe.is_package(d):
        return d, False
    while d.parent != d and probe.is_package(d.parent):
        d = d.parent
    return d.parent, True
 
 
class _Resolver:
    """First-party names: workspace packages, plus a script's local modules.
 
    *local_roots* are the directories a script outside any package can
    import from: its own directory (``sys.path[0]``) and, as the runners
    in this workspace arrange, the one above. Their modules and their
    namespace directories (holding ``.py`` files, not tests or docs) are
    first-party.
    """
 
    def __init__(self, roots: tuple[Path, ...], local_roots: tuple[Path, ...],
                 probe: _Probe) -> None:
        self.roots = roots
        self.local_roots = local_roots
        self.probe = probe
        self.tops: set[str] = set()
        for r in roots:
            self.tops.update(n for n in probe.listing(r) if probe.is_package(r / n))
        for r in local_roots:
            for n in probe.listing(r):
                if n.endswith(".py"):
                    self.tops.add(n[:-3])
                elif (n not in _NOT_NAMESPACE and not n.startswith(".") and probe.is_dir(r, n)
                      and any(x.endswith(".py") for x in probe.listing(r / n))):
                    self.tops.add(n)
 
    def is_first_party(self, dotted: str) -> bool:
        return dotted.split(".")[0] in self.tops
 
    def exists(self, dotted: str) -> bool:
        return (any(self.probe.module_exists(r, dotted) for r in self.roots)
                or any(self.probe.module_exists(r, dotted, True) for r in self.local_roots))
 
 
def _resolver_for(roots: tuple[Path, ...], local_roots: tuple[Path, ...],
                  probe: _Probe) -> _Resolver:
    resolvers = probe.memo.setdefault("resolvers", {})
    got = resolvers.get((roots, local_roots))
    if got is None:
        got = resolvers[(roots, local_roots)] = _Resolver(roots, local_roots, probe)
    return got
 
 
# -- imports ----------------------------------------------------------------------------------
 
 
def _type_checking_kind(test: ast.expr) -> str | None:
    """``"body"`` for ``if TYPE_CHECKING:``, ``"orelse"`` for ``if not TYPE_CHECKING:``."""
    def is_tc(node: ast.expr) -> bool:
        return ((isinstance(node, ast.Name) and node.id == "TYPE_CHECKING")
                or (isinstance(node, ast.Attribute) and node.attr == "TYPE_CHECKING"))
    if is_tc(test):
        return "body"
    if isinstance(test, ast.UnaryOp) and isinstance(test.op, ast.Not) and is_tc(test.operand):
        return "orelse"
    return None
 
 
def _imports_with_laziness(tree: ast.Module) -> list[tuple[ast.Import | ast.ImportFrom, bool]]:
    """Every import statement with whether it is lazy, in one linear pass."""
    out: list[tuple[ast.Import | ast.ImportFrom, bool]] = []
    stack: list[tuple[ast.AST, bool]] = [(stmt, False) for stmt in tree.body]
    while stack:
        node, lazy = stack.pop()
        if isinstance(node, (ast.Import, ast.ImportFrom)):
            out.append((node, lazy))
        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            stack.extend((child, True) for child in node.body)
        elif isinstance(node, ast.If):
            kind = _type_checking_kind(node.test)
            stack.extend((child, lazy or kind == "body") for child in node.body)
            stack.extend((child, lazy or kind == "orelse") for child in node.orelse)
        else:
            stack.extend((child, lazy) for child in ast.iter_child_nodes(node)
                         if isinstance(child, (ast.stmt, ast.excepthandler, ast.match_case)))
    return out
 
 
def collect_imports(tree: ast.Module, module: str, is_package: bool,
                    resolver: _Resolver) -> tuple[dict[str, bool], set[str]]:
    """``(non-stdlib dependency -> imported eagerly, standard-library names)``.
 
    Names are recorded as written: ``import a.b`` is ``a.b``; ``from X import
    y`` is ``X.y`` when that is a first-party module, else ``X``.
    """
    found: dict[str, bool] = {}
    stdlib: set[str] = set()
 
    def add(name: str, eager: bool) -> None:
        if name and name != module:
            found[name] = found.get(name, False) or eager
 
    for node, lazy in _imports_with_laziness(tree):
        eager = not lazy
        if isinstance(node, ast.Import):
            for alias in node.names:
                if alias.name.split(".")[0] in STDLIB:
                    stdlib.add(alias.name)
                else:
                    add(alias.name, eager)
            continue
        if node.level:
            base = _resolve_relative(module, is_package, node.level, node.module)
            if base is None:
                continue
        else:
            base = node.module or ""
        if not base:
            continue
        if base.split(".")[0] in STDLIB:
            stdlib.add(base)
            continue
        if resolver.is_first_party(base):
            for alias in node.names:
                sub = f"{base}.{alias.name}"
                add(sub if alias.name != "*" and resolver.exists(sub) else base, eager)
        else:
            add(base, eager)
    return found, stdlib
 
 
# -- the check ------------------------------------------------------------------------------------
 
 
def _covers(entry: str, dep: str, first_party: bool) -> bool:
    if entry == dep:
        return True
    return not first_party and dep.startswith(entry + ".")
 
 
def _is_test_module(py_file: Path, naming_root: Path, project_root: Path) -> bool:
    return (_is_test_path(py_file, naming_root) or _is_test_path(py_file, project_root)
            or py_file.name.startswith("test_") or py_file.name == "conftest.py")
 
 
def _stdlib_satisfied(entry: str, stdlib: set[str]) -> bool:
    """A declared standard-library entry is satisfied by an import of it,
    of a submodule, or of its package (``import os`` gives ``os.path``)."""
    return any(_covers(entry, s, False) or _covers(s, entry, False) for s in stdlib)
 
 
def check_dependencies_match(py_file: Path, project_root: Path, tree: ast.Module | None,
                             docstring: str | None = None, *,
                             memo: dict | None = None) -> tuple[bool, str]:
    """The ``dependencies_match`` verdict for one file: ``(ok, message)``.
 
    *docstring* defaults to the module docstring of *tree*. *memo* holds the
    filesystem probes across calls (the linter passes one per run); without
    it every call probes afresh. Test modules are exempt. Never raises.
    """
    func = "check_dependencies_match"
    try:
        probe = _Probe(memo if memo is not None else {})
        py_file = py_file.resolve()
        project_root = project_root.resolve()
        roots = _workspace_roots(project_root, probe)
        naming, packaged = _naming_root(py_file, roots, probe)
        if _is_test_module(py_file, naming, project_root):
            return True, "exempt: test module"
        if tree is None:
            return True, "not run: the file does not parse (syntax_error reports it)"
        if docstring is None:
            docstring = ast.get_docstring(tree, clean=False) or ""
        local = () if packaged and naming in roots else tuple(dict.fromkeys((naming, naming.parent)))
        resolver = _resolver_for(roots if naming in roots else roots + (naming,), local, probe)
        module = module_name_for(py_file, naming)
        imports, stdlib = collect_imports(tree, module, py_file.name == "__init__.py", resolver)
        entries, unrecognised, _present = parse_entries(docstring)
 
        f = Findings(unrecognised=unrecognised)
        declared: list[tuple[Entry, bool]] = []
        for e in entries:
            fp = resolver.is_first_party(e.name)
            if fp and not e.not_imported and e.name not in imports and not resolver.exists(e.name):
                f.unresolvable.append(e.name)
                continue
            declared.append((e, fp))
 
        # A package entry can cover an eager import (click) and a lazy one
        # (click.utils) at once; the lazy one then needs no marker.
        covers_eager = {
            id(e): any(imports[d] and _covers(e.name, d, fp) for d in imports)
            for e, fp in declared
        }
        for dep, eager in sorted(imports.items()):
            covering = [e for e, fp in declared if _covers(e.name, dep, fp)]
            if covering and all(e.not_imported for e in covering):
                f.marked_not_imported_but_imported.append(dep)
            elif not covering:
                (f.missing if eager else f.missing_lazy).append(dep)
            elif not eager and not any(e.lazy or e.not_imported or covers_eager[id(e)] for e in covering):
                f.unmarked_lazy.append(dep)
            elif eager and all(e.lazy for e in covering):
                f.marked_lazy_but_eager.append(dep)
        for e, fp in declared:
            if e.not_imported:
                continue
            if any(_covers(e.name, dep, fp) for dep in imports):
                continue
            if e.name.split(".")[0] in STDLIB and _stdlib_satisfied(e.name, stdlib):
                continue
            if e.name == "oc_diagnostics" and _SHIM in imports:
                continue
            f.stale.append(e.name)
        _dbg("OCT-DEPS", func, f"{module}: imports={len(imports)} entries={len(entries)} ok={f.ok}", 4)
        return f.ok, f.message()
    except Exception as exc:  # never raise: see Contracts
        return True, f"not run: internal error ({type(exc).__name__}: {exc})"
 
 
__all__ = [
    "Entry",
    "Findings",
    "STDLIB",
    "check_dependencies_match",
    "collect_imports",
    "parse_entries",
]