test(mcp): sample-project fixture + assertion contract (T3 #650)

New `tests/mcp/fixtures/`:
- `sample_project/python/` — canonical call graph
  `entrypoint -> service -> {UserRepo,OrderRepo}.repo -> db`
  plus a small class hierarchy (BaseRepo <- UserRepo, OrderRepo)
  and inter-file imports so IMPORTS edges exist.
- `expected.yaml` — single source of truth for every per-tool
  ticket's integration assertions: minimum per-label counts, named
  callers / callees, known paths, prefix-search hits.

New `tests/mcp/conftest.py`:
- `expected_contract` (pure-Python, always available) loads the
  YAML once per session.
- `indexed_fixture` (session-scoped) indexes the fixture into a
  unique `code:sample_project:test-<uuid>` graph so parallel CI
  shards don't contend. Self-skips when FalkorDB is unreachable.
  Uses `SourceAnalyzer.analyze_local_folder` directly so the
  fixture doesn't need to be a git repository.

New `tests/mcp/test_fixture_contract.py` — regression-tests the
fixture itself: contract shape, on-disk files, and that the
integration fixture indexes cleanly and meets the minimum count
contract.

Multilingual coverage (Java + C#) was dropped from the spec: both
multilspy analyzers demand a Maven / .NET project layout at the
indexed root, which would force this fixture into an awkward shape.
Deferred to a follow-up ticket (likely T16 which adds languages).

All 4 contract tests pass against FalkorDB on 6390.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: DvirDukhan

pyproject.toml

@@ -49,4 +49,5 @@ where = ["."]
 dev = [
     "pytest>=9.0.2",
     "pytest-anyio>=0.0.0",
+    "pyyaml>=6.0.3",
 ]

tests/mcp/conftest.py

@@ -0,0 +1,116 @@
+"""Shared fixtures for the MCP test suite.
+
+Every per-tool integration test from T4 onward reuses the
+``indexed_fixture`` fixture below: it indexes ``fixtures/sample_project``
+into a uniquely named FalkorDB graph once per test session and yields a
+descriptor (project name, branch, graph name) the test can pass straight
+to the MCP tool under test.
+
+The integration fixture is opt-in — it requires a reachable FalkorDB
+(see ``api/graph.py``) and the optional language analyzers. Tests that
+only need the static contract (counts, named callers / callees / paths)
+can depend on ``expected_contract`` alone, which is pure-Python and
+always available.
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import pytest
+import yaml
+
+
+FIXTURE_DIR = Path(__file__).parent / "fixtures"
+SAMPLE_PROJECT = FIXTURE_DIR / "sample_project"
+EXPECTED_PATH = FIXTURE_DIR / "expected.yaml"
+
+
+# ---------------------------------------------------------------------------
+# Pure-Python contract (no FalkorDB required)
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class IndexedFixture:
+    """Descriptor for an indexed fixture graph."""
+
+    project: str
+    branch: str
+    graph_name: str
+    path: Path
+
+
+@pytest.fixture(scope="session")
+def expected_contract() -> dict[str, Any]:
+    """Load ``fixtures/expected.yaml`` once per session."""
+    with EXPECTED_PATH.open() as fh:
+        return yaml.safe_load(fh)
+
+
+@pytest.fixture(scope="session")
+def sample_project_path() -> Path:
+    """Filesystem path to the fixture project."""
+    return SAMPLE_PROJECT
+
+
+# ---------------------------------------------------------------------------
+# Integration fixture — indexes into a real FalkorDB
+# ---------------------------------------------------------------------------
+
+
+def _falkordb_reachable() -> bool:
+    """Cheap probe so the integration fixture can self-skip in dev."""
+    try:
+        import socket
+
+        host = os.getenv("FALKORDB_HOST", "localhost")
+        port = int(os.getenv("FALKORDB_PORT", 6379))
+        with socket.create_connection((host, port), timeout=1):
+            return True
+    except OSError:
+        return False
+
+
+@pytest.fixture(scope="session")
+def indexed_fixture(sample_project_path: Path) -> IndexedFixture:
+    """Index the sample project into a unique per-session graph.
+
+    Each test session creates a new graph named
+    ``code:sample_project:test-<uuid>`` so parallel CI shards never
+    contend on the same graph. The graph is intentionally **not**
+    cleaned up — short-lived CI runners discard the FalkorDB volume,
+    and keeping it around helps post-mortem debugging on developer
+    machines.
+
+    Uses :class:`api.analyzers.SourceAnalyzer` directly (instead of
+    ``Project.from_local_repository``) so the fixture doesn't need to
+    be a git repository — analyzing a plain directory is exactly the
+    code path the ``index_repo`` MCP tool exercises for non-git
+    folders.
+    """
+
+    if not _falkordb_reachable():
+        pytest.skip("FalkorDB not reachable on $FALKORDB_HOST:$FALKORDB_PORT")
+
+    # Import locally so unit-only tests don't pay the import cost.
+    from api.analyzers.source_analyzer import SourceAnalyzer
+    from api.graph import Graph
+
+    project_name = sample_project_path.name  # "sample_project"
+    branch = f"test-{uuid.uuid4().hex[:8]}"
+    graph = Graph(project_name, branch=branch)
+
+    analyzer = SourceAnalyzer()
+    analyzer.analyze_local_folder(str(sample_project_path), graph)
+
+    return IndexedFixture(
+        project=project_name,
+        branch=branch,
+        graph_name=graph.name,
+        path=sample_project_path,
+    )

tests/mcp/fixtures/expected.yaml

@@ -0,0 +1,48 @@
+# MCP fixture assertion contract.
+#
+# Anything declared here is treated as load-bearing by tests/mcp/*. The
+# precise numeric counts depend on which analyzers run in CI (some test
+# environments skip the multilspy passes), so the per-label counts are
+# expressed as minimums (`>=`) rather than equalities. Named-symbol
+# assertions are exact.
+
+# The graph is named after the fixture directory (per Project.from_local_repository).
+project_name: sample_project
+
+# Minimum counts produced by the Python tree-sitter analyzer alone.
+# Java + C# add more when multilspy is available.
+counts_min:
+  File: 4        # 4 python files (java/csharp may bump this)
+  Class: 3       # BaseRepo, UserRepo, OrderRepo
+  Function: 6    # entrypoint, service, db, BaseRepo.repo, UserRepo.repo, OrderRepo.repo
+
+# Named callers / callees (used by T5 — get_callers / get_callees).
+calls:
+  service:
+    callers: ["entrypoint"]
+    # service() instantiates UserRepo + OrderRepo and calls .repo() on each;
+    # the analyzer encodes that as CALLS edges on the method.
+    callees_any_of: ["repo", "UserRepo", "OrderRepo"]
+  entrypoint:
+    callers: []
+    callees_any_of: ["service"]
+  db:
+    # db() is called by both subclasses via super().repo() -> BaseRepo.repo().
+    callers_any_of: ["repo"]
+    callees: []
+
+# Known path between two named symbols (used by T7 — find_path).
+paths:
+  - source: entrypoint
+    dest: db
+    min_paths: 1
+
+# Prefix-search hits (used by T8 — search_code).
+search_prefixes:
+  ent:
+    must_include: ["entrypoint"]
+  serv:
+    must_include: ["service"]
+  Repo:
+    # case-insensitive prefix match on Searchable label
+    must_include_any_of: ["BaseRepo", "UserRepo", "OrderRepo"]

tests/mcp/fixtures/sample_project/README.md

@@ -0,0 +1,36 @@
+# Sample fixture project for the MCP test suite
+
+This directory is consumed by `tests/mcp/conftest.py::indexed_fixture`. Every
+MCP tool ticket from T4 onward asserts against the assertions declared in
+`expected.yaml`.
+
+## Canonical Python call graph
+
+```
+entrypoint() -> service() -> {UserRepo,OrderRepo}.repo() -> db()
+```
+
+Plus a small class hierarchy:
+
+```
+BaseRepo
+  ├── UserRepo
+  └── OrderRepo
+```
+
+## Why three languages? (deferred)
+
+The original T3 spec called for one Java + one C# file so multilspy's
+second-pass code paths would be exercised. In practice both analyzers
+demand a real Maven / .NET project layout at the **root** of the indexed
+tree, which would make this fixture awkward to co-host with the Python
+sample. The multilingual coverage is therefore deferred to a follow-up
+ticket (likely T16, which already pulls in additional languages).
+
+T4-T8 only need Python, which this fixture covers in full.
+
+## Stability contract
+
+If you change this fixture, you must also update `expected.yaml`. Tests
+read counts and named symbols directly from that file so the assertion
+contract stays in lock-step.

tests/mcp/fixtures/sample_project/python/__init__.py

@@ -0,0 +1 @@
+"""Marks ``sample_project/python`` as a package so IMPORTS edges resolve."""

tests/mcp/fixtures/sample_project/python/db.py

@@ -0,0 +1,6 @@
+"""Bottom of the canonical call chain."""
+
+
+def db() -> str:
+    """Leaf function — entrypoint -> service -> repo -> db."""
+    return "db"

tests/mcp/fixtures/sample_project/python/entrypoint.py

@@ -0,0 +1,17 @@
+"""Entrypoint for the MCP test fixture project.
+
+Call graph (must match ``expected.yaml``):
+
+    entrypoint() -> service() -> repo() -> db()
+"""
+
+from .service import service
+
+
+def entrypoint() -> str:
+    """Top of the canonical call chain used by every MCP integration test."""
+    return service()
+
+
+if __name__ == "__main__":
+    entrypoint()

tests/mcp/fixtures/sample_project/python/repo.py

@@ -0,0 +1,23 @@
+"""Repository layer for the MCP fixture project.
+
+Exercises a small class hierarchy: ``BaseRepo`` <- ``UserRepo`` / ``OrderRepo``.
+"""
+
+from .db import db
+
+
+class BaseRepo:
+    """Base class so the analyzer emits an INHERITS edge."""
+
+    def repo(self) -> str:
+        return db()
+
+
+class UserRepo(BaseRepo):
+    def repo(self) -> str:
+        return "user:" + super().repo()
+
+
+class OrderRepo(BaseRepo):
+    def repo(self) -> str:
+        return "order:" + super().repo()

tests/mcp/fixtures/sample_project/python/service.py

@@ -0,0 +1,10 @@
+"""Service layer for the MCP fixture project."""
+
+from .repo import UserRepo, OrderRepo
+
+
+def service() -> str:
+    """Middle of the canonical call chain: entrypoint -> service -> repo."""
+    users = UserRepo()
+    orders = OrderRepo()
+    return users.repo() + ":" + orders.repo()

tests/mcp/test_fixture_contract.py

@@ -0,0 +1,81 @@
+"""T3 — assertion contract sanity check.
+
+These tests validate the *fixture itself*: that the YAML contract parses,
+that the sample-project tree on disk matches what the contract declares,
+and that the integration fixture can actually index the project against
+a live FalkorDB. Tool-specific assertions live in the per-tool ticket
+test modules (T4, T5, T7, T8 ...).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.mcp.conftest import EXPECTED_PATH, SAMPLE_PROJECT
+
+
+# ---------------------------------------------------------------------------
+# Pure-Python contract checks (always run)
+# ---------------------------------------------------------------------------
+
+
+def test_expected_yaml_exists():
+    assert EXPECTED_PATH.is_file(), "fixtures/expected.yaml must exist"
+
+
+def test_expected_contract_shape(expected_contract):
+    """Required top-level keys are present and well-typed."""
+
+    assert expected_contract["project_name"] == "sample_project"
+
+    counts = expected_contract["counts_min"]
+    for label in ("File", "Class", "Function"):
+        assert label in counts and isinstance(counts[label], int) and counts[label] >= 0
+
+    calls = expected_contract["calls"]
+    for sym in ("service", "entrypoint", "db"):
+        assert sym in calls, f"calls.{sym} missing from expected.yaml"
+
+    paths = expected_contract["paths"]
+    assert isinstance(paths, list) and len(paths) >= 1
+    for p in paths:
+        for k in ("source", "dest", "min_paths"):
+            assert k in p, f"path entry missing key: {k}"
+
+    prefixes = expected_contract["search_prefixes"]
+    assert "ent" in prefixes
+
+
+def test_sample_project_python_files_present():
+    """The Python tree the contract references must exist on disk."""
+    py = SAMPLE_PROJECT / "python"
+    for name in ("entrypoint.py", "service.py", "repo.py", "db.py", "__init__.py"):
+        assert (py / name).is_file(), f"missing fixture file: python/{name}"
+
+
+# ---------------------------------------------------------------------------
+# Integration check — requires FalkorDB; self-skips when unreachable
+# ---------------------------------------------------------------------------
+
+
+def test_indexed_fixture_loads_minimum_counts(indexed_fixture, expected_contract):
+    """The fixture indexes cleanly and meets the minimum count contract.
+
+    Subsequent per-tool tickets (T4+) use ``indexed_fixture`` directly;
+    this test exists so the fixture itself is regression-tested in
+    isolation.
+    """
+
+    from api.graph import Graph
+
+    g = Graph(indexed_fixture.project, branch=indexed_fixture.branch)
+    counts_min = expected_contract["counts_min"]
+
+    for label, minimum in counts_min.items():
+        rows = g.g.query(f"MATCH (n:{label}) RETURN count(n) AS c").result_set
+        actual = rows[0][0] if rows else 0
+        assert actual >= minimum, (
+            f"label {label}: expected >={minimum}, got {actual}"
+        )