test: language-packs step 0 — identity net, forbidden-edge gate, plane reports

.github/workflows/ci.yml

@@ -162,6 +162,18 @@ jobs:
           # this exemption hides it — that is the cost, and it is the reason the
           # rule stays on for src/.
           #
+          # pythonsecurity:S8706 (connection-string injection) is the same
+          # finding at sqlite3.connect: graph_identity.py opens a database named
+          # on its own command line. `dump()` requires an existing .db/.sqlite/
+          # .sqlite3 regular file before connecting, and passes no `uri=True`, so
+          # the argument is a file name and never a connection string. Exempted
+          # for scripts/ on the S2083 reasoning above — the engine does not model
+          # the check — and kept on for src/.
+          #
+          # tests/fixtures/ is excluded outright: it holds input corpora, not
+          # code. Its package.json files exist to be parsed by cgis, and a lock
+          # file next to them (text:S8564) would be a fixture nothing reads.
+          #
           # `sonar.newCode.referenceBranch` overrides the server-side new-code
           # definition for this analysis. The instance default is
           # `previous_version` (confirmed: sonar.leak.period=previous_version,
@@ -218,13 +230,17 @@ jobs:
             -Dsonar.newCode.referenceBranch=main
             -Dsonar.sources=src,scripts,.github
             -Dsonar.tests=tests
+            -Dsonar.exclusions=tests/fixtures/**
+            -Dsonar.test.exclusions=tests/fixtures/**
             -Dsonar.python.version=3.12
             -Dsonar.python.coverage.reportPaths=coverage.xml
-            -Dsonar.issue.ignore.multicriteria=e1,e2,e3,e4
+            -Dsonar.issue.ignore.multicriteria=e1,e2,e3,e4,e5
             -Dsonar.issue.ignore.multicriteria.e1.ruleKey=pythonsecurity:S8707
             -Dsonar.issue.ignore.multicriteria.e1.resourceKey=scripts/**/*.py
             -Dsonar.issue.ignore.multicriteria.e4.ruleKey=pythonsecurity:S2083
             -Dsonar.issue.ignore.multicriteria.e4.resourceKey=scripts/**/*.py
+            -Dsonar.issue.ignore.multicriteria.e5.ruleKey=pythonsecurity:S8706
+            -Dsonar.issue.ignore.multicriteria.e5.resourceKey=scripts/**/*.py
             -Dsonar.issue.ignore.multicriteria.e2.ruleKey=githubactions:S8541
             -Dsonar.issue.ignore.multicriteria.e2.resourceKey=.github/workflows/*.yml
             -Dsonar.issue.ignore.multicriteria.e3.ruleKey=githubactions:S8544

.gitignore

@@ -15,6 +15,8 @@ downloads/
 eggs/
 .eggs/
 lib/
+# The graph-identity corpus has lib/ directories that are inputs, not build output.
+!tests/fixtures/graph_identity/corpus/**/lib/
 lib64/
 parts/
 sdist/

pyproject.toml

@@ -101,6 +101,9 @@ codegraph-brain = "cgis.api.mcp_server:mcp.run"
 # Target Python 3.12+
 target-version = "py312"
 line-length = 100
+# Fixture corpora are inputs, not code: reformatting one changes what the
+# graph-identity snapshot measures (docs/plans/2026-10-14-language-packs.md).
+extend-exclude = ["tests/fixtures"]
 
 [tool.ruff.lint]
 # Enable many plugins: 
