scripts.graph_test_preflight

Reference › scripts › graph_test_preflight

scripts/graph_test_preflight.py validates the Cypher queries embedded in test files against the graph's uniqueness constraints before the tests run, so a test that would violate a constraint is caught statically. A human or a pre-test step runs it. It imports orchestrata_ultimate.core.graph_factory and orchestrata_ultimate.knowledge_graph.canonical_schema.

main takes an optional test directory (default tests) and --offline. Online, load_constraints_from_graph fetches the uniqueness constraints from the running operational graph and _normalise_constraint reshapes each record; offline, load_cached_constraints derives the set from the canonical schema (_constraint_sensitive_props gives the guarded properties per label) with no graph connection. scan_file runs find_cypher_queries over the Python string literals in each test file, skipping __pycache__ and generated-broken directories, and check_query_against_constraints flags queries that could violate a constraint. Lines carrying a # gate-ok: declaration are exempt (_gate_ok_lines). It prints violations per file and exits 1 on a missing directory or any violation.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/graph_test_preflight.py

#!/usr/bin/env python3
"""Validate test Cypher queries against GRAPH constraints before running tests.

Prevents ConstraintError failures by checking if test CREATE statements could
violate uniqueness constraints on the operational KG.

There is no Neo4j: the containers were retired in the 2026-07-27 cutover and
both graphs are databases on the Memgraph-based orchgraph cluster. The
endpoint was migrated then, but the RESULT SHAPE was not — see
``load_constraints_from_graph``.

Usage:
    python scripts/graph_test_preflight.py [tests/path/]
    python scripts/graph_test_preflight.py --offline  # Use cached constraints
"""

import argparse
import re
import sys
from collections import defaultdict
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.graph_factory import resolve_kg_database, resolve_uri

# Patterns to find Cypher queries in Python strings
CYPHER_PATTERNS = [
    # CREATE (n:Label {prop: value})
    (
        re.compile(
            r"CREATE\s*\(\s*(\w+)\s*:\s*(\w+)\s*\{([^}]*)\}\s*\)",
            re.IGNORECASE | re.MULTILINE,
        ),
        "CREATE",
    ),
    # MERGE (n:Label {prop: value})
    (
        re.compile(
            r"MERGE\s*\(\s*(\w+)\s*:\s*(\w+)\s*\{([^}]*)\}\s*\)",
            re.IGNORECASE | re.MULTILINE,
        ),
        "MERGE",
    ),
    # CREATE (n:Label)-[:REL]->(m:Label {prop: value})
    (
        re.compile(
            r"CREATE\s*\(\s*\w+\s*:\s*(\w+)\s*\).*?\(\s*\w+\s*:\s*(\w+)\s*\{([^}]*)\}\s*\)",
            re.IGNORECASE | re.MULTILINE | re.DOTALL,
        ),
        "CREATE_REL",
    ),
]


def _constraint_sensitive_props() -> dict[str, list[str]]:
    """label -> the properties a canonical uniqueness constraint guards.

    DERIVED from ``canonical_schema.CANONICAL_MERGE_KEYS`` (SPEC-106 FR-001),
    never hand-listed. The table this replaced was a fourth independent copy
    of the same fact, and copies drift: measured 2026-08-06 against the live
    test cluster, **8 of its 21 labels disagreed with the schema actually
    installed** — it claimed ``Project`` was keyed on ``[id, name]`` when the
    live constraint is on ``slug``, and ``File``/``Directory`` on ``path``
    when both are keyed on the project-qualified ``id``. Every violation this
    module reported was therefore scored against a schema that does not
    exist.

    Deriving also means a label added to the canonical ontology is covered
    here immediately, with no second edit to remember.
    """
    from orchestrata_ultimate.knowledge_graph.canonical_schema import (
        CANONICAL_MERGE_KEYS,
    )

    return {label: [prop] for label, prop in CANONICAL_MERGE_KEYS.items()}


CONSTRAINT_SENSITIVE_PROPS = _constraint_sensitive_props()


def _normalise_constraint(record: Any) -> dict:
    """Map one live constraint record onto this module's internal shape.

    Reads Memgraph's field names (``constraint type``, ``label``,
    ``properties``). Falls back to the Neo4j spellings so the function stays
    correct if pointed at a Neo4j instance, rather than silently emptying out
    the way the pre-2026-08-06 version did against Memgraph.
    """
    labels = record.get("label") or record.get("labelsOrTypes") or []
    if isinstance(labels, str):
        labels = [labels]
    props = record.get("properties") or []
    if isinstance(props, str):
        props = [props]
    name = record.get("name") or "_".join([*labels, *props])
    return {
        "name": name,
        "type": record.get("constraint type") or record.get("type") or "",
        "entity_type": record.get("entityType", "NODE"),
        "labels": list(labels),
        "properties": list(props),
    }


