"""M05 — Python AST Extractor. Red/green tests derived from the module spec. Chosen public API (documented here, asserted below):: extract_python_file( source: str, relative_path: str, *, source_id: str = "", ) -> tuple[list[Node], list[Edge], list[RawImport]] ``source_id`` defaults to the normalised ``relative_path`` when empty. Extraction is pure on text: ``SyntaxError`` is caught internally and yields ``([], [], [])``. A ``PythonExtractor(ast.NodeVisitor)`` with a scope stack does the work internally; a ``read_source(path)`` helper enforces the 1 MiB / encoding skip rules for callers. Metadata stringification convention (``Node.metadata`` is ``dict[str, str]``): - Booleans -> the lowercase strings ``"true"`` / ``"false"`` (``is_method``, ``is_async``, ``is_property``). - Lists -> compact ``json.dumps`` (``lambda_names``, ``pytest_markers``). """ from __future__ import annotations import json from pathlib import Path from loosecanvas.contracts import Edge, Node, RawImport from loosecanvas.extractors.ast_extractor import extract_python_file, read_source _REL = "src/auth/providers.py" SAMPLE = '''\ import os from collections import OrderedDict from . import sibling from .pkg import thing as t from m import * from x import a, b class Animal: """base""" def __init__(self, name): self.name = name @property def label(self): return self.name async def fetch(self): return 1 class Meta: ordering = "name" class Dog(Animal): def bark(self): adder = lambda value: value + 1 # noqa: E731 return adder(1) ''' def _ids(nodes: list[Node]) -> list[str]: return [n.id for n in nodes] def _by_id(nodes: list[Node], node_id: str) -> Node: for n in nodes: if n.id == node_id: return n raise AssertionError(f"node not found: {node_id}\navailable: {_ids(nodes)}") def _has_edge(edges: list[Edge], source: str, target: str, kind: str) -> bool: return any( e.source == source and e.target == target and e.kind == kind for e in edges ) def test_class_method_and_async_nodes() -> None: nodes, _edges, _imports = extract_python_file(SAMPLE, _REL) ids = set(_ids(nodes)) assert f"{_REL}::class::Animal" in ids assert f"{_REL}::function::Animal.__init__" in ids assert f"{_REL}::function::Animal.label" in ids assert f"{_REL}::function::Animal.fetch" in ids init = _by_id(nodes, f"{_REL}::function::Animal.__init__") assert init.kind == "function" assert init.metadata["is_method"] == "true" assert init.metadata["is_async"] == "false" fetch = _by_id(nodes, f"{_REL}::function::Animal.fetch") assert fetch.metadata["is_async"] == "true" assert fetch.metadata["is_method"] == "true" def test_nested_class_qualified_name_uses_dot() -> None: nodes, _edges, _imports = extract_python_file(SAMPLE, _REL) assert f"{_REL}::class::Animal.Meta" in set(_ids(nodes)) def test_property_metadata() -> None: nodes, _edges, _imports = extract_python_file(SAMPLE, _REL) label = _by_id(nodes, f"{_REL}::function::Animal.label") assert label.metadata["is_property"] == "true" def test_inherits_edge() -> None: nodes, edges, _imports = extract_python_file(SAMPLE, _REL) dog = f"{_REL}::class::Dog" assert dog in set(_ids(nodes)) assert any( e.source == dog and e.kind == "inherits" and "Animal" in e.target for e in edges ) def test_contains_edges_link_module_class_function() -> None: nodes, edges, _imports = extract_python_file(SAMPLE, _REL) module_id = f"{_REL}::module::providers" animal = f"{_REL}::class::Animal" init = f"{_REL}::function::Animal.__init__" assert module_id in set(_ids(nodes)) assert _has_edge(edges, module_id, animal, "contains") assert _has_edge(edges, animal, init, "contains") def test_lambda_names_recorded_on_parent() -> None: nodes, _edges, _imports = extract_python_file(SAMPLE, _REL) bark = _by_id(nodes, f"{_REL}::function::Dog.bark") assert json.loads(bark.metadata["lambda_names"]) == ["adder"] def test_imports_collected_as_rawimport() -> None: _nodes, _edges, imports = extract_python_file(SAMPLE, _REL) def find(pred: object) -> RawImport: # noqa: ANN401 - test helper return next(i for i in imports if pred(i)) # type: ignore[operator] os_imp = find(lambda i: i.module == "os") assert os_imp.names == [] and os_imp.level == 0 from_x = find(lambda i: i.module == "x") assert from_x.names == ["a", "b"] and from_x.level == 0 rel = find(lambda i: i.level == 1 and i.module == "") assert rel.names == ["sibling"] star = find(lambda i: i.names == ["*"]) assert star.module == "m" assert all(i.current_file == _REL for i in imports) def test_parametrize_marker_recorded() -> None: src = ( "import pytest\n\n\n" '@pytest.mark.parametrize("x", [1, 2])\n' "def test_thing(x):\n" " assert x\n" ) nodes, _edges, _imports = extract_python_file(src, "tests/test_thing.py") node = _by_id(nodes, "tests/test_thing.py::function::test_thing") assert "parametrize" in json.loads(node.metadata["pytest_markers"]) def test_every_node_has_valid_evidence() -> None: nodes, _edges, _imports = extract_python_file(SAMPLE, _REL) assert nodes for n in nodes: assert len(n.evidence_refs) >= 1 ev = n.evidence_refs[0] assert ev.snippet_hash assert ev.start_line <= ev.end_line def test_node_ids_have_no_backslash_and_three_parts() -> None: nodes, _edges, _imports = extract_python_file(SAMPLE, "src\\auth\\providers.py") for n in nodes: assert "\\" not in n.id assert len(n.id.split("::")) == 3 def test_syntax_error_handled_gracefully() -> None: nodes, edges, imports = extract_python_file("def broken(:\n pass\n", _REL) assert nodes == [] assert edges == [] assert imports == [] def test_determinism_no_id_collisions() -> None: first, _e1, _i1 = extract_python_file(SAMPLE, _REL) second, _e2, _i2 = extract_python_file(SAMPLE, _REL) ids = _ids(first) assert ids == _ids(second) assert len(ids) == len(set(ids)) def test_init_py_dual_nodes(tmp_path: Path) -> None: nodes, _edges, _imports = extract_python_file("x = 1\n", "pkg/sub/__init__.py") ids = set(_ids(nodes)) # module node uses the package (parent dir) qualified name, never "__init__". assert "pkg/sub/__init__.py::module::sub" in ids assert not any(i.endswith("::module::__init__") for i in ids) def test_read_source_skips_large(tmp_path: Path) -> None: big = tmp_path / "big.py" big.write_bytes(b"# " + b"x" * 1_100_000) assert read_source(big) is None def test_read_source_latin1_fallback(tmp_path: Path) -> None: p = tmp_path / "lat.py" p.write_bytes(b"# calf\xe9\n") # invalid utf-8, valid latin-1 text = read_source(p) assert text is not None and "calf" in text