@@ -237,6 +240,8 @@ color = true
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+# pytest's defaults plus `fixtures`: the identity corpus holds a test file of its own.
+norecursedirs = [".*", "*.egg", "_darcs", "build", "CVS", "dist", "node_modules", "venv", "{arch}", "fixtures"]
 asyncio_mode = "auto"
 markers = [
     "slow: marks tests as slow — run with '-m slow' or skip with '-m \"not slow\"'",

scripts/graph_identity.py

@@ -0,0 +1,172 @@
+"""Canonical dump and diff of a stored cgis graph — the language-packs identity net.
+
+Steps 1-5 of docs/specs/2026-10-14-language-packs-design.md move code without
+changing behaviour. This is how that is checked: the same input ingested by two
+cgis versions must dump identically. Absolute paths become `<ROOT>` and the
+mtime-derived `ingested_at` is dropped, so a dump compares across machines and
+temp directories.
+"""
+
+import argparse
+import json
+import shutil
+import sqlite3
+import sys
+import tempfile
+from collections.abc import Mapping
+from contextlib import closing
+from pathlib import Path
+from typing import cast
+
+from cgis.extractors.registry import build_extractors
+from cgis.pipeline import IngestionPipeline
+from cgis.storage.sqlite_store import SQLiteStore
+
+_NODE_COLUMNS = (
+    "id", "type", "name", "file_path", "start_line", "end_line", "language",
+    "namespace", "is_test", "is_generated", "metadata", "ontology_class",
+    "domains", "confidence_score",
+)  # fmt: skip
+_EDGE_COLUMNS = (
+    "id", "source", "target", "type", "weight", "confidence", "context",
+    "file_path", "line_number", "type_only",
+)  # fmt: skip
+_VOLATILE_STATE = frozenset({"ingested_at"})
+_DB_SUFFIXES = frozenset({".db", ".sqlite", ".sqlite3"})
+
+#: The corpus roots a snapshot ingests, by name. `sub` reads configs above its
+#: root (#555); `py_src` is the src layout, where the local `alembic` package
+#: meets the declared `alembic` dependency.
+SNAPSHOT_ROOTS: dict[str, str] = {"whole": ".", "sub": "ts/apps/web/src", "py_src": "py/src"}
+
+
+def _normalise(value: object, root: str) -> object:
+    """`value` with the corpus root replaced by `<ROOT>`, JSON text parsed to data."""
+    if isinstance(value, str):
+        text = value.replace(root, "<ROOT>")
+        if text[:1] in "[{":
+            try:
+                return _normalise(json.loads(text), root)
+            except ValueError:
+                return text
+        return text
+    if isinstance(value, list):
+        return [_normalise(v, root) for v in value]
+    if isinstance(value, dict):
+        return {k: _normalise(v, root) for k, v in sorted(value.items())}
+    return value
+
+
+def _rows(
+    conn: sqlite3.Connection, table: str, columns: tuple[str, ...], root: str
+) -> list[dict[str, object]]:
+    """Every row of `table` as a normalised dict, sorted by id."""
+    query = f"SELECT {', '.join(columns)} FROM {table} ORDER BY id"  # table/columns are constants
+    return [
+        {col: _normalise(val, root) for col, val in zip(columns, row, strict=True)}
+        for row in conn.execute(query)
+    ]
+
+
+def dump(db_path: Path, root: Path) -> dict[str, object]:
+    """The canonical content of one stored graph, with `root` normalised away.
+
+    `db_path` must be an existing database file: `sqlite3.connect` would create
+    a missing one, and its URI-free path keeps the argument a plain file name.
+    """
+    if db_path.suffix not in _DB_SUFFIXES:
+        msg = f"not a graph database (expected {sorted(_DB_SUFFIXES)}): {db_path}"
+        raise ValueError(msg)
+    if not db_path.is_file():
+        msg = f"no graph database at {db_path}"
+        raise FileNotFoundError(msg)
+    prefix = str(root.resolve())
+    with closing(sqlite3.connect(db_path)) as conn:  # `with conn` alone never closes
+        state = {
+            key: _normalise(value, prefix)
+            for key, value in conn.execute("SELECT key, value FROM ingest_state ORDER BY key")
+            if key not in _VOLATILE_STATE
+        }
+        return {
+            "nodes": _rows(conn, "nodes", _NODE_COLUMNS, prefix),
+            "edges": _rows(conn, "edges", _EDGE_COLUMNS, prefix),
+            "ingest_state": state,
+        }
+
+
+Row = dict[str, object]
+
+
+def _by_id(dumped: Mapping[str, object], table: str) -> dict[object, Row]:
+    """One table of a dump, keyed by row id."""
+    return {row["id"]: row for row in cast("list[Row]", dumped[table])}
+
+
+def diff(before: Mapping[str, object], after: Mapping[str, object]) -> list[str]:
+    """One line per row that differs between two dumps, keyed by table and id."""
+    lines: list[str] = []
+    for table in ("nodes", "edges"):
+        old, new = _by_id(before, table), _by_id(after, table)
+        lines.extend(
+            f"{table} {key}: {json.dumps(old.get(key))} → {json.dumps(new.get(key))}"
+            for key in sorted(old.keys() | new.keys(), key=str)
+            if old.get(key) != new.get(key)
+        )
+    if before["ingest_state"] != after["ingest_state"]:
+        lines.append(f"ingest_state: {before['ingest_state']} → {after['ingest_state']}")
+    return lines
+
+
+def _ingest(root: Path, db: Path) -> None:
+    """One full ingest of `root` into `db`, as `cgis ingest` does it, ingest state included."""
+    with SQLiteStore(str(db)) as store:
+        pipeline = IngestionPipeline(build_extractors([]))
+        pipeline.run(str(root), store=store)
+        store.record_ingest(str(root), pipeline.observed_mtimes)
+        store.record_ingest_options([], None)
+
+
+def snapshot(corpus: Path) -> dict[str, object]:
+    """Every snapshot root, ingested from a temp copy of `corpus` with a `.git` marker.
+
+    The copy keeps the checked-in corpus free of `.git` (which git cannot hold)
+    and makes the repository boundary of the upward config search (#555) the
+    corpus itself, wherever the tests run.
+    """
+    with tempfile.TemporaryDirectory() as tmp:
+        copy = Path(tmp) / "corpus"
+        shutil.copytree(corpus, copy, ignore=shutil.ignore_patterns("__pycache__"))
+        (copy / ".git").mkdir()
+        dumps: dict[str, object] = {}
+        for name, relative in SNAPSHOT_ROOTS.items():
+            db = Path(tmp) / f"{name}.db"
+            _ingest(copy / relative, db)
+            dumps[name] = dump(db, copy)
+        return dumps
+
+
+def main(argv: list[str] | None = None) -> int:
+    """`diff` two stored graphs (exit 1 when they differ), or `snapshot` the corpus."""
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    sub = parser.add_subparsers(dest="command", required=True)
+    d = sub.add_parser("diff")
+    for name in ("db_a", "root_a", "db_b", "root_b"):
+        d.add_argument(name, type=Path)
+    s = sub.add_parser("snapshot")
+    s.add_argument("corpus", type=Path)
+    s.add_argument("--out", type=Path)
+    args = parser.parse_args(argv)
+    if args.command == "snapshot":
+        text = json.dumps(snapshot(args.corpus), indent=1, sort_keys=True, ensure_ascii=False)
+        if args.out:
+            args.out.write_text(text + "\n", encoding="utf-8")
+        else:
+            print(text)
+        return 0
+    lines = diff(dump(args.db_a, args.root_a), dump(args.db_b, args.root_b))
+    print("\n".join(lines) if lines else "identical")
+    return 1 if lines else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/fixtures/graph_identity/corpus/py/lib/legacy/tool.py

@@ -0,0 +1,5 @@
+from shop import place
+
+
+def main() -> None:
+    place(None, "x")

tests/fixtures/graph_identity/corpus/py/pyproject.toml

@@ -0,0 +1,3 @@
+[project]
+name = "shop"
+dependencies = ["fastapi>=0.110", "alembic>=1.13"]

tests/fixtures/graph_identity/corpus/py/src/alembic/env.py

@@ -0,0 +1,2 @@
+def run_migrations() -> None:
+    pass

tests/fixtures/graph_identity/corpus/py/src/shop/__init__.py

@@ -0,0 +1 @@
+from .service import *  # noqa: F403 — star-import chain under test

tests/fixtures/graph_identity/corpus/py/src/shop/cli.py

@@ -0,0 +1,5 @@
+from shop import place
+
+
+def run() -> None:
+    place(None, "x")

tests/fixtures/graph_identity/corpus/py/src/shop/deps.py

@@ -0,0 +1,12 @@
+from typing import Annotated
+
+from fastapi import Depends
+
+from .repo import Repo, SqlRepo
+
+
+def get_repo() -> Repo:
+    return SqlRepo()
+
+
+RepoDep = Annotated[Repo, Depends(get_repo)]

tests/fixtures/graph_identity/corpus/py/src/shop/repo.py

@@ -0,0 +1,11 @@
+from abc import ABC, abstractmethod
+
+
+class Repo(ABC):
+    @abstractmethod
+    def save(self, item: str) -> None: ...
+
+
+class SqlRepo(Repo):
+    def save(self, item: str) -> None:
+        print(item)

tests/fixtures/graph_identity/corpus/py/src/shop/service.py

@@ -0,0 +1,16 @@
+from .deps import RepoDep
+from .repo import Repo
+
+__all__ = ["OrderService", "place"]
+
+
+class OrderService:
+    def __init__(self, repo: RepoDep) -> None:
+        self.repo = repo
+
+    def place(self, item: str) -> None:
+        self.repo.save(item)
+
+
+def place(repo: Repo, item: str) -> None:
+    repo.save(item)

tests/fixtures/graph_identity/corpus/py/tests/test_service.py

@@ -0,0 +1,5 @@
+from shop.service import place
+
+
+def test_place() -> None:
+    place(None, "x")

tests/fixtures/graph_identity/corpus/ts/apps/web/package.json

@@ -0,0 +1 @@
+{"name": "web", "dependencies": {"@x/lib": "^1.0.0", "react": "^19.0.0", "lodash.debounce": "^4.0.8"}}

tests/fixtures/graph_identity/corpus/ts/apps/web/src/main.ts

@@ -0,0 +1,16 @@
+import { randomUUID } from "node:crypto";
+import debounce from "lodash.debounce";
+import { useState } from "react";
+import { libFn } from "@x/lib";
+import { helper } from "@app/util";
+
+function setTimeout(fn: () => void): void {
+  fn();
+}
+
+export function main(): number {
+  useState(0);
+  randomUUID();
+  setTimeout(() => debounce(() => helper(libFn()), 10));
+  return libFn();
+}

tests/fixtures/graph_identity/corpus/ts/apps/web/src/types.d.ts

@@ -0,0 +1,3 @@
+export interface Shape {
+  area(): number;
+}

tests/fixtures/graph_identity/corpus/ts/apps/web/src/util.ts

@@ -0,0 +1,3 @@
+export function helper(n: number): number {
+  return n + 1;
+}

tests/fixtures/graph_identity/corpus/ts/package.json

@@ -0,0 +1 @@
+{"name": "mono", "private": true, "workspaces": ["packages/*"], "dependencies": {"react": "^19.0.0"}}

tests/fixtures/graph_identity/corpus/ts/packages/lib/index.ts

@@ -0,0 +1,3 @@
+export function libFn(): number {
+  return 1;
+}

tests/fixtures/graph_identity/corpus/ts/packages/lib/package.json

@@ -0,0 +1 @@
+{"name": "@x/lib", "version": "1.0.0"}

tests/fixtures/graph_identity/corpus/ts/pnpm-workspace.yaml

@@ -0,0 +1,3 @@
+packages:
+  - "apps/*"
+  - "!**/fixtures/**"

tests/fixtures/graph_identity/corpus/ts/tsconfig.json

@@ -0,0 +1 @@
+{"compilerOptions": {"baseUrl": ".", "paths": {"@app/*": ["apps/web/src/*"]}}}