test(schema): canonical-schema conformance gate (L1)

Adds assert_conformant() check verifying schema_version, relative symbol_table
keys, module sources, callable span validity, and absent null values. Fixes
core.py to emit relative paths for symbol_table keys per v2 schema spec.

Author: rahlk

codeanalyzer/core.py

@@ -564,9 +564,9 @@ def _build_symbol_table(self, cached_symbol_table: Optional[Dict[str, PyModule]]
         if self.file_name is not None:
             single_file = self.project_dir / self.file_name
             logger.info(f"Analyzing single file: {single_file}")
-            
+
             # Check if file is in cache and unchanged
-            file_key = str(single_file)
+            file_key = str(single_file.relative_to(self.project_dir))
             if file_key in cached_symbol_table and not self.rebuild_analysis:
                 # Compute file checksum to see if it changed
                 if self._file_unchanged(single_file, cached_symbol_table[file_key]):
@@ -616,7 +616,7 @@ def _build_symbol_table(self, cached_symbol_table: Optional[Dict[str, PyModule]]
             # Separate files into cached and new/changed
             files_to_process = []
             for py_file in py_files:
-                file_key = str(py_file)
+                file_key = str(py_file.relative_to(self.project_dir))
                 if file_key in cached_symbol_table and not self.rebuild_analysis:
                     if self._file_unchanged(py_file, cached_symbol_table[file_key]):
                         # Use cached version
@@ -644,7 +644,7 @@ def _build_symbol_table(self, cached_symbol_table: Optional[Dict[str, PyModule]]
 
             with ProgressBar(len(py_files), "Building symbol table") as progress:
                 for py_file in py_files:
-                    file_key = str(py_file)
+                    file_key = str(py_file.relative_to(self.project_dir))
 
                     # Check if file is cached and unchanged
                     if file_key in cached_symbol_table and not self.rebuild_analysis:

test/__init__.py

@@ -0,0 +1,52 @@
+def _assert_no_nulls(obj, path="$"):
+    if obj is None:
+        raise AssertionError(f"unexpected null at {path} (exclude_none must drop it)")
+    if isinstance(obj, dict):
+        for k, v in obj.items():
+            _assert_no_nulls(v, f"{path}.{k}")
+    elif isinstance(obj, list):
+        for i, v in enumerate(obj):
+            _assert_no_nulls(v, f"{path}[{i}]")
+
+
+def _iter_callables(app):
+    def walk_callable(c):
+        yield c
+        for ic in (c.get("inner_callables") or {}).values():
+            yield from walk_callable(ic)
+        for cl in (c.get("inner_classes") or {}).values():
+            yield from walk_class(cl)
+
+    def walk_class(cl):
+        for m in (cl.get("methods") or {}).values():
+            yield from walk_callable(m)
+        for ic in (cl.get("inner_classes") or {}).values():
+            yield from walk_class(ic)
+
+    for mod in app["symbol_table"].values():
+        for fn in (mod.get("functions") or {}).values():
+            yield mod, fn
+        for cl in (mod.get("classes") or {}).values():
+            for m in walk_class(cl):
+                yield mod, m
+
+
+def assert_conformant(payload: dict, max_level: int) -> None:
+    _assert_no_nulls(payload)
+    assert payload["schema_version"] == "2.0.0"
+    app = payload["application"]
+    for key, mod in app["symbol_table"].items():
+        assert not key.startswith("/") and ".." not in key, f"non-relative key {key}"
+        assert isinstance(mod.get("source"), str) and mod["source"], f"module {key} missing source"
+    for mod, c in _iter_callables(app):
+        lo, hi = c["span"]["bytes"]
+        text = mod["source"].encode("utf-8")[lo:hi].decode("utf-8")
+        assert text.lstrip().startswith(("def ", "async def ", "@")), f"{c['id']} span mismatch"
+        for node in c.get("body", {}).values():
+            if node["kind"] == "call" and max_level >= 2:
+                assert node.get("callee") is None or isinstance(node["callee"], str)
+    for mod, c in _iter_callables(app):
+        node_ids = set(c.get("body", {}).keys())
+        for lst in ("cfg", "cdg", "ddg", "summary"):
+            for e in c.get(lst, []):
+                assert e["src"] in node_ids and e["dst"] in node_ids, f"dangling {lst} in {c['id']}"

test/conftest_v2.py

@@ -33,3 +33,15 @@ def test_cli_emits_v2_envelope(tmp_path: Path):
     assert payload["max_level"] == 1
     assert payload["application"]["kind"] == "application"
     assert "program_graphs" not in payload  # dissolved into the tree
+
+
+def test_l1_output_is_conformant(tmp_path: Path):
+    from test.conftest_v2 import assert_conformant
+    proj = tmp_path / "proj"; proj.mkdir()
+    (proj / "pkg").mkdir()
+    (proj / "pkg" / "m.py").write_text("def f(a):\n    return a\n", encoding="utf-8")
+    out = subprocess.run(
+        [sys.executable, "-m", "codeanalyzer", "-i", str(proj), "-a", "1", "--no-venv"],
+        capture_output=True, text=True, check=True,
+    ).stdout
+    assert_conformant(json.loads(out), max_level=1)