orchestrata_ultimate.verification.review.checks_safety

Reference › orchestrata_ultimate › verification › review › checks_safety

The safety checks of the deterministic review: security patterns, configuration discipline, test hygiene and lint. It is called by the review engine, by the refactor codemod's review checks and by the orchestrator's verification_gate. Each check returns issue tuples.

• check_security (check 30) looks for eval and exec and hardcoded secrets.
• check_env_access (check 30c, CONFIG) flags bare os.environ["X"] subscript reads.
• check_infra_env_defaults (check 30d) flags literal fallbacks for required infrastructure-connection environment keys, using the key list in core.infra_env_rule. Both configuration checks mirror rules enforced by the prolog-gate and the env-contract gate.
• check_test_isolation (check 34) flags global graph or Redis teardowns in tests; the static commit-time equivalent is the parallel-safety gate.
• check_no_mocks and validate_code_no_mocks flag mock, stub or fake usage in test files or in a code string, matching the test-purity gate.
• check_todo_comments (check 12b) flags TODO, FIXME and HACK comments.
• check_lint runs ruff, or flake8, for undefined names and unused imports; see the ruff gate.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/review/checks_safety.py

"""Safety checks: security, TODO comments, mocks, lint.

Covers checklist items 30, 12b, and mock/lint detection.
"""

import ast
import os
import re
import subprocess

from orchestrata_ultimate.core.infra_env_rule import find_infra_env_defaults
from orchestrata_ultimate.verification.review.types import Issue

_MOCK_PATTERNS = [
    (r"from\s+unittest\.mock\s+import", "unittest.mock import"),
    (r"import\s+unittest\.mock", "unittest.mock import"),
    (r"from\s+unittest\s+import\s+mock", "unittest import mock"),
    (r"Mock\s*\(", "Mock() instantiation"),
    (r"MagicMock\s*\(", "MagicMock() instantiation"),
    (r"AsyncMock\s*\(", "AsyncMock() instantiation"),
    (r"@\s*(mock\.)?patch\s*\(", "@patch decorator"),
    (r"mocker\.(patch|spy|stub|mock)\s*\(", "pytest-mock usage"),
    (r"create_autospec\s*\(", "create_autospec()"),
    (r"patch\.object\s*\(", "patch.object()"),
    (r"patch\.dict\s*\(", "patch.dict()"),
    (r"patch\.multiple\s*\(", "patch.multiple()"),
    (r"class\s+(Mock|Fake|Stub|Dummy)\w*\s*[\(:]", "custom mock/stub/fake class"),
    (r"monkeypatch\.\w+\s*\(", "monkeypatch usage"),
]

_SECRET_PATTERNS = [
    (
        r"(?:password|passwd|secret|api_key|apikey|token|auth)"
        r'\s*=\s*["\'][^"\']{8,}["\']',
        "Possible hardcoded secret",
    ),
    (
        # Left boundary: a prefix only counts at a token start, so the ``sk-``
        # inside "task-management-tool" or "ask-openrouter" is not a key.
        r"(?<![A-Za-z0-9-])(?:sk-|pk_live_|sk_live_|ghp_|gho_|AKIA)"
        r"[A-Za-z0-9]{10,}",
        "Possible API key/token",
    ),
]


def check_security(
    filepath: str,
    tree: ast.Module,
    lines: list[str],
) -> list[Issue]:
    """Check 30: SECURITY -- eval/exec, hardcoded secrets.

    Args:
        filepath: Absolute path to the file.
        tree: Parsed AST module.
        lines: File lines.

    Returns:
        List of security issues found.
    """
    issues: list[Issue] = []
    rel_name = os.path.basename(filepath)
    for node in ast.walk(tree):
        if (
            isinstance(node, ast.Call)
            and isinstance(node.func, ast.Name)
            and node.func.id in ("eval", "exec")
        ):
            issues.append(
                (
                    "SECURITY",
                    rel_name,
                    f"L{node.lineno}: `{node.func.id}()` call -- "
                    f"potential code injection risk.",
                )
            )
    for i, line in enumerate(lines, 1):
        for pattern, desc in _SECRET_PATTERNS:
            if re.search(pattern, line, re.IGNORECASE):
                stripped = line.strip()
                if stripped.startswith("#"):
                    continue
                if "os.getenv" in line or "os.environ" in line:
                    continue
                issues.append(("SECURITY", rel_name, f"L{i}: {desc}."))
                break
    return issues