def load_constraints_from_graph() -> list[dict]:
    """Fetch uniqueness constraints from the running operational KG.

    Memgraph answers ``SHOW CONSTRAINTS`` (it accepts the Neo4j spelling as an
    alias for ``SHOW CONSTRAINT INFO``) but returns a DIFFERENT record shape:
    ``constraint type`` / ``label`` / ``properties`` / ``data_type``, where
    Neo4j returns ``name`` / ``type`` / ``entityType`` / ``labelsOrTypes`` /
    ``properties``.

    Until 2026-08-06 this function read the Neo4j field names against the live
    Memgraph cluster. Only ``properties`` overlapped, so it loaded 103
    constraints whose name and labels were all empty — and every label-keyed
    comparison downstream then matched nothing. The preflight reported success
    while validating against an empty constraint set: a silent false green, and
    exactly the class this same session fixed in code-review.py (62349326).

    Constraint names are synthesised (Memgraph does not name constraints) as
    ``<label>_<prop>_<prop>`` so downstream reporting stays stable.
    """
    try:
        from neo4j import GraphDatabase
    except ImportError:
        print("ERROR: neo4j package not installed")
        sys.exit(1)

    # KG_ROUTING_URI/KG_URI via resolve_uri(); the NEO4J_* keys were retired
    # with the Neo4j containers on 2026-07-27. A preflight that silently
    # dialled bolt://localhost:7687 would report the wrong instance healthy.
    uri = resolve_uri()

    try:
        driver = GraphDatabase.driver(uri, auth=None)
        with driver.session(database=resolve_kg_database()) as session:
            constraints = [
                _normalise_constraint(record)
                for record in session.run("SHOW CONSTRAINTS")
            ]
            driver.close()
            return constraints
    except Exception as e:
        print(f"ERROR: Failed to connect to the graph: {e}")
        print("Use --offline to run without a graph connection")
        sys.exit(1)


def load_cached_constraints() -> list[dict]:
    """Offline constraint set, DERIVED from the canonical schema.

    This was a hand-written list of 21 constraint dicts — a FIFTH independent
    copy of the merge keys, carrying the same drift as the table above
    (``File`` on ``path``, ``Project`` on ``id``). Offline runs therefore
    scored every file against a schema that does not exist, and did so
    silently, because an offline run looks identical to an online one.

    Deriving from ``CANONICAL_MERGE_KEYS`` means online and offline validate
    the same contract by construction.
    """
    return [
        {
            "name": f"{label.lower()}_{prop}",
            "type": "UNIQUENESS",
            "entity_type": "NODE",
            "labels": [label],
            "properties": [prop],
        }
        for label, props in CONSTRAINT_SENSITIVE_PROPS.items()
        for prop in props
    ]


def find_cypher_queries(content: str) -> list[tuple[int, str, str, str]]:
    """Find Cypher queries in Python string literals.

    Returns: list of (line_number, operation, label, properties_string)
    """
    queries = []
    lines = content.splitlines()

    # Find multi-line strings that might contain Cypher
    in_string = False
    string_start = 0
    string_lines = []

    for i, line in enumerate(lines, 1):
        # Look for triple-quoted strings
        if '"""' in line or "'''" in line:
            if not in_string:
                in_string = True
                string_start = i
                string_lines = [line]
            else:
                string_lines.append(line)
                # End of string
                string_content = "\n".join(string_lines)
                for pattern, op in CYPHER_PATTERNS:
                    for match in pattern.finditer(string_content):
                        # Calculate approximate line number
                        rel_pos = match.start()
                        line_offset = string_content[:rel_pos].count("\n")
                        actual_line = string_start + line_offset
                        if op == "CREATE_REL":
                            queries.append(
                                (actual_line, op, match.group(2), match.group(3))
                            )
                        else:
                            queries.append(
                                (actual_line, op, match.group(2), match.group(3))
                            )
                in_string = False
                string_lines = []
        elif in_string:
            string_lines.append(line)

        # Also check single-line strings
        if not in_string:
            for pattern, op in CYPHER_PATTERNS:
                for match in pattern.finditer(line):
                    if op == "CREATE_REL":
                        queries.append((i, op, match.group(2), match.group(3)))
                    else:
                        queries.append((i, op, match.group(2), match.group(3)))

    return queries


def check_query_against_constraints(
    line: int, op: str, label: str, props_str: str, constraints: list[dict]
) -> list[dict]:
    """Check if a Cypher query could violate constraints."""
    violations: list[dict] = []

    for constraint in constraints:
        if constraint["type"] != "UNIQUENESS":
            continue
        if not constraint["labels"]:
            continue

        # Check if this label has a constraint
        if label not in constraint["labels"]:
            continue

        # Check if the constrained property is in the query
        for prop in constraint["properties"]:
            if prop.lower() in props_str.lower():
                # Check if it's a CREATE (MERGE is safe)
                if op == "CREATE":
                    # Check if the property value is dynamic (has $param or an
                    # f-string). Matching a bare "f" anywhere in props_str is
                    # too broad -- most literal property strings contain the
                    # letter "f" incidentally (e.g. `path: "foo"`), which
                    # silently downgraded almost every hardcoded CREATE to
                    # MEDIUM and defeated the HIGH-risk check. Require an
                    # actual f-string prefix (`f"..."` / `f'...'`) instead.
                    is_dynamic = bool(
                        "$" in props_str
                        or "{" in props_str
                        or re.search(r"f['\"]", props_str)
                    )
                    risk = "HIGH" if not is_dynamic else "MEDIUM"
                    violations.append(
                        {
                            "line": line,
                            "label": label,
                            "property": prop,
                            "constraint": constraint["name"],
                            "risk": risk,
                            "reason": f"CREATE {label} with {prop} — uniqueness constraint '{constraint['name']}'",
                        }
                    )

    return violations


def _gate_ok_lines(content: str) -> set[int]:
    """Lines covered by a ``# gate-ok:`` declaration.

    The marker is this project's established way of saying "this Cypher is
    DATA, not a query anyone runs" — five other gates already honour it
    (code-review.py, gate_parallel_safety.py, agent_check.py,
    comprehensive_review.py, diagnose_test_failure.py). This module did not,
    so it reported its OWN test fixtures as HIGH-risk violations: strings the
    tool exists to scan, each already annotated as never executed.

    A marker covers its own line and the one following, which is where the
    annotated literal sits.
    """
    marked = {
        n for n, line in enumerate(content.splitlines(), start=1) if "gate-ok:" in line
    }
    return marked | {n + 1 for n in marked}


def scan_file(py_file: Path, constraints: list[dict]) -> list[dict]:
    """Scan a single test file for constraint violations."""
    try:
        content = py_file.read_text(encoding="utf-8")
    except Exception:
        return []

    exempt = _gate_ok_lines(content)
    return [
        violation
        for line, op, label, props_str in find_cypher_queries(content)
        if line not in exempt
        for violation in check_query_against_constraints(
            line, op, label, props_str, constraints
        )
    ]


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Validate test Cypher queries against graph constraints"
    )
    parser.add_argument(
        "path", nargs="?", default="tests", help="Path to test directory"
    )
    parser.add_argument(
        "--offline",
        action="store_true",
        help="Use cached constraints (no graph connection)",
    )
    args = parser.parse_args(argv)

    # Load constraints
    if args.offline:
        constraints = load_cached_constraints()
        print("Using cached constraints (offline mode)")
    else:
        constraints = load_constraints_from_graph()

    print(f"Loaded {len(constraints)} graph constraints\n")

    test_dir = Path(args.path)
    if not test_dir.is_dir():
        print(f"ERROR: {test_dir} is not a directory")
        return 1

    all_violations = defaultdict(list)
    file_count = 0
    total_violations = 0

    for py_file in sorted(test_dir.rglob("*.py")):
        if "__pycache__" in str(py_file):
            continue
        if "_generated_broken" in str(py_file) or "generated_recovery" in str(py_file):
            continue

        violations = scan_file(py_file, constraints)
        if violations:
            file_count += 1
            rel_path = py_file
            if py_file.is_absolute():
                try:
                    rel_path = py_file.relative_to(Path(".").resolve())
                except ValueError:
                    # py_file isn't under the CWD (e.g. an out-of-repo temp
                    # dir passed as `path`) -- fall back to the absolute
                    # path instead of crashing the whole scan.
                    rel_path = py_file
            all_violations[rel_path] = violations
            total_violations += len(violations)

    if total_violations == 0:
        print("✅ No constraint violations found!")
        return 0

    print(f"{'=' * 70}")
    print("GRAPH TEST PREFLIGHT REPORT")
    print(f"{'=' * 70}")
    print(f"Files scanned: {len(list(test_dir.rglob('*.py')))}")
    print(f"Files with violations: {file_count}")
    print(f"Total violations: {total_violations}")
    print(f"{'=' * 70}\n")

    for rel_path, violations in sorted(all_violations.items()):
        print(f"\n{rel_path}")
        print("-" * len(str(rel_path)))
        for v in violations:
            print(f"  Line {v['line']:4d} [{v['risk']:6s}] {v['reason']}")
            if v["risk"] == "HIGH":
                print("           Fix: Use MERGE or DELETE existing node first")
            else:
                print("           Note: Dynamic value reduces risk, but MERGE is safer")

    print(f"\n{'=' * 70}")
    print(f"SUMMARY: {file_count} files with {total_violations} potential violations")
    print(f"{'=' * 70}")

    # Gate on ACTIONABLE findings only. HIGH means a hardcoded value against a
    # uniqueness constraint — it will collide. MEDIUM means the value is
    # dynamic (`$param` / f-string), so it cannot collide; this tool's own
    # output calls that "dynamic value reduces risk, but MERGE is safer",
    # which is advice, not a defect.
    #
    # Gating on both made the exit code useless in the other direction from a
    # false green: 135 advisory notes held the gate red, so a real HIGH would
    # have changed nothing anyone could see. Advisory findings are still
    # printed in full.
    high = sum(
        1
        for violations in all_violations.values()
        for v in violations
        if v["risk"] == "HIGH"
    )
    print(f"Gating (HIGH) violations: {high}")
    return 1 if high > 0 else 0


if __name__ == "__main__":
    sys.exit(main())