Spaces:
Configuration error
Configuration error
| """Unit tests for the FastAPI backend endpoints.""" | |
| import sys | |
| import os | |
| import json | |
| from unittest.mock import patch | |
| from fastapi.testclient import TestClient | |
| # Ensure root directory is in sys.path | |
| sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| from backend.api import app | |
| client = TestClient(app) | |
| def test_get_examples() -> None: | |
| """Verifies example repos list is returned correctly.""" | |
| response = client.get("/api/repos/examples") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert len(data) > 0 | |
| assert "name" in data[0] | |
| assert "url" in data[0] | |
| def test_get_recent() -> None: | |
| """Verifies recent repos list returns empty or populated list.""" | |
| response = client.get("/api/repos/recent") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert isinstance(data, list) | |
| def test_map_issue() -> None: | |
| """Verifies issue mapping endpoint works and returns plan structure.""" | |
| payload = { | |
| "repo": "test/repo", | |
| "title": "Fix memory leaks in cache", | |
| "description": "The cache store is leaking reference counts.", | |
| } | |
| response = client.post("/api/issues/map", json=payload) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "issue_summary" in data | |
| assert "relevant_files" in data | |
| assert "implementation_plan" in data | |
| assert "issue_type" in data | |
| assert "complexity" in data | |
| assert "confidence" in data | |
| assert "verified" in data | |
| assert "sources" in data | |
| assert "affected_components" in data | |
| def test_index_invalid_repo_url_returns_400() -> None: | |
| """Invalid GitHub repo URL should not crash and should return 4xx.""" | |
| response = client.post("/api/index", json={"repo_url": "hello"}) | |
| assert response.status_code in (400, 422) | |
| def test_index_nonexistent_repo_returns_404() -> None: | |
| """Nonexistent repository should map to 404 not 500.""" | |
| response = client.post( | |
| "/api/index", | |
| json={ | |
| "repo_url": "https://github.com/this-repo-should-not-exist-1234567890/does-not-exist" | |
| }, | |
| ) | |
| # Depending on environment/network, 404 mapping should happen; otherwise this test may fail. | |
| assert response.status_code in (404,) | |
| def test_analyze_repository_success() -> None: | |
| """Verifies analyze endpoint works end-to-end and returns correct schema.""" | |
| with ( | |
| patch("backend.routers.repositories.github_service") as mock_gh, | |
| patch("backend.routers.repositories.embedding_service") as mock_embed, | |
| patch("backend.routers.repositories.chroma_store"), | |
| patch("backend.routers.repositories.symbol_service"), | |
| patch("backend.routers.repositories.architecture_service") as mock_arch, | |
| patch("backend.dependencies.call_graph_service") as mock_call, | |
| patch("backend.dependencies.api_surface_service") as mock_api, | |
| patch( | |
| "backend.routers.repositories.generate_architecture_summary" | |
| ) as mock_summary, | |
| patch("backend.routers.repositories.snapshot_store") as mock_store, | |
| patch("backend.routers.repositories._persist_analysis_store"), | |
| ): | |
| mock_gh.clone_repository.return_value = "/dummy/path" | |
| mock_gh.extract_source_files.return_value = [ | |
| {"path": "main.py", "content": "print('hello')"} | |
| ] | |
| mock_store.load.return_value = None | |
| mock_embed.generate_embeddings.return_value = [[0.1] * 1536] | |
| mock_arch.build_full.return_value = { | |
| "files_parsed": 1, | |
| "dependencies_found": 0, | |
| "entry_points": [], | |
| } | |
| mock_call.build_full.return_value = [] | |
| mock_api.build_full.return_value = [] | |
| from models.schemas import ArchitectureSummary | |
| mock_summary.return_value = ArchitectureSummary( | |
| summary="Dummy summary", reading_order=[], relationships=[] | |
| ) | |
| payload = {"url": "https://github.com/test-owner/test-repo", "branch": "main"} | |
| response = client.post("/api/analyze", json=payload) | |
| assert response.status_code == 200 | |
| events = [] | |
| for line in response.iter_lines(): | |
| if line.startswith("data: "): | |
| events.append(json.loads(line[6:])) | |
| assert any(e.get("status") == "complete" for e in events) | |
| done_event = next(e for e in events if e.get("status") == "done") | |
| assert done_event["repo"] == "test-owner/test-repo" | |
| # Verify frontend parser accepts response and extracts correct owner/repo | |
| repo_path = ( | |
| done_event.get("repo") | |
| or done_event.get("repository") | |
| or ( | |
| done_event.get("owner") | |
| and done_event.get("repo_name") | |
| and f"{done_event.get('owner')}/{done_event.get('repo_name')}" | |
| ) | |
| ) | |
| assert repo_path is not None, ( | |
| "Frontend parser would reject: missing repo in analysis result" | |
| ) | |
| parts = repo_path.split("/") | |
| assert len(parts) == 2 | |
| owner, repo = parts | |
| assert owner == "test-owner" | |
| assert repo == "test-repo" | |
| # Verify navigation URL generates correctly | |
| nav_url = f"/analysis?owner={owner}&repo={repo}" | |
| assert nav_url == "/analysis?owner=test-owner&repo=test-repo" | |
| # Verify repository loads after analysis | |
| response_detail = client.get(f"/api/analysis/{owner}/{repo}") | |
| assert response_detail.status_code == 200 | |
| detail_data = response_detail.json() | |
| assert "analysis" in detail_data | |
| assert "architecture" in detail_data | |
| def test_analyze_repository_failure() -> None: | |
| """Verifies analyze endpoint handles exceptions gracefully by yielding status: error and status: done.""" | |
| with ( | |
| patch("backend.routers.repositories.github_service") as mock_gh, | |
| patch("backend.routers.repositories.snapshot_store"), | |
| ): | |
| mock_gh.clone_repository.side_effect = Exception("Mock clone error") | |
| payload = {"url": "https://github.com/test-owner/test-repo", "branch": "main"} | |
| response = client.post("/api/analyze", json=payload) | |
| assert response.status_code == 200 | |
| events = [] | |
| for line in response.iter_lines(): | |
| if line.startswith("data: "): | |
| events.append(json.loads(line[6:])) | |
| assert any(e.get("status") == "error" for e in events) | |
| error_event = next(e for e in events if e.get("status") == "error") | |
| # Raw exception text stays in server logs only; clients receive the | |
| # structured, sanitized error envelope. | |
| assert "Mock clone error" not in error_event["message"] | |
| assert "An unexpected internal error occurred." in error_event["message"] | |
| assert "Stage:" in error_event["message"] | |
| done_event = next(e for e in events if e.get("status") == "done") | |
| assert "repo" not in done_event | |
| def test_chat_reload_resolves_pipeline_without_attribute_error() -> None: | |
| """Regression test for H-1: /chat/reload must not raise AttributeError. | |
| The endpoint previously referenced `dependencies._retrieval_pipeline` | |
| which does not exist in the module, causing a silent failure or crash. | |
| This test verifies: | |
| 1. The endpoint does not raise AttributeError. | |
| 2. When a retrieval pipeline exists, the ProviderManager is replaced. | |
| 3. The endpoint returns a structured response (success or provider error). | |
| """ | |
| from backend import dependencies | |
| from unittest.mock import AsyncMock, MagicMock | |
| # Simulate a cached retrieval pipeline with a ProviderManager | |
| mock_pipeline = MagicMock() | |
| mock_pipeline.provider_manager = MagicMock() | |
| original_pm = mock_pipeline.provider_manager | |
| with patch.dict(dependencies._SINGLETONS, {"retrieval_pipeline": mock_pipeline}): | |
| with patch( | |
| "services.llm.provider_factory.ProviderFactory.get_provider" | |
| ) as mock_get: | |
| mock_provider = AsyncMock() | |
| mock_provider.generate = AsyncMock(return_value="ready") | |
| mock_get.return_value = mock_provider | |
| response = client.post("/api/v1/chat/reload") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["status"] == "ok" | |
| # Verify the ProviderManager was replaced (not the original instance) | |
| assert mock_pipeline.provider_manager is not original_pm | |
| def test_openapi_version_matches_release() -> None: | |
| """M-2 regression: OpenAPI metadata must report the current release version.""" | |
| response = client.get("/openapi.json") | |
| assert response.status_code == 200 | |
| schema = response.json() | |
| assert schema["info"]["version"] == "1.5.0" | |
| def test_production_startup_skips_embedding_warmup() -> None: | |
| """Verify that in production mode, startup warmup does not eagerly load or encode with embedding model.""" | |
| from backend.api import _warmup_services | |
| with ( | |
| patch("backend.settings.settings.app_env", "production"), | |
| patch("services.embedding_service._get_model") as mock_get_model, | |
| patch("services.tree_sitter_service.TreeSitterService") as mock_ts, | |
| ): | |
| _warmup_services() | |
| mock_get_model.assert_not_called() | |
| mock_ts.assert_called_once() | |