diff --git a/tools/code_index/README.md b/tools/code_index/README.md index 0918190..ab1a171 100644 --- a/tools/code_index/README.md +++ b/tools/code_index/README.md @@ -135,6 +135,32 @@ python -m code_index.search --corpus docs "limitation du contrat foudre" En Python : `from code_index import search_docs ; search_docs("anti-scraping", k=6)`. +## Graphe d'appels — `code_impact` (Phase 4) + +`graph.py` extrait un **graphe d'appels statique** (tree-sitter, PHP/Python/TS/JS) : +définitions (fonctions/classes/méthodes) reliées par leurs sites d'appel. Il répond à +**« qu'est-ce qui casse si je change X ? »** (`callers`, rayon d'impact) et **« de quoi +dépend X ? »** (`callees`). C'est le pendant *code* du `lineage` *data* curé — aucun +embedding, **aucun appel API** (pur AST), donc l'artefact se (re)construit n'importe où. + +Résolution **par nom** (multi-langage, pragmatique) : un appel `foo()` pointe vers toute +définition nommée `foo`. Deux garde-fous data-driven contre les collisions avec des +primitives : `--max-fanout` (nom trop défini → non relié) et un seuil de fréquence d'appel +(nom appelé partout, ex. `.push()`, → non relié) ; les fichiers minifiés/bundlés sont exclus. +Limitation assumée et signalée dans la sortie (champ `ambiguous`). + +```bash +# construire l'artefact (JSON ou JSON.gz — ~1,5 Mo gz pour ~37k nœuds) : +python -m code_index.graph build --out graph.json.gz +# interroger : rayon d'impact (callers) / dépendances (callees) : +python -m code_index.graph impact --graph graph.json.gz "Accumulation" --direction callers --depth 2 +python -m code_index.graph impact --graph graph.json.gz "PluviometrieService" --direction callees +``` + +En Python : `from code_index.graph import load_graph, code_impact`. Un nœud `class` « tire » +ses méthodes comme graines (changer la classe = changer ses membres). `--meta sidecar.json` +ajoute les **URLs source** (permaliens) à chaque résultat. + ## Configuration (variables d'environnement) | Variable | Défaut | Rôle | diff --git a/tools/code_index/chunk_ast.py b/tools/code_index/chunk_ast.py index ef56526..4918846 100644 --- a/tools/code_index/chunk_ast.py +++ b/tools/code_index/chunk_ast.py @@ -12,6 +12,7 @@ """ from __future__ import annotations +from . import tsutil from .chunk import Chunk, chunk_text # Étiquette de langage (walk.LANG_BY_EXT) → nom tree-sitter. @@ -31,64 +32,14 @@ _COMMENT_PREFIXES = ("#", "//", "/*", "*", "*/") - -def _get_parser(ts_lang: str): - """Parser tree-sitter ou None si la lib/grammaire est absente (→ repli char-window).""" - try: - from tree_sitter_language_pack import get_parser - return get_parser(ts_lang) - except Exception: # noqa: BLE001 — lib absente, grammaire inconnue, etc. - return None - - -# ── Accès NŒUD/ARBRE agnostique au binding ─────────────────────────────────── -# Les bindings tree-sitter divergent : py-tree-sitter expose node.type / -# node.start_point / node.named_children / tree.root_node (propriété) ; d'autres -# (ex. language-pack 1.9) exposent node.kind / node.start_position / node.named_child(i) -# / tree.root_node() (méthode). On lisse les deux pour rester portable. - -def _v(obj, name): - """Attribut `name`, appelé s'il s'agit d'une méthode (les bindings exposent tantôt des - propriétés, tantôt des méthodes pour les mêmes infos).""" - a = getattr(obj, name, None) - return a() if callable(a) else a - - -def _root(tree): - return _v(tree, "root_node") - - -def _kind(node) -> str: - return _v(node, "type") or _v(node, "kind") - - -def _row(pt) -> int: - """Ligne 0-based d'un point tree-sitter : tuple (row, col) ou objet Point(.row).""" - return pt.row if hasattr(pt, "row") else pt[0] - - -def _start_row(node) -> int: - return _row(_v(node, "start_point") or _v(node, "start_position")) - - -def _end_row(node) -> int: - return _row(_v(node, "end_point") or _v(node, "end_position")) - - -def _named_children(node) -> list: - kids = _v(node, "named_children") - if kids is not None: - return list(kids) - cnt = _v(node, "named_child_count") or 0 - return [node.named_child(i) for i in range(cnt)] - - -def _parse(parser, text: str): - """parse(bytes) (py-tree-sitter) ou parse(str) (language-pack 1.9) — on tente les deux.""" - try: - return parser.parse(text.encode("utf-8", errors="replace")) - except TypeError: - return parser.parse(text) +# Accès tree-sitter agnostique au binding : factorisé dans `tsutil` (partagé avec `graph`). +_get_parser = tsutil.get_parser +_root = tsutil.root +_kind = tsutil.kind +_start_row = tsutil.start_row +_end_row = tsutil.end_row +_named_children = tsutil.named_children +_parse = tsutil.parse def _emit_window(lines: list[str], a: int, b: int, max_chars: int, out: list[Chunk]) -> None: diff --git a/tools/code_index/graph.py b/tools/code_index/graph.py new file mode 100644 index 0000000..4feebc1 --- /dev/null +++ b/tools/code_index/graph.py @@ -0,0 +1,416 @@ +"""Graphe d'appels du code (Phase 4) : « qu'est-ce qui casse si je change X ? » + +Complète le `lineage` data curé (flux de tables/pipelines) par un graphe **de code** +extrait statiquement via tree-sitter (PHP/Python/TS/JS) : définitions (fonctions, classes, +méthodes) reliées par leurs sites d'appel. L'outil `code_impact` parcourt ce graphe : +- ``callers`` (défaut) : qui appelle X, transitivement = **rayon d'impact** d'un changement ; +- ``callees`` : ce dont X dépend (ses appels sortants). + +Résolution **par nom** (pragmatique, multi-langage) : un appel à ``foo()`` pointe vers +toute définition nommée ``foo``. C'est une SUR-approximation ; pour rester utile on élague +les noms trop ambigus (``--max-fanout``, défaut 6) — un nom défini partout (``run``, ``get``) +ne produit pas d'arêtes. Limitation assumée et signalée dans la sortie. + +Le graphe est un **artefact JSON** (comme l'index LanceDB) : construit une fois (pur +tree-sitter, AUCUN appel API), livré à côté de l'index, lu par le bot/MCP. Sans +``tree_sitter_language_pack`` (CI) l'extraction renvoie un graphe vide — rien ne casse. +""" +from __future__ import annotations + +import argparse +import gzip +import json +import sys +from collections import deque +from pathlib import Path +from typing import Iterable, Iterator + +from . import tsutil, walk +from .meta import chunk_url + +GRAPH_VERSION = "graph-v1" + +# Étiquette de langage (walk.LANG_BY_EXT) → nom tree-sitter. +_TS_LANG = {"php": "php", "python": "python", "ts": "typescript", "js": "javascript"} + +# Nœuds « définition » → genre lisible, par grammaire tree-sitter. +_DEF_TYPES = { + "python": {"function_definition": "function", "class_definition": "class"}, + "php": {"function_definition": "function", "method_declaration": "method", + "class_declaration": "class", "interface_declaration": "interface", + "trait_declaration": "trait", "enum_declaration": "enum"}, + "typescript": {"function_declaration": "function", "method_definition": "method", + "class_declaration": "class", "interface_declaration": "interface", + "enum_declaration": "enum", "abstract_class_declaration": "class"}, + "javascript": {"function_declaration": "function", "method_definition": "method", + "class_declaration": "class"}, +} + +# Nœuds « site d'appel », par grammaire. +_CALL_TYPES = { + "python": {"call"}, + "php": {"function_call_expression", "member_call_expression", "scoped_call_expression"}, + "typescript": {"call_expression"}, + "javascript": {"call_expression"}, +} + +MODULE_QNAME = "" # portée fichier (appels hors de toute définition) + + +# ── Extraction par fichier ──────────────────────────────────────────────────── + +def _leaf_name(node, src: bytes) -> str: + """Dernier segment d'un nom qualifié/membre : ``os.path.join`` → ``join``, + ``Foo\\Bar`` → ``Bar``, ``$this->helper`` → ``helper``.""" + if node is None: + return "" + if tsutil.kind(node) in ("identifier", "name"): + return tsutil.node_text(node, src).strip() + kids = tsutil.named_children(node) + if kids: + return _leaf_name(kids[-1], src) + return tsutil.node_text(node, src).strip() + + +def _callee_name(node, ts_lang: str, src: bytes) -> str: + """Nom de la fonction/méthode appelée par un nœud d'appel (champ selon la grammaire).""" + if ts_lang == "php" and tsutil.kind(node) in ("member_call_expression", + "scoped_call_expression"): + return _leaf_name(tsutil.child_by_field(node, "name"), src) # méthode : champ "name" + return _leaf_name(tsutil.child_by_field(node, "function"), src) # py/ts/js + php libre + + +def extract_file(lang: str, text: str) -> tuple[list[dict], list[dict]]: + """(définitions, appels) d'un fichier. Définitions : {name, qname, kind, start_line, + end_line}. Appels : {caller (qname englobant ou ""), callee (nom), line}. Vide si le + langage n'est pas outillé ou si le parsing échoue (jamais d'exception).""" + ts_lang = _TS_LANG.get(lang) + parser = tsutil.get_parser(ts_lang) if ts_lang else None + if parser is None or not text.strip(): + return [], [] + try: + tree = tsutil.parse(parser, text) + root = tsutil.root(tree) + except Exception: # noqa: BLE001 + return [], [] + src = text.encode("utf-8", errors="replace") + def_types = _DEF_TYPES.get(ts_lang, {}) + call_types = _CALL_TYPES.get(ts_lang, set()) + defs: list[dict] = [] + calls: list[dict] = [] + stack: list[str] = [] # pile des noms englobants → qname (Classe.méthode) + + def visit(node) -> None: + k = tsutil.kind(node) + pushed = False + if k in def_types: + nm = tsutil.child_by_field(node, "name") + name = tsutil.node_text(nm, src).strip() if nm is not None else "" + if name: + defs.append({"name": name, "qname": ".".join(stack + [name]), + "kind": def_types[k], + "start_line": tsutil.start_row(node) + 1, + "end_line": tsutil.end_row(node) + 1}) + stack.append(name) + pushed = True + elif k in call_types: + cn = _callee_name(node, ts_lang, src) + if cn: + calls.append({"caller": ".".join(stack), "callee": cn, + "line": tsutil.start_row(node) + 1}) + for c in tsutil.named_children(node): + visit(c) + if pushed: + stack.pop() + + visit(root) + return defs, calls + + +# ── Construction du graphe ───────────────────────────────────────────────────── + +def _node_id(repo: str, path: str, start_line: int, qname: str) -> str: + return f"{repo}/{path}:{start_line}:{qname}" + + +def build_graph(files: Iterable[tuple[walk.SourceFile, str]], *, + max_name_fanout: int = 6, max_call_freq: int = 200) -> dict: + """Construit le graphe d'appels à partir d'un itérable (SourceFile, texte). + + Résolution par nom : une arête caller→callee est créée pour CHAQUE définition portant + le nom appelé. Deux garde-fous data-driven contre les collisions avec des primitives + (``.push()``, ``.map()``, ``substr()``…) qui parasitent une résolution par nom : + - `max_name_fanout` : un nom défini plus de N fois est trop ambigu → non relié ; + - `max_call_freq` : un nom appelé depuis plus de N sites est quasi sûrement une + primitive du langage/bibliothèque (ex. ``push`` ~2900×) → non relié. + Les appels vers des noms inconnus (stdlib/externe) sont aussi ignorés (on ne graphe + que les symboles internes).""" + nodes: dict[str, dict] = {} + by_name: dict[str, list[str]] = {} + call_freq: dict[str, int] = {} + # passe 1 : toutes les définitions deviennent des nœuds (by_name complet AVANT les arêtes) + pending: list[tuple[walk.SourceFile, list[dict], dict[str, str]]] = [] + for sf, text in files: + defs, calls = extract_file(sf.lang, text) + qmap: dict[str, str] = {} + for d in defs: + nid = _node_id(sf.repo, sf.path, d["start_line"], d["qname"]) + nodes[nid] = {"name": d["name"], "qname": d["qname"], "repo": sf.repo, + "path": sf.path, "lang": sf.lang, "kind": d["kind"], + "start_line": d["start_line"], "end_line": d["end_line"]} + by_name.setdefault(d["name"], []).append(nid) + qmap[d["qname"]] = nid + for c in calls: + call_freq[c["callee"]] = call_freq.get(c["callee"], 0) + 1 + if calls: + pending.append((sf, calls, qmap)) + + # noms à ne PAS relier : trop de définitions (ambigu) ou trop d'appels (primitive) + blocked = {nm for nm, ids in by_name.items() if len(ids) > max_name_fanout} + blocked |= {nm for nm, f in call_freq.items() if f > max_call_freq} + + # passe 2 : arêtes caller→callee, résolues par nom + out: dict[str, set[str]] = {} + for sf, calls, qmap in pending: + module_id: str | None = None + for c in calls: + if c["callee"] in blocked: + continue + targets = by_name.get(c["callee"]) + if not targets: + continue # nom inconnu (externe) → pas d'arête + caller = c["caller"] + if caller and caller in qmap: + src_id = qmap[caller] + elif caller and any(caller.startswith(q + ".") or q == caller for q in qmap): + # appel dans une portée englobante connue (méthode imbriquée non capturée) + src_id = next(qmap[q] for q in qmap + if caller == q or caller.startswith(q + ".")) + else: + # appel au niveau module : nœud de portée fichier, créé à la demande + if module_id is None: + module_id = _node_id(sf.repo, sf.path, 1, MODULE_QNAME) + nodes.setdefault(module_id, { + "name": MODULE_QNAME, "qname": MODULE_QNAME, "repo": sf.repo, + "path": sf.path, "lang": sf.lang, "kind": "module", + "start_line": 1, "end_line": 1}) + src_id = module_id + for tid in targets: + if tid != src_id: # on garde la récursion directe ? non : self-edge inutile ici + out.setdefault(src_id, set()).add(tid) + + return { + "version": GRAPH_VERSION, + "nodes": nodes, + "by_name": by_name, + "out": {k: sorted(v) for k, v in out.items()}, + } + + +def _is_minified(path: str, text: str) -> bool: + """Fichier minifié/bundlé (junk pour un graphe : identifiants mutilés, tout sur une + ligne) : nom révélateur, ou une ligne anormalement longue (source jamais > ~2 000 car.).""" + low = path.lower() + if any(m in low for m in (".min.", "-min.", ".bundle.")): + return True + if {"vendor", "dist", "node_modules"} & set(low.split("/")): + return True + return any(len(ln) > 2000 for ln in text.split("\n", 50)[:50]) + + +def _iter_supported(manifest: dict, base_dir: Path, repos: list[str] | None, + max_file_bytes: int) -> Iterator[tuple[walk.SourceFile, str]]: + """Fichiers des langages outillés (PHP/Py/TS/JS), texte décodé, hors minifiés/bundles.""" + for sf in walk.iter_files(manifest, base_dir, repos=repos, max_file_bytes=max_file_bytes): + if sf.lang not in _TS_LANG: + continue + try: + text = sf.abspath.read_bytes().decode("utf-8", errors="replace") + except OSError: + continue + if text.strip() and not _is_minified(sf.path, text): + yield sf, text + + +# ── Requête : code_impact ────────────────────────────────────────────────────── + +def load_graph(path: str | Path) -> dict: + """Charge un graphe (`.json` ou `.json.gz`) et calcule l'index inverse `in` + (callers) à partir de `out`.""" + p = Path(path) + raw = (gzip.decompress(p.read_bytes()).decode("utf-8") if p.suffix == ".gz" + else p.read_text(encoding="utf-8")) + g = json.loads(raw) + rev: dict[str, list[str]] = {} + for src_id, tgts in g.get("out", {}).items(): + for t in tgts: + rev.setdefault(t, []).append(src_id) + g["in"] = {k: sorted(v) for k, v in rev.items()} + return g + + +def resolve_symbol(graph: dict, symbol: str) -> list[str]: + """Identifiants de nœuds correspondant à `symbol` : nom exact, ou suffixe de qname + (``Classe.methode`` / ``methode``), ou ``chemin:ligne``. Insensible à la casse en repli.""" + s = symbol.strip() + if not s: + return [] + nodes = graph.get("nodes", {}) + # 1) nom exact (le plus courant) + hits = list(graph.get("by_name", {}).get(s, [])) + if hits: + return hits + # 2) chemin:ligne + if ":" in s: + head = s.rsplit(":", 1) + if head[1].isdigit(): + ln = int(head[1]) + hits = [nid for nid, n in nodes.items() + if n["path"].endswith(head[0]) and n["start_line"] == ln] + if hits: + return hits + # 3) qname exact ou suffixe (Classe.methode), insensible à la casse + sl = s.lower() + return [nid for nid, n in nodes.items() + if n["qname"].lower() == sl or n["qname"].lower().endswith("." + sl)] + + +def code_impact(graph: dict, symbol: str, *, direction: str = "callers", depth: int = 2, + max_nodes: int = 40, sidecar: dict | None = None) -> dict: + """Parcours BFS borné du graphe depuis `symbol`. + + direction = ``callers`` (qui appelle X, transitif = rayon d'impact) ou ``callees`` + (ce dont X dépend). Renvoie {roots, impacted, truncated, ambiguous} ; chaque nœud est + enrichi de son URL source (permalien) si un `sidecar` méta est fourni.""" + if "in" not in graph: # graphe chargé sans load_graph (ex. fraîchement construit) + graph = {**graph, "in": _reverse(graph.get("out", {}))} + edges = graph["in"] if direction == "callers" else graph.get("out", {}) + roots = resolve_symbol(graph, symbol) + nodes = graph.get("nodes", {}) + + # Un conteneur (classe/interface/trait/enum) « tire » ses membres comme graines : changer + # la classe, c'est changer ses méthodes ⇒ impact = celui de la classe OU d'un membre. + seeds = set(roots) + for r in roots: + n = nodes.get(r) + if n and n["kind"] in ("class", "interface", "trait", "enum"): + pref = n["qname"] + "." + seeds.update(nid for nid, nn in nodes.items() + if nn["repo"] == n["repo"] and nn["path"] == n["path"] + and nn["qname"].startswith(pref)) + + seen: dict[str, int] = {s: 0 for s in seeds} # graines en prof 0 → exclues des résultats + order: list[tuple[str, int]] = [] + frontier = list(seeds) + d = 0 + while frontier and d < depth: + d += 1 + nxt: list[str] = [] + for nid in frontier: + for m in edges.get(nid, []): + if m not in seen: + seen[m] = d + nxt.append(m) + order.append((m, d)) + frontier = nxt + + truncated = len(order) > max_nodes + impacted = [_describe(nodes[i], dep, sidecar) for i, dep in order[:max_nodes] if i in nodes] + return { + "symbol": symbol, + "direction": direction, + "depth": depth, + "roots": [_describe(nodes[r], 0, sidecar) for r in roots if r in nodes], + "ambiguous": len(roots) > 1, + "impacted": impacted, + "truncated": truncated, + } + + +def _reverse(out: dict) -> dict: + rev: dict[str, list[str]] = {} + for s, tgts in out.items(): + for t in tgts: + rev.setdefault(t, []).append(s) + return rev + + +def _describe(node: dict, depth: int, sidecar: dict | None) -> dict: + url = (chunk_url(sidecar, node["repo"], node["path"], + node["start_line"], node["end_line"]) if sidecar else "") + return {"qname": node["qname"], "kind": node["kind"], "repo": node["repo"], + "path": node["path"], "lang": node["lang"], "start_line": node["start_line"], + "end_line": node["end_line"], "depth": depth, "source_url": url} + + +# ── CLI ───────────────────────────────────────────────────────────────────────── + +def _cmd_build(args: argparse.Namespace) -> int: + from .config import load_config, load_manifest + cfg = load_config() + base = args.base_dir or cfg.base_dir + manifest = load_manifest(args.manifest) + g = build_graph(_iter_supported(manifest, base, args.repos, cfg.max_file_bytes), + max_name_fanout=args.max_fanout) + n_edges = sum(len(v) for v in g["out"].values()) + out_path = Path(args.out) + data = json.dumps(g, ensure_ascii=False) + if out_path.suffix == ".gz": + out_path.write_bytes(gzip.compress(data.encode("utf-8"))) + else: + out_path.write_text(data, encoding="utf-8") + print(f"Graphe écrit : {out_path} — {len(g['nodes'])} nœuds, {n_edges} arêtes, " + f"{len(g['by_name'])} noms (fanout max {args.max_fanout}).") + return 0 + + +def _cmd_impact(args: argparse.Namespace) -> int: + graph = load_graph(args.graph) + sidecar = json.loads(Path(args.meta).read_text(encoding="utf-8")) if args.meta else None + res = code_impact(graph, args.symbol, direction=args.direction, depth=args.depth, + sidecar=sidecar) + if not res["roots"]: + print(f"Symbole introuvable dans le graphe : {args.symbol}", file=sys.stderr) + return 1 + verb = "appelé par" if args.direction == "callers" else "dépend de" + print(f"# {args.symbol} — {verb} (profondeur {args.depth})") + if res["ambiguous"]: + print(f" ⚠ {len(res['roots'])} définitions portent ce nom (résolution par nom).") + for r in res["roots"]: + print(f" ⌖ {r['qname']} ({r['kind']}) {r['repo']}/{r['path']}:{r['start_line']}") + for n in res["impacted"]: + url = f" {n['source_url']}" if n["source_url"] else "" + print(f" {'·' * n['depth']} {n['qname']} ({n['kind']}) " + f"{n['repo']}/{n['path']}:{n['start_line']}{url}") + if res["truncated"]: + print(" … (résultats tronqués)") + return 0 + + +def main(argv: list[str]) -> int: + ap = argparse.ArgumentParser(description="Graphe d'appels du code Infoclimat (tree-sitter).") + sub = ap.add_subparsers(dest="cmd", required=True) + + b = sub.add_parser("build", help="Construire l'artefact graphe (JSON).") + b.add_argument("--out", required=True, help="Fichier de sortie (.json ou .json.gz).") + b.add_argument("--base-dir", type=Path, default=None) + b.add_argument("--manifest", type=Path, default=None) + b.add_argument("--repo", action="append", dest="repos", default=None) + b.add_argument("--max-fanout", type=int, default=6, + help="Au-delà de N définitions homonymes, le nom n'est pas relié (anti-bruit).") + b.set_defaults(func=_cmd_build) + + q = sub.add_parser("impact", help="Rayon d'impact / dépendances d'un symbole.") + q.add_argument("symbol") + q.add_argument("--graph", required=True, help="Artefact graphe (.json/.json.gz).") + q.add_argument("--direction", choices=["callers", "callees"], default="callers") + q.add_argument("--depth", type=int, default=2) + q.add_argument("--meta", default=None, help="Sidecar méta (pour les URLs source).") + q.set_defaults(func=_cmd_impact) + + args = ap.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/tools/code_index/tsutil.py b/tools/code_index/tsutil.py new file mode 100644 index 0000000..2e6d9f3 --- /dev/null +++ b/tools/code_index/tsutil.py @@ -0,0 +1,93 @@ +"""Accès tree-sitter agnostique au binding (partagé par le chunking AST et le graphe). + +Les bindings tree-sitter divergent : py-tree-sitter expose ``node.type`` / +``node.start_point`` / ``node.named_children`` / ``tree.root_node`` (propriété) ; +d'autres (ex. language-pack 1.9) exposent ``node.kind`` / ``node.start_position`` / +``node.named_child(i)`` / ``tree.root_node()`` (méthode), avec des points tantôt +tuples ``(row, col)`` tantôt objets ``Point(.row)``. On lisse les deux pour rester +portable, et on importe la lib PARESSEUSEMENT : sans elle (CI/tests stdlib), tout +appelant retombe sur son propre repli — rien ne casse. +""" +from __future__ import annotations + + +def get_parser(ts_lang: str): + """Parser tree-sitter pour `ts_lang`, ou None si la lib/grammaire est absente.""" + try: + from tree_sitter_language_pack import get_parser as _gp + return _gp(ts_lang) + except Exception: # noqa: BLE001 — lib absente, grammaire inconnue, etc. + return None + + +def v(obj, name): + """Attribut `name`, appelé s'il s'agit d'une méthode (les bindings exposent tantôt + des propriétés, tantôt des méthodes pour les mêmes infos).""" + a = getattr(obj, name, None) + return a() if callable(a) else a + + +def root(tree): + return v(tree, "root_node") + + +def kind(node) -> str: + return v(node, "type") or v(node, "kind") + + +def _row(pt) -> int: + """Ligne 0-based d'un point tree-sitter : tuple (row, col) ou objet Point(.row).""" + return pt.row if hasattr(pt, "row") else pt[0] + + +def start_row(node) -> int: + return _row(v(node, "start_point") or v(node, "start_position")) + + +def end_row(node) -> int: + return _row(v(node, "end_point") or v(node, "end_position")) + + +def named_children(node) -> list: + kids = v(node, "named_children") + if kids is not None: + return list(kids) + cnt = v(node, "named_child_count") or 0 + return [node.named_child(i) for i in range(cnt)] + + +def child_by_field(node, field: str): + """Enfant nommé via son champ de grammaire (ex. `name`), ou None. + + py-tree-sitter : ``child_by_field_name(field)`` ; language-pack 1.9 idem. On lisse + aussi l'éventuel accès par propriété (rare).""" + fn = getattr(node, "child_by_field_name", None) + if callable(fn): + try: + return fn(field) + except Exception: # noqa: BLE001 + return None + return None + + +def node_text(node, src_bytes: bytes) -> str: + """Texte source d'un nœud. Utilise ``node.text`` si présent, sinon découpe les octets + via ``start_byte``/``end_byte`` (lissés) — décodage tolérant.""" + t = getattr(node, "text", None) + if isinstance(t, (bytes, bytearray)): + return bytes(t).decode("utf-8", errors="replace") + if isinstance(t, str): + return t + a = v(node, "start_byte") + b = v(node, "end_byte") + if a is None or b is None: + return "" + return src_bytes[a:b].decode("utf-8", errors="replace") + + +def parse(parser, text: str): + """parse(bytes) (py-tree-sitter) ou parse(str) (language-pack 1.9) — on tente les deux.""" + try: + return parser.parse(text.encode("utf-8", errors="replace")) + except TypeError: + return parser.parse(text) diff --git a/tools/tests/test_code_index_graph.py b/tools/tests/test_code_index_graph.py new file mode 100644 index 0000000..d5df421 --- /dev/null +++ b/tools/tests/test_code_index_graph.py @@ -0,0 +1,175 @@ +"""Tests du graphe d'appels (code_index.graph) : extraction, construction, code_impact.""" +from __future__ import annotations + +import pytest + +from code_index import graph +from code_index.walk import SourceFile + + +def _sf(repo: str, path: str, lang: str) -> SourceFile: + from pathlib import Path + return SourceFile(repo=repo, path=path, abspath=Path("/dev/null"), lang=lang) + + +# ── Extraction (nécessite tree-sitter) ───────────────────────────────────────── + +PY_SRC = ( + "import os\n\n" + "def alpha(a):\n" + " return beta(a)\n\n" + "def beta(a):\n" + " return os.path.join(a)\n\n" + "class Gamma:\n" + " def m(self):\n" + " return self.helper() + alpha(1)\n" + " def helper(self):\n" + " return 0\n" +) + + +def test_extract_python_defs_and_calls() -> None: + pytest.importorskip("tree_sitter_language_pack") + defs, calls = graph.extract_file("python", PY_SRC) + qnames = {d["qname"] for d in defs} + assert {"alpha", "beta", "Gamma", "Gamma.m", "Gamma.helper"} <= qnames + # alpha appelle beta ; Gamma.m appelle helper (membre) et alpha + pairs = {(c["caller"], c["callee"]) for c in calls} + assert ("alpha", "beta") in pairs + assert ("Gamma.m", "helper") in pairs # self.helper() → leaf 'helper' + assert ("Gamma.m", "alpha") in pairs + assert ("beta", "join") in pairs # os.path.join → leaf 'join' + + +def test_extract_php_member_and_free_calls() -> None: + pytest.importorskip("tree_sitter_language_pack") + php = ("helper() + alpha(1); }\n" + " public function helper() { return 0; }\n" + "}\n") + defs, calls = graph.extract_file("php", php) + qnames = {d["qname"] for d in defs} + assert {"alpha", "Gamma", "Gamma.m", "Gamma.helper"} <= qnames + pairs = {(c["caller"], c["callee"]) for c in calls} + assert ("alpha", "beta") in pairs + assert ("Gamma.m", "helper") in pairs # $this->helper() + assert ("Gamma.m", "alpha") in pairs + + +def test_extract_unsupported_lang_is_empty() -> None: + # Langage non outillé → aucune extraction, jamais d'exception. + assert graph.extract_file("ncl", "a = 1\n") == ([], []) + assert graph.extract_file("python", "") == ([], []) + + +# ── Construction du graphe ────────────────────────────────────────────────────── + +def test_build_graph_resolves_internal_calls() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = graph.build_graph([(_sf("r", "mod.py", "python"), PY_SRC)]) + assert g["version"] == graph.GRAPH_VERSION + # beta est défini → il a un nœud, et alpha→beta est une arête + beta_ids = g["by_name"]["beta"] + assert len(beta_ids) == 1 + alpha_id = g["by_name"]["alpha"][0] + assert beta_ids[0] in g["out"][alpha_id] + # join (os.path.join) n'est PAS défini en interne → pas de nœud, pas d'arête + assert "join" not in g["by_name"] + + +def test_build_graph_prunes_ambiguous_names() -> None: + pytest.importorskip("tree_sitter_language_pack") + # 'run' défini dans 3 fichiers, appelé par caller ; fanout=2 → élagué (>2 homonymes) + files = [] + for i in range(3): + files.append((_sf("r", f"w{i}.py", "python"), f"def run():\n return {i}\n")) + files.append((_sf("r", "main.py", "python"), "def go():\n return run()\n")) + g = graph.build_graph(files, max_name_fanout=2) + go_id = g["by_name"]["go"][0] + assert go_id not in g["out"] # 'run' a 3 défs > 2 → aucune arête créée + g2 = graph.build_graph(files, max_name_fanout=5) + assert len(g2["out"][g2["by_name"]["go"][0]]) == 3 # 5 ≥ 3 → reliées + + +def test_build_graph_prunes_ubiquitous_call_names() -> None: + pytest.importorskip("tree_sitter_language_pack") + # 'push' défini une seule fois (méthode user) mais appelé partout (primitive .push()) : + # au-delà de max_call_freq, aucune arête ne pointe vers lui (anti-collision builtin). + files = [(_sf("r", "lib.py", "python"), "class S:\n def push(self):\n return 1\n")] + callers = "\n".join(f"def c{i}():\n x.push()\n" for i in range(5)) + files.append((_sf("r", "many.py", "python"), callers)) + g = graph.build_graph(files, max_call_freq=3) # 5 appels > 3 → 'push' bloqué + push_id = g["by_name"]["push"][0] + loaded = {**g, "in": graph._reverse(g["out"])} + assert push_id not in loaded["in"] # aucun caller relié + g2 = graph.build_graph(files, max_call_freq=10) # 5 ≤ 10 → reliés + assert graph._reverse(g2["out"]).get(push_id) + + +def test_is_minified_detection() -> None: + assert graph._is_minified("web/js/app.min.js", "x") + assert graph._is_minified("vendor/lib/foo.js", "x") + assert graph._is_minified("a.js", "var x=1;" + "a" * 3000) # ligne géante + assert not graph._is_minified("src/app.js", "const x = 1\nconst y = 2\n") + + +# ── code_impact (BFS) ─────────────────────────────────────────────────────────── + +def _chain_graph(): + """a → b → c (a appelle b, b appelle c), en python.""" + src = ("def c():\n return 1\n\n" + "def b():\n return c()\n\n" + "def a():\n return b()\n") + return graph.build_graph([(_sf("r", "chain.py", "python"), src)]) + + +def test_code_impact_callers_blast_radius() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = _chain_graph() + # qui casse si on change c ? → b (prof 1) puis a (prof 2) + res = graph.code_impact(g, "c", direction="callers", depth=2) + names = {(n["qname"], n["depth"]) for n in res["impacted"]} + assert ("b", 1) in names and ("a", 2) in names + # profondeur 1 : seulement b + res1 = graph.code_impact(g, "c", direction="callers", depth=1) + assert {n["qname"] for n in res1["impacted"]} == {"b"} + + +def test_code_impact_callees_dependencies() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = _chain_graph() + # de quoi dépend a ? → b puis c + res = graph.code_impact(g, "a", direction="callees", depth=2) + assert {n["qname"] for n in res["impacted"]} == {"b", "c"} + + +def test_code_impact_unknown_symbol() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = _chain_graph() + res = graph.code_impact(g, "inexistant", direction="callers") + assert res["roots"] == [] and res["impacted"] == [] + + +def test_load_graph_roundtrip_and_reverse(tmp_path) -> None: + pytest.importorskip("tree_sitter_language_pack") + import json + g = _chain_graph() + p = tmp_path / "graph.json" + p.write_text(json.dumps(g), encoding="utf-8") + loaded = graph.load_graph(p) + assert "in" in loaded # index inverse calculé + res = graph.code_impact(loaded, "c", direction="callers", depth=2) + assert {n["qname"] for n in res["impacted"]} == {"a", "b"} + + +def test_load_graph_gzip(tmp_path) -> None: + pytest.importorskip("tree_sitter_language_pack") + import gzip + import json + g = _chain_graph() + p = tmp_path / "graph.json.gz" + p.write_bytes(gzip.compress(json.dumps(g).encode("utf-8"))) + loaded = graph.load_graph(p) + assert loaded["nodes"] == g["nodes"]