Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -225,9 +225,11 @@ codeprism stats --verbose # per-file breakdown
codeprism stats --json
```

Give enough of the path to be unique (`api/utils.py`, not `utils.py`). The MCP tools report an
ambiguous path together with its candidates; the CLI commands do not, and `callers` even prints
"No callers found" for a path it can't resolve ([#44](https://github.com/knight22-21/CodePrism/issues/44)).
Give enough of the path to be unique (`api/utils.py`, not `utils.py`). The MCP tools and CLI
`callers`, `context`, `impact` and `summary` commands report ambiguous paths with candidate files
and a hint to use a longer path. These CLI commands write lookup errors to stderr and exit with
status 1 for an unindexed file, ambiguous path or unknown symbol. `callers` prints "No callers
found" and exits successfully only when the symbol exists and has no callers.

### Visualization

Expand Down
41 changes: 34 additions & 7 deletions codeprism/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import asyncio
import re
from pathlib import Path
from typing import TYPE_CHECKING

import typer
from rich.console import Console
Expand All @@ -14,6 +15,9 @@
from .core.languages import SUPPORTED_LANGUAGES, normalize_language
from .core.models import SYMBOL_KINDS

if TYPE_CHECKING:
from .query.engine import QueryEngine

# Allowlist for git ref characters — prevents argument injection via diff_range
_SAFE_GIT_REF_RE = re.compile(r"^(?!-)[\w./~^@{}:+\-]{1,200}$")

Expand Down Expand Up @@ -65,6 +69,22 @@ async def _open_session(project_path: str):
return QueryEngine(graph, storage), storage


async def _resolve_query_file(engine: QueryEngine, file: str, project: str) -> str:
"""Report failed file resolution on stderr before running a CLI query."""
path, error = await engine.resolve_file(file, project)
if error is not None:
if "candidates" in error:
typer.echo(error["error"], err=True)
for candidate in error["candidates"]:
typer.echo(f" {candidate}", err=True)
typer.echo(error["hint"], err=True)
else:
typer.echo(f"Not found: {error['error']}", err=True)
raise typer.Exit(1)
assert path is not None
return path


def _parse_target(target: str) -> tuple[str, str]:
"""Parse 'file.py::symbol' → (file_path, symbol_name)."""
if "::" not in target:
Expand Down Expand Up @@ -197,12 +217,13 @@ async def _context(target: str, depth: int, project: str) -> None:
file_path, sym_name = _parse_target(target)
engine, storage = await _open_session(project)
try:
result = await engine.get_context(file_path, sym_name, depth)
resolved = await _resolve_query_file(engine, file_path, project)
result = await engine.get_context(resolved, sym_name, depth)
finally:
await storage.close()

if result is None:
console.print(f"[red]Not found:[/red] {sym_name} in {file_path}")
typer.echo(f"Not found: {sym_name} in {file_path}", err=True)
raise typer.Exit(1)

s = result.symbol
Expand Down Expand Up @@ -243,12 +264,13 @@ async def _impact(target: str, project: str) -> None:
file_path, sym_name = _parse_target(target)
engine, storage = await _open_session(project)
try:
result = await engine.get_impact(file_path, sym_name)
resolved = await _resolve_query_file(engine, file_path, project)
result = await engine.get_impact(resolved, sym_name)
finally:
await storage.close()

if result is None:
console.print(f"[red]Not found:[/red] {sym_name} in {file_path}")
typer.echo(f"Not found: {sym_name} in {file_path}", err=True)
raise typer.Exit(1)

severity_colour = {"LOW": "green", "MEDIUM": "yellow", "HIGH": "red", "CRITICAL": "bright_red"}
Expand Down Expand Up @@ -291,12 +313,13 @@ def summary(
async def _summary(file: str, project: str) -> None:
engine, storage = await _open_session(project)
try:
result = await engine.get_module_summary(file)
resolved = await _resolve_query_file(engine, file, project)
result = await engine.get_module_summary(resolved)
finally:
await storage.close()

if result is None:
console.print(f"[red]Not found:[/red] {file}")
typer.echo(f"Not found: {file}", err=True)
raise typer.Exit(1)

console.print(Panel(result.purpose, title=Path(file).name))
Expand Down Expand Up @@ -335,7 +358,11 @@ async def _callers(target: str, project: str) -> None:
file_path, sym_name = _parse_target(target)
engine, storage = await _open_session(project)
try:
syms = await engine.get_callers(file_path, sym_name)
resolved = await _resolve_query_file(engine, file_path, project)
if await engine.find_symbol(resolved, sym_name) is None:
typer.echo(f"Not found: {sym_name} in {file_path}", err=True)
raise typer.Exit(1)
syms = await engine.get_callers(resolved, sym_name)
finally:
await storage.close()

Expand Down
23 changes: 1 addition & 22 deletions codeprism/mcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,28 +89,7 @@ async def _resolve_file(file: str) -> tuple[str | None, dict[str, Any] | None]:
Tries the path under the project root first, then as a relative suffix
("utils.py", "api/utils.py") matched against every indexed file.
"""
storage = _get()._storage
matches = await storage.find_files_by_path(_abs(file))
if not matches:
matches = await storage.find_files_by_path(file)
if len(matches) == 1:
return matches[0].path, None
root = Path(_project_path).resolve()
if not matches:
return None, {"error": f"File '{file}' not indexed", "project_path": str(root)}

def rel(p: str) -> str:
try:
return Path(p).relative_to(root).as_posix()
except ValueError:
return p

candidates = sorted(rel(m.path) for m in matches)
return None, {
"error": f"Path '{file}' is ambiguous: {len(candidates)} indexed files match",
"candidates": candidates[:20],
"hint": f"Pass a longer path, e.g. '{candidates[0]}'",
}
return await _get().resolve_file(file, _project_path)


async def _symbol_refs(symbols) -> list[dict[str, Any]]:
Expand Down
28 changes: 28 additions & 0 deletions codeprism/query/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,34 @@ async def find_symbol(self, file_path: str, name: str) -> SymbolRecord | None:
async def get_file(self, file_path: str) -> FileRecord | None:
return await self._storage.get_file_by_path(file_path)

async def resolve_file(
self, file: str, project_path: str
) -> tuple[str | None, dict[str, Any] | None]:
"""Resolve against the project root, then by suffix, or explain a failed lookup."""
root = Path(project_path).resolve()
path = Path(file)
absolute = str((path if path.is_absolute() else root / path).resolve()) if file else file
matches = await self._storage.find_files_by_path(absolute)
if not matches:
matches = await self._storage.find_files_by_path(file)
if len(matches) == 1:
return matches[0].path, None
if not matches:
return None, {"error": f"File '{file}' not indexed", "project_path": str(root)}

def rel(p: str) -> str:
try:
return Path(p).relative_to(root).as_posix()
except ValueError:
return p

candidates = sorted(rel(m.path) for m in matches)
return None, {
"error": f"Path '{file}' is ambiguous: {len(candidates)} indexed files match",
"candidates": candidates[:20],
"hint": f"Pass a longer path, e.g. '{candidates[0]}'",
}

# ── Callers / callees ─────────────────────────────────────────────────────

async def get_callers(self, file_path: str, symbol_name: str) -> list[SymbolRecord]:
Expand Down
4 changes: 4 additions & 0 deletions docs/changelog.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,10 @@ Versioning follows [Semantic Versioning](https://semver.org/).
## [Unreleased]

### Fixed
- CLI `callers`, `context`, `impact` and `summary` distinguish missing files and ambiguous paths,
report candidates on stderr, and exit with status 1 for lookup errors. `callers` no longer
reports an unknown symbol as a successful zero-callers result. MCP responses stay unchanged.
(#44)
- `index --languages` now rejects unknown names with a list of supported languages. Names remain
case-insensitive, and supported aliases such as `py` and `C++` still work. (#55)
- `codeprism --version` now exists, and `codeprism.__version__` is read from the installed
Expand Down
160 changes: 160 additions & 0 deletions tests/test_cli_query_errors.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,160 @@
"""CLI query errors remain distinct from valid empty results on a real index."""

from collections.abc import AsyncIterator
from pathlib import Path

import pytest
from typer.testing import CliRunner

import codeprism.mcp.server as srv
from codeprism.cli import app
from codeprism.core.graph import GraphEngine
from codeprism.core.paths import get_db_path
from codeprism.core.storage import StorageManager
from codeprism.query.engine import QueryEngine

runner = CliRunner()
FILE_COMMANDS = ("callers", "context", "impact", "summary")


@pytest.fixture
def query_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Index two same-named files without touching the user's data directory."""
monkeypatch.setattr("codeprism.core.paths.get_data_dir", lambda: tmp_path / "data")
monkeypatch.setenv("GIT_CEILING_DIRECTORIES", str(tmp_path))
project = tmp_path / "project"
for directory in ("api", "core"):
source = project / directory / "utils.py"
source.parent.mkdir(parents=True)
source.write_text("def helper():\n return 1\n", encoding="utf-8")
with (project / "api" / "utils.py").open("a", encoding="utf-8") as stream:
stream.write("\ndef used():\n return 2\n\ndef invoke():\n return used()\n")
result = runner.invoke(app, ["index", str(project), "--languages", "python", "--workers", "1"])
assert result.exit_code == 0, result.output
return project


def _target(command: str, file: str, symbol: str = "helper") -> str:
"""Summary takes a file; the other query commands take file::symbol."""
return file if command == "summary" else f"{file}::{symbol}"


@pytest.mark.parametrize("command", FILE_COMMANDS)
def test_cli_ambiguous_file_reports_candidates_on_stderr(query_project: Path, command: str) -> None:
"""Every affected command explains an ambiguous file and exits nonzero."""
result = runner.invoke(
app, [command, _target(command, "utils.py"), "--project", str(query_project)]
)
assert result.exit_code == 1, result.output
assert result.stdout == ""
assert "ambiguous" in result.stderr
assert "2 indexed files match" in result.stderr
assert result.stderr.index("api/utils.py") < result.stderr.index("core/utils.py")
assert "Pass a longer path" in result.stderr
assert "No callers found" not in result.output


@pytest.mark.parametrize("command", FILE_COMMANDS)
def test_cli_missing_file_reports_not_found_on_stderr(query_project: Path, command: str) -> None:
"""An unindexed file is an error rather than a successful empty answer."""
result = runner.invoke(
app, [command, _target(command, "nowhere.py"), "--project", str(query_project)]
)
assert result.exit_code == 1, result.output
assert result.stdout == ""
assert "Not found:" in result.stderr and "nowhere.py" in result.stderr
assert "No callers found" not in result.output


@pytest.mark.parametrize("command", ["callers", "context", "impact"])
def test_cli_missing_symbol_reports_not_found_on_stderr(query_project: Path, command: str) -> None:
"""A known file with an unknown symbol remains distinct from zero callers."""
result = runner.invoke(
app,
[command, _target(command, "api/utils.py", "nope"), "--project", str(query_project)],
)
assert result.exit_code == 1, result.output
assert result.stdout == ""
assert "Not found:" in result.stderr and "nope" in result.stderr
assert "No callers found" not in result.output


def test_cli_existing_symbol_with_zero_callers_succeeds(query_project: Path) -> None:
"""The genuine empty answer from the issue reproduction still exits zero."""
result = runner.invoke(
app, ["callers", "api/utils.py::helper", "--project", str(query_project)]
)
assert result.exit_code == 0, result.output
assert result.stderr == ""
assert "No callers found for helper" in result.stdout


def test_cli_existing_symbol_with_callers_keeps_results(query_project: Path) -> None:
"""Validation preserves the actual graph caller rather than fabricating an answer."""
result = runner.invoke(app, ["callers", "api/utils.py::used", "--project", str(query_project)])
assert result.exit_code == 0, result.output
assert result.stderr == ""
assert "Callers of used" in result.stdout and "invoke" in result.stdout


@pytest.mark.parametrize("command", FILE_COMMANDS)
def test_cli_longer_path_resolves_against_project_not_cwd(
query_project: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, command: str
) -> None:
"""All four commands accept an unambiguous path from an unrelated working directory."""
cwd = tmp_path / "other"
cwd.mkdir()
monkeypatch.chdir(cwd)
result = runner.invoke(
app, [command, _target(command, "api/utils.py"), "--project", "../project"]
)
assert result.exit_code == 0, result.output
assert result.stderr == ""
expected = {
"callers": "No callers found for helper",
"context": "helper",
"impact": "Impact: helper",
"summary": "utils.py",
}
assert expected[command] in result.stdout


@pytest.fixture
async def query_engine(query_project: Path) -> AsyncIterator[QueryEngine]:
"""Load the persisted project graph with real SQLite storage for resolver tests."""
storage = StorageManager(get_db_path(query_project))
await storage.initialize()
try:
graph = GraphEngine()
await graph.load_from_storage(storage)
yield QueryEngine(graph, storage)
finally:
await storage.close()


@pytest.mark.parametrize("file", ["utils.py", "nowhere.py", "api/utils.py", ""])
async def test_shared_resolver_preserves_mcp_response_contract(
query_project: Path, query_engine: QueryEngine, monkeypatch: pytest.MonkeyPatch, file: str
) -> None:
"""The existing MCP wrapper keeps its exact response keys, messages and candidates."""
monkeypatch.setattr(srv, "_engine", query_engine)
monkeypatch.setattr(srv, "_project_path", str(query_project))
expected: tuple[str | None, dict[str, object] | None]
if file == "utils.py":
expected = (
None,
{
"error": "Path 'utils.py' is ambiguous: 2 indexed files match",
"candidates": ["api/utils.py", "core/utils.py"],
"hint": "Pass a longer path, e.g. 'api/utils.py'",
},
)
elif file == "api/utils.py":
expected = (str((query_project / file).resolve()), None)
else:
expected = (
None,
{"error": f"File '{file}' not indexed", "project_path": str(query_project.resolve())},
)
assert await query_engine.resolve_file(file, str(query_project)) == expected
assert await srv._resolve_file(file) == expected
Loading