diff --git a/scripts/notebook_tools/audit_engine_named_not_invoked.py b/scripts/notebook_tools/audit_engine_named_not_invoked.py index 5489e6d252..b118b1daf4 100644 --- a/scripts/notebook_tools/audit_engine_named_not_invoked.py +++ b/scripts/notebook_tools/audit_engine_named_not_invoked.py @@ -194,13 +194,12 @@ class SurfaceHits: simulation_hits: List[Tuple[int, str]] = field(default_factory=list) -def _strip_comments(src: str) -> str: - """Retire les commentaires `#` et les docstrings pour le scan wiring/proof. +def _strip_comments_and_strings(src: str) -> str: + """Retire commentaires, docstrings, et chaines de caracteres pour le scan wiring. - Un match dans un commentaire n'est pas une preuve d'invocation -- c'est - juste une mention discursive. On garde les chaines de caracteres - (print("openai...")) parce qu'elles peuvent etre des appels caches, - mais on retire les commentaires Python purs. + Un match dans un commentaire OU dans une string n'est pas une preuve + d'invocation SDK -- c'est une mention discursive ou un exemple pedagogique. + On garde uniquement le code actif (appels reels, imports reels). """ lines = src.split("\n") out = [] @@ -208,19 +207,38 @@ def _strip_comments(src: str) -> str: stripped = line.lstrip() if stripped.startswith("#"): continue - # Retrait d'un commentaire en milieu de ligne (apres du code) - # Heuristique : `#` non precede d'un caractere alphanumerique (donc pas dans une string) - # Simplification : on retire tout apres le premier `#` qui n'est pas dans une string. - # Les strings simples avec `#` (`"#hashtag"`) ne sont pas affectees par notre regex wiring - # (qui cherche `google.adk` etc.), donc cette approximation suffit. - in_str = False - for i, c in enumerate(line): - if c == '"' or c == "'": - in_str = not in_str - elif c == "#" and not in_str: - line = line[:i] + # Parse caractere par caractere pour reperer les commentaires et strings. + in_str = None # None si hors string, '"' ou "'" sinon + new_line_chars = [] + i = 0 + while i < len(line): + c = line[i] + if in_str: + if c == in_str: + in_str = None + new_line_chars.append(c) if False else None # on jette le contenu des strings + i += 1 + elif c == "#": + # Commentaire in-line : on tronque break - out.append(line) + elif c == '"' or c == "'": + in_str = c + # On remplace le contenu de la string par des espaces (preserve positions pour l'erreur) + new_line_chars.append(c) + # Skip le contenu de la string + j = i + 1 + while j < len(line) and line[j] != in_str: + if line[j] == "\\" and j + 1 < len(line): + j += 2 + else: + j += 1 + # Ajouter des espaces a la place du contenu + new_line_chars.append(" " * (j - i - 1)) + i = j + 1 if j < len(line) else j + else: + new_line_chars.append(c) + i += 1 + out.append("".join(new_line_chars)) return "\n".join(out) @@ -233,7 +251,7 @@ def _scan_engine(notebook: dict, spec: EngineSpec) -> SurfaceHits: out_text = _cell_output_text(cell) if ctype == "code" else "" # Code sans commentaires : utilise pour wiring (un import dans un # commentaire n'est pas un import reel). - code_clean = _strip_comments(src) if ctype == "code" else "" + code_clean = _strip_comments_and_strings(src) if ctype == "code" else "" # --- claim : markdown prose --- if ctype == "markdown": @@ -363,10 +381,14 @@ def classify_notebook( verdict = "ENGINE_EXEC_PROVED" elif disclosed and (siblings is None or _detect_disclosed_sequence(notebook_path, spec, hits, siblings) is True): verdict = "DISCLOSED_SEQUENCE_PROVED" - elif has_wiring and not has_proof: - verdict = "WIRING_ONLY" elif has_simulation: + # Simulation detectee : un import SDK + un output simule = le wiring + # est cosmétique, le moteur reel n'est pas invoqué. Prend precedence + # sur WIRING_ONLY car le verdict `SIMULATED_TERMINAL` est plus + # informatif pour le lecteur (cycle 93 feedback). verdict = "SIMULATED_TERMINAL" + elif has_wiring and not has_proof: + verdict = "WIRING_ONLY" elif not has_wiring: verdict = "NAMED_NOT_INVOKED" else: diff --git a/scripts/notebook_tools/tests/test_audit_engine_named_not_invoked.py b/scripts/notebook_tools/tests/test_audit_engine_named_not_invoked.py index ca889fe446..ba6fca8109 100644 --- a/scripts/notebook_tools/tests/test_audit_engine_named_not_invoked.py +++ b/scripts/notebook_tools/tests/test_audit_engine_named_not_invoked.py @@ -388,4 +388,204 @@ def test_scan_repo_finds_track2_adk_baseline(tmp_path): if "google_adk" in nb_results: assert nb_results["google_adk"]["verdict"] == "NAMED_NOT_INVOKED" found = True - assert found, "Lab10 Track2 devrait declencher NAMED_NOT_INVOKED" \ No newline at end of file + assert found, "Lab10 Track2 devrait declencher NAMED_NOT_INVOKED" + + +# --- Coverage gaps identified by empirical --scan-all cycle 92 ------------- + +def test_wiring_inside_string_literal_ignored(): + """Un import SDK place dans une string (print('import openai')) n'est PAS + un wiring reel. Le scanner doit ignorer les chaines et considerer que + la cellule n'a pas d'import. + """ + nb = _nb( + md_cells=["# GPT-4 integration"], + code_cells=[ + ( + "msg = 'import openai'\nprint(msg)", + _out_stream("import openai"), + ), + ], + ) + p = Path("/tmp/_test_string_import.ipynb") + p.write_text(json.dumps(nb)) + try: + results = classify_notebook(p, nb, ["openai_llm"], {p.parent.resolve(): [p]}) + # L'output textuel contient 'import openai' mais ce n'est pas un wiring + # car il est dans une string. La cellule n'a pas d'import SDK reel. + # Cependant l'output (print(msg)) est reel, et il y a un claim LLM : + # c'est WIRING_ONLY si on considere le wiring absent, ou NAMED_NOT_INVOKED + # si on est strict. Le scanner doit signaler NAMED_NOT_INVOKED + # (pas de vrai wiring SDK). + assert results["openai_llm"]["verdict"] == "NAMED_NOT_INVOKED" + finally: + p.unlink() + + +def test_simulation_marker_only_in_output(): + """Si seul l'output contient 'Reponse simulee' (pas le code source), + le scanner doit toujours detecter la simulation. + """ + nb = _nb( + md_cells=["# GPT-4 integration via API"], + code_cells=[ + ( + "import openai\nresponse = openai.ChatCompletion.create()\nprint(response.choices[0].message.content)", + _out_stream("Reponse simulee : bonjour"), + ), + ], + ) + p = Path("/tmp/_test_simulation_in_output.ipynb") + p.write_text(json.dumps(nb)) + try: + results = classify_notebook(p, nb, ["openai_llm"], {p.parent.resolve(): [p]}) + assert results["openai_llm"]["verdict"] == "SIMULATED_TERMINAL" + assert results["openai_llm"]["simulation"], "simulation_marker devrait etre detecte" + assert not results["openai_llm"]["proof"], "proof devrait etre vide a cause de la simulation" + finally: + p.unlink() + + +def test_claim_only_in_objectives_metadata(): + """Le claim peut etre dans la cellule objectifs (premiere markdown) au lieu + du titre. Le scanner doit scanner toutes les cellules markdown du notebook, + pas seulement la premiere. + """ + nb = _nb( + md_cells=[ + "# Lab standard\nObjectifs pedagogiques.", + "## Details\nCe notebook integre le **Google ADK** pour demonstrer les patterns.", + ], + code_cells=[ + ("print('placeholder')", _out_stream("placeholder")), + ], + ) + p = Path("/tmp/_test_claim_in_objectives.ipynb") + p.write_text(json.dumps(nb)) + try: + results = classify_notebook(p, nb, ["google_adk"], {p.parent.resolve(): [p]}) + assert "google_adk" in results + assert results["google_adk"]["verdict"] == "NAMED_NOT_INVOKED" + finally: + p.unlink() + + +def test_engine_filter_limits_scan(): + """Le filtre --engine doit limiter le scan au(x) moteur(s) demande(s).""" + nb = _nb( + md_cells=["# Lab\nGoogle ADK et BigQuery sont mentionnes ici."], + code_cells=[ + ("from google.cloud import bigquery", _out_stream("Client() OK")), + ], + ) + p = Path("/tmp/_test_engine_filter.ipynb") + p.write_text(json.dumps(nb)) + try: + # Scan limite a bigquery : seul bigquery devrait apparaitre dans les results + results_bq = classify_notebook(p, nb, ["bigquery"], {p.parent.resolve(): [p]}) + assert "bigquery" in results_bq + assert "google_adk" not in results_bq + # Scan complet (tous moteurs) : les deux devraient apparaitre + results_all = classify_notebook(p, nb, None, {p.parent.resolve(): [p]}) + assert "google_adk" in results_all + assert "bigquery" in results_all + finally: + p.unlink() + + +def test_disclosed_deterministic_with_successor_executed(): + """Le verdict DISCLOSED_SEQUENCE_PROVED exige un successeur dans la meme + serie-sibling qui ait wiring+proof. Ce test verifie que le scanner cherche + bien dans le cache sibling, pas seulement le notebook courant. + """ + p_deterministic = Path("/tmp/_test_seq_det.ipynb") + p_real = Path("/tmp/_test_seq_real.ipynb") + + # Notebook deterministe qui mentionne Google ADK dans sa prose comme + # etant realise dans le prochain notebook. + nb_det = _nb( + md_cells=[ + "# Setup deterministe\nCe notebook est deterministe (sans LLM).\n" + "Il prepare les inputs pour le **Google ADK** runtime du notebook suivant.", + ], + code_cells=[ + ("x = 1\nprint(x)", _out_stream("1")), + ], + ) + # Notebook suivant dans la serie : wiring et proof reels. + nb_real = _nb( + md_cells=["# Real Google ADK runtime"], + code_cells=[ + ("from google.adk import Agent\nprint('agent ready')", _out_stream("agent ready")), + ], + ) + p_deterministic.write_text(json.dumps(nb_det)) + p_real.write_text(json.dumps(nb_real)) + try: + # Le cache sibling DOIT inclure les deux notebooks dans le bon ordre. + siblings = [p_deterministic, p_real] + cache = {p_deterministic.parent.resolve(): siblings} + results = classify_notebook(p_deterministic, nb_det, ["google_adk"], cache) + assert "google_adk" in results + # Verdict attendu : DISCLOSED_SEQUENCE_PROVED (le notebook suivant dans + # la meme serie a wiring+proof reels). + assert results["google_adk"]["verdict"] == "DISCLOSED_SEQUENCE_PROVED" + finally: + p_deterministic.unlink() + p_real.unlink() + + +def test_disclosed_deterministic_without_successor_falls_back(): + """Si un notebook est deterministe ET claim Google ADK mais qu'il n'y a + PAS de successeur dans la serie, le scanner doit retomber sur + NAMED_NOT_INVOKED (pas inventer un successeur). + """ + p = Path("/tmp/_test_det_no_successor.ipynb") + nb = _nb( + md_cells=[ + "# Lab isole\nDeterministe, sans LLM.\nMais le **Google ADK** est mentionne.", + ], + code_cells=[ + ("x = 1", _out_stream("1")), + ], + ) + p.write_text(json.dumps(nb)) + try: + # Pas de successeur dans la cache (un seul notebook) + siblings = [p] + cache = {p.parent.resolve(): siblings} + results = classify_notebook(p, nb, ["google_adk"], cache) + assert "google_adk" in results + # Pas de successeur -> NAMED_NOT_INVOKED (le claim n'est pas prouve) + assert results["google_adk"]["verdict"] == "NAMED_NOT_INVOKED" + finally: + p.unlink() + + +def test_scan_repo_excludes_nested_output_dir(tmp_path): + """_iter_notebooks doit exclure les sous-dossiers _output/_executed + pour eviter de scanner des artefacts en double. + """ + # Un notebook normal + (tmp_path / "main.ipynb").write_text(json.dumps(_nb( + md_cells=["# Lab\nGoogle ADK integration"], + code_cells=[("print(1)", _out_stream("1"))], + ))) + # Un artefact _output au meme endroit + (tmp_path / "main_output.ipynb").write_text(json.dumps(_nb( + md_cells=["# _output artefact"], + code_cells=[("print(1)", _out_stream("1"))], + ))) + # Un sous-dossier _archive (doit etre scanne si pas exclu par le suffix) + archive_dir = tmp_path / "_archive" + archive_dir.mkdir() + (archive_dir / "old.ipynb").write_text(json.dumps(_nb( + md_cells=["# Archive"], + code_cells=[("print(1)", _out_stream("1"))], + ))) + results = scan_repo(tmp_path) + # Doit inclure main mais pas main_output + found_main = any("main.ipynb" in nb_path and "_output" not in nb_path for nb_path in results) + found_output_artefact = any("main_output.ipynb" in nb_path for nb_path in results) + assert found_main, "main.ipynb doit etre scanne" + assert not found_output_artefact, "main_output.ipynb doit etre exclu" \ No newline at end of file