Joshua Sundance Bailey
loosecanvas: local AI thought-mapping canvas with a trust-tagged knowledge graph
6d1438c | """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 | |