diff --git a/CHANGELOG.md b/CHANGELOG.md index 47d174ca..77adf74a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,25 @@ uses [Semantic Versioning](https://semver.org/). `addModule`; `std`, `builtin`, and third-party packages are dropped. `test` blocks and `zig-cache/`, `.zig-cache/`, `zig-out/` are skipped, `build.zig` is also a build artifact, and `build.zig.zon` is a manifest. +- **Lean 4 analyzer** ([#154](https://github.com/FSoft-AI4Code/CodeWiki/issues/154)). + `def`, `theorem`/`lemma`, `abbrev`, `instance`, `structure`, `class`, + `inductive`, `axiom` and `opaque` declarations become components, keyed by + their fully qualified name (`Geo.Point.add`). A header scanner reads the + structure from the source with comments and strings masked: declarations, + the `namespace`/`section`/`mutual`/`end` stack, `open` (also `open X in`), + `import`, spans and `/-- -/` docstrings. So a region that the tree-sitter + grammar cannot parse does not hide a declaration. References come from the + tree-sitter identifiers, without binders, `let`/`have`/`fun`/`intro` names + and match patterns. A reference resolves only to an exact name in the + declaration's namespaces, the root or the open namespaces, in a file that + the caller imports (transitively); `private` names stay in their file, and + a reference with no match or more than one is dropped. Constructor and + field names (`Point.mk`, `Point.x`, `Shape.dot`) link to their structure or + inductive, and `origin.x` links to `origin`. `structure` is a `struct`, + `class` an `interface`, `inductive` a `class`, and the rest are functions. + `.lake/` and `lake-packages/` are skipped, `lakefile.lean` and + `lakefile.toml` are build artifacts, and `lake-manifest.json` and + `lean-toolchain` are manifests. - **Vue single-file components.** Each `.vue` file becomes a component (`ArticleList.vue::ArticleList`), and its options-API members (`methods`, `computed`, `watch`, lifecycle hooks such as `mounted` or `setup`) become @@ -93,6 +112,17 @@ uses [Semantic Versioning](https://semver.org/). `--update` keeps it, so existing flat docs stay flat. The GitHub Pages viewer, the web app and the MCP tools (`doc_path` in `processing_order.json`) follow the layout. +- **tree-sitter-language-pack 1.x.** The pin moves from `>=0.8.0,<0.9.0` to + `>=1.23.1,<2.0.0`, which brings grammars 0.8 does not ship (Lean, among + others). 1.x downloads a grammar on first use and caches it under + `tree_sitter_language_pack.cache_dir()`, so the first run of the Dart, Go, + JavaScript/TypeScript-in-Vue, Vue and Zig analyzers needs network access; + the Docker image fetches them at build time. Without network access and + without a cached grammar, the analysis stops before the first file with one + error that names the grammar, the cache folder and the `t.download([...])` + command, instead of a traceback per file. The 1.x Dart grammar parses + `Foo(x)` as a call instead of a relational expression, and the Dart + analyzer now takes the type-argument edges from that call too. ## [2.0.0] - 2026-09-18 diff --git a/README.md b/README.md index 594eeb31..bab8597e 100644 --- a/README.md +++ b/README.md @@ -120,8 +120,8 @@ Full list in the [CHANGELOG](./CHANGELOG.md). ## Features **Languages.** Python, Java, JavaScript, TypeScript, C, C++, C#, Kotlin, -PHP, Ruby, Scala, Rust, Dart (Flutter-aware), Go, Zig, Vue (single-file -components). +PHP, Ruby, Scala, Rust, Dart (Flutter-aware), Go, Zig, Lean 4, Vue +(single-file components). **Hierarchical decomposition.** The dependency graph is clustered into a module tree, recursively, so a 1.4M-line repository gets the same treatment diff --git a/codewiki/cli/utils/repo_validator.py b/codewiki/cli/utils/repo_validator.py index 3a56587a..ebd93f3f 100644 --- a/codewiki/cli/utils/repo_validator.py +++ b/codewiki/cli/utils/repo_validator.py @@ -37,6 +37,7 @@ ".dart", # Dart ".go", # Go ".zig", # Zig + ".lean", # Lean 4 ".vue", # Vue single-file components } diff --git a/codewiki/cli/utils/validation.py b/codewiki/cli/utils/validation.py index d5bddfd0..82f77b6b 100644 --- a/codewiki/cli/utils/validation.py +++ b/codewiki/cli/utils/validation.py @@ -171,6 +171,7 @@ def detect_supported_languages(directory: Path) -> list[tuple[str, int]]: "Dart": [".dart"], "Go": [".go"], "Zig": [".zig"], + "Lean": [".lean"], "Vue": [".vue"], } diff --git a/codewiki/mcp/tools/analysis.py b/codewiki/mcp/tools/analysis.py index 0d05040d..1a1e7859 100644 --- a/codewiki/mcp/tools/analysis.py +++ b/codewiki/mcp/tools/analysis.py @@ -232,6 +232,7 @@ def _detect_via_mtime( ".dart", ".go", ".zig", + ".lean", ".vue", } diff --git a/codewiki/src/be/dependency_analyzer/analysis/analysis_service.py b/codewiki/src/be/dependency_analyzer/analysis/analysis_service.py index 0be19e62..7b41b7fb 100644 --- a/codewiki/src/be/dependency_analyzer/analysis/analysis_service.py +++ b/codewiki/src/be/dependency_analyzer/analysis/analysis_service.py @@ -365,6 +365,7 @@ def _filter_supported_languages(self, code_files: list[dict]) -> list[dict]: "scala", "dart", "zig", + "lean", "vue", } @@ -392,6 +393,7 @@ def _get_supported_languages(self) -> list[str]: "dart", "go", "zig", + "lean", "vue", ] diff --git a/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py b/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py index c52d090f..47bc73b4 100644 --- a/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py +++ b/codewiki/src/be/dependency_analyzer/analysis/call_graph_analyzer.py @@ -15,6 +15,7 @@ from collections import defaultdict from contextlib import contextmanager from pathlib import Path +from typing import TYPE_CHECKING from codewiki.src.be.dependency_analyzer.models.core import CallRelationship, Node from codewiki.src.be.dependency_analyzer.utils.external_symbols import ( @@ -22,9 +23,14 @@ is_external_symbol, is_macro_name, ) +from codewiki.src.be.dependency_analyzer.utils.grammars import ensure_pack_grammars from codewiki.src.be.dependency_analyzer.utils.patterns import CODE_EXTENSIONS from codewiki.src.be.dependency_analyzer.utils.security import safe_open_text +if TYPE_CHECKING: + from codewiki.src.be.dependency_analyzer.analyzers.lean import LeanFile, LeanScope + from codewiki.src.be.dependency_analyzer.analyzers.lean_resolution import LeanNameResolver + logger = logging.getLogger(__name__) @@ -75,6 +81,7 @@ def __init__(self): self._dart_unscoped = False self._go_visible: dict[str, set[str]] = {} self._go_index = None + self._reset_lean_state() self.call_relationships: list[CallRelationship] = [] self._python_project_modules: set = set() self._python_external_import_roots: set = set() @@ -96,6 +103,9 @@ def analyze_code_files(self, code_files: list[dict], base_dir: str) -> dict: self.functions = {} self.call_relationships = [] code_files = self._route_contextual_headers(code_files, base_dir) + # Fail once, with a clear message, when a pack grammar cannot be + # downloaded, instead of once per file inside the per-file handlers. + ensure_pack_grammars(f["language"] for f in code_files) self._python_project_modules = self._collect_python_modules(code_files) self._python_external_import_roots = set() self._dart_directives = {} @@ -104,6 +114,7 @@ def analyze_code_files(self, code_files: list[dict], base_dir: str) -> dict: self._dart_unscoped = False self._go_visible = {} self._go_index = None + self._reset_lean_state() files_analyzed = 0 files_failed = 0 @@ -289,6 +300,8 @@ def _analyze_code_file(self, repo_dir: str, file_info: dict): self._analyze_go_file(file_path, content, repo_dir) elif language == "zig": self._analyze_zig_file(file_path, content, repo_dir) + elif language == "lean": + self._analyze_lean_file(file_path, content, repo_dir) elif language == "vue": self._analyze_vue_file(file_path, content, repo_dir) # else: @@ -664,6 +677,59 @@ def _analyze_vue_file(self, file_path: str, content: str, repo_dir: str): except Exception: logger.exception(f"Failed to analyze Vue file {file_path}") + def _reset_lean_state(self) -> None: + self._lean_files: dict[str, LeanFile] = {} + self._lean_scopes: dict[str, LeanScope] = {} + self._lean_aliases: dict[str, list[str]] = defaultdict(list) + self._lean_private: set[str] = set() + self._lean_protected: set[str] = set() + self._lean_resolver: LeanNameResolver | None = None + + def _analyze_lean_file(self, file_path: str, content: str, repo_dir: str): + """ + Analyze a Lean 4 file. + + Besides components and references, records the file's imports, each + declaration's namespaces and open namespaces, and the constructor and + field names of its structures, for Lean resolution by exact name. + + Args: + file_path: Path to the Lean file + content: File content string + repo_dir: Repository base directory + """ + from codewiki.src.be.dependency_analyzer.analyzers.lean import TreeSitterLeanAnalyzer + + try: + analyzer = TreeSitterLeanAnalyzer(str(file_path), content, repo_path=repo_dir) + + for func in analyzer.nodes: + self.functions[func.id] = func + + self.call_relationships.extend(analyzer.call_relationships) + self._lean_files[analyzer.relative_path] = analyzer.scan + self._lean_scopes.update(analyzer.scopes) + for alias, node_id in analyzer.aliases.items(): + self._lean_aliases[alias].append(node_id) + self._lean_private.update(analyzer.private_ids) + self._lean_protected.update(analyzer.protected_ids) + except Exception: + logger.exception(f"Failed to analyze Lean file {file_path}") + + def _resolve_lean_callee(self, callee: str, caller: Node) -> str | None: + from codewiki.src.be.dependency_analyzer.analyzers.lean_resolution import LeanNameResolver + + if self._lean_resolver is None: + self._lean_resolver = LeanNameResolver( + self.functions, + self._lean_files, + self._lean_scopes, + self._lean_aliases, + self._lean_private, + self._lean_protected, + ) + return self._lean_resolver.resolve(callee, caller) + def _analyze_go_file(self, file_path: str, content: str, repo_dir: str): """ Analyze Go file using tree-sitter based analyzer. @@ -856,11 +922,11 @@ def _resolve_call_relationships(self): self.call_relationships.extend(link_go_interfaces(self.functions)) def _has_authoritative_resolution(self, relationship: CallRelationship) -> bool: - """A Dart or Go caller's callee that the language's scoped resolution + """A Dart, Go or Lean caller's callee that the language's scoped resolution rejected is external or invisible; drop it so ast_parser's name-only fallback cannot bind it to an unrelated component.""" caller = self.functions.get(relationship.caller) - return caller is not None and caller.language in ("dart", "go") + return caller is not None and caller.language in ("dart", "go", "lean") def _dotted_project_packages(self) -> dict[str, set]: """Project packages/namespaces, partitioned by language. Java and C# @@ -1003,6 +1069,9 @@ def _resolve_callee( if caller_language == "go" and caller is not None: # So is Go's: a name resolves only in the package it names. return self._resolve_go_callee(relationship.callee, caller) + if caller_language == "lean" and caller is not None: + # And Lean's: only an exact FQN in a visible file. + return self._resolve_lean_callee(relationship.callee, caller) if lang_indexes: match = self._resolve_callee_in( relationship, lang_indexes["exact"], lang_indexes["simple"] @@ -1148,6 +1217,8 @@ def _generate_visualization_data(self) -> dict: node_classes.append("lang-go") elif file_ext == ".zig": node_classes.append("lang-zig") + elif file_ext == ".lean": + node_classes.append("lang-lean") elif file_ext == ".vue": node_classes.append("lang-vue") diff --git a/codewiki/src/be/dependency_analyzer/analyzers/artifact.py b/codewiki/src/be/dependency_analyzer/analyzers/artifact.py index d7be67a9..e87d4d8b 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/artifact.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/artifact.py @@ -244,6 +244,8 @@ def is_artifact_file_node(node: Any) -> bool: "build.sbt", "Package.swift", "build.zig.zon", + "lake-manifest.json", + "lean-toolchain", "pubspec.yaml", "Pipfile", "environment.yml", @@ -296,6 +298,8 @@ def is_artifact_file_node(node: Any) -> bool: "Gruntfile.js", "Herebyfile.mjs", "build.zig", + "lakefile.lean", + "lakefile.toml", # Autotools inputs; the generated configure, Makefile.in and aclocal.m4 # are not build sources. "configure.ac", @@ -451,6 +455,7 @@ def classify_artifact( ".scala", ".sc", ".zig", + ".lean", ".vue", }: if name in _TEST_INFRA_NAMES: diff --git a/codewiki/src/be/dependency_analyzer/analyzers/dart.py b/codewiki/src/be/dependency_analyzer/analyzers/dart.py index c773265d..6af1b0c9 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/dart.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/dart.py @@ -36,7 +36,7 @@ from functools import lru_cache from pathlib import Path -from tree_sitter_language_pack import get_parser +from codewiki.src.be.dependency_analyzer.utils.grammars import get_pack_parser from codewiki.src.be.dependency_analyzer.analyzers.dart_flutter import ( COMPOSITION_METHOD_RE, @@ -212,7 +212,7 @@ def _local_declared_types(*roots) -> dict[str, str]: @lru_cache(maxsize=1) def _dart_parser(): - return get_parser("dart") + return get_pack_parser("dart") def _text(node) -> str: @@ -762,6 +762,17 @@ def _process_chain(self, target, head, selectors, local_types, lift_to): return caller, owner = target.caller, target.owner line = head.start_point[0] + 1 + # `Foo(x)` / `foo(x)`: the type arguments sit in the call's + # argument_part (grammars that misparse it as a relational expression + # are handled by _handle_generic_invocation). + for token_kind, _, token_node in tokens: + if token_kind != "call": + continue + type_args = _first_child(token_node, "type_arguments") + if type_args is not None: + for ref in _descendants(type_args): + if ref.type == "type_identifier": + self._add_type_edge(caller, _text(ref), line) name = _text(head) if head.type == "identifier" and name in self.import_prefixes and tokens[0][0] == "member": name, tokens = tokens[0][1] or "", tokens[1:] diff --git a/codewiki/src/be/dependency_analyzer/analyzers/go.py b/codewiki/src/be/dependency_analyzer/analyzers/go.py index 7a241812..8c0042fa 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/go.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/go.py @@ -48,7 +48,7 @@ from functools import lru_cache from pathlib import Path -from tree_sitter_language_pack import get_parser +from codewiki.src.be.dependency_analyzer.utils.grammars import get_pack_parser from codewiki.src.be.dependency_analyzer.models.core import CallRelationship, Node from codewiki.src.be.dependency_analyzer.utils.file_selection import is_skipped_go_file @@ -326,7 +326,7 @@ def _default_import_names(import_path: str) -> set[str]: @lru_cache(maxsize=1) def _go_parser(): - return get_parser("go") + return get_pack_parser("go") class TreeSitterGoAnalyzer: diff --git a/codewiki/src/be/dependency_analyzer/analyzers/lean.py b/codewiki/src/be/dependency_analyzer/analyzers/lean.py new file mode 100644 index 00000000..dd1e5e12 --- /dev/null +++ b/codewiki/src/be/dependency_analyzer/analyzers/lean.py @@ -0,0 +1,967 @@ +"""Lean 4 analyzer: a header scanner for structure, tree-sitter for references. + +The tree-sitter Lean grammar (``lean`` in tree-sitter-language-pack) parses +most declarations, but it fails on many commands and notations (``#eval``, +``¬``, ``←``, ``do`` blocks, user notation), and it gives ``end`` as a flat +node, so the tree does not show namespace nesting. This analyzer therefore +uses two layers: + +1. **Structure, from a header scanner.** Comments and strings are masked, + and the Lean commands are read line by line: declarations (``def``, + ``theorem``/``lemma``, ``abbrev``, ``instance``, ``structure``, ``class``, + ``inductive``, ``axiom``, ``opaque``), the ``namespace``/``section``/ + ``mutual``/``end`` stack, ``variable``, scoped/restricted ``open`` + (also ``open X in``), ``export``, ``import``, + ``/-- -/`` docstrings and spans. The scanner does not depend on the + parse tree, so an ``ERROR`` region does not hide a declaration. +2. **References, from tree-sitter.** Every ``identifier`` node in a + declaration's lines is a reference candidate, also inside ``ERROR`` + nodes. Text that an ``ERROR`` node does not cover with child nodes is + tokenized from the masked source. Local names (binders, ``let``, + ``have``, ``fun``, ``intro``, match patterns) are removed, and so is a + dotted name whose first part is local (``p.x``). + +Each declaration is keyed by its fully qualified name (FQN). Structures and +classes also answer to their constructor and field names (``Point.mk``, +``Point.x``), and inductives to their constructors (``Tree.node``). The +analyzer emits each reference as written. ``CallGraphAnalyzer`` resolves it +to an exact FQN only, in Lean's order: the namespaces of the declaration +(innermost first), the root namespace, then the open namespaces. Only the +file itself and the files it imports (transitively) are visible, and a +``private`` name is visible only in its own file. A reference that matches +no declaration, or more than one, is dropped. A bare name is never guessed. + +Open directives are bound in their command context and retain selection, hiding, +renaming and protected-name rules. Scoped opens do not introduce ordinary names. +Exports persist as aliases of their original components, including across imports; +closing a section restores its open directives and variable bindings. + +Known limitations: +- Generalized field notation (``v.norm`` for ``Vec.norm``) needs type + information, so it does not resolve. +- References made through user ``notation``/``macro`` are lost. +- The scanner assumes that each command starts a line, which is the usual + Lean style. A declaration ends at the next command at the same or a + smaller indentation. +- An anonymous ``instance`` gets the synthetic name ``instance_L``, + not the name that Lean generates. ``example`` makes no component. +- Local names are removed by name for the whole declaration, so a global + with the same simple name as a local of that declaration is not linked. +""" + +import bisect +import logging +import os +import re +from dataclasses import dataclass, field +from functools import lru_cache +from pathlib import Path + +from codewiki.src.be.dependency_analyzer.models.core import CallRelationship, Node +from codewiki.src.be.dependency_analyzer.utils.file_selection import is_skipped_lean_file +from codewiki.src.be.dependency_analyzer.utils.grammars import get_pack_parser + +logger = logging.getLogger(__name__) + +# `component_type` per declaration keyword; `node_type` keeps the keyword. +COMPONENT_TYPES = { + "structure": "struct", + "class": "interface", + "inductive": "class", + "class inductive": "class", +} + +_IDENT_PART = r"(?:«[^»\n]*»|[^\W\d][\w'!?₀-₉ₐ-ₜ]*)" +IDENT = rf"{_IDENT_PART}(?:\.(?:{_IDENT_PART}|\d+))*" +_IDENT_RE = re.compile(IDENT) + +_MODIFIERS = r"(?:private|protected|noncomputable|partial|unsafe|nonrec|scoped|local|public|meta)" +_ATTRS = r"(?:@\[[^\]\n]*\]\s*)*" +_DECL_RE = re.compile( + rf"^(?P\s*){_ATTRS}(?P(?:{_MODIFIERS}\s+)*)" + r"(?Pclass\s+inductive|def|theorem|lemma|abbrev|instance|structure|class|inductive" + r"|axiom|opaque|example)\b(?P.*)$" +) +_ATTR_ONLY_RE = re.compile(r"^\s*@\[[^\]\n]*\]\s*$") +_IMPORT_RE = re.compile(r"^\s*(?:(?:public|meta|private)\s+)*import\s+(?P.+)$") +_NAMESPACE_RE = re.compile(rf"^\s*namespace\s+(?P{IDENT})") +_SECTION_RE = re.compile(rf"^\s*(?:noncomputable\s+)?section\b\s*(?P{IDENT})?") +_MUTUAL_RE = re.compile(r"^\s*mutual\b") +_END_RE = re.compile(rf"^\s*end\b\s*(?P{IDENT})?\s*$") +_OPEN_RE = re.compile(r"^\s*open\b\s*(?P.*)$") +_EXPORT_RE = re.compile(r"^\s*export\b\s*(?P.*)$") +_VARIABLE_RE = re.compile(r"^\s*variable\b\s*(?P.*)$") +# Commands that end the span of the declaration before them. +_COMMAND_RE = re.compile( + r"^\s*(?:@\[|#|(?:" + r"import|namespace|section|end|open|export|mutual|attribute|variable|universe|set_option" + r"|notation|infix|infixl|infixr|prefix|postfix|macro|macro_rules|syntax|elab|elab_rules" + r"|deriving|initialize|builtin_initialize|declare_syntax_cat|add_decl_doc|omit|include" + r"|run_cmd" + r")\b)" +) +_NAME_AFTER_KIND_RE = re.compile(rf"^\s*(?P{IDENT})") +_INSTANCE_NAME_RE = re.compile(rf"^\s*(?:\(\s*priority\s*:=[^)]*\)\s*)?(?P{IDENT})?\s*") +_EXTENDS_RE = re.compile(r"\bextends\s+(?P[^:=]*?)(?:\bwhere\b|:=|:|$)") +_STRUCT_CTOR_RE = re.compile(rf"^\s*(?P{IDENT})\s*::\s*$") +_FIELD_RE = re.compile(rf"^\s*\(?\s*(?P{IDENT}(?:\s+{IDENT})*)\s*(?::(?!=)|:=)") + +# Local names introduced in a declaration's text. Tree rules cover the parsed +# parts; these also cover `ERROR` regions and tactic blocks. +_LOCAL_RES = [ + re.compile(rf"\blet\s+(?:mut\s+|rec\s+)?(?P{IDENT})"), + re.compile(rf"\bhave\s+(?P{IDENT})\s*:"), + re.compile(rf"\bfor\s+(?P{IDENT})\s+in\b"), + re.compile(rf"\(\s*(?P{IDENT})\s*:="), + re.compile( + r"\b(?:intro|intros|rintro|rename_i|next|case|obtain|rcases|cases'|induction')\s+(?P[^\n:=]*)" + ), + re.compile(rf"(?P{IDENT})\s*←"), +] + +_KEYWORDS = { + "fun", "let", "have", "show", "from", "by", "do", "then", "else", "if", "match", "with", + "at", "in", "for", "return", "where", "deriving", "mut", "rec", "Type", "Sort", "Prop", + "true", "false", "True", "False", "this", "_", "sorry", "calc", "suffices", "using", +} # fmt: skip + +_LOCAL_BINDER_PARENTS = {"binders"} +_BINDER_TYPES = { + "explicit_binder", + "implicit_binder", + "inst_binder", + "strict_implicit_binder", +} +_INTRO_TACTICS = {"intro", "intros", "rintro", "obtain", "rcases", "rename_i", "next", "case"} + + +@lru_cache(maxsize=1) +def _lean_parser(): + return get_pack_parser("lean") + + +# ---------------------------------------------------------------------- +# Masking +# ---------------------------------------------------------------------- + + +@dataclass +class _Masked: + text: str # same lines and the same UTF-8 byte offsets as the source + docstrings: list = field(default_factory=list) # (end line index, text) + + +_RAW_STRING_RE = re.compile(r'r(#*)"') +_MASK_START_RE = re.compile(r"--|/-|r#*\"|\"|'") +_BLOCK_TOKEN_RE = re.compile(r"/-|-/") +_NOT_NEWLINE_RE = re.compile(rb"[^\n]") +_CHAR_RE = re.compile(r"'(?:\\[^']*|[^\\'\n])'") + + +def _blank(text: str) -> str: + return _NOT_NEWLINE_RE.sub(b" ", text.encode("utf8")).decode("ascii") + + +def mask_source(source: str) -> _Masked: + """Blank comments, strings and char literals with spaces of the same + UTF-8 length, so lines and byte offsets do not move. Docstrings + (``/-- -/``) are kept aside with the line on which they end.""" + out: list[str] = [] + docstrings = [] + i, n = 0, len(source) + line, counted = 0, 0 # newlines before `counted`, so line numbers stay linear + + while i < n: + m = _MASK_START_RE.search(source, i) + if m is None: + out.append(source[i:]) + break + start, token = m.start(), m.group(0) + out.append(source[i:start]) + if token == "--": + j = source.find("\n", start) + j = n if j < 0 else j + elif token == "/-": + depth, j = 1, start + 2 + while depth: + b = _BLOCK_TOKEN_RE.search(source, j) + if b is None: + j = n + break + depth += 1 if b.group(0) == "/-" else -1 + j = b.end() + if source.startswith("/--", start): + line += source.count("\n", counted, j) + counted = j + docstrings.append((line, _clean_docstring(source[start:j]))) + elif token != '"' and _ident_before(source, start): + # `add_comm'` or `r"` after a name: not a literal. + out.append(source[start]) + i = start + 1 + continue + elif token == "'": + c = _CHAR_RE.match(source, start) + if c is None: + out.append(token) + i = start + 1 + continue + j = c.end() + else: + j = _string_end(source, start) + out.append(_blank(source[start:j])) + i = j + return _Masked("".join(out), docstrings) + + +def _ident_before(source: str, i: int) -> bool: + return i > 0 and (source[i - 1].isalnum() or source[i - 1] in "_'!?»") + + +def _string_end(source: str, i: int) -> int: + n = len(source) + if source[i] == "r": + hashes = len(_RAW_STRING_RE.match(source, i).group(1)) + close = '"' + "#" * hashes + j = source.find(close, i + 2 + hashes) + return n if j < 0 else j + len(close) + j = i + 1 + while j < n: + if source[j] == "\\": + j += 2 + elif source[j] == '"': + return j + 1 + else: + j += 1 + return n + + +def _clean_docstring(block: str) -> str: + body = block[3:-2] if block.endswith("-/") else block[3:] + return "\n".join(line.strip() for line in body.strip().splitlines()).strip() + + +# ---------------------------------------------------------------------- +# Header scanner +# ---------------------------------------------------------------------- + + +@dataclass(frozen=True) +class LeanOpen: + """An opening captured at its command, before entering any later scope.""" + + namespace: str + line: int + context: str + preceding: tuple[int, ...] + mode: str = "all" + names: tuple[str, ...] = () + renames: tuple[tuple[str, str], ...] = () + + +@dataclass(frozen=True) +class LeanExport: + namespace: str + names: tuple[str, ...] + destination: str + line: int + opens: tuple[int, ...] + + +@dataclass(frozen=True) +class LeanScope: + namespaces: tuple[str, ...] + opens: tuple[int, ...] + line: int + + +@dataclass +class LeanDecl: + fqn: str + kind: str + start: int # 0-based line of the first attribute or keyword line + keyword_line: int + end: int # 0-based, inclusive + indent: int + private: bool + namespaces: list[str] # the declaration's namespace prefixes, innermost first + opens: list[int] + protected: bool = False + variables: frozenset[str] = frozenset() + docstring: str = "" + aliases: list[str] = field(default_factory=list) # constructor and field FQNs + bases: list[str] = field(default_factory=list) + + +@dataclass +class LeanFile: + declarations: list[LeanDecl] = field(default_factory=list) + imports: list[str] = field(default_factory=list) + opens: list[LeanOpen] = field(default_factory=list) + exports: list[LeanExport] = field(default_factory=list) + namespaces: list[tuple[int, str]] = field(default_factory=list) + + +@dataclass +class _Scope: + kind: str # "namespace", "section" or "mutual" + name: str + opens: list[int] = field(default_factory=list) + variables: set[str] = field(default_factory=set) + + +def _qualify(prefix: str, name: str) -> str: + return f"{prefix}.{name}" if prefix else name + + +def _strip_guillemets(name: str) -> str: + return name.replace("«", "").replace("»", "") + + +_TOKEN_RE = re.compile(rf"{IDENT}|:=|=>|->|[^\s]") +_DELIMITERS = {"(": ")", "{": "}", "⦃": "⦄", "[": "]"} + + +def _tokens(text: str) -> list[str]: + return _TOKEN_RE.findall(text) + + +def _group_end(tokens: list[str], start: int) -> int: + """Index after a balanced group, or the end of an incomplete group.""" + stack = [_DELIMITERS[tokens[start]]] + for i in range(start + 1, len(tokens)): + token = tokens[i] + if token in _DELIMITERS: + stack.append(_DELIMITERS[token]) + elif token == stack[-1]: + stack.pop() + if not stack: + return i + 1 + return len(tokens) + + +def _binder_names(tokens: list[str]) -> set[str]: + """Read binder names, never identifiers in their types or defaults.""" + names: set[str] = set() + i = 0 + while i < len(tokens): + token = tokens[i] + if token in (":", ":=", "=>", ",", "where", "extends", "deriving", "|"): + break + if token in _DELIMITERS: + end = _group_end(tokens, i) + body = tokens[i + 1 : end - 1] + # `[Class α]` binds an anonymous instance, not `Class` or `α`. + if token != "[" or (len(body) > 1 and _IDENT_RE.fullmatch(body[0]) and body[1] == ":"): + names.update(_binder_names(body)) + i = end + continue + if _IDENT_RE.fullmatch(token) and "." not in token: + names.add(_strip_guillemets(token)) + i += 1 + return names - _KEYWORDS + + +def _parse_open(body: str, line: int, context: str, preceding: list[int]) -> list[LeanOpen]: + tokens = _tokens(body) + mode = "all" + if tokens and tokens[0] == "scoped": + mode = tokens.pop(0) + if not tokens or not _IDENT_RE.fullmatch(tokens[0]): + return [] + if len(tokens) > 1 and tokens[1] in ("(", "hiding", "renaming"): + namespace = _strip_guillemets(tokens[0]) + shape = tokens[1] + names: tuple[str, ...] = () + renames: tuple[tuple[str, str], ...] = () + if shape == "(": + mode = "select" + if tokens[-1] != ")": + return [] + names = tuple(_strip_guillemets(t) for t in tokens[2:-1] if _IDENT_RE.fullmatch(t)) + elif shape == "hiding": + mode = "hiding" + names = tuple(_strip_guillemets(t) for t in tokens[2:] if _IDENT_RE.fullmatch(t)) + else: + mode = "renaming" + pairs = [] + rest = tokens[2:] + while rest: + if ( + len(rest) < 3 + or not _IDENT_RE.fullmatch(rest[0]) + or rest[1] not in ("→", "->") + or not _IDENT_RE.fullmatch(rest[2]) + ): + return [] + pairs.append((_strip_guillemets(rest[0]), _strip_guillemets(rest[2]))) + rest = rest[3:] + if rest: + if rest[0] != ",": + return [] + rest = rest[1:] + renames = tuple(pairs) + return [LeanOpen(namespace, line, context, tuple(preceding), mode, names, renames)] + if any(not _IDENT_RE.fullmatch(t) for t in tokens): + return [] + return [LeanOpen(_strip_guillemets(t), line, context, tuple(preceding), mode) for t in tokens] + + +def scan_headers(masked: _Masked) -> tuple[list[LeanDecl], list[str]]: + """Compatibility interface for callers that only need headers and imports.""" + scanned = _scan_file(masked) + return scanned.declarations, scanned.imports + + +def _scan_file(masked: _Masked) -> LeanFile: + lines = masked.text.split("\n") + docs = {end: text for end, text in masked.docstrings} + scanned = LeanFile() + decls = scanned.declarations + stack: list[_Scope] = [] + root_scope = _Scope("section", "") + pending_open_in: list[int] = [] + attr_start: int | None = None + consumed_until = -1 + + def namespace() -> str: + return ".".join(s.name for s in stack if s.kind == "namespace") + + def active_opens() -> list[int]: + return root_scope.opens + [o for s in stack for o in s.opens] + + def add_decl(match, idx: int, start: int, extra_opens: list[int]) -> None: + decl = _decl_from_match(match, idx, start, namespace(), extra_opens) + if decl is not None: + decl.opens = active_opens() + decl.opens + decl.variables = frozenset(root_scope.variables.union(*(s.variables for s in stack))) + decl.docstring = _docstring_before(docs, lines, start) + decls.append(decl) + + def close_previous(line_idx: int, indent: int) -> None: + if decls and decls[-1].end < 0 and indent <= decls[-1].indent: + end = line_idx - 1 + while end > decls[-1].keyword_line and not lines[end].strip(): + end -= 1 + decls[-1].end = end + + def close_scope(name: str | None) -> None: + if not stack: + logger.warning("Unmatched Lean end %s", name or "") + return + if name is None: + if not stack[-1].name: + stack.pop() + return + else: + parts = _strip_guillemets(name).split(".") + kind = stack[-1].kind + if ( + len(stack) >= len(parts) + and all(s.kind == kind for s in stack[-len(parts) :]) + and [s.name for s in stack[-len(parts) :]] == parts + ): + del stack[-len(parts) :] + return + logger.warning("Mismatched Lean end %s; retaining scope", name or "") + + def process_command( + command: str, idx: int, start: int, indent: int, temporary_opens: list[int] + ) -> list[int]: + # Peel right-associative `open ... in` wrappers without changing the + # spelling of the inner command (attributes and quoted names included). + temporary = bool(temporary_opens) + while m := _OPEN_RE.match(command): + body = m.group("body") + in_token = next((t for t in _TOKEN_RE.finditer(body) if t.group() == "in"), None) + opening_text = body[: in_token.start()] if in_token else body + preceding = active_opens() + temporary_opens + opening_ids = [] + for opening in _parse_open(opening_text, idx, namespace(), preceding): + opening_id = len(scanned.opens) + scanned.opens.append( + LeanOpen( + opening.namespace, + opening.line, + opening.context, + tuple(preceding), + opening.mode, + opening.names, + opening.renames, + ) + ) + opening_ids.append(opening_id) + preceding.append(opening_id) + if in_token is None: + if not temporary: + (stack[-1] if stack else root_scope).opens.extend(opening_ids) + return [] + temporary = True + temporary_opens = temporary_opens + opening_ids + command = " " * indent + body[in_token.end() :].strip() + if not command.strip(): + return temporary_opens + + if m := _DECL_RE.match(command): + add_decl(m, idx, start, temporary_opens) + elif m := _IMPORT_RE.match(command): + scanned.imports.extend( + _strip_guillemets(x.group(0)) for x in _IDENT_RE.finditer(m.group("mods")) + ) + elif m := _NAMESPACE_RE.match(command): + for name in _strip_guillemets(m.group("name")).split("."): + stack.append(_Scope("namespace", name)) + scanned.namespaces.append((idx, namespace())) + stack[-1].opens.extend(temporary_opens) + elif _MUTUAL_RE.match(command): + stack.append(_Scope("mutual", "", opens=temporary_opens)) + elif m := _END_RE.match(command): + close_scope(m.group("name")) + elif m := _SECTION_RE.match(command): + for name in (m.group("name") or "").split("."): + stack.append(_Scope("section", _strip_guillemets(name))) + stack[-1].opens.extend(temporary_opens) + elif m := _VARIABLE_RE.match(command): + if not temporary: + (stack[-1] if stack else root_scope).variables.update( + _binder_names(_tokens(m.group("body"))) + ) + elif m := _EXPORT_RE.match(command): + tokens = _tokens(m.group("body")) + if ( + len(tokens) >= 3 + and _IDENT_RE.fullmatch(tokens[0]) + and tokens[1] == "(" + and tokens[-1] == ")" + and all(_IDENT_RE.fullmatch(t) for t in tokens[2:-1]) + ): + scanned.exports.append( + LeanExport( + _strip_guillemets(tokens[0]), + tuple(_strip_guillemets(t) for t in tokens[2:-1]), + namespace(), + idx, + tuple(active_opens() + temporary_opens), + ) + ) + return [] + + for idx, line in enumerate(lines): + if idx <= consumed_until or not line.strip(): + continue + indent = len(line) - len(line.lstrip()) + if decls and decls[-1].end < 0 and indent > decls[-1].indent: + continue + decl_m = _DECL_RE.match(line) + if decl_m is None and not _COMMAND_RE.match(line): + continue + close_previous(idx, indent) + if _ATTR_ONLY_RE.match(line): + if attr_start is None: + attr_start = idx + continue + start = attr_start if attr_start is not None else idx + attr_start = None + if decl_m is not None: + add_decl(decl_m, idx, start, pending_open_in) + pending_open_in = [] + continue + + # Gather indented continuations of directives, excluding a following command. + command = line + if _OPEN_RE.match(line) or _EXPORT_RE.match(line) or _VARIABLE_RE.match(line): + end = idx + 1 + while end < len(lines): + continuation = lines[end] + if continuation.strip(): + continuation_indent = len(continuation) - len(continuation.lstrip()) + if ( + continuation_indent <= indent + or _DECL_RE.match(continuation) + or _COMMAND_RE.match(continuation) + ): + break + command += " " + continuation.strip() + consumed_until = end + end += 1 + + pending_open_in = process_command(command, idx, start, indent, pending_open_in) + + if decls and decls[-1].end < 0: + end = len(lines) - 1 + while end > decls[-1].keyword_line and not lines[end].strip(): + end -= 1 + decls[-1].end = end + for decl in decls: + if decl.kind in ("structure", "class", "inductive", "class inductive"): + decl.aliases = _member_aliases(decl, lines) + return scanned + + +def _decl_from_match(m, idx, start, ns, open_in) -> LeanDecl | None: + kind = re.sub(r"\s+", " ", m.group("kind")) + rest = m.group("rest") + if kind == "example": + name = None + elif kind == "instance": + nm = _INSTANCE_NAME_RE.match(rest) + name = nm.group("name") if nm else None + if name in _KEYWORDS: + name = None + if name is None: + name = f"instance_L{idx + 1}" + else: + nm = _NAME_AFTER_KIND_RE.match(rest) + name = nm.group("name") if nm else None + if not name: + # `example` and unnamed forms still end the previous declaration. + return None + name = _strip_guillemets(name) + if name.startswith("_root_."): + fqn = name[len("_root_.") :] + else: + fqn = _qualify(ns, name) + parts = fqn.split(".") + namespaces = [".".join(parts[:k]) for k in range(len(parts) - 1, 0, -1)] + bases = [] + if kind in ("structure", "class") and (em := _EXTENDS_RE.search(rest)): + bases = [_strip_guillemets(b.group(0)) for b in _IDENT_RE.finditer(em.group("bases"))] + return LeanDecl( + fqn=fqn, + kind=kind, + start=start, + keyword_line=idx, + end=-1, + indent=len(m.group("indent")), + private="private" in m.group("mods").split(), + protected="protected" in m.group("mods").split(), + namespaces=namespaces, + opens=list(open_in), + bases=bases, + ) + + +def _docstring_before(docs: dict, lines: list[str], start: int) -> str: + line = start - 1 + while line >= 0 and not lines[line].strip(): + if line in docs: + return docs[line] + line -= 1 + return docs.get(line, "") if line >= 0 else "" + + +def _member_aliases(decl: LeanDecl, lines: list[str]) -> list[str]: + aliases: list[str] = [] + body = lines[decl.keyword_line + 1 : decl.end + 1] + header = lines[decl.keyword_line] + if decl.kind in ("inductive", "class inductive"): + for line in [header.split(" where", 1)[-1], *body]: + for chunk in line.split("|")[1:]: + if m := _IDENT_RE.match(chunk.strip()): + aliases.append(_qualify(decl.fqn, _strip_guillemets(m.group(0)))) + return aliases + ctor = "mk" + if " where" in header or header.rstrip().endswith("where"): + for line in body: + if line.strip().startswith("deriving"): + break + if m := _STRUCT_CTOR_RE.match(line): + ctor = _strip_guillemets(m.group("name")) + elif m := _FIELD_RE.match(line): + for name in m.group("names").split(): + aliases.append(_qualify(decl.fqn, _strip_guillemets(name))) + aliases.append(_qualify(decl.fqn, ctor)) + return aliases + + +# ---------------------------------------------------------------------- +# References +# ---------------------------------------------------------------------- + + +def _node_text(node) -> str: + return node.text.decode("utf8", errors="replace") + + +def _collect_tree_names(root, masked_bytes: bytes): + """Return (references, locals): references as (line, name); locals as + (line, simple name).""" + refs: list[tuple[int, str]] = [] + locals_: list[tuple[int, str]] = [] + line_starts = [0] + [m.end() for m in re.finditer(rb"\n", masked_bytes)] + skip: set[int] = set() # ids of identifier nodes that are not references + + def mark_local(node) -> None: + for ident in _identifiers(node): + skip.add(ident.id) + text = _node_text(ident) + if "." not in text: + locals_.append((ident.start_point[0], text)) + + stack = [root] + while stack: + node = stack.pop() + t = node.type + children = node.children + if t in _LOCAL_BINDER_PARENTS: + for child in children: + if child.type == "identifier": + mark_local(child) + elif t in _BINDER_TYPES: + for child in children: + if child.type == ":": + break + if child.type == "identifier": + mark_local(child) + elif t in ("let", "have"): + for child in children[1:]: + if child.type == "identifier": + mark_local(child) + break + if child.type in (":=", ":", "←"): + break + elif t == "match_alt": + for child in children: + if child.type == "=>": + break + for ident in _identifiers(child): + if "." not in _node_text(ident): + mark_local(ident) + elif t in ("struct_field", "field"): + for child in children: + if child.type in (":=", ":"): + break + if child.type == "identifier": + skip.add(child.id) + elif t == "app" and children and children[0].type == "identifier": + if _node_text(children[0]) in _INTRO_TACTICS: + for child in children[1:]: + mark_local(child) + elif t == "ERROR": + covered = 0 + for child in children: + if child.start_byte > covered: + _tokens_in( + masked_bytes, + line_starts, + max(covered, node.start_byte), + child.start_byte, + refs, + ) + covered = max(covered, child.end_byte) + if covered < node.end_byte: + _tokens_in( + masked_bytes, line_starts, max(covered, node.start_byte), node.end_byte, refs + ) + if t == "identifier" and node.id not in skip: + refs.append((node.start_point[0], _node_text(node))) + stack.extend(reversed(children)) + return refs, locals_ + + +def _identifiers(node): + if node.type == "identifier": + yield node + return + for child in node.children: + yield from _identifiers(child) + + +def _tokens_in( + masked_bytes: bytes, line_starts: list[int], start: int, end: int, out: list +) -> None: + if end <= start: + return + chunk = masked_bytes[start:end].decode("utf8", errors="replace") + base_line = bisect.bisect_right(line_starts, start) - 1 + for m in _IDENT_RE.finditer(chunk): + out.append((base_line + chunk.count("\n", 0, m.start()), m.group(0))) + + +def _text_locals(text: str) -> set[str]: + names: set[str] = set() + tokens = _tokens(text) + declaration_kinds = { + "def", + "theorem", + "lemma", + "abbrev", + "instance", + "structure", + "class", + "inductive", + "axiom", + "opaque", + } + for start, token in enumerate(tokens): + if token in declaration_kinds: + binder_start = start + 1 + if token == "class" and tokens[binder_start : binder_start + 1] == ["inductive"]: + binder_start += 1 + if binder_start < len(tokens) and _IDENT_RE.fullmatch(tokens[binder_start]): + binder_start += 1 # the declaration name, if present + names.update(_binder_names(tokens[binder_start:])) + break + for start, token in enumerate(tokens): + if token not in ("fun", "λ", "∀", "∃", "Σ", "Π"): + continue + end = start + 1 + terminator = "=>" if token in ("fun", "λ") else "," + while end < len(tokens): + if tokens[end] == terminator: + names.update(_binder_names(tokens[start + 1 : end])) + break + if tokens[end] in _DELIMITERS: + end = _group_end(tokens, end) + else: + end += 1 + for regex in _LOCAL_RES: + for m in regex.finditer(text): + for ident in _IDENT_RE.finditer(m.group("names")): + if "." not in ident.group(0): + names.add(ident.group(0)) + return names - _KEYWORDS + + +def normalize_reference(name: str) -> str | None: + """Strip `@`, guillemets and numeric projections; drop `.foo` and keywords.""" + name = name.lstrip("@") + if not name or name.startswith("."): + return None + parts = [p for p in _strip_guillemets(name).split(".") if p] + while parts and parts[-1].isdigit(): + parts.pop() + if not parts or parts[0] in _KEYWORDS: + return None + return ".".join(parts) + + +# ---------------------------------------------------------------------- +# Analyzer +# ---------------------------------------------------------------------- + + +class TreeSitterLeanAnalyzer: + def __init__(self, file_path: str, content: str, repo_path: str | None = None): + self.file_path = Path(file_path) + self.content = content + self.repo_path = repo_path or "" + self.nodes: list[Node] = [] + self.call_relationships: list[CallRelationship] = [] + self.imports: list[str] = [] + self.scopes: dict[str, LeanScope] = {} + self.scan = LeanFile() + # constructor/field FQN -> node id of the structure or inductive + self.aliases: dict[str, str] = {} + self.private_ids: set[str] = set() + self.protected_ids: set[str] = set() + self.relative_path = Path(self._relative_path()).as_posix() + if is_skipped_lean_file(self.relative_path): + return + try: + self._analyze() + except Exception as e: # noqa: BLE001 — a broken file must not abort the sweep + logger.error(f"Error parsing Lean file {self.file_path}: {e}") + + def _relative_path(self) -> str: + if self.repo_path: + try: + return os.path.relpath(str(self.file_path), self.repo_path) + except ValueError: + return str(self.file_path) + return str(self.file_path) + + def _component_id(self, fqn: str) -> str: + return f"{self.relative_path}::{fqn}" + + def _analyze(self) -> None: + masked = mask_source(self.content) + self.scan = _scan_file(masked) + decls, self.imports = self.scan.declarations, self.scan.imports + source_lines = self.content.split("\n") + by_fqn: dict[str, Node] = {} + owners: list[tuple[LeanDecl, Node]] = [] + for decl in decls: + if decl.fqn in by_fqn: + continue + node = self._add_node(decl, source_lines) + by_fqn[decl.fqn] = node + owners.append((decl, node)) + for alias in decl.aliases: + self.aliases.setdefault(alias, node.id) + if not owners: + return + + tree = _lean_parser().parse(self.content.encode("utf8")) + masked_bytes = masked.text.encode("utf8") + refs, tree_locals = _collect_tree_names(tree.root_node, masked_bytes) + line_owner: dict[int, tuple[LeanDecl, Node]] = {} + for decl, node in owners: + for line in range(decl.keyword_line, decl.end + 1): + line_owner.setdefault(line, (decl, node)) + + locals_by_decl: dict[str, set[str]] = {} + masked_lines = masked.text.split("\n") + for decl, node in owners: + text = "\n".join(masked_lines[decl.keyword_line : decl.end + 1]) + locals_by_decl[node.id] = _text_locals(text) | set(decl.variables) + for line, name in tree_locals: + if line in line_owner: + locals_by_decl[line_owner[line][1].id].add(name) + + seen: set[tuple[str, str]] = set() + for line, raw in refs: + owner = line_owner.get(line) + if owner is None: + continue + decl, node = owner + name = normalize_reference(raw) + if name is None or name.split(".")[0] in locals_by_decl[node.id]: + continue + if line == decl.keyword_line and name == decl.fqn.split(".")[-1]: + continue # the declaration's own name + key = (node.id, name) + if key in seen: + continue + seen.add(key) + self.call_relationships.append( + CallRelationship(caller=node.id, callee=name, call_line=line + 1) + ) + + def _add_node(self, decl: LeanDecl, source_lines: list[str]) -> Node: + component_id = self._component_id(decl.fqn) + component_type = COMPONENT_TYPES.get(decl.kind, "function") + node_type = "theorem" if decl.kind == "lemma" else decl.kind + node = Node( + id=component_id, + name=decl.fqn, + component_type=component_type, + file_path=str(self.file_path), + relative_path=self.relative_path, + source_code="\n".join(source_lines[decl.start : decl.end + 1]), + start_line=decl.start + 1, + end_line=decl.end + 1, + has_docstring=bool(decl.docstring), + docstring=decl.docstring, + node_type=node_type, + base_classes=decl.bases or None, + display_name=f"{node_type} {decl.fqn}", + component_id=component_id, + language="lean", + qualified_name=decl.fqn, + ) + self.nodes.append(node) + self.scopes[component_id] = LeanScope( + tuple(decl.namespaces), tuple(decl.opens), decl.keyword_line + ) + if decl.private: + self.private_ids.add(component_id) + if decl.protected: + self.protected_ids.add(component_id) + return node + + +def analyze_lean_file( + file_path: str, content: str, repo_path: str | None = None +) -> tuple[list[Node], list[CallRelationship]]: + analyzer = TreeSitterLeanAnalyzer(file_path, content, repo_path) + return analyzer.nodes, analyzer.call_relationships diff --git a/codewiki/src/be/dependency_analyzer/analyzers/lean_resolution.py b/codewiki/src/be/dependency_analyzer/analyzers/lean_resolution.py new file mode 100644 index 00000000..81aedc42 --- /dev/null +++ b/codewiki/src/be/dependency_analyzer/analyzers/lean_resolution.py @@ -0,0 +1,264 @@ +"""Repository-local Lean name resolution, including scoped opens and exports. + +Commands are bound in import/source order. Bindings retain their defining file +and position: importing a target alone must not expose an unrelated export. +No type elaboration, external declarations, or notation expansion is attempted. +""" + +from collections import defaultdict +from dataclasses import dataclass + +from codewiki.src.be.dependency_analyzer.analyzers.lean import ( + LeanFile, + LeanOpen, + LeanScope, + _qualify, +) +from codewiki.src.be.dependency_analyzer.models.core import Node + +NO_MATCH = object() + + +@dataclass(frozen=True) +class _Binding: + target: str + origin: str + line: int + protected: bool = False + exported: bool = False + + +def _prefixes(name: str) -> list[str]: + parts = name.split(".") if name else [] + return [".".join(parts[:end]) for end in range(len(parts), 0, -1)] + [""] + + +class LeanNameResolver: + def __init__( + self, + functions: dict[str, Node], + files: dict[str, LeanFile], + scopes: dict[str, LeanScope], + aliases: dict[str, list[str]], + private: set[str], + protected: set[str], + ): + self.functions = functions + self.files = files + self.scopes = scopes + self.private = private + self.index: dict[str, list[_Binding]] = defaultdict(list) + # Earliest introduction per file is enough to test command visibility. + self.namespaces: dict[str, dict[str, int]] = defaultdict(dict) + self.modules: dict[str, set[str]] = defaultdict(set) + self.visible: dict[str, frozenset[str]] = {} + self.open_targets: dict[tuple[str, int], tuple[str, ...]] = {} + + for rel, scanned in files.items(): + parts = rel.removesuffix(".lean").split("/") + for start in range(len(parts)): + self.modules[".".join(parts[start:])].add(rel) + for line, namespace in scanned.namespaces: + self._add_namespace(namespace, rel, line) + + # Declaration names are indexed as before; directive binding additionally + # filters by position so a later namespace cannot change an earlier open. + for node_id, node in functions.items(): + if node.language != "lean" or not node.qualified_name: + continue + scope = scopes.get(node_id) + line = scope.line if scope else node.start_line - 1 + self._add_binding( + node.qualified_name, + _Binding(node_id, node.relative_path, line, node_id in protected), + ) + for name, targets in aliases.items(): + for target in targets: + node = functions[target] + scope = scopes.get(target) + self._add_binding( + name, + _Binding( + target, + node.relative_path, + scope.line if scope else node.start_line - 1, + target in protected, + ), + ) + + done: set[str] = set() + visiting: set[str] = set() + + def bind_file(rel: str) -> None: + if rel in done or rel in visiting: + return # invalid import cycles must not abort repository analysis + visiting.add(rel) + scanned = files[rel] + for module in scanned.imports: + for dependency in sorted(self.modules.get(module, ())): + bind_file(dependency) + events = [(opening.line, 0, i) for i, opening in enumerate(scanned.opens)] + events += [(export.line, 1, i) for i, export in enumerate(scanned.exports)] + for _, kind, i in sorted(events): + if kind == 0: + self._bind_open(rel, i) + else: + export = scanned.exports[i] + namespaces = self._namespace_targets( + export.namespace, export.destination, export.opens, rel, export.line + ) + for name in export.names: + targets = { + binding.target + for namespace in namespaces + for binding in self._matches( + _qualify(namespace, name), rel, export.line + ) + } + if len(targets) == 1: + self._add_binding( + _qualify(export.destination, name), + _Binding(next(iter(targets)), rel, export.line, exported=True), + ) + visiting.remove(rel) + done.add(rel) + + for rel in sorted(files): + bind_file(rel) + + def _add_binding(self, name: str, binding: _Binding) -> None: + if binding not in self.index[name]: + self.index[name].append(binding) + for namespace in _prefixes(name)[1:-1]: + self._add_namespace(namespace, binding.origin, binding.line) + + def _add_namespace(self, namespace: str, origin: str, line: int) -> None: + introductions = self.namespaces[namespace] + introductions[origin] = min(line, introductions.get(origin, line)) + + def _visible_files(self, rel: str) -> frozenset[str]: + if rel not in self.visible: + seen = {rel} + queue = [rel] + while queue: + scanned = self.files.get(queue.pop()) + if scanned is None: + continue + for module in scanned.imports: + for dependency in self.modules.get(module, ()): + if dependency not in seen: + seen.add(dependency) + queue.append(dependency) + self.visible[rel] = frozenset(seen) + return self.visible[rel] + + def _matches( + self, name: str, rel: str, line: int, *, declarations_after: bool = False + ) -> list[_Binding]: + visible = self._visible_files(rel) + return [ + binding + for binding in self.index.get(name, ()) + if binding.origin in visible + and self.functions[binding.target].relative_path in visible + and ( + binding.origin != rel + or binding.line <= line + or declarations_after + and not binding.exported + ) + and ( + binding.target not in self.private + or self.functions[binding.target].relative_path == rel + ) + ] + + @staticmethod + def _opened_name(opening: LeanOpen, name: str) -> str | None: + if opening.mode == "scoped": + return None + if opening.mode == "select": + return name if name in opening.names else None + if opening.mode == "hiding": + return None if name in opening.names else name + if opening.mode == "renaming": + return next((source for source, alias in opening.renames if alias == name), None) + return name + + def _namespace_targets( + self, name: str, context: str, opens: tuple[int, ...], rel: str, line: int + ) -> tuple[str, ...]: + visible = self._visible_files(rel) + + def exists(namespace: str) -> bool: + return any( + origin in visible and (origin != rel or position <= line) + for origin, position in self.namespaces.get(namespace, {}).items() + ) + + if name.startswith("_root_."): + target = name[len("_root_.") :] + return (target,) if exists(target) else () + candidates = {_qualify(prefix, name) for prefix in _prefixes(context)} + for opening_id in opens: + opening = self.files[rel].opens[opening_id] + suffix = self._opened_name(opening, name) + if suffix is not None: + candidates.update( + _qualify(namespace, suffix) + for namespace in self.open_targets.get((rel, opening_id), ()) + ) + return tuple(sorted(candidate for candidate in candidates if exists(candidate))) + + def _bind_open(self, rel: str, opening_id: int) -> None: + opening = self.files[rel].opens[opening_id] + targets = self._namespace_targets( + opening.namespace, opening.context, opening.preceding, rel, opening.line + ) + # `hiding` and `renaming` require one unambiguous namespace. + if opening.mode in ("hiding", "renaming") and len(targets) != 1: + targets = () + self.open_targets[rel, opening_id] = targets + + def resolve(self, callee: str, caller: Node) -> str | None: + """Find the longest constant prefix, preserving ambiguity as a stop.""" + parts = callee.split(".") + for end in range(len(parts), 0, -1): + found = self._resolve_name(".".join(parts[:end]), caller) + if found is not NO_MATCH: + return found + return None + + def _resolve_name(self, name: str, caller: Node): + rel = caller.relative_path + scope = self.scopes.get(caller.id, LeanScope((), (), caller.start_line - 1)) + + def matches(fqn: str) -> list[_Binding]: + # Keep the existing declaration lookup behavior (including mutual + # definitions), but never expose an export before its command. + return self._matches(fqn, rel, scope.line, declarations_after=True) + + def targets(fqn: str) -> set[str]: + return {binding.target for binding in matches(fqn)} + + def unique(found: set[str]) -> str | None: + return next(iter(found)) if len(found) == 1 and caller.id not in found else None + + if name.startswith("_root_."): + found = targets(name[len("_root_.") :]) + return unique(found) if found else NO_MATCH + for prefix in (*scope.namespaces, ""): + found = targets(_qualify(prefix, name)) + if found: + return unique(found) + opened: set[str] = set() + for opening_id in scope.opens: + opening = self.files[rel].opens[opening_id] + suffix = self._opened_name(opening, name) + if suffix is None: + continue + for namespace in self.open_targets.get((rel, opening_id), ()): + for binding in matches(_qualify(namespace, suffix)): + if opening.mode in ("select", "renaming") or not binding.protected: + opened.add(binding.target) + return unique(opened) if opened else NO_MATCH diff --git a/codewiki/src/be/dependency_analyzer/analyzers/vue.py b/codewiki/src/be/dependency_analyzer/analyzers/vue.py index 1a17f1ab..4934f72c 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/vue.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/vue.py @@ -37,7 +37,7 @@ from functools import lru_cache from pathlib import Path -from tree_sitter_language_pack import get_parser +from codewiki.src.be.dependency_analyzer.utils.grammars import get_pack_parser from codewiki.src.be.dependency_analyzer.models.core import CallRelationship, Node @@ -107,7 +107,7 @@ @lru_cache(maxsize=None) def _parser(language: str): - return get_parser(language) + return get_pack_parser(language) def _is_sfc(target: str) -> bool: diff --git a/codewiki/src/be/dependency_analyzer/analyzers/zig.py b/codewiki/src/be/dependency_analyzer/analyzers/zig.py index cd5f1b7c..feb60e2e 100644 --- a/codewiki/src/be/dependency_analyzer/analyzers/zig.py +++ b/codewiki/src/be/dependency_analyzer/analyzers/zig.py @@ -47,7 +47,7 @@ from functools import lru_cache from pathlib import Path -from tree_sitter_language_pack import get_parser +from codewiki.src.be.dependency_analyzer.utils.grammars import get_pack_parser from codewiki.src.be.dependency_analyzer.models.core import CallRelationship, Node from codewiki.src.be.dependency_analyzer.utils.file_selection import is_skipped_zig_file @@ -107,7 +107,7 @@ def _build_zig_modules(repo_path: str) -> dict[str, str]: def _zig_parser(): - return get_parser("zig") + return get_pack_parser("zig") def _alias_chain(var_decl) -> tuple[tuple[str, str], tuple[str, ...]] | None: diff --git a/codewiki/src/be/dependency_analyzer/ast_parser.py b/codewiki/src/be/dependency_analyzer/ast_parser.py index e5f67c40..8e626112 100644 --- a/codewiki/src/be/dependency_analyzer/ast_parser.py +++ b/codewiki/src/be/dependency_analyzer/ast_parser.py @@ -230,6 +230,7 @@ def _file_to_module_path(self, file_path: str) -> str: ".dart", ".go", ".zig", + ".lean", ".vue", ] for ext in extensions: diff --git a/codewiki/src/be/dependency_analyzer/utils/file_selection.py b/codewiki/src/be/dependency_analyzer/utils/file_selection.py index 02fa5942..6793a970 100644 --- a/codewiki/src/be/dependency_analyzer/utils/file_selection.py +++ b/codewiki/src/be/dependency_analyzer/utils/file_selection.py @@ -5,6 +5,7 @@ PHP_TEMPLATE_PATTERNS = {".blade.php", ".phtml", ".twig.php"} PHP_TEMPLATE_DIRECTORIES = {"views", "templates", "resources/views"} ZIG_SKIPPED_DIRS = {"zig-cache", ".zig-cache", "zig-out"} +LEAN_SKIPPED_DIRS = {".lake", "lake-packages"} def is_skipped_go_file(relative_path: str) -> bool: @@ -17,6 +18,11 @@ def is_skipped_zig_file(relative_path: str) -> bool: return any(part in ZIG_SKIPPED_DIRS for part in Path(relative_path).parts[:-1]) +def is_skipped_lean_file(relative_path: str) -> bool: + """Lake's build folder and its package checkouts are not project code.""" + return any(part in LEAN_SKIPPED_DIRS for part in Path(relative_path).parts[:-1]) + + def is_php_template_file(file_path: str) -> bool: file_str = str(file_path) return any(file_str.endswith(pattern) for pattern in PHP_TEMPLATE_PATTERNS) or any( @@ -32,6 +38,8 @@ def is_skipped_source_file(file_info: dict, base_dir: str) -> bool: return is_skipped_go_file(file_info["path"]) if language == "zig": return is_skipped_zig_file(file_info["path"]) + if language == "lean": + return is_skipped_lean_file(file_info["path"]) if language == "php": return is_php_template_file(str(Path(base_dir) / file_info["path"])) return False diff --git a/codewiki/src/be/dependency_analyzer/utils/grammars.py b/codewiki/src/be/dependency_analyzer/utils/grammars.py new file mode 100644 index 00000000..bdbae313 --- /dev/null +++ b/codewiki/src/be/dependency_analyzer/utils/grammars.py @@ -0,0 +1,64 @@ +"""Parsers from ``tree-sitter-language-pack``. + +Pack 1.x does not ship its grammars in the wheel. It downloads each grammar +the first time it is loaded and caches it under ``cache_dir()``. Without +network access that first load fails with a ``DownloadError`` that names a +URL but not the cause or the fix. The per-file ``except`` blocks of the +analyzers would also swallow it, and the language would silently drop out of +the graph. So the analysis checks the grammars it needs once, before it +parses any file, and stops with one clear message. +""" + +from functools import lru_cache + +import tree_sitter_language_pack as tslp + +# Pack grammars each analyzed language loads. ``docker/Dockerfile`` downloads +# the same set at build time (``tests/test_pack_grammars.py`` checks this). +PACK_GRAMMARS_BY_LANGUAGE: dict[str, tuple[str, ...]] = { + "dart": ("dart",), + "go": ("go",), + "lean": ("lean",), + "vue": ("vue", "javascript", "typescript"), + "zig": ("zig",), +} + +PACK_GRAMMARS: tuple[str, ...] = tuple( + sorted({g for grammars in PACK_GRAMMARS_BY_LANGUAGE.values() for g in grammars}) +) + + +class GrammarUnavailableError(RuntimeError): + """A pack grammar is not cached and could not be downloaded.""" + + def __init__(self, grammar: str, cause: Exception): + try: + cache = tslp.cache_dir() + except Exception: # noqa: BLE001 — the path only improves the message + cache = "the tree-sitter-language-pack cache" + names = ", ".join(repr(g) for g in PACK_GRAMMARS) + super().__init__( + f"The tree-sitter grammar '{grammar}' is not in {cache} and could not be " + f"downloaded ({cause}). tree-sitter-language-pack downloads each grammar on " + "first use, so the first analysis of these files needs network access. " + "To prepare a machine without network access, run this once on a machine " + "with access and copy the cache folder: " + f'python -c "import tree_sitter_language_pack as t; t.download([{names}])"' + ) + self.grammar = grammar + + +@lru_cache(maxsize=None) +def get_pack_parser(grammar: str): + """Return the pack parser for ``grammar``, with a clear error when it is unavailable.""" + try: + return tslp.get_parser(grammar) + except tslp.DownloadError as e: + raise GrammarUnavailableError(grammar, e) from e + + +def ensure_pack_grammars(languages) -> None: + """Load every pack grammar that ``languages`` need, or raise ``GrammarUnavailableError``.""" + for language in sorted(set(languages)): + for grammar in PACK_GRAMMARS_BY_LANGUAGE.get(language, ()): + get_pack_parser(grammar) diff --git a/codewiki/src/be/dependency_analyzer/utils/patterns.py b/codewiki/src/be/dependency_analyzer/utils/patterns.py index 6692df36..84c4ffe2 100644 --- a/codewiki/src/be/dependency_analyzer/utils/patterns.py +++ b/codewiki/src/be/dependency_analyzer/utils/patterns.py @@ -70,6 +70,11 @@ "zig-cache/", "zig-out", "zig-out/", + # Lean: Lake build folder and package checkouts + ".lake", + ".lake/", + "lake-packages", + "lake-packages/", # IDEs and editors / Java ".project", # C/C++ @@ -184,6 +189,7 @@ "*.sc", "*.dart", "*.zig", + "*.lean", "*.vue", "*.clj", "*.hs", @@ -250,6 +256,9 @@ "Procfile", "go.mod", "build.zig.zon", + "lakefile.toml", + "lake-manifest.json", + "lean-toolchain", ".editorconfig", ".pre-commit-config.yaml", ".eslintrc*", @@ -321,6 +330,7 @@ ".dart": "dart", ".cs": "csharp", ".zig": "zig", + ".lean": "lean", ".vue": "vue", } @@ -556,6 +566,15 @@ "const {name} = enum", "const {name} = union", ], + "lean": [ + "def {name}", + "theorem {name}", + "lemma {name}", + "abbrev {name}", + "structure {name}", + "class {name}", + "inductive {name}", + ], "c": ["void {name}", "int {name}", "{name}("], "cpp": ["void {name}", "int {name}", "{name}("], "php": [ diff --git a/codewiki/src/be/prompt_template.py b/codewiki/src/be/prompt_template.py index db16a55d..f649cead 100644 --- a/codewiki/src/be/prompt_template.py +++ b/codewiki/src/be/prompt_template.py @@ -400,6 +400,7 @@ ".dart": "dart", ".go": "go", ".zig": "zig", + ".lean": "lean", ".vue": "vue", ".php": "php", ".phtml": "php", diff --git a/docker/Dockerfile b/docker/Dockerfile index d0a70d16..38fca3c2 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -18,6 +18,10 @@ COPY requirements.txt . # Install Python dependencies RUN pip install --no-cache-dir -r requirements.txt +# tree-sitter-language-pack 1.x downloads each grammar on first use. Fetch the +# ones the analyzers load at build time, so the image runs offline. +RUN python -c "import tree_sitter_language_pack as t; t.download(['dart', 'go', 'javascript', 'lean', 'typescript', 'vue', 'zig'])" + # Copy application code COPY codewiki ./codewiki COPY img ./img diff --git a/guides/artifact-aware-generation.md b/guides/artifact-aware-generation.md index 90db8486..17d7212b 100644 --- a/guides/artifact-aware-generation.md +++ b/guides/artifact-aware-generation.md @@ -20,8 +20,8 @@ Files are classified by name, extension, and directory. Nine classes: | Class | Examples | | --- | --- | -| `manifest` | `package.json`, `pyproject.toml`, `setup.py`, `setup.cfg`, `Cargo.toml`, `go.mod`, `build.zig.zon`, `Gemfile`, `pnpm-workspace.yaml` | -| `build` | `Makefile`, `CMakeLists.txt`, `configure.ac`, `Makefile.am`, `m4/*.m4`, `build.zig`, `*.gradle`, `BUILD.gn`, `*.gni`, `DEPS`, `Rakefile`, `rakelib/*.rake`, `build/` | +| `manifest` | `package.json`, `pyproject.toml`, `setup.py`, `setup.cfg`, `Cargo.toml`, `go.mod`, `build.zig.zon`, `lake-manifest.json`, `lean-toolchain`, `Gemfile`, `pnpm-workspace.yaml` | +| `build` | `Makefile`, `CMakeLists.txt`, `configure.ac`, `Makefile.am`, `m4/*.m4`, `build.zig`, `lakefile.lean`, `lakefile.toml`, `*.gradle`, `BUILD.gn`, `*.gni`, `DEPS`, `Rakefile`, `rakelib/*.rake`, `build/` | | `container` | `Dockerfile*`, `Containerfile`, `docker-compose*.yml`, files under `docker/` | | `ci` | `.github/workflows/*.yml`, `.github/actions/*/action.yml`, `.gitlab-ci.yml`, `Jenkinsfile`, `ci/`, `.circleci/` | | `packaging` | `debian/`, `rpm/`, `pkg/`, `packages/`, `*.spec`, `*.service`, install and lifecycle scripts | diff --git a/guides/development.md b/guides/development.md index 4c88fb32..73a1bcd5 100644 --- a/guides/development.md +++ b/guides/development.md @@ -122,7 +122,12 @@ Analyzers are standalone classes, one file per language. Follow 1. **Grammar.** Add the `tree-sitter-` package to `pyproject.toml` and `requirements.txt`, or take the grammar from `tree-sitter-language-pack` (already a dependency) with - `get_parser("")`, as `analyzers/dart.py` and `analyzers/go.py` do. + `get_pack_parser("")` from `utils/grammars.py`, as + `analyzers/dart.py` and `analyzers/go.py` do. Add the language and its + grammars to `PACK_GRAMMARS_BY_LANGUAGE` in the same file. The pack (1.x) downloads a grammar on first + use, so also add the name to the `t.download([...])` list in + `docker/Dockerfile`. `tests/test_pack_grammars.py` checks that the two + lists match. 2. **Analyzer.** Create `analyzers/.py` with a `TreeSitterAnalyzer` class that extracts components and relationships, plus a module-level `analyze__file(...)` function. @@ -138,8 +143,8 @@ Analyzers are standalone classes, one file per language. Follow snippets that check component and edge extraction. See `tests/test_scala_analyzer.py`, `tests/test_ruby_analyzer.py`, `tests/test_rust_analyzer.py`, `tests/test_dart_analyzer.py`, - `tests/test_go_analyzer.py`, `tests/test_zig_analyzer.py`, and - `tests/test_vue_analyzer.py`. + `tests/test_go_analyzer.py`, `tests/test_zig_analyzer.py`, + `tests/test_lean_analyzer.py`, and `tests/test_vue_analyzer.py`. 6. **Docs.** Add the language to the README list. ## Tests and lint diff --git a/pyproject.toml b/pyproject.toml index a407109a..a96d9216 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,7 @@ dependencies = [ "GitPython>=3.1.40", "Jinja2>=3.1.6", "tree-sitter>=0.25", - "tree-sitter-language-pack>=0.8.0,<0.9.0", + "tree-sitter-language-pack>=1.23.1,<2.0.0", "tree-sitter-python>=0.23.6", "tree-sitter-java>=0.23.5", "tree-sitter-javascript>=0.21.4", diff --git a/requirements.txt b/requirements.txt index 7551bf8a..1d764c05 100644 --- a/requirements.txt +++ b/requirements.txt @@ -148,7 +148,7 @@ tree-sitter-embedded-template==0.23.2 tree-sitter-java==0.23.5 tree-sitter-javascript==0.21.4 tree-sitter-kotlin==1.1.0 -tree-sitter-language-pack==0.8.0 +tree-sitter-language-pack==1.23.1 tree-sitter-python==0.23.6 tree-sitter-ruby==0.23.1 tree-sitter-rust==0.23.2 diff --git a/tests/test_artifact_analyzer.py b/tests/test_artifact_analyzer.py index 6ba521a2..8c73e8bf 100644 --- a/tests/test_artifact_analyzer.py +++ b/tests/test_artifact_analyzer.py @@ -130,6 +130,11 @@ def _artifact_nodes(components: dict[str, Node]) -> dict[str, Node]: ("build.zig", "build"), ("build.zig.zon", "manifest"), ("src/main.zig", None), + ("lakefile.lean", "build"), + ("lakefile.toml", "build"), + ("lake-manifest.json", "manifest"), + ("lean-toolchain", "manifest"), + ("Geo/Point.lean", None), ("src/App.vue", None), ("pytest.ini", "test_infra"), ("conftest.py", "test_infra"), diff --git a/tests/test_lean_analyzer.py b/tests/test_lean_analyzer.py new file mode 100644 index 00000000..49709269 --- /dev/null +++ b/tests/test_lean_analyzer.py @@ -0,0 +1,739 @@ +"""Tests for the Lean 4 analyzer: header-scanner structure, tree-sitter +references, and exact-name resolution in CallGraphAnalyzer.""" + +from pathlib import Path + +import pytest + +pytest.importorskip("tree_sitter_language_pack") + +from codewiki.src.be.dependency_analyzer.analysis.call_graph_analyzer import ( # noqa: E402 + CallGraphAnalyzer, +) +from codewiki.src.be.dependency_analyzer.analyzers.lean import ( # noqa: E402 + analyze_lean_file, + mask_source, + normalize_reference, + scan_headers, +) +from codewiki.src.be.dependency_analyzer.utils.file_selection import is_skipped_lean_file # noqa: E402 + +POINT = """\ +/-! Geometry primitives. -/ +namespace Geo + +/-- A point in the plane. +Coordinates are naturals. -/ +structure Point where + x : Nat + y : Nat + deriving Repr + +def Point.add (p q : Point) : Point := + let s := p.x + q.x + { x := s, y := p.y + q.y } + +@[simp] +theorem Point.add_zero (p : Point) : Point.add p ⟨0, 0⟩ = p := by + cases p + simp [Point.add, Nat.add_zero] + +inductive Shape where + | dot (p : Point) + | segment (a b : Point) + +def Shape.start : Shape → Point + | .dot p => p + | .segment a _ => a + +private def secret : Nat := 1 + +instance : Add Point := ⟨Point.add⟩ + +section Ops +open Point in +def double (p : Point) : Point := add p p +end Ops + +end Geo +""" + +USE = """\ +import Geo.Point + +open Geo + +def origin : Point := Point.mk 0 0 + +def first (s : Shape) : Point := Shape.start s + +def shapes : List Shape := [Shape.dot origin, .segment origin origin] + +def leak : Nat := secret + +lemma origin_x : origin.x = 0 := rfl + +def report : IO Unit := do + let p ← pure origin + IO.println s!"{p.x}" + #eval first (Shape.dot p) + +def after : Point := double origin + +def external := Nat.add_comm 1 2 +""" + +ISLAND = """\ +def lonely : Nat := Geo.secret + +def notImported := Geo.Point.add +""" + + +def _write(root: Path, rel: str, text: str) -> str: + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return rel + + +@pytest.fixture +def lean_repo(tmp_path: Path): + files = [ + _write(tmp_path, "src/Geo/Point.lean", POINT), + _write(tmp_path, "src/Geo/Use.lean", USE), + _write(tmp_path, "src/Island.lean", ISLAND), + _write(tmp_path, ".lake/packages/dep/Dep.lean", "def dep := 1\n"), + ] + analyzer = CallGraphAnalyzer() + analyzer.analyze_code_files([{"path": f, "language": "lean"} for f in files], str(tmp_path)) + edges = {(r.caller, r.callee) for r in analyzer.call_relationships} + return analyzer.functions, edges + + +P = "src/Geo/Point.lean::" +U = "src/Geo/Use.lean::" +I = "src/Island.lean::" # noqa: E741 + + +def test_declarations_namespaces_and_types(lean_repo) -> None: + functions, _ = lean_repo + kinds = {f.qualified_name: (f.component_type, f.node_type) for f in functions.values()} + assert kinds["Geo.Point"] == ("struct", "structure") + assert kinds["Geo.Point.add"] == ("function", "def") + assert kinds["Geo.Point.add_zero"] == ("function", "theorem") + assert kinds["Geo.Shape"] == ("class", "inductive") + assert kinds["Geo.double"] == ("function", "def") + assert kinds["origin_x"] == ("function", "theorem") # `lemma` is a theorem + assert any(name.startswith("Geo.instance_L") for name in kinds) + assert all(f.language == "lean" for f in functions.values()) + + +def test_spans_docstrings_and_attributes(lean_repo) -> None: + functions, _ = lean_repo + point = functions[P + "Geo.Point"] + assert point.docstring == "A point in the plane.\nCoordinates are naturals." + assert (point.start_line, point.end_line) == (6, 9) + theorem = functions[P + "Geo.Point.add_zero"] + assert theorem.start_line == 15 # the `@[simp]` line + assert "simp [Point.add" in (theorem.source_code or "") + assert not functions[P + "Geo.Point.add"].docstring + + +def test_lake_folders_are_skipped(lean_repo) -> None: + functions, _ = lean_repo + assert not any(".lake" in f.relative_path for f in functions.values()) + assert is_skipped_lean_file(".lake/packages/dep/Dep.lean") + assert is_skipped_lean_file("lake-packages/std/Std.lean") + assert not is_skipped_lean_file("src/Geo/Point.lean") + + +def test_same_file_references_resolve_in_the_namespace(lean_repo) -> None: + _, edges = lean_repo + assert (P + "Geo.Point.add", P + "Geo.Point") in edges + assert (P + "Geo.Point.add_zero", P + "Geo.Point.add") in edges + assert (P + "Geo.Shape", P + "Geo.Point") in edges # constructor argument types + assert (P + "Geo.Shape.start", P + "Geo.Shape") in edges + + +def test_open_in_applies_to_the_next_declaration(lean_repo) -> None: + functions, edges = lean_repo + assert (P + "Geo.double", P + "Geo.Point.add") in edges # `add` through `open Point in` + instance = next(i for i in functions if i.startswith(P + "Geo.instance_L")) + assert (instance, P + "Geo.Point.add") in edges + + +def test_imports_open_and_member_aliases(lean_repo) -> None: + _, edges = lean_repo + assert (U + "origin", P + "Geo.Point") in edges # `Point.mk` is the structure + assert (U + "first", P + "Geo.Shape.start") in edges + assert (U + "shapes", P + "Geo.Shape") in edges # `Shape.dot` is the inductive + assert (U + "shapes", U + "origin") in edges + assert (U + "origin_x", U + "origin") in edges + assert (U + "after", P + "Geo.double") in edges + + +def test_error_regions_still_give_references(lean_repo) -> None: + _, edges = lean_repo + assert (U + "report", U + "origin") in edges + assert (U + "report", U + "first") in edges + assert (U + "report", P + "Geo.Shape") in edges + + +def test_locals_externals_and_recursion_give_no_edges(lean_repo) -> None: + functions, edges = lean_repo + callees = {callee for caller, callee in edges if caller == P + "Geo.Point.add"} + assert callees == {P + "Geo.Point"} # p, q, s, p.x are locals + assert not any(caller == U + "external" for caller, _ in edges) # Nat.add_comm + assert all(caller != callee for caller, callee in edges) + assert all(callee in functions for _, callee in edges) + + +def test_private_and_unimported_names_are_invisible(lean_repo) -> None: + _, edges = lean_repo + assert not any(caller == U + "leak" for caller, _ in edges) # private in another file + assert not any(caller.startswith(I) for caller, _ in edges) # Island imports nothing + + +def test_ambiguous_name_is_dropped(tmp_path: Path) -> None: + files = [ + _write(tmp_path, "A.lean", "namespace X\ndef f := 1\nend X\n"), + _write(tmp_path, "B.lean", "namespace Y\ndef f := 2\nend Y\n"), + _write(tmp_path, "C.lean", "import A\nimport B\nopen X Y\ndef g := f\ndef h := X.f\n"), + ] + analyzer = CallGraphAnalyzer() + analyzer.analyze_code_files([{"path": f, "language": "lean"} for f in files], str(tmp_path)) + edges = {(r.caller, r.callee) for r in analyzer.call_relationships} + assert not any(caller == "C.lean::g" for caller, _ in edges) + assert ("C.lean::h", "A.lean::X.f") in edges + + +def test_root_prefix_mutual_and_nested_namespaces(tmp_path: Path) -> None: + source = """\ +namespace A.B +def _root_.top := 1 +mutual + def even : Nat → Bool + | 0 => true + | n + 1 => odd n + def odd : Nat → Bool + | 0 => false + | n + 1 => even n +end +end A.B +def last := top +""" + nodes, _ = analyze_lean_file(str(tmp_path / "M.lean"), source, repo_path=str(tmp_path)) + names = {n.qualified_name for n in nodes} + assert names == {"top", "A.B.even", "A.B.odd", "last"} + + +def test_masking_keeps_lines_and_byte_offsets() -> None: + source = "def a := \"-- not a comment\" -- é comment\n/- block /- nested -/ -/ def b := 'x'\n" + masked = mask_source(source) + assert masked.text.count("\n") == source.count("\n") + assert len(masked.text.encode("utf8")) == len(source.encode("utf8")) + assert "not a comment" not in masked.text and "nested" not in masked.text + assert "def b" in masked.text + + +def test_interpolated_strings_and_primes_are_masked_correctly() -> None: + masked = mask_source('def c := s!"x {y} -- z"\ndef d\' := r#"q"# ++ a\'\n') + assert "-- z" not in masked.text and '"q"' not in masked.text + assert "def d' :=" in masked.text and "a'" in masked.text + + +def test_header_scanner_without_the_parse_tree() -> None: + decls, imports = scan_headers( + mask_source("import Mathlib.Data.Nat\nimport Foo\n#eval 1\ntheorem t : True := trivial\n") + ) + assert imports == ["Mathlib.Data.Nat", "Foo"] + assert [d.fqn for d in decls] == ["t"] + + +@pytest.mark.parametrize( + "raw, expected", + [ + ("Foo.bar", "Foo.bar"), + ("@Foo.bar", "Foo.bar"), + ("«weird name».x", "weird name.x"), + ("p.1", "p"), + (".dot", None), + ("fun", None), + ], +) +def test_normalize_reference(raw: str, expected: str | None) -> None: + assert normalize_reference(raw) == expected + + +def test_empty_and_broken_files(tmp_path: Path) -> None: + assert analyze_lean_file(str(tmp_path / "E.lean"), "", repo_path=str(tmp_path)) == ([], []) + nodes, _ = analyze_lean_file( + str(tmp_path / "B.lean"), "def broken (x : := ⟨\n#eval ¬¬\n", repo_path=str(tmp_path) + ) + assert [n.qualified_name for n in nodes] == ["broken"] + + +def _analyze_sources(tmp_path: Path, sources: dict[str, str], *, reverse: bool = False): + files = [_write(tmp_path, rel, source) for rel, source in sources.items()] + analyzer = CallGraphAnalyzer() + analyzer.analyze_code_files( + [{"path": rel, "language": "lean"} for rel in (reversed(files) if reverse else files)], + str(tmp_path), + ) + return analyzer, {(r.caller, r.callee) for r in analyzer.call_relationships} + + +@pytest.mark.parametrize( + "binder", + [ + "fun (x : Foo) => x", + "λ (x : Foo) => x", + "fun x : Foo => x", + "fun (x y : Foo) => x", + "fun {x : Foo} => x", + "fun ⦃x : Foo⦄ => x", + "fun [inst : Foo] => inst", + "fun [Foo] => 1", + "∀ (x : Foo), x = x", + "∃ (x : Foo), x = x", + "∃! (x : Foo), x = x", + "Σ (x : Foo), Foo", + "Π (x : Foo), Foo", + ], +) +def test_typed_binders_preserve_type_dependencies(tmp_path: Path, binder: str) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": f"structure Foo where\n value : Nat\ndef use := {binder}\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::Foo")} + + +def test_nested_binder_types_are_not_local_names(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "structure Foo where\n value : Nat\n" + "structure Bar where\n value : Nat\n" + "def use := fun (x : Foo × (List Bar)) (y : Bar := ⟨1⟩) => x\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::Foo"), ("M.lean::use", "M.lean::Bar")} + + +def test_untyped_binders_suppress_globals(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "def x := 1\ndef y := 2\ndef use := fun x y => x + y\n", + }, + ) + assert not edges + + +@pytest.mark.parametrize( + "header, ending", + [ + ("namespace A\nnamespace B", "end A.B"), + ("namespace A.B", "end A.B"), + ("section A\nsection B", "end A.B"), + ("section A.B", "end A.B"), + ], +) +def test_named_end_closes_matching_suffix(tmp_path: Path, header: str, ending: str) -> None: + analyzer, _ = _analyze_sources( + tmp_path, + { + "M.lean": f"{header}\ndef inside := 1\n{ending}\ndef outside := 2\n", + }, + ) + assert "M.lean::outside" in analyzer.functions + + +def test_partial_namespace_end_and_section_barrier(tmp_path: Path, caplog) -> None: + analyzer, _ = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B\nnamespace C\nend B.C\ndef first := 1\n" + "section\nnamespace D.E\nend A.D.E\ndef retained := 2\n" + "end D.E\nend\nend A\ndef last := 3\n", + }, + ) + assert set(analyzer.functions) == {"M.lean::A.first", "M.lean::A.D.E.retained", "M.lean::last"} + assert "Mismatched Lean end" in caplog.text + + +def test_section_variables_are_scoped_and_restored(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "def x := 1\ndef y := 2\nvariable (x : Nat)\n" + "section Outer.Inner\nvariable {y : Nat}\ndef inside := x + y\n" + "end Inner\ndef middle := x + y\nend Outer\ndef outside := x + y\n", + }, + ) + assert edges == {("M.lean::middle", "M.lean::y"), ("M.lean::outside", "M.lean::y")} + + +def test_namespace_variables_expire_and_types_are_not_binders(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "structure Foo where\n value : Nat\ndef x := 1\n" + "namespace A\nvariable\n (x : Foo)\n [Inhabited Foo]\n" + "def inside := (x, (⟨1⟩ : Foo))\nend A\ndef outside := x\n", + }, + ) + assert edges == {("M.lean::A.inside", "M.lean::Foo"), ("M.lean::outside", "M.lean::x")} + + +_OPEN_FIXTURE = "namespace A\ndef f := 1\ndef g := 2\nend A\nnamespace B\ndef g := 3\nend B\n" + + +@pytest.mark.parametrize("opening", ["open A (f)", "open A hiding g", "open scoped A"]) +def test_open_restrictions_do_not_create_ambiguity(tmp_path: Path, opening: str) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + f"{opening}\nopen B\ndef use := g\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::B.g")} + + +@pytest.mark.parametrize( + "opening, expected", + [ + ("open A (f)", {"A.f"}), + ("open A hiding g", {"A.f"}), + ("open scoped A", set()), + ("open A renaming f → h", {"A.f"}), + ("open A renaming f -> h, g -> k", {"A.f", "A.g"}), + ("open A\n renaming\n f → h,\n g → k", {"A.f", "A.g"}), + ("open A\n hiding g", {"A.f"}), + ("open A\n (f)", {"A.f"}), + ], +) +def test_open_modes_only_expose_the_selected_names( + tmp_path: Path, opening: str, expected: set +) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + f"{opening}\ndef use := (f, g, h, k)\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::" + name) for name in expected} + + +@pytest.mark.parametrize( + "opening, exposed", + [ + ("open A", False), + ("open A hiding other", False), + ("open A (secret)", True), + ("open A renaming secret → value", True), + ], +) +def test_protected_names_require_explicit_open(tmp_path: Path, opening: str, exposed: bool) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A\nprotected def secret := 1\ndef other := 2\nend A\n" + + opening + + "\ndef use := (secret, value)\n", + }, + ) + assert edges == ({("M.lean::use", "M.lean::A.secret")} if exposed else set()) + + +@pytest.mark.parametrize("same_line", [False, True]) +def test_restricted_open_in_expires(tmp_path: Path, same_line: bool) -> None: + separator = " " if same_line else "\n" + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + + "open A renaming f → value in" + + separator + + "def inside := value\ndef outside := value\n", + }, + ) + assert edges == {("M.lean::inside", "M.lean::A.f")} + + +def test_open_in_applies_to_a_namespace_block(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "open A in\nnamespace Inner\n" + "def inside := f\nend Inner\ndef outside := f\n", + }, + ) + assert edges == {("M.lean::Inner.inside", "M.lean::A.f")} + + +def test_open_uses_the_command_namespace(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A\ndef f := 1\nend A\n" + "namespace B.A\ndef f := 2\nend B.A\nopen A\n" + "namespace B\ndef use := f\nend B\n", + }, + ) + assert edges == {("M.lean::B.use", "M.lean::A.f")} + + +def test_relative_and_sequential_opens(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B.C\ndef f := 1\nend A.B.C\n" + "namespace A\nopen B C\ndef inside := f\nend A\n" + "def outside := f\n", + }, + ) + assert edges == {("M.lean::A.inside", "M.lean::A.B.C.f")} + + +def test_later_namespace_does_not_rebind_an_open(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A\ndef f := 1\nend A\nnamespace B\nopen A\n" + "namespace A\ndef f := 2\nend A\ndef use := f\nend B\n", + }, + ) + assert edges == {("M.lean::B.use", "M.lean::A.f")} + + +@pytest.mark.parametrize("reverse", [False, True]) +def test_exports_are_persistent_visible_aliases(tmp_path: Path, reverse: bool) -> None: + analyzer, edges = _analyze_sources( + tmp_path, + { + "Base.lean": "namespace A\ndef f := 1\ndef g := 2\nend A\n", + "Exports.lean": "import Base\nnamespace B\nsection\nexport A (f)\nend\nend B\n", + "Reexport.lean": "import Exports\nnamespace C\nexport B (f)\n" + "def inside := f\nend C\ndef outside := C.f\n", + "Use.lean": "import Reexport\nopen C\ndef use := (f, B.f, C.f)\n", + "Island.lean": "import Base\ndef missing := (B.f, C.f)\n", + }, + reverse=reverse, + ) + assert edges == { + ("Reexport.lean::C.inside", "Base.lean::A.f"), + ("Reexport.lean::outside", "Base.lean::A.f"), + ("Use.lean::use", "Base.lean::A.f"), + } + assert len(analyzer.functions) == 6 # aliases introduce no components + + +def test_exports_only_expose_selected_names_and_keep_command_context(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "namespace Outer\nopen A\n" + "namespace Inner\nexport _root_.A (f)\nend Inner\nend Outer\n" + "open Outer.Inner\ndef use := (f, g, Outer.Inner.f)\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::A.f")} + + +def test_root_and_constructor_exports_resolve_to_original_components(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A\ninductive Choice where\n | first\n | second\n" + "export Choice (first)\nend A\nexport A (first)\ndef use := first\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::A.Choice")} + + +def test_export_is_not_visible_before_its_command(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "def before := f\nexport A (f)\ndef after := f\n", + }, + ) + assert edges == {("M.lean::after", "M.lean::A.f")} + + +def test_private_and_ambiguous_exports_are_not_linked(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "Base.lean": "namespace A\nprivate def secret := 1\ndef f := 2\nend A\n", + "Other.lean": "namespace A\ndef f := 3\nend A\n", + "Exports.lean": "import Base\nimport Other\nexport A (secret f)\ndef use := (secret, f)\n", + "Use.lean": "import Exports\ndef use := (secret, f)\n", + }, + ) + assert not edges + + +def test_lean_resolution_state_is_reset_between_runs(tmp_path: Path) -> None: + analyzer, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "export A (f)\ndef use := f\n", + }, + ) + assert edges + _write(tmp_path, "M.lean", "def use := f\n") + analyzer.analyze_code_files([{"path": "M.lean", "language": "lean"}], str(tmp_path)) + assert not analyzer.call_relationships + + +def test_malformed_directives_do_not_abort_the_file(tmp_path: Path) -> None: + analyzer, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "open A renaming f →\nexport A (\nend Nope\n" + "def after := A.f\n", + }, + ) + assert "M.lean::after" in analyzer.functions + assert edges == {("M.lean::after", "M.lean::A.f")} + + +@pytest.mark.parametrize("separator", [" ", "\n"]) +def test_nested_open_in_wrappers_expire(tmp_path: Path, separator: str) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B\ndef f := 1\nend A.B\n" + + "open A in" + + separator + + "open B in" + + separator + + "@[simp] def inside := f\ndef outside := f\n", + }, + ) + assert edges == {("M.lean::inside", "M.lean::A.B.f")} + + +def test_open_in_does_not_leak_an_inner_open_command(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B\ndef f := 1\nend A.B\nopen A in open B\ndef outside := f\n", + }, + ) + assert not edges + + +def test_open_in_is_consumed_by_a_non_declaration(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "open A in\n#check f\ndef outside := f\n", + }, + ) + assert not edges + + +def test_named_end_restores_opens_and_variables(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": _OPEN_FIXTURE + "def x := 1\nsection Outer.Inner\n" + "open A\nvariable (x : Nat)\ndef inside := (f, x)\n" + "end Outer.Inner\ndef outside := (f, x)\n", + }, + ) + assert edges == {("M.lean::inside", "M.lean::A.f"), ("M.lean::outside", "M.lean::x")} + + +def test_selective_open_disambiguates_each_selected_name(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B\ndef f := 1\nend A.B\nnamespace B\ndef g := 2\nend B\n" + "open A\nopen B (f g)\ndef use := (f, g)\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::A.B.f"), ("M.lean::use", "M.lean::B.g")} + + +def test_duplicate_exports_to_the_same_target_are_not_ambiguous(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "Base.lean": _OPEN_FIXTURE, + "First.lean": "import Base\nnamespace Shared\nexport A (f)\nend Shared\n", + "Second.lean": "import Base\nnamespace Shared\nexport A (f)\nend Shared\n", + "Use.lean": "import First\nimport Second\ndef use := Shared.f\n", + }, + ) + assert edges == {("Use.lean::use", "Base.lean::A.f")} + + +def test_export_source_resolves_through_an_open_namespace(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A.B\ndef f := 1\nend A.B\n" + "namespace C\nopen A\nexport B (f)\nend C\ndef use := C.f\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::A.B.f")} + + +def test_export_keeps_the_namespace_at_its_command(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "namespace A\ndef f := 1\nend A\nnamespace C\nexport A (f)\n" + "namespace A\ndef f := 2\nend A\ndef use := f\nend C\n", + }, + ) + assert edges == {("M.lean::C.use", "M.lean::A.f")} + + +def test_invalid_import_cycle_and_unresolved_exports_do_not_crash(tmp_path: Path) -> None: + analyzer, edges = _analyze_sources( + tmp_path, + { + "A.lean": "import B\nexport Missing (f)\ndef a := 1\n", + "B.lean": "import A\nopen Missing\ndef b := 2\n", + }, + ) + assert set(analyzer.functions) == {"A.lean::a", "B.lean::b"} + assert not edges + + +@pytest.mark.parametrize( + "body", + [ + "fun (x : (Foo : Type)) => x", + "fun [Inhabited (Foo : Type)] => (⟨1⟩ : Foo)", + "((⟨1⟩ : Foo), (Foo : Type))", + ], +) +def test_type_ascriptions_are_not_mistaken_for_binders(tmp_path: Path, body: str) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "structure Foo where\n value : Nat\ndef use := " + body + "\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::Foo")} + + +def test_declaration_binders_still_work_in_error_regions(tmp_path: Path) -> None: + _, edges = _analyze_sources( + tmp_path, + { + "M.lean": "structure Foo where\n value : Nat\ndef x := 1\n" + "def use\n (x : Foo)\n : Foo := by\n exact x\n", + }, + ) + assert edges == {("M.lean::use", "M.lean::Foo")} diff --git a/tests/test_pack_grammars.py b/tests/test_pack_grammars.py new file mode 100644 index 00000000..88803fd9 --- /dev/null +++ b/tests/test_pack_grammars.py @@ -0,0 +1,80 @@ +"""Grammars from tree-sitter-language-pack: a failed download stops the +analysis once, with a clear message, and the Docker image fetches every +grammar the analyzers load.""" + +import re +from pathlib import Path + +import pytest +import tree_sitter_language_pack as tslp + +from codewiki.src.be.dependency_analyzer.analysis.call_graph_analyzer import ( + CallGraphAnalyzer, +) +from codewiki.src.be.dependency_analyzer.utils import grammars +from codewiki.src.be.dependency_analyzer.utils.grammars import ( + PACK_GRAMMARS, + GrammarUnavailableError, + ensure_pack_grammars, + get_pack_parser, +) + +DOCKERFILE = Path(__file__).resolve().parents[1] / "docker" / "Dockerfile" + + +@pytest.fixture +def offline(monkeypatch): + """Every pack grammar load fails as it does without network access.""" + calls = [] + + def fail(name): + calls.append(name) + raise tslp.DownloadError(f"Failed to fetch manifest for {name}: Connection refused") + + get_pack_parser.cache_clear() + monkeypatch.setattr(grammars.tslp, "get_parser", fail) + yield calls + get_pack_parser.cache_clear() + + +def test_download_failure_gives_a_clear_error(offline): + with pytest.raises(GrammarUnavailableError) as info: + get_pack_parser("go") + message = str(info.value) + assert info.value.grammar == "go" + assert "'go'" in message + assert "network access" in message + assert "Connection refused" in message + assert "t.download([" in message + assert isinstance(info.value.__cause__, tslp.DownloadError) + + +def test_only_the_grammars_of_present_languages_load(offline, monkeypatch): + monkeypatch.setattr(grammars.tslp, "get_parser", lambda name: offline.append(name)) + ensure_pack_grammars(["python", "vue", "java", "vue"]) + assert sorted(offline) == ["javascript", "typescript", "vue"] + + +def test_languages_without_pack_grammars_need_no_network(offline): + ensure_pack_grammars(["python", "java", "rust"]) + assert offline == [] + + +def test_analysis_stops_before_the_first_file(offline, tmp_path, monkeypatch): + (tmp_path / "main.go").write_text("package main\n\nfunc main() {}\n", encoding="utf-8") + analyzed = [] + monkeypatch.setattr( + CallGraphAnalyzer, "_analyze_code_file", lambda self, base, info: analyzed.append(info) + ) + with pytest.raises(GrammarUnavailableError, match="'go'"): + CallGraphAnalyzer().analyze_code_files( + [{"path": "main.go", "language": "go"}], str(tmp_path) + ) + assert analyzed == [] + + +def test_dockerfile_downloads_every_pack_grammar(): + match = re.search(r"t\.download\(\[([^\]]*)\]\)", DOCKERFILE.read_text(encoding="utf-8")) + assert match, "docker/Dockerfile must download the pack grammars at build time" + in_dockerfile = sorted(re.findall(r"'([^']+)'", match.group(1))) + assert in_dockerfile == list(PACK_GRAMMARS)