def check_env_access(
    filepath: str,
    tree: ast.Module,
) -> list[Issue]:
    """Check 30c: CONFIG -- bare ``os.environ["X"]`` subscript reads.

    Reading ``os.environ["X"]`` raises ``KeyError`` when the variable is
    unset (fresh checkout, CI), which is the most common config-time crash.
    Env reads MUST use ``os.environ.get("X", default)`` so a sensible
    fallback applies (see standards rule 30/30c and the AGENTS.md table).

    Args:
        filepath: Absolute path to the file.
        tree: Parsed AST module.

    Returns:
        One CONFIG issue per bare ``os.environ[...]`` read.
    """
    issues: list[Issue] = []
    rel_name = os.path.basename(filepath)
    for node in ast.walk(tree):
        if (
            isinstance(node, ast.Subscript)
            and isinstance(node.ctx, ast.Load)
            and isinstance(node.value, ast.Attribute)
            and node.value.attr == "environ"
            and isinstance(node.value.value, ast.Name)
            and node.value.value.id == "os"
        ):
            key = (
                node.slice.value
                if isinstance(node.slice, ast.Constant)
                and isinstance(node.slice.value, str)
                else ""
            )
            target = f'"{key}"' if key else "..."
            issues.append(
                (
                    "CONFIG",
                    rel_name,
                    f"L{node.lineno}: bare `os.environ[{target}]` raises "
                    f"KeyError when unset -- use "
                    f"`os.environ.get({target}, default)`.",
                )
            )
    return issues


# Parallel test isolation (standards rule 34). Mirrors the blocking
# patterns of scripts/gate_parallel_safety.py so the QA volk's review and
# the CI gate agree on the same standard.
_ISO_DELETE_RE = re.compile(r"DETACH\s+DELETE|\)\s*DELETE\s+\w", re.IGNORECASE)
_ISO_FLUSH_RE = re.compile(r"\.flush(?:db|all)\s*\(")
_ISO_SCOPE_RE = re.compile(
    r"\$\w+|test_id|_TEST_|__test\b|project_id|session_id|tmp_|uuid",
    re.IGNORECASE,
)


def check_test_isolation(
    filepath: str,
    lines: list[str],
) -> list[Issue]:
    """Check 34: PARALLEL TEST ISOLATION -- global graph/Redis teardowns.

    Tests run concurrently against ONE shared Memgraph database and Redis
    instance (per-worker isolation is by database name, ``orchkg_w<N>``,
    not by address). A global ``DETACH DELETE`` or ``flushdb()`` in one
    test deletes rows a concurrent test is mid-assertion on, which
    surfaces as flakiness. Teardowns MUST be scoped to this test (a Cypher
    ``$param`` / ``test_id`` / ``project_id`` filter); Redis must never be
    flushed. See standards rule 34 and scripts/gate_parallel_safety.py
    (the CI gate enforcing the same patterns over the whole tests/ tree).

    Args:
        filepath: Absolute path to the file (expected to be a test).
        lines: Source lines of the file.

    Returns:
        One ISOLATION issue per unscoped global teardown.
    """
    issues: list[Issue] = []
    rel_name = os.path.basename(filepath)
    for i, raw in enumerate(lines):
        line = raw.split("#", 1)[0]
        if not line.strip():
            continue
        if _ISO_FLUSH_RE.search(line):
            issues.append(
                (
                    "ISOLATION",
                    rel_name,
                    f"L{i + 1}: `flushdb`/`flushall` wipes the Redis DB "
                    f"shared by concurrent workers -- delete only this "
                    f"test's keys (scope by test_id).",
                )
            )
            continue
        if _ISO_DELETE_RE.search(line):
            window = "".join(lines[max(0, i - 6) : i + 5])
            if not _ISO_SCOPE_RE.search(window):
                issues.append(
                    (
                        "ISOLATION",
                        rel_name,
                        f"L{i + 1}: unscoped graph DELETE can wipe data a "
                        f"concurrent test is asserting on -- add a "
                        f"$param/test_id/project_id filter.",
                    )
                )
    return issues


def check_todo_comments(
    filepath: str,
    lines: list[str],
) -> list[Issue]:
    """Check 12b: No TODO/FIXME/HACK comments left behind.

    Args:
        filepath: Absolute path to the file.
        lines: File lines.

    Returns:
        List of TODO/FIXME/HACK comment issues.
    """
    issues: list[Issue] = []
    rel_name = os.path.basename(filepath)
    tags = ("TODO", "FIXME", "HACK", "XXX")
    for i, line in enumerate(lines, 1):
        stripped = line.strip()
        if stripped.startswith("#"):
            upper = stripped.upper()
            if any(tag in upper for tag in tags):
                issues.append(
                    (
                        "TODO",
                        rel_name,
                        f"L{i}: {stripped[:60]}",
                    )
                )
    return issues


def check_no_mocks(
    filepath: str,
    lines: list[str],
) -> list[Issue]:
    """Check test files for mock/stub/fake usage.

    Args:
        filepath: Absolute path to the test file.
        lines: File lines.

    Returns:
        List of mock violation issues.
    """
    issues: list[Issue] = []
    rel_name = os.path.basename(filepath)
    for i, line in enumerate(lines, 1):
        raw = line.strip()
        if not raw or raw.startswith("#"):
            continue
        for pattern, desc in _MOCK_PATTERNS:
            if re.search(pattern, raw):
                issues.append(
                    (
                        "MOCK",
                        rel_name,
                        f"L{i}: {desc} -- `{raw[:60]}`",
                    )
                )
                break  # One violation per line
    return issues


def validate_code_no_mocks(
    code: str,
) -> list[tuple[int, str]]:
    """Check a code string for mock/stub usage.

    Unlike check_no_mocks which works on file lines and returns
    Issue tuples, this variant works on a raw code string and
    returns (line_number, description) pairs.

    Args:
        code: Source code string to validate.

    Returns:
        List of (line_number, description) for each violation.
    """
    violations: list[tuple[int, str]] = []
    for line_num, line in enumerate(code.split("\n"), 1):
        for pattern, description in _MOCK_PATTERNS:
            if re.search(pattern, line):
                violations.append((line_num, description))
                break
    return violations


def check_lint(filepath: str) -> list[Issue]:
    """Run ruff or flake8 for undefined names and unused imports.

    Args:
        filepath: Absolute path to the Python file.

    Returns:
        List of lint issues found.
    """
    issues: list[Issue] = []
    rel_name = os.path.basename(filepath)
    ruff_sel = "F821,F811,F401"
    flake8_sel = "E999,F821,F811,F401"
    linter_cmds = [
        ("ruff check", ruff_sel),
        ("flake8", flake8_sel),
        ("/opt/orchestrata_ultimate/.venv/bin/ruff check", ruff_sel),
        ("/opt/orchestrata_ultimate/.venv/bin/flake8", flake8_sel),
    ]
    for linter_cmd, select in linter_cmds:
        cmd = linter_cmd.split() + [f"--select={select}", filepath]
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
            if result.returncode != 0 and result.stdout.strip():
                for line in result.stdout.strip().splitlines()[:5]:
                    msg = line.strip()
                    if filepath in msg:
                        msg = msg.replace(filepath, rel_name)
                    issues.append(("LINT", rel_name, msg))
            return issues
        except FileNotFoundError:
            continue
        except subprocess.TimeoutExpired:
            return issues
    return issues


def check_infra_env_defaults(
    filepath: str,
    tree: ast.Module,
) -> list[Issue]:
    """Check 30d: NO literal fallback for REQUIRED infra-connection env keys
    (RUL-003).

    The deliberate inverse of :func:`check_env_access`: ordinary config SHOULD
    give ``os.environ.get`` a default to avoid ``KeyError``, but a required
    infra ENDPOINT key must never carry a literal fallback — an unset key must
    RAISE, never silently connect to some default host/URI/port (the classic
    failure being a default of the PRODUCTION Redis/graph, so a mis-configured
    run quietly writes to prod). Which keys count and why is owned by
    ``core.infra_env_rule`` so this check and the ``validate_ci`` gate can never
    drift. Fix a hit by routing through the raising seam
    (``core.graph_factory.resolve_uri`` / ``core.redis_factory``) or a one-arg
    ``.get(KEY)`` + explicit raise.

    Args:
        filepath: Absolute path to the file.
        tree: Parsed AST module.

    Returns:
        One CONFIG issue per forbidden ``<mapping>.get(KEY, default)`` /
        ``setdefault(KEY, default)`` on a required infra key.
    """
    rel_name = os.path.basename(filepath)
    return [
        (
            "CONFIG",
            rel_name,
            f'L{v.lineno}: `.{v.method}("{v.key}", <default>)` gives a '
            f"literal fallback for the required infra key {v.key} -- forbidden "
            f"(RUL-003). Read it with no default and raise when unset (route "
            f"through core.graph_factory.resolve_uri / core.redis_factory).",
        )
        for v in find_infra_env_defaults(tree)
    ]