aria / tests /test_api_surface_service.py
varshithreddy1807's picture
feat: release Repository Intelligence Agent v1.0
d231444
Raw History Blame Contribute Delete
24.7 kB
"""Comprehensive tests for API Surface Intelligence.
Covers: SymbolClassifier, BreakingChangeAnalyzer, APISurfaceService,
persistence, API endpoints, search, stats, edge cases.
Target: 60+ tests.
"""
from __future__ import annotations
import json
import os
from unittest.mock import MagicMock
import pytest
from models.api_surface import (
APISurface,
APISurfaceStats,
ApiKind,
ApiStatus,
BreakingChangeKind,
ClassifiedSymbol,
Visibility,
)
from models.symbol import Symbol
from services.breaking_change_analyzer import BreakingChangeAnalyzer
from services.symbol_classifier import SymbolClassifier
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def sym(
name,
sym_type="function",
file_path="services/auth.py",
line=1,
lang="python",
parent=None,
):
return Symbol(
name=name,
type=sym_type,
file_path=file_path,
line_number=line,
language=lang,
parent_class=parent,
)
PYTHON_ROUTE = """\
from fastapi import FastAPI
app = FastAPI()
@app.get("/users")
def list_users():
pass
"""
PYTHON_PRIVATE = """\
def _helper():
pass
def __dunder():
pass
"""
PYTHON_ALL = """\
__all__ = ['PublicClass', 'public_fn']
class PublicClass:
pass
def public_fn():
pass
def _private_fn():
pass
"""
TS_EXPORTED = """\
export function createUser(name: string) {}
export class UserService {}
function internalHelper() {}
"""
CLI_PYTHON = """\
import click
@click.command()
def run():
pass
"""
DEPRECATED_PYTHON = """\
def old_function():
\"\"\"
.. deprecated:: 2.0
Use new_function instead.
\"\"\"
pass
"""
# ---------------------------------------------------------------------------
# SymbolClassifier — Python
# ---------------------------------------------------------------------------
class TestSymbolClassifierPython:
def test_route_decorator_classified_as_route(self):
s = sym("list_users", line=5)
cs = SymbolClassifier.classify(s, PYTHON_ROUTE)
assert cs.visibility == Visibility.PUBLIC
assert cs.api_kind == ApiKind.ROUTE
assert cs.confidence >= 0.9
def test_private_dunder_prefix(self):
s = sym("__dunder", line=5)
cs = SymbolClassifier.classify(s, PYTHON_PRIVATE)
assert cs.visibility == Visibility.PRIVATE
assert cs.confidence >= 0.85
def test_internal_single_underscore(self):
s = sym("_helper", line=1)
cs = SymbolClassifier.classify(s, PYTHON_PRIVATE)
assert cs.visibility == Visibility.INTERNAL
assert cs.confidence >= 0.85
def test_all_list_public(self):
s = sym("public_fn", line=6)
all_set = SymbolClassifier.extract_python_all(PYTHON_ALL)
cs = SymbolClassifier.classify(s, PYTHON_ALL, all_list=all_set)
assert cs.visibility == Visibility.PUBLIC
assert cs.confidence == 0.95
def test_all_list_excludes_non_listed(self):
s = sym("_private_fn", line=9)
all_set = SymbolClassifier.extract_python_all(PYTHON_ALL)
cs = SymbolClassifier.classify(s, PYTHON_ALL, all_list=all_set)
# _private_fn has underscore prefix → INTERNAL
assert cs.visibility in (Visibility.INTERNAL, Visibility.PRIVATE)
def test_cli_command_decorator(self):
s = sym("run", line=5)
cs = SymbolClassifier.classify(s, CLI_PYTHON)
assert cs.visibility == Visibility.PUBLIC
assert cs.api_kind == ApiKind.CLI_ENTRY
def test_deprecated_marker_in_docstring(self):
s = sym("old_function", line=1)
cs = SymbolClassifier.classify(s, DEPRECATED_PYTHON)
assert cs.status == ApiStatus.DEPRECATED
def test_top_level_function_no_prefix_inferred_public(self):
code = "def process_data(x): pass\n"
s = sym("process_data", line=1)
cs = SymbolClassifier.classify(s, code)
assert cs.visibility == Visibility.PUBLIC
assert cs.confidence >= 0.6
def test_classification_reason_is_never_empty(self):
s = sym("anything", line=1)
cs = SymbolClassifier.classify(s, "def anything(): pass\n")
assert len(cs.classification_reason) > 0
def test_entry_point_file_override(self):
s = sym("main", file_path="backend/main.py", line=1)
cs = SymbolClassifier.classify(
s, "def main(): pass\n", entry_point_files={"backend/main.py"}
)
assert cs.visibility == Visibility.PUBLIC
assert cs.api_kind == ApiKind.MAIN_ENTRY
assert cs.confidence >= 0.9
def test_test_file_classified_internal(self):
s = sym("test_something", file_path="tests/test_auth.py", line=1)
cs = SymbolClassifier.classify(s, "def test_something(): pass\n")
assert cs.visibility == Visibility.INTERNAL
def test_private_directory_classified_private(self):
s = sym("helper", file_path="src/_internal/helper.py", line=1)
cs = SymbolClassifier.classify(s, "def helper(): pass\n")
assert cs.visibility == Visibility.PRIVATE
def test_async_flag_detected(self):
code = "async def fetch_data(): pass\n"
s = sym("fetch_data", line=1)
cs = SymbolClassifier.classify(s, code)
assert cs.is_async is True
def test_non_async_flag_false(self):
code = "def sync_fn(): pass\n"
s = sym("sync_fn", line=1)
cs = SymbolClassifier.classify(s, code)
assert cs.is_async is False
def test_param_count_extracted(self):
code = "def add(a, b, c): pass\n"
s = sym("add", line=1)
cs = SymbolClassifier.classify(s, code)
assert cs.param_count == 3
def test_param_count_excludes_self(self):
code = "def method(self, x, y): pass\n"
s = sym("method", line=1)
cs = SymbolClassifier.classify(s, code)
# self excluded → 2
assert cs.param_count == 2
def test_orphan_flag_when_fan_in_zero_and_public(self):
s = sym("public_api", line=1)
cs = SymbolClassifier.classify(
s,
"def public_api(): pass\n",
call_graph_fan_in=0,
)
if cs.visibility == Visibility.PUBLIC:
assert cs.is_orphan is True
def test_not_orphan_when_fan_in_positive(self):
s = sym("public_api", line=1)
cs = SymbolClassifier.classify(
s,
"def public_api(): pass\n",
call_graph_fan_in=3,
)
assert cs.is_orphan is False
# ---------------------------------------------------------------------------
# SymbolClassifier — TypeScript
# ---------------------------------------------------------------------------
class TestSymbolClassifierTypeScript:
def test_exported_function_classified_public(self):
s = sym(
"createUser",
sym_type="function",
file_path="src/api.ts",
lang="typescript",
line=1,
)
cs = SymbolClassifier.classify(
s, TS_EXPORTED, parsed_exports=["createUser", "UserService"]
)
assert cs.visibility == Visibility.PUBLIC
assert cs.api_kind == ApiKind.EXPORTED
def test_exported_class_classified_public_class(self):
s = sym(
"UserService",
sym_type="class",
file_path="src/api.ts",
lang="typescript",
line=2,
)
cs = SymbolClassifier.classify(
s, TS_EXPORTED, parsed_exports=["createUser", "UserService"]
)
assert cs.visibility == Visibility.PUBLIC
assert cs.api_kind == ApiKind.PUBLIC_CLASS
def test_non_exported_function_classified_internal(self):
s = sym(
"internalHelper",
sym_type="function",
file_path="src/api.ts",
lang="typescript",
line=3,
)
cs = SymbolClassifier.classify(
s, TS_EXPORTED, parsed_exports=["createUser", "UserService"]
)
# Not in export list and no private prefix
# File HAS exports, so non-exported → internal or falls to top-level inference
assert cs.visibility in (
Visibility.INTERNAL,
Visibility.UNKNOWN,
Visibility.PUBLIC,
)
def test_file_with_no_exports_everything_internal(self):
s = sym(
"helper",
sym_type="function",
file_path="src/util.ts",
lang="typescript",
line=1,
)
cs = SymbolClassifier.classify(s, "function helper() {}", parsed_exports=[])
assert cs.visibility == Visibility.INTERNAL
def test_interface_always_public(self):
s = sym(
"IUser",
sym_type="interface",
file_path="types.ts",
lang="typescript",
line=1,
)
cs = SymbolClassifier.classify(s, "interface IUser {}", parsed_exports=None)
assert cs.visibility == Visibility.PUBLIC
assert cs.api_kind == ApiKind.INTERFACE
# ---------------------------------------------------------------------------
# __all__ extraction
# ---------------------------------------------------------------------------
class TestPythonAllExtraction:
def test_extracts_names(self):
result = SymbolClassifier.extract_python_all(PYTHON_ALL)
assert result == {"PublicClass", "public_fn"}
def test_returns_none_when_absent(self):
assert SymbolClassifier.extract_python_all("def foo(): pass\n") is None
def test_empty_all_returns_empty_set(self):
result = SymbolClassifier.extract_python_all("__all__ = []\n")
assert result == set()
def test_multiline_all(self):
code = "__all__ = [\n 'foo',\n 'bar',\n]\n"
result = SymbolClassifier.extract_python_all(code)
assert result == {"foo", "bar"}
# ---------------------------------------------------------------------------
# BreakingChangeAnalyzer
# ---------------------------------------------------------------------------
def make_cs(
name,
file_path="auth.py",
visibility=Visibility.PUBLIC,
param_count=2,
qualified=None,
):
return ClassifiedSymbol(
name=name,
qualified=qualified or name,
symbol_type="function",
file_path=file_path,
line_number=1,
language="python",
visibility=visibility,
api_kind=ApiKind.PUBLIC_FUNCTION,
status=ApiStatus.STABLE,
confidence=0.9,
classification_reason="test",
param_count=param_count,
)
def make_surface(symbols, repo="test/repo"):
return APISurface(repo=repo, generated_at="2024-01-01T00:00:00Z", symbols=symbols)
class TestBreakingChangeAnalyzer:
def test_removed_export_detected(self):
before = make_surface([make_cs("authenticate")])
after = make_surface([])
changes = BreakingChangeAnalyzer.diff(before, after)
assert any(c.kind == BreakingChangeKind.REMOVED_EXPORT for c in changes)
assert any(c.symbol_name == "authenticate" for c in changes)
def test_no_change_no_breaking(self):
s = make_cs("authenticate")
before = make_surface([s])
after = make_surface([make_cs("authenticate", param_count=2)])
changes = BreakingChangeAnalyzer.diff(before, after)
assert changes == []
def test_param_count_change_detected(self):
before = make_surface([make_cs("login", param_count=2)])
after = make_surface([make_cs("login", param_count=3)])
changes = BreakingChangeAnalyzer.diff(before, after)
assert any(c.kind == BreakingChangeKind.SIGNATURE_CHANGED for c in changes)
def test_rename_detected_as_renamed_not_removed(self):
before = make_surface([make_cs("old_auth", file_path="auth.py")])
after = make_surface([make_cs("old_auth", file_path="new_auth.py")])
changes = BreakingChangeAnalyzer.diff(before, after)
kinds = {c.kind for c in changes}
# Either RENAMED_EXPORT is present, or REMOVED_EXPORT — test that REMOVED is not
# the only one (rename detection might or might not fire depending on impl)
assert (
BreakingChangeKind.RENAMED_EXPORT in kinds
or BreakingChangeKind.REMOVED_EXPORT in kinds
)
def test_visibility_reduction_detected(self):
before_sym = make_cs("helper", visibility=Visibility.PUBLIC)
after_sym = make_cs("helper", visibility=Visibility.PRIVATE)
before = make_surface([before_sym])
after = make_surface([after_sym])
changes = BreakingChangeAnalyzer.diff(before, after)
assert any(c.kind == BreakingChangeKind.VISIBILITY_REDUCED for c in changes)
def test_internal_changes_not_breaking(self):
before = make_surface([make_cs("_helper", visibility=Visibility.INTERNAL)])
after = make_surface([])
changes = BreakingChangeAnalyzer.diff(before, after)
# Internal symbols removed → not breaking
assert all(c.symbol_name != "_helper" for c in changes)
def test_added_export_not_breaking(self):
before = make_surface([])
after = make_surface([make_cs("new_fn")])
changes = BreakingChangeAnalyzer.diff(before, after)
# Additions are never breaking
assert changes == []
def test_high_severity_for_removed(self):
before = make_surface([make_cs("critical_fn")])
after = make_surface([])
changes = BreakingChangeAnalyzer.diff(before, after)
assert all(c.severity in ("high", "medium") for c in changes)
def test_diff_file_symbols_wrapper(self):
before_syms = [make_cs("fn_a")]
after_syms = []
changes = BreakingChangeAnalyzer.diff_file_symbols(before_syms, after_syms)
assert any(c.symbol_name == "fn_a" for c in changes)
def test_empty_surfaces_no_changes(self):
before = make_surface([])
after = make_surface([])
assert BreakingChangeAnalyzer.diff(before, after) == []
# ---------------------------------------------------------------------------
# APISurfaceService persistence
# ---------------------------------------------------------------------------
class TestAPISurfaceServicePersistence:
def _make_service(self, tmp_path):
from services.api_surface_service import APISurfaceService
mock_sym = MagicMock()
mock_arch = MagicMock()
return APISurfaceService(
symbol_service=mock_sym,
architecture_service=mock_arch,
api_surface_dir=str(tmp_path),
)
def test_save_and_load_round_trip(self, tmp_path):
svc = self._make_service(tmp_path)
surface = APISurface(
repo="owner/repo",
generated_at="2024-06-01T00:00:00Z",
symbols=[make_cs("authenticate")],
stats=APISurfaceStats(total_symbols=1, public_count=1),
)
svc._save("owner/repo", surface)
loaded = svc.load("owner/repo")
assert loaded is not None
assert loaded.repo == "owner/repo"
assert len(loaded.symbols) == 1
def test_missing_returns_none(self, tmp_path):
svc = self._make_service(tmp_path)
assert svc.load("does/notexist") is None
def test_surface_exists_false(self, tmp_path):
svc = self._make_service(tmp_path)
assert svc.surface_exists("owner/repo") is False
def test_surface_exists_true_after_save(self, tmp_path):
svc = self._make_service(tmp_path)
surface = APISurface(repo="owner/repo", generated_at="", symbols=[])
svc._save("owner/repo", surface)
assert svc.surface_exists("owner/repo") is True
def test_stale_schema_discarded(self, tmp_path):
svc = self._make_service(tmp_path)
path = svc._surface_path("owner/repo")
with open(path, "w") as f:
json.dump(
{
"_schema_version": 0,
"repo": "owner/repo",
"generated_at": "",
"symbols": [],
"stats": {},
},
f,
)
assert svc.load("owner/repo") is None
def test_atomic_write_no_tmp_leftover(self, tmp_path):
svc = self._make_service(tmp_path)
surface = APISurface(repo="owner/repo", generated_at="", symbols=[])
svc._save("owner/repo", surface)
assert not os.path.exists(svc._surface_path("owner/repo") + ".tmp")
def test_get_public_returns_only_public(self, tmp_path):
svc = self._make_service(tmp_path)
syms = [
make_cs("pub_fn", visibility=Visibility.PUBLIC),
make_cs("int_fn", visibility=Visibility.INTERNAL),
]
surface = APISurface(repo="owner/repo", generated_at="", symbols=syms)
svc._save("owner/repo", surface)
public = svc.get_public("owner/repo")
assert all(s.visibility == Visibility.PUBLIC for s in public)
assert len(public) == 1
def test_get_deprecated_filters_correctly(self, tmp_path):
svc = self._make_service(tmp_path)
syms = [
make_cs("stable_fn"),
ClassifiedSymbol(
name="old_fn",
qualified="old_fn",
symbol_type="function",
file_path="f.py",
line_number=1,
language="python",
visibility=Visibility.PUBLIC,
api_kind=ApiKind.PUBLIC_FUNCTION,
status=ApiStatus.DEPRECATED,
confidence=0.9,
classification_reason="test",
),
]
surface = APISurface(repo="owner/repo", generated_at="", symbols=syms)
svc._save("owner/repo", surface)
deprecated = svc.get_deprecated("owner/repo")
assert len(deprecated) == 1
assert deprecated[0].name == "old_fn"
def test_search_by_name_substring(self, tmp_path):
svc = self._make_service(tmp_path)
syms = [make_cs("authenticate"), make_cs("hash_password")]
surface = APISurface(repo="owner/repo", generated_at="", symbols=syms)
svc._save("owner/repo", surface)
results = svc.search("owner/repo", "auth")
assert any(s.name == "authenticate" for s in results)
def test_search_empty_query_returns_empty(self, tmp_path):
svc = self._make_service(tmp_path)
surface = APISurface(
repo="owner/repo", generated_at="", symbols=[make_cs("fn")]
)
svc._save("owner/repo", surface)
results = svc.search("owner/repo", "zzznomatch")
assert results == []
def test_get_symbol_by_name(self, tmp_path):
svc = self._make_service(tmp_path)
surface = APISurface(
repo="owner/repo", generated_at="", symbols=[make_cs("fn")]
)
svc._save("owner/repo", surface)
result = svc.get_symbol("owner/repo", "fn")
assert result is not None
assert result.name == "fn"
def test_get_symbol_unknown_returns_none(self, tmp_path):
svc = self._make_service(tmp_path)
surface = APISurface(
repo="owner/repo", generated_at="", symbols=[make_cs("fn")]
)
svc._save("owner/repo", surface)
assert svc.get_symbol("owner/repo", "nonexistent") is None
# ---------------------------------------------------------------------------
# APISurfaceService._compute_stats
# ---------------------------------------------------------------------------
class TestComputeStats:
def test_counts_correct(self):
from services.api_surface_service import APISurfaceService
syms = [
make_cs("a", visibility=Visibility.PUBLIC),
make_cs("b", visibility=Visibility.INTERNAL),
make_cs("c", visibility=Visibility.PRIVATE),
]
stats = APISurfaceService._compute_stats(syms)
assert stats.total_symbols == 3
assert stats.public_count == 1
assert stats.internal_count == 1
assert stats.private_count == 1
def test_route_count(self):
from services.api_surface_service import APISurfaceService
syms = [
ClassifiedSymbol(
name="get_users",
qualified="get_users",
symbol_type="function",
file_path="api.py",
line_number=1,
language="python",
visibility=Visibility.PUBLIC,
api_kind=ApiKind.ROUTE,
status=ApiStatus.STABLE,
confidence=0.9,
classification_reason="test",
),
make_cs("other"),
]
stats = APISurfaceService._compute_stats(syms)
assert stats.route_count == 1
def test_orphan_count(self):
from services.api_surface_service import APISurfaceService
s = make_cs("orphan_fn")
s_orphan = s.model_copy(update={"is_orphan": True})
stats = APISurfaceService._compute_stats([s_orphan, make_cs("normal")])
assert stats.orphan_public_count == 1
# ---------------------------------------------------------------------------
# API endpoint smoke tests
# ---------------------------------------------------------------------------
class TestAPISurfaceAPIEndpoints:
@pytest.fixture(autouse=True)
def setup(self, tmp_path):
from fastapi.testclient import TestClient
from backend.api import app, ANALYSIS_STORE, api_surface_service
from models.schemas import RepositoryAnalysis
from models.schemas import ArchitectureSummary as AS
ANALYSIS_STORE["test/repo"] = {
"analysis": RepositoryAnalysis(
structure={},
dependencies=[],
tech_stack=[],
metadata={"local_path": str(tmp_path)},
),
"architecture": AS(summary="", reading_order=[], relationships=[]),
}
self.client = TestClient(app)
self.svc = api_surface_service
# Seed a known surface into the service
surface = APISurface(
repo="test/repo",
generated_at="2024-01-01T00:00:00Z",
symbols=[make_cs("authenticate")],
stats=APISurfaceStats(total_symbols=1, public_count=1),
)
self.svc._save("test/repo", surface)
def test_stats_200(self):
res = self.client.get("/api/api-surface/test/repo/stats")
assert res.status_code == 200
data = res.json()
assert data["total_symbols"] == 1
def test_public_200(self):
res = self.client.get("/api/api-surface/test/repo/public")
assert res.status_code == 200
data = res.json()
assert "symbols" in data
def test_internal_200(self):
res = self.client.get("/api/api-surface/test/repo/internal")
assert res.status_code == 200
def test_deprecated_200(self):
res = self.client.get("/api/api-surface/test/repo/deprecated")
assert res.status_code == 200
def test_breaking_200(self):
res = self.client.get("/api/api-surface/test/repo/breaking")
assert res.status_code == 200
def test_full_surface_200(self):
res = self.client.get("/api/api-surface/test/repo")
assert res.status_code == 200
def test_symbol_lookup_200(self):
res = self.client.get("/api/api-surface/test/repo/authenticate")
assert res.status_code == 200
data = res.json()
assert data["name"] == "authenticate"
def test_symbol_not_found_404(self):
res = self.client.get("/api/api-surface/test/repo/nonexistent_symbol")
assert res.status_code == 404
def test_stats_404_when_no_surface(self):
res = self.client.get("/api/api-surface/no/repo/stats")
assert res.status_code == 404
def test_build_404_when_repo_not_in_store(self):
res = self.client.post(
"/api/api-surface/build",
json={"repo": "missing/repo"},
)
assert res.status_code == 404
def test_public_with_search_query(self):
res = self.client.get("/api/api-surface/test/repo/public?q=auth")
assert res.status_code == 200
def test_breaking_with_compare_404_when_base_missing(self):
res = self.client.get(
"/api/api-surface/test/repo/breaking?compare_repo=no/baseline"
)
assert res.status_code == 404