Spaces:
Sleeping
Sleeping
File size: 7,067 Bytes
b84ea83 | 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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 | """
Unit Tests — AnalyticsService
Tests:
- get_summary: correct counts, completion rate calculation
- get_visit_analytics: per-visit stats, correct rates
- get_visit_analytics: returns None for missing visit
- get_visit_trends: returns trend points, respects days param
"""
import pytest
from sqlalchemy.ext.asyncio import AsyncSession
from models.visit_model import Visit
from models.checklist_model import Checklist
from models.photo_model import Photo
from services.analytics_service import AnalyticsService
@pytest.mark.asyncio
class TestAnalyticsSummary:
"""Tests for AnalyticsService.get_summary()"""
async def test_empty_db_returns_zeros(self, async_session: AsyncSession):
svc = AnalyticsService()
result = await svc.get_summary(async_session)
data = result["data"]
assert data["visits"]["total"] == 0
assert data["checklists"]["total_items"] == 0
assert data["checklists"]["completion_rate_pct"] == 0.0
assert data["photos"]["total"] == 0
async def test_counts_visits_by_status(self, async_session: AsyncSession):
from datetime import datetime, timezone
# Insert 2 visits with different statuses
for status in ("pending", "completed"):
async_session.add(Visit(
title=f"Visit {status}",
visit_date=datetime(2026, 5, 1, 9, 0, tzinfo=timezone.utc),
status=status,
))
await async_session.flush()
svc = AnalyticsService()
result = await svc.get_summary(async_session)
stats = result["data"]["visits"]
assert stats["total"] == 2
assert stats["by_status"]["pending"] == 1
assert stats["by_status"]["completed"] == 1
assert stats["by_status"]["in_progress"] == 0
async def test_checklist_completion_rate(
self, async_session: AsyncSession, sample_visit: Visit
):
# Add 4 items, 3 checked → 75%
from datetime import datetime, timezone
for i, checked in enumerate([True, True, True, False]):
async_session.add(Checklist(
visit_id=sample_visit.id,
item_name=f"Item {i}",
is_checked=checked,
))
await async_session.flush()
svc = AnalyticsService()
result = await svc.get_summary(async_session)
cl = result["data"]["checklists"]
assert cl["checked_items"] == 3
assert cl["total_items"] >= 4
assert cl["completion_rate_pct"] > 0.0
async def test_response_structure(
self, async_session: AsyncSession
):
svc = AnalyticsService()
result = await svc.get_summary(async_session)
assert result["status"] == "success"
assert "data" in result
assert "meta" in result
assert "generated_at" in result["meta"]
@pytest.mark.asyncio
class TestVisitAnalytics:
"""Tests for AnalyticsService.get_visit_analytics()"""
async def test_returns_none_for_missing_visit(
self, async_session: AsyncSession
):
import uuid
svc = AnalyticsService()
result = await svc.get_visit_analytics(async_session, uuid.uuid4())
assert result is None
async def test_visit_with_no_checklists(
self, async_session: AsyncSession, sample_visit: Visit
):
svc = AnalyticsService()
result = await svc.get_visit_analytics(async_session, sample_visit.id)
assert result is not None
cl = result["data"]["checklists"]
assert cl["total_items"] == 0
assert cl["completion_rate_pct"] == 0.0
async def test_partial_checklist_completion(
self, async_session: AsyncSession, sample_visit: Visit
):
# 1 checked, 1 unchecked → 50%
for checked in (True, False):
async_session.add(Checklist(
visit_id=sample_visit.id,
item_name="Item",
is_checked=checked,
))
await async_session.flush()
svc = AnalyticsService()
result = await svc.get_visit_analytics(async_session, sample_visit.id)
cl = result["data"]["checklists"]
assert cl["checked_items"] == 1
assert cl["unchecked_items"] == 1
assert cl["completion_rate_pct"] == 50.0
async def test_photo_count(
self, async_session: AsyncSession, sample_visit: Visit
):
from datetime import datetime, timezone
for _ in range(3):
async_session.add(Photo(
visit_id=sample_visit.id,
file_path="/test.jpg",
file_name="test.jpg",
))
await async_session.flush()
svc = AnalyticsService()
result = await svc.get_visit_analytics(async_session, sample_visit.id)
assert result["data"]["photos"]["total"] == 3
async def test_response_contains_visit_info(
self, async_session: AsyncSession, sample_visit: Visit
):
svc = AnalyticsService()
result = await svc.get_visit_analytics(async_session, sample_visit.id)
data = result["data"]
assert data["visit_id"] == str(sample_visit.id)
assert data["title"] == sample_visit.title
assert data["status"] == sample_visit.status
@pytest.mark.asyncio
class TestVisitTrends:
"""Tests for AnalyticsService.get_visit_trends()
NOTE: date_trunc() is PostgreSQL-specific.
These tests are skipped when running against SQLite (unit test DB).
They run against the real PostgreSQL in integration tests.
"""
@pytest.mark.skip(reason="date_trunc is PostgreSQL-only; covered by integration tests")
async def test_empty_returns_empty_list(self, async_session: AsyncSession):
svc = AnalyticsService()
result = await svc.get_visit_trends(async_session, days=7)
assert result["status"] == "success"
assert isinstance(result["data"]["trends"], list)
@pytest.mark.skip(reason="date_trunc is PostgreSQL-only; covered by integration tests")
async def test_period_days_reflected_in_response(
self, async_session: AsyncSession
):
svc = AnalyticsService()
result = await svc.get_visit_trends(async_session, days=90)
assert result["data"]["period_days"] == 90
@pytest.mark.skip(reason="date_trunc is PostgreSQL-only; covered by integration tests")
async def test_recent_visits_appear_in_trends(
self, async_session: AsyncSession
):
from datetime import datetime, timezone
v = Visit(
title="Today's visit",
visit_date=datetime.now(timezone.utc),
status="pending",
)
v.created_at = datetime.now(timezone.utc)
async_session.add(v)
await async_session.flush()
svc = AnalyticsService()
result = await svc.get_visit_trends(async_session, days=7)
trends = result["data"]["trends"]
assert len(trends) >= 1
total_count = sum(p["count"] for p in trends)
assert total_count >= 1
|