| from datetime import UTC, datetime |
|
|
| from leaderboard_analytics.repositories import AnalyticsRepository |
| from leaderboard_analytics.schemas import QueryFilters |
|
|
|
|
| class CapturingCollection: |
| def __init__(self, rows: list[dict] | None = None) -> None: |
| self.rows = rows or [] |
| self.pipeline: list[dict] | None = None |
|
|
| def aggregate(self, pipeline: list[dict]): |
| self.pipeline = pipeline |
| return iter(self.rows) |
|
|
|
|
| def _filters() -> QueryFilters: |
| return QueryFilters( |
| start_time=datetime(2026, 1, 1, tzinfo=UTC), |
| end_time=datetime(2026, 1, 31, tzinfo=UTC), |
| ) |
|
|
|
|
| def test_funnel_pipeline_preserves_ordered_step_logic() -> None: |
| collection = CapturingCollection() |
| repository = AnalyticsRepository(collection) |
|
|
| repository.funnel(_filters()) |
|
|
| assert collection.pipeline is not None |
| assert {"$sort": {"session_id": 1, "event_ts": 1}} in collection.pipeline |
| assert any( |
| "$push" in stage.get("$group", {}).get("events", {}) for stage in collection.pipeline |
| ) |
| assert not any( |
| "$addToSet" in str(stage) and "events" in str(stage) for stage in collection.pipeline |
| ) |
| assert any( |
| "table_download_at" in str(stage) and "$filter_change_at" in str(stage) |
| for stage in collection.pipeline |
| ) |
|
|
|
|
| def test_new_vs_returning_pipeline_computes_first_seen_before_range_match() -> None: |
| collection = CapturingCollection() |
| repository = AnalyticsRepository(collection) |
|
|
| repository.visitors_new_vs_returning(_filters()) |
|
|
| assert collection.pipeline is not None |
| window_index = next( |
| i for i, stage in enumerate(collection.pipeline) if "$setWindowFields" in stage |
| ) |
| range_match_index = next( |
| i |
| for i, stage in enumerate(collection.pipeline) |
| if stage.get("$match", {}).get("event_ts") is not None |
| ) |
| assert window_index < range_match_index |
|
|
|
|
| def test_overview_totals_filters_empty_identifiers() -> None: |
| collection = CapturingCollection([{"pv": 1, "uv": 1, "sessions": 1, "events": 2}]) |
| repository = AnalyticsRepository(collection) |
|
|
| totals = repository.overview_totals(_filters()) |
|
|
| assert totals == {"pv": 1, "uv": 1, "sessions": 1, "events": 2} |
| assert collection.pipeline is not None |
| pipeline_text = str(collection.pipeline) |
| assert '"$sessions"' in pipeline_text or "'$sessions'" in pipeline_text |
| assert '"$visitors"' in pipeline_text or "'$visitors'" in pipeline_text |
| assert "$$s" in pipeline_text |
| assert "$$v" in pipeline_text |
|
|
