Spaces:
Sleeping
Sleeping
File size: 2,888 Bytes
af5af97 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | from unittest.mock import AsyncMock, patch
import pytest
from app.models.issue import IssueReport
from app.models.report import EngineeringReport
from app.models.repository import RepositoryMetadata
from app.models.review import GeneratedTests, ReviewSuggestions
from app.services import analysis_service
VALID_URL = "https://github.com/owner/repo"
def _fake_report() -> EngineeringReport:
return EngineeringReport(
repository=RepositoryMetadata(
url=VALID_URL,
name="repo",
local_path="/tmp/fake",
language="Python",
frameworks=[],
architecture="",
entry_points=[],
summary="",
),
issues=IssueReport(),
review=ReviewSuggestions(summary="ok", overall_score=8.0),
tests=GeneratedTests(),
full_report="# Report",
generated_at="2026-06-15T00:00:00Z",
)
@pytest.mark.asyncio
async def test_run_full_analysis_returns_cached_report_on_hit() -> None:
cached = _fake_report()
with (
patch("app.services.analysis_service.resolve_remote_head_sha", return_value="abc123"),
patch("app.services.analysis_service.get_cached_report", return_value=cached),
patch("app.services.analysis_service._run_pipeline", new=AsyncMock()) as mock_pipeline,
):
result = await analysis_service.run_full_analysis(VALID_URL)
assert result.from_cache is True
assert result.full_report == "# Report"
mock_pipeline.assert_not_called()
@pytest.mark.asyncio
async def test_run_full_analysis_runs_pipeline_and_caches_on_miss() -> None:
fresh = _fake_report()
with (
patch("app.services.analysis_service.resolve_remote_head_sha", return_value="abc123"),
patch("app.services.analysis_service.get_cached_report", return_value=None),
patch("app.services.analysis_service._run_pipeline", new=AsyncMock(return_value=fresh)),
patch("app.services.analysis_service.set_cached_report") as mock_set,
):
result = await analysis_service.run_full_analysis(VALID_URL)
assert result.from_cache is False
assert result.repository.head_sha == "abc123"
mock_set.assert_called_once()
@pytest.mark.asyncio
async def test_run_full_analysis_skips_cache_when_sha_unresolvable() -> None:
fresh = _fake_report()
with (
patch("app.services.analysis_service.resolve_remote_head_sha", return_value=None),
patch("app.services.analysis_service.get_cached_report") as mock_get,
patch("app.services.analysis_service._run_pipeline", new=AsyncMock(return_value=fresh)),
patch("app.services.analysis_service.set_cached_report") as mock_set,
):
result = await analysis_service.run_full_analysis(VALID_URL)
assert result.full_report == "# Report"
mock_get.assert_not_called()
mock_set.assert_not_called() |