diff --git a/tools/code_index/README.md b/tools/code_index/README.md index ab1a171..2712ac0 100644 --- a/tools/code_index/README.md +++ b/tools/code_index/README.md @@ -143,11 +143,23 @@ définitions (fonctions/classes/méthodes) reliées par leurs sites d'appel. Il 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`). +**Résolution par cascade à score de confiance** (v2 — état de l'art accessible sans +compilateur : scope/stack graphs, SCIP, cascade « Codebase-Memory ») au lieu d'une simple +correspondance de nom : + +1. `$this->m()` / `self::m()` / `parent::m()` → résolu dans la classe englobante et sa + **hiérarchie** (extends/implements/use trait) — confiance 0.95 ; +2. `Class::m()` / `new Class()` → classe résolue via `use`/namespace (PHP) ou import (TS/PY) + puis méthode/constructeur dans la classe+hiérarchie — 0.90 ; +3. fonction libre `foo()` : import-exact (`use`, import relatif → fichier) 0.95, + même-namespace/même-fichier 0.90, nom unique projet 0.75, sinon **ambigu** 0.35. + +Toute résolution est **contrainte à la même langue** que l'appel (un `new Date()` JS ne vise +pas une classe PHP `Date`). Garde-fous data-driven contre les primitives (`--max-fanout`, +fréquence d'appel, ex. `.push()`/`range()`), fichiers minifiés exclus ; les arêtes fortes +(≥0.7) ne sont jamais élaguées. La sortie bucketise en **Certain / Probable / Incertain** +(certitude d'un chemin = min des confiances), regroupe par **sous-système** (repo + dossier de +tête) et classe par **centralité PageRank** + fan-in. Ambiguïté de nom affichée (`ambiguous`). ```bash # construire l'artefact (JSON ou JSON.gz — ~1,5 Mo gz pour ~37k nœuds) : diff --git a/tools/code_index/graph.py b/tools/code_index/graph.py index 4feebc1..1856444 100644 --- a/tools/code_index/graph.py +++ b/tools/code_index/graph.py @@ -1,18 +1,29 @@ -"""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 +"""Graphe d'appels du code (Phase 4, v2) : « qu'est-ce qui casse si je change X ? » + +Complète le `lineage` data curé par un graphe **de code** extrait statiquement via +tree-sitter (PHP/Python/TS/JS). `code_impact` le parcourt : ``callers`` (qui appelle X, +transitivement = rayon d'impact), ``callees`` (ce dont X dépend). + +v2 — résolution **par cascade à score de confiance** (état de l'art accessible sans +compilateur : scope/stack graphs, SCIP, cascade « Codebase-Memory ») plutôt que la simple +correspondance de nom : + + 1. ``$this->m()`` / ``self::m()`` / ``parent::m()`` → résolu dans la classe englobante + et sa **hiérarchie** (extends/implements/use trait) — conf 0.95 ; + 2. ``Class::m()`` / ``new Class()`` → classe résolue via ``use``/namespace (PHP) ou + import (TS/PY) puis méthode/constructeur dans la classe+hiérarchie — conf 0.9 ; + 3. fonction libre ``foo()`` : import-exact (``use`` PHP, import relatif TS/PY → fichier) + 0.95, même-namespace/même-fichier 0.9, nom unique projet 0.75, sinon ambigu 0.35 ; + 4. ``$obj->m()`` (receveur de type inconnu) → name-based. + +Chaque arête porte sa **confiance**. Les arêtes fortes (≥0.7) ne sont JAMAIS élaguées ; le +fallback name-based bas-confiance reste élagué par fanout/fréquence d'appel (anti-collision +avec les primitives ``.push()``). La sortie bucketise en **Certain/Probable/Incertain** +(certitude d'un chemin = min des confiances) — ambiguïté affichée honnêtement. Au build on +calcule aussi **fan-in** et **centralité PageRank** par nœud (classement du rayon d'impact, +façon repo-map Aider). + +Pur AST, **aucun appel API** → artefact reconstructible partout. Sans ``tree_sitter_language_pack`` (CI) l'extraction renvoie un graphe vide — rien ne casse. """ from __future__ import annotations @@ -20,20 +31,18 @@ import argparse import gzip import json +import posixpath 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" +GRAPH_VERSION = "graph-v2" -# É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", @@ -45,149 +54,363 @@ "javascript": {"function_declaration": "function", "method_definition": "method", "class_declaration": "class"}, } +_CLASS_KINDS = {"class", "interface", "trait", "enum"} -# 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"}, + "php": {"function_call_expression", "member_call_expression", + "scoped_call_expression", "object_creation_expression"}, + "typescript": {"call_expression", "new_expression"}, + "javascript": {"call_expression", "new_expression"}, } -MODULE_QNAME = "" # portée fichier (appels hors de toute définition) +MODULE_QNAME = "" + +# Confiance par stratégie de résolution. +C_HIER = 0.95 # this/self/parent résolu dans la hiérarchie de classe +C_STATIC = 0.90 # Class::m / new Class résolu +C_IMPORT = 0.95 # fonction importée (use / import relatif) → fichier exact +C_SAMENS = 0.90 # même namespace / même fichier +C_UNIQUE = 0.75 # nom unique dans tout le projet +C_MEMBER1 = 0.60 # $obj->m receveur inconnu, mais nom unique +C_AMBIG = 0.35 # plusieurs définitions homonymes (name-based) +STRONG = 0.70 # au-dessus : jamais élagué +TIER_CERTAIN, TIER_PROBABLE = 0.80, 0.50 + +# ── Helpers d'extraction ──────────────────────────────────────────────────────── + +def _name_of(node, src: bytes) -> str: + nm = tsutil.child_by_field(node, "name") + return tsutil.node_text(nm, src).strip() if nm is not None else "" -# ── 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``.""" + """Dernier segment d'un nom qualifié/membre (os.path.join→join, Foo\\Bar→Bar).""" if node is None: return "" - if tsutil.kind(node) in ("identifier", "name"): + if tsutil.kind(node) in ("identifier", "name", "property_identifier", "type_identifier"): 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).""" + return _leaf_name(kids[-1], src) if kids else tsutil.node_text(node, src).strip() + + +def _receiver(node, ts_lang: str, src: bytes) -> tuple[str, str, str]: + """(callee_name, receiver_kind, receiver_class). kind ∈ free|this|self|parent|static|new|member.""" + k = tsutil.kind(node) + if ts_lang == "php": + if k == "object_creation_expression": # new Class() + cls = next((c for c in tsutil.named_children(node) + if tsutil.kind(c) in ("name", "qualified_name")), None) + return ("__construct", "new", _leaf_name(cls, src)) + if k == "scoped_call_expression": # self::/parent::/Class:: + scope = tsutil.child_by_field(node, "scope") + meth = _leaf_name(tsutil.child_by_field(node, "name"), src) + st = tsutil.node_text(scope, src).strip() if scope is not None else "" + if st in ("self", "static"): + return (meth, "self", "") + if st == "parent": + return (meth, "parent", "") + return (meth, "static", _leaf_name(scope, src)) + if k == "member_call_expression": # $x->m() + obj = tsutil.child_by_field(node, "object") + meth = _leaf_name(tsutil.child_by_field(node, "name"), src) + if obj is not None and tsutil.node_text(obj, src).strip() == "$this": + return (meth, "this", "") + return (meth, "member", "") + return (_leaf_name(tsutil.child_by_field(node, "function"), src), "free", "") + # python / ts / js + if k == "new_expression": + return ("__construct", "new", + _leaf_name(tsutil.child_by_field(node, "constructor"), src)) + fn = tsutil.child_by_field(node, "function") + if fn is not None and tsutil.kind(fn) in ("attribute", "member_expression"): + kids = tsutil.named_children(fn) + base = tsutil.node_text(kids[0], src).strip() if kids else "" + meth = _leaf_name(fn, src) + return (meth, "this", "") if base in ("self", "this") else (meth, "member", "") + return (_leaf_name(fn, src), "free", "") + + +def _bases(node, ts_lang: str, src: bytes) -> list[str]: + """Noms (bruts) des classes/interfaces/traits parents d'une déclaration de classe.""" + out: list[str] = [] + if ts_lang == "python": + al = next((c for c in tsutil.named_children(node) + if tsutil.kind(c) == "argument_list"), None) + if al is not None: + out += [tsutil.node_text(c, src).strip().split(".")[-1] + for c in tsutil.named_children(al) if tsutil.kind(c) == "identifier"] + return out + for c in tsutil.named_children(node): + ck = tsutil.kind(c) + if ck in ("base_clause", "class_interface_clause", "class_heritage"): + for d in tsutil.named_children(c): + dk = tsutil.kind(d) + if dk in ("name", "identifier", "type_identifier", "qualified_name"): + out.append(_leaf_name(d, src)) + elif dk in ("extends_clause", "implements_clause"): + out += [_leaf_name(e, src) for e in tsutil.named_children(d)] + elif ck == "use_declaration": # php : use TraitX; dans le corps de classe + out += [_leaf_name(d, src) for d in tsutil.named_children(c) + if tsutil.kind(d) in ("name", "qualified_name")] + return out + + +def _php_namespace(root, src: bytes) -> str: + for c in tsutil.named_children(root): + if tsutil.kind(c) == "namespace_definition": + nm = tsutil.child_by_field(c, "name") + return tsutil.node_text(nm, src).strip().strip("\\") if nm is not None else "" + return "" + + +def _imports(root, ts_lang: str, src: bytes, file_dir: str) -> dict[str, dict]: + """alias → {fqn} (PHP use) ou {name, rel} (import relatif PY/TS).""" + imp: dict[str, dict] = {} + if ts_lang == "php": + def walk_use(n): + if tsutil.kind(n) == "namespace_use_clause": + txt = tsutil.node_text(n, src).strip() + fqn, alias = txt, None + if " as " in txt: + fqn, alias = [p.strip() for p in txt.split(" as ", 1)] + fqn = fqn.strip("\\") + imp[alias or fqn.split("\\")[-1]] = {"fqn": fqn} + for c in tsutil.named_children(n): + walk_use(c) + walk_use(root) + return imp + + for c in tsutil.named_children(root): + ck = tsutil.kind(c) + if ts_lang == "python" and ck == "import_from_statement": + mod = next((tsutil.node_text(d, src).strip() for d in tsutil.named_children(c) + if tsutil.kind(d) == "dotted_name"), "") + rel = _py_rel(mod, file_dir) + if rel is None: + continue + for d in tsutil.named_children(c): + if tsutil.kind(d) == "aliased_import": + nm, al = _name_of(d, src), _leaf_name(tsutil.named_children(d)[-1], src) + if nm: + imp[al] = {"name": nm, "rel": rel} + elif tsutil.kind(d) == "dotted_name" and tsutil.node_text(d, src).strip() != mod: + nm = tsutil.node_text(d, src).strip().split(".")[-1] + imp[nm] = {"name": nm, "rel": rel} + elif ts_lang in ("typescript", "javascript") and ck == "import_statement": + srcstr = next((tsutil.node_text(s, src).strip().strip("'\"") + for s in tsutil.named_children(c) if tsutil.kind(s) == "string"), "") + rel = _ts_rel(srcstr, file_dir) + if rel is None: + continue + clause = next((s for s in tsutil.named_children(c) + if tsutil.kind(s) == "import_clause"), None) + if clause is None: + continue + for d in tsutil.named_children(clause): + if tsutil.kind(d) == "identifier": # import Default from … + imp[tsutil.node_text(d, src).strip()] = {"name": "default", "rel": rel} + elif tsutil.kind(d) == "named_imports": # { A, B as C } + for spec in tsutil.named_children(d): + if tsutil.kind(spec) == "import_specifier": + orig = _name_of(spec, src) + al = tsutil.child_by_field(spec, "alias") + alias = tsutil.node_text(al, src).strip() if al is not None else orig + if orig: + imp[alias] = {"name": orig, "rel": rel} + return imp + + +def _py_rel(mod: str, file_dir: str) -> str | None: + if not mod.startswith("."): + return None + up = len(mod) - len(mod.lstrip(".")) + rest = mod[up:].replace(".", "/") + base = file_dir + for _ in range(up - 1): + base = posixpath.dirname(base) + return posixpath.normpath(posixpath.join(base, rest)) if rest else base + + +def _ts_rel(srcstr: str, file_dir: str) -> str | None: + return posixpath.normpath(posixpath.join(file_dir, srcstr)) if srcstr.startswith(".") else None + + +# ── Extraction par fichier ────────────────────────────────────────────────────── + +def extract_file(lang: str, text: str, path: str = "") -> dict: + """Infos structurelles d'un fichier : ns, imports, defs (avec hiérarchie), calls (avec + receveur). Vide si langage non outillé / parsing impossible (jamais d'exception).""" + empty = {"ns": "", "imports": {}, "defs": [], "calls": []} 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 [], [] + return empty try: tree = tsutil.parse(parser, text) root = tsutil.root(tree) except Exception: # noqa: BLE001 - return [], [] + return empty src = text.encode("utf-8", errors="replace") def_types = _DEF_TYPES.get(ts_lang, {}) call_types = _CALL_TYPES.get(ts_lang, set()) + ns = _php_namespace(root, src) if ts_lang == "php" else "" + imports = _imports(root, ts_lang, src, posixpath.dirname(path)) + defs: list[dict] = [] calls: list[dict] = [] - stack: list[str] = [] # pile des noms englobants → qname (Classe.méthode) + stack: list[str] = [] + class_stack: list[str] = [] def visit(node) -> None: k = tsutil.kind(node) - pushed = False + pushed = is_class = 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 "" + name = _name_of(node, src) 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}) + kind = def_types[k] + d = {"name": name, "qname": ".".join(stack + [name]), "kind": kind, + "start_line": tsutil.start_row(node) + 1, + "end_line": tsutil.end_row(node) + 1} + if kind in _CLASS_KINDS: + d["bases"] = _bases(node, ts_lang, src) + is_class = True + defs.append(d) stack.append(name) pushed = True + if is_class: + class_stack.append(d["qname"]) elif k in call_types: - cn = _callee_name(node, ts_lang, src) - if cn: - calls.append({"caller": ".".join(stack), "callee": cn, + callee, rkind, rclass = _receiver(node, ts_lang, src) + if callee: + calls.append({"callee": callee, "rkind": rkind, "rclass": rclass, + "caller": ".".join(stack), + "class": class_stack[-1] if class_stack else "", "line": tsutil.start_row(node) + 1}) for c in tsutil.named_children(node): visit(c) if pushed: stack.pop() + if is_class: + class_stack.pop() visit(root) - return defs, calls + return {"ns": ns, "imports": imports, "defs": defs, "calls": calls} -# ── Construction du graphe ───────────────────────────────────────────────────── +# ── Construction du graphe ─────────────────────────────────────────────────────── def _node_id(repo: str, path: str, start_line: int, qname: str) -> str: return f"{repo}/{path}:{start_line}:{qname}" +def _fqn(ns: str, qname: str) -> str: + return f"{ns}\\{qname}" if ns else 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).""" + """Graphe d'appels à résolution par cascade. Voir le docstring du module.""" nodes: dict[str, dict] = {} by_name: dict[str, list[str]] = {} + by_fqn: dict[str, str] = {} + classes: dict[str, dict] = {} + file_defs: dict[tuple, dict] = {} + pending: list[tuple] = [] 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) + info = extract_file(sf.lang, text, sf.path) + ns = info["ns"] qmap: dict[str, str] = {} - for d in defs: + local: dict[str, list[str]] = {} + class_fqn_by_q: dict[str, str] = {} + for d in info["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) + local.setdefault(d["name"], []).append(nid) qmap[d["qname"]] = nid - for c in calls: + by_fqn.setdefault(_fqn(ns, d["qname"]), []).append(nid) + if d["kind"] in _CLASS_KINDS: + cfqn = _fqn(ns, d["qname"]) + classes[cfqn] = {"id": nid, "bases": d.get("bases", []), "methods": {}, + "ns": ns, "imports": info["imports"], "lang": sf.lang} + class_fqn_by_q[d["qname"]] = cfqn + for d in info["defs"]: + if d["kind"] in ("method", "function") and "." in d["qname"]: + cfqn = class_fqn_by_q.get(d["qname"].rsplit(".", 1)[0]) + if cfqn: + classes[cfqn]["methods"][d["name"]] = qmap[d["qname"]] + file_defs[(sf.repo, sf.path)] = local + for c in info["calls"]: call_freq[c["callee"]] = call_freq.get(c["callee"], 0) + 1 - if calls: - pending.append((sf, calls, qmap)) + if info["calls"]: + pending.append((sf, ns, info["imports"], info["calls"], qmap, class_fqn_by_q)) - # 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: + def resolve_class(name: str, ns: str, imports: dict, lang: str) -> str | None: + """FQN d'une classe `name` visible depuis (`ns`, `imports`), MÊME LANGAGE que l'appel + (le registre est multi-langage : un `new Date()` JS ne doit pas viser une classe PHP).""" + if not name: + return None + def ok(cfqn): + return cfqn in classes and classes[cfqn]["lang"] == lang + spec = imports.get(name) + if spec and "fqn" in spec and ok(spec["fqn"]): + return spec["fqn"] + if ok(_fqn(ns, name)): + return _fqn(ns, name) + if ok(name): + return name + cand = [c for c in classes if classes[c]["lang"] == lang + and c.rsplit("\\", 1)[-1].rsplit(".", 1)[-1] == name] + return cand[0] if len(cand) == 1 else None + + def method_in_hierarchy(cfqn: str | None, meth: str, seen=None) -> str | None: + seen = seen or set() + if not cfqn or cfqn in seen or cfqn not in classes: + return None + seen.add(cfqn) + cl = classes[cfqn] + if meth in cl["methods"]: + return cl["methods"][meth] + for b in cl["bases"]: + bf = resolve_class(b, cl.get("ns", ""), cl.get("imports", {}), cl["lang"]) + r = method_in_hierarchy(bf, meth, seen) if bf else None + if r: + return r + return None + + out: dict[str, dict[str, float]] = {} + + def add_edge(src_id: str, tid: str | None, conf: float) -> bool: + if tid and tid != src_id: + d = out.setdefault(src_id, {}) + d[tid] = max(d.get(tid, 0.0), conf) + return True + return False + + def _same_lang(ids, lang): + return [i for i in ids if nodes[i]["lang"] == lang] + + for sf, ns, imports, calls, qmap, class_fqn_by_q in pending: module_id: str | None = None + clang = sf.lang 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) + callee, rkind, rclass = c["callee"], c["rkind"], c["rclass"] + caller_q = c["caller"] + if caller_q and caller_q in qmap: + src_id = qmap[caller_q] + elif caller_q and any(caller_q == q or caller_q.startswith(q + ".") for q in qmap): src_id = next(qmap[q] for q in qmap - if caller == q or caller.startswith(q + ".")) + if caller_q == q or caller_q.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, { @@ -195,32 +418,116 @@ def build_graph(files: Iterable[tuple[walk.SourceFile, str]], *, "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) + + cur_class = class_fqn_by_q.get(c["class"], "") + if rkind in ("this", "self"): + if add_edge(src_id, method_in_hierarchy(cur_class, callee), C_HIER): + continue + elif rkind == "parent": + done = False + for b in classes.get(cur_class, {}).get("bases", []): + bf = resolve_class(b, ns, imports, clang) + if bf and add_edge(src_id, method_in_hierarchy(bf, callee), C_HIER): + done = True + break + if done: + continue + elif rkind in ("static", "new"): + cf = resolve_class(rclass, ns, imports, clang) + if cf: + tid = (method_in_hierarchy(cf, "__construct") or classes[cf]["id"] + if rkind == "new" else method_in_hierarchy(cf, callee)) + if add_edge(src_id, tid, C_STATIC): + continue + + # fonction libre / membre inconnu → import / namespace / fichier / nom + if rkind == "free" and callee in imports: + spec = imports[callee] + fq = _same_lang(by_fqn.get(spec.get("fqn", ""), []), clang) + if len(fq) == 1: + add_edge(src_id, fq[0], C_IMPORT) + continue + if spec.get("rel"): + tgt = _lookup_rel(file_defs, sf.repo, spec["rel"], spec.get("name", callee)) + if tgt and add_edge(src_id, tgt, C_IMPORT): + continue + # même fichier (locality forte) : on résout même un nom ubiquitaire local + same = file_defs.get((sf.repo, sf.path), {}).get(callee) + if same and len(same) == 1: + add_edge(src_id, same[0], C_SAMENS) + continue + # nom ubiquitaire (primitive du langage) ou trop défini → on s'arrête AVANT la + # résolution globale par nom (sinon une fonction interne homonyme d'un builtin, + # ex. `range`, absorberait tous les appels du builtin via le FQN unique). + if callee in blocked: + continue + ns_ids = _same_lang(by_fqn.get(_fqn(ns, callee), []), clang) if rkind == "free" else [] + if len(ns_ids) == 1: + add_edge(src_id, ns_ids[0], C_SAMENS) + continue + cand = _same_lang(by_name.get(callee, []), clang) + if not cand: + continue + if len(cand) == 1: + add_edge(src_id, cand[0], C_UNIQUE if rkind == "free" else C_MEMBER1) + else: + for tid2 in cand: + add_edge(src_id, tid2, C_AMBIG) + + out_lists = {s: list(d.items()) for s, d in out.items()} + fan_in: dict[str, int] = {} + for d in out.values(): + for t in d: + fan_in[t] = fan_in.get(t, 0) + 1 + centrality = _pagerank(nodes, out_lists) + for nid, node in nodes.items(): + node["fan_in"] = fan_in.get(nid, 0) + node["centrality"] = round(centrality.get(nid, 0.0), 7) return { "version": GRAPH_VERSION, "nodes": nodes, "by_name": by_name, - "out": {k: sorted(v) for k, v in out.items()}, + "by_fqn": by_fqn, + "out": {s: [[t, round(c, 2)] for t, c in d.items()] for s, d 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 _lookup_rel(file_defs: dict, repo: str, rel: str, name: str) -> str | None: + rel = rel.lstrip("/") + for ext in (".py", ".ts", ".tsx", ".js", ".jsx", "/__init__.py", "/index.ts", "/index.js"): + cand = file_defs.get((repo, rel + ext)) + if cand and name in cand and len(cand[name]) == 1: + return cand[name][0] + return None + + +def _pagerank(nodes: dict, out_lists: dict, d: float = 0.85, iters: int = 20) -> dict: + """PageRank pondéré par la confiance des arêtes (importance structurelle d'un symbole).""" + n = len(nodes) + if not n: + return {} + pr = {k: 1.0 / n for k in nodes} + wsum = {s: (sum(c for _, c in lst) or 1.0) for s, lst in out_lists.items()} + base = (1.0 - d) / n + dangling_keys = [k for k in nodes if k not in out_lists] + for _ in range(iters): + nxt = {k: base for k in nodes} + dshare = d * sum(pr[k] for k in dangling_keys) / n + for s, lst in out_lists.items(): + ps = pr[s] * d / wsum[s] + for t, c in lst: + if t in nxt: + nxt[t] += ps * c + for k in nodes: + nxt[k] += dshare + pr = nxt + return pr 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.""" + """Fichiers 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 @@ -232,35 +539,46 @@ def _iter_supported(manifest: dict, base_dir: Path, repos: list[str] | None, yield sf, text +def _is_minified(path: str, text: str) -> bool: + 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]) + + # ── 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`.""" + """Charge un graphe (.json/.json.gz) et calcule l'index inverse `in` (avec confiance).""" 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()} + g["in"] = _reverse(g.get("out", {})) return g +def _reverse(out: dict) -> dict: + rev: dict[str, list] = {} + for s, lst in out.items(): + for t, c in lst: + rev.setdefault(t, []).append([s, c]) + return rev + + 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.""" + """Nœuds correspondant à `symbol` : nom exact, FQN, suffixe de qname, ou chemin:ligne.""" 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 s in graph.get("by_fqn", {}): + return list(graph["by_fqn"][s]) if ":" in s: head = s.rsplit(":", 1) if head[1].isdigit(): @@ -269,80 +587,95 @@ def resolve_symbol(graph: dict, symbol: str) -> list[str]: 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 _subsystem(node: dict) -> str: + seg = node["path"].split("/", 1)[0] + return f"{node['repo']}/{seg}" if seg else node["repo"] + + +def _tier(conf: float) -> str: + return ("certain" if conf >= TIER_CERTAIN + else "probable" if conf >= TIER_PROBABLE else "incertain") + + +def _describe(node: dict, depth: int, conf: float, 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, + "confidence": round(conf, 2), "tier": _tier(conf), "subsystem": _subsystem(node), + "centrality": node.get("centrality", 0.0), "fan_in": node.get("fan_in", 0)} + + 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`. + max_nodes: int = 40, min_confidence: float = 0.0, + sidecar: dict | None = None) -> dict: + """BFS borné depuis `symbol`. direction=callers (impact) | callees (deps). - 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) + Certitude d'un nœud = **min des confiances** sur le meilleur chemin l'atteignant ; + résultats triés (certitude, centralité), groupés par sous-système.""" + if "in" not in graph: 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"): + if n and n["kind"] in _CLASS_KINDS: 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]] = [] + best: dict[str, tuple[int, float]] = {s: (0, 1.0) for s in seeds} + order: list[str] = [] 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)) + pconf = best[nid][1] + for tgt, c in edges.get(nid, []): + if c < min_confidence: + continue + path_conf = min(pconf, c) + if tgt not in best: + best[tgt] = (d, path_conf) + nxt.append(tgt) + order.append(tgt) + elif path_conf > best[tgt][1]: + best[tgt] = (best[tgt][0], path_conf) frontier = nxt - truncated = len(order) > max_nodes - impacted = [_describe(nodes[i], dep, sidecar) for i, dep in order[:max_nodes] if i in nodes] + impacted = [_describe(nodes[i], best[i][0], best[i][1], sidecar) + for i in order if i in nodes] + impacted.sort(key=lambda x: ({"certain": 0, "probable": 1, "incertain": 2}[x["tier"]], + -x["centrality"])) + truncated = len(impacted) > max_nodes + impacted = impacted[:max_nodes] + by_sub: dict[str, int] = {} + tiers = {"certain": 0, "probable": 0, "incertain": 0} + for x in impacted: + by_sub[x["subsystem"]] = by_sub.get(x["subsystem"], 0) + 1 + tiers[x["tier"]] += 1 return { - "symbol": symbol, - "direction": direction, - "depth": depth, - "roots": [_describe(nodes[r], 0, sidecar) for r in roots if r in nodes], + "symbol": symbol, "direction": direction, "depth": depth, + "roots": [_describe(nodes[r], 0, 1.0, sidecar) for r in roots if r in nodes], "ambiguous": len(roots) > 1, - "impacted": impacted, - "truncated": truncated, + "impacted": impacted, "truncated": truncated, + "by_subsystem": dict(sorted(by_sub.items(), key=lambda kv: -kv[1])), + "tiers": tiers, } -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: @@ -353,14 +686,15 @@ def _cmd_build(args: argparse.Namespace) -> int: 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()) + strong = sum(1 for v in g["out"].values() for _, c in v if c >= STRONG) 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}).") + print(f"Graphe v2 écrit : {out_path} — {len(g['nodes'])} nœuds, {n_edges} arêtes " + f"({strong} fortes ≥{STRONG}), {len(g['by_fqn'])} FQN.") return 0 @@ -370,44 +704,42 @@ def _cmd_impact(args: argparse.Namespace) -> int: 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) + print(f"Symbole introuvable : {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})") + t = res["tiers"] + print(f"# {args.symbol} — {verb} (prof {args.depth}) — " + f"{t['certain']} certains / {t['probable']} probables / {t['incertain']} incertains") 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']}") + print(f" ⚠ {len(res['roots'])} définitions portent ce nom.") + if res["by_subsystem"]: + print(" sous-systèmes : " + ", ".join(f"{k} ({v})" for k, v in res["by_subsystem"].items())) for n in res["impacted"]: url = f" {n['source_url']}" if n["source_url"] else "" - print(f" {'·' * n['depth']} {n['qname']} ({n['kind']}) " + print(f" [{n['tier']}] {n['qname']} ({n['kind']}) " f"{n['repo']}/{n['path']}:{n['start_line']}{url}") if res["truncated"]: - print(" … (résultats tronqués)") + print(" … (tronqué)") 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("--out", required=True) 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.add_argument("--max-fanout", type=int, default=6) 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("--graph", required=True) 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.add_argument("--meta", default=None) q.set_defaults(func=_cmd_impact) - args = ap.parse_args(argv) return args.func(args) diff --git a/tools/tests/test_code_index_graph.py b/tools/tests/test_code_index_graph.py index d5df421..c5a3654 100644 --- a/tools/tests/test_code_index_graph.py +++ b/tools/tests/test_code_index_graph.py @@ -1,6 +1,9 @@ -"""Tests du graphe d'appels (code_index.graph) : extraction, construction, code_impact.""" +"""Tests du graphe d'appels v2 (code_index.graph) : extraction riche, cascade de +résolution à confiance, code_impact (tiers/sous-système).""" from __future__ import annotations +from pathlib import Path + import pytest from code_index import graph @@ -8,131 +11,161 @@ 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) ───────────────────────────────────────── +def _edges(g, src_id): + """{tid: conf} des arêtes sortantes d'un nœud.""" + return {t: c for t, c in g["out"].get(src_id, [])} + + +# ── Extraction ────────────────────────────────────────────────────────────────── 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" + "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 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} + info = graph.extract_file("python", PY_SRC, "mod.py") + qnames = {d["qname"] for d in info["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' + triples = {(c["caller"], c["callee"], c["rkind"]) for c in info["calls"]} + assert ("alpha", "beta", "free") in triples + assert ("Gamma.m", "helper", "this") in triples # self.helper() → receveur this + assert ("Gamma.m", "alpha", "free") in triples -def test_extract_php_member_and_free_calls() -> None: +def test_extract_php_receivers() -> 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 + php = ("h(); self::s(); parent::b(); Geo::load(); new Geo(); }\n" + " function h() {}\n}\n") + info = graph.extract_file("php", php, "Svc.php") + assert info["ns"] == "App" + assert info["imports"].get("Geo", {}).get("fqn") == "App\\Util\\Geo" + svc = next(d for d in info["defs"] if d["qname"] == "Svc") + assert "Base" in svc["bases"] + kinds = {(c["callee"], c["rkind"], c["rclass"]) for c in info["calls"]} + assert ("h", "this", "") in kinds + assert ("s", "self", "") in kinds + assert ("b", "parent", "") in kinds + assert ("load", "static", "Geo") in kinds + assert ("__construct", "new", "Geo") in kinds 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", "") == ([], []) + assert graph.extract_file("ncl", "a = 1\n") == {"ns": "", "imports": {}, "defs": [], "calls": []} + assert graph.extract_file("python", "")["defs"] == [] -# ── Construction du graphe ────────────────────────────────────────────────────── +# ── Cascade de résolution ───────────────────────────────────────────────────────── -def test_build_graph_resolves_internal_calls() -> None: +def test_resolve_this_and_parent_via_hierarchy() -> None: + pytest.importorskip("tree_sitter_language_pack") + php = ("shared() + $this->own(); }\n" + " function own() { return 2; }\n}\n") + g = graph.build_graph([(_sf("r", "svc.php", "php"), php)]) + run_id = g["by_name"]["run"][0] + e = _edges(g, run_id) + shared_id = g["by_name"]["shared"][0] + own_id = g["by_name"]["own"][0] + # $this->shared() résolu dans la classe PARENT (hiérarchie) → confiance forte + assert e.get(shared_id) == graph.C_HIER + assert e.get(own_id) == graph.C_HIER + + +def test_resolve_static_and_new_via_use() -> 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: + files = [ + (_sf("r", "geo.php", "php"), + " 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 + # deux 'dup' GLOBAUX (ns="") + un appelant → ambigu (pas de fausse précision FQN) + files = [ + (_sf("r", "a.php", "php"), " None: + pytest.importorskip("tree_sitter_language_pack") + files = [ + (_sf("r", "u.php", "php"), " 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é + files.append((_sf("r", "many.py", "python"), + "\n".join(f"def c{i}():\n x.push()\n" for i in range(5)))) + g = graph.build_graph(files, max_call_freq=3) 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) + assert push_id not in graph._reverse(g["out"]) # 'push' bloqué (5 appels > 3) + + +def test_pagerank_and_fanin_present() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = _chain_graph() + for n in g["nodes"].values(): + assert "centrality" in n and "fan_in" in n 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 graph._is_minified("a.js", "var x=1;" + "a" * 3000) assert not graph._is_minified("src/app.js", "const x = 1\nconst y = 2\n") -# ── code_impact (BFS) ─────────────────────────────────────────────────────────── +# ── code_impact ─────────────────────────────────────────────────────────────────── 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)]) + return graph.build_graph([(_sf("r", "pkg/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"} @@ -140,11 +173,33 @@ def test_code_impact_callers_blast_radius() -> None: 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_tiers_and_subsystem() -> None: + pytest.importorskip("tree_sitter_language_pack") + g = _chain_graph() + res = graph.code_impact(g, "c", direction="callers", depth=2) + # b et c sont résolus en nom unique (≥0.75) → tier certain/probable, jamais d'erreur + assert set(res["tiers"]) == {"certain", "probable", "incertain"} + assert all(n["tier"] in ("certain", "probable", "incertain") for n in res["impacted"]) + assert res["by_subsystem"] # regroupé par repo/dossier de tête + assert all("/" in k for k in res["by_subsystem"]) + + +def test_code_impact_ambiguous_is_incertain() -> None: + pytest.importorskip("tree_sitter_language_pack") + files = [ + (_sf("r", "a.php", "php"), " None: pytest.importorskip("tree_sitter_language_pack") g = _chain_graph() @@ -152,24 +207,17 @@ def test_code_impact_unknown_symbol() -> None: assert res["roots"] == [] and res["impacted"] == [] -def test_load_graph_roundtrip_and_reverse(tmp_path) -> None: +def test_load_graph_roundtrip_and_gzip(tmp_path) -> None: pytest.importorskip("tree_sitter_language_pack") + import gzip 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é + assert "in" in loaded 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"] + pgz = tmp_path / "graph.json.gz" + pgz.write_bytes(gzip.compress(json.dumps(g).encode("utf-8"))) + assert graph.load_graph(pgz)["nodes"] == g["nodes"]