242 lines
8.7 KiB
Python
242 lines
8.7 KiB
Python
"""Tests for AnalyticsMixin gRPC endpoints.
|
|
|
|
Tests cover:
|
|
- GetAnalyticsOverview: success, with filters (project, time range)
|
|
- ListSpeakerStats: success, with filters
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import TYPE_CHECKING
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from noteflow.domain.entities.analytics import (
|
|
AnalyticsOverview,
|
|
DailyMeetingStats,
|
|
SpeakerStat,
|
|
)
|
|
from noteflow.domain.identity import OperationContext, UserContext, WorkspaceContext
|
|
from noteflow.domain.identity.roles import WorkspaceRole
|
|
from noteflow.grpc.mixins._types import GrpcContext
|
|
from noteflow.grpc.mixins.analytics_mixin import AnalyticsMixin
|
|
from noteflow.grpc.proto import noteflow_pb2
|
|
|
|
WORKSPACE_ID = uuid4()
|
|
PROJECT_ID = uuid4()
|
|
EXPECTED_TOTAL_WORDS = 8500
|
|
"""Expected total word count for sample analytics overview."""
|
|
|
|
|
|
class MockServicerHost(AnalyticsMixin):
|
|
def __init__(self, analytics_service: MagicMock) -> None:
|
|
self.analytics_service = analytics_service
|
|
self._op_context = OperationContext(
|
|
user=UserContext(user_id=uuid4(), display_name="Test User"),
|
|
workspace=WorkspaceContext(
|
|
workspace_id=WORKSPACE_ID,
|
|
workspace_name="Test Workspace",
|
|
role=WorkspaceRole.OWNER,
|
|
),
|
|
)
|
|
|
|
def get_operation_context(self, _ctx: GrpcContext) -> OperationContext:
|
|
return self._op_context
|
|
|
|
if TYPE_CHECKING:
|
|
|
|
async def GetAnalyticsOverview(
|
|
self,
|
|
request: noteflow_pb2.GetAnalyticsOverviewRequest,
|
|
context: GrpcContext,
|
|
) -> noteflow_pb2.GetAnalyticsOverviewResponse: ...
|
|
|
|
async def ListSpeakerStats(
|
|
self,
|
|
request: noteflow_pb2.ListSpeakerStatsRequest,
|
|
context: GrpcContext,
|
|
) -> noteflow_pb2.ListSpeakerStatsResponse: ...
|
|
|
|
|
|
def create_sample_overview() -> AnalyticsOverview:
|
|
return AnalyticsOverview(
|
|
daily=[
|
|
DailyMeetingStats(
|
|
date="2026-01-20",
|
|
meetings=3,
|
|
total_duration=3600.0,
|
|
word_count=5000,
|
|
),
|
|
DailyMeetingStats(
|
|
date="2026-01-21",
|
|
meetings=2,
|
|
total_duration=2400.0,
|
|
word_count=3500,
|
|
),
|
|
],
|
|
total_meetings=5,
|
|
total_duration=6000.0,
|
|
total_words=EXPECTED_TOTAL_WORDS,
|
|
total_segments=150,
|
|
speaker_count=4,
|
|
)
|
|
|
|
|
|
def create_sample_speaker_stats() -> list[SpeakerStat]:
|
|
return [
|
|
SpeakerStat(
|
|
speaker_id="SPEAKER_00",
|
|
display_name="Alice",
|
|
total_time=1800.0,
|
|
segment_count=50,
|
|
meeting_count=3,
|
|
avg_confidence=0.92,
|
|
),
|
|
SpeakerStat(
|
|
speaker_id="SPEAKER_01",
|
|
display_name="Bob",
|
|
total_time=1200.0,
|
|
segment_count=35,
|
|
meeting_count=2,
|
|
avg_confidence=0.88,
|
|
),
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_analytics_service() -> MagicMock:
|
|
service = MagicMock()
|
|
service.get_overview = AsyncMock()
|
|
service.get_speaker_stats = AsyncMock()
|
|
return service
|
|
|
|
|
|
class TestGetAnalyticsOverview:
|
|
"""Tests for GetAnalyticsOverview endpoint."""
|
|
|
|
async def test_grpc_get_overview_success(
|
|
self,
|
|
mock_analytics_service: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""Test successful analytics overview retrieval."""
|
|
sample_overview = create_sample_overview()
|
|
mock_analytics_service.get_overview = AsyncMock(return_value=sample_overview)
|
|
servicer = MockServicerHost(mock_analytics_service)
|
|
request = noteflow_pb2.GetAnalyticsOverviewRequest()
|
|
|
|
response = await servicer.GetAnalyticsOverview(request, mock_grpc_context)
|
|
|
|
assert response.total_meetings == 5, "Total meetings should match"
|
|
assert response.total_words == EXPECTED_TOTAL_WORDS, "Total words should match"
|
|
assert len(response.daily) == 2, "Should have 2 daily entries"
|
|
assert response.speaker_count == 4, "Speaker count should match"
|
|
|
|
async def test_grpc_get_overview_with_project_filter(
|
|
self,
|
|
mock_analytics_service: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""Test overview with project ID filter."""
|
|
sample_overview = create_sample_overview()
|
|
mock_analytics_service.get_overview = AsyncMock(return_value=sample_overview)
|
|
servicer = MockServicerHost(mock_analytics_service)
|
|
request = noteflow_pb2.GetAnalyticsOverviewRequest(
|
|
project_id=str(PROJECT_ID),
|
|
)
|
|
|
|
await servicer.GetAnalyticsOverview(request, mock_grpc_context)
|
|
|
|
mock_analytics_service.get_overview.assert_called_once()
|
|
call_kwargs = mock_analytics_service.get_overview.call_args.kwargs
|
|
assert call_kwargs["project_ids"] is not None, "Project IDs should be passed"
|
|
assert PROJECT_ID in call_kwargs["project_ids"], "Project ID should be in filter"
|
|
|
|
async def test_grpc_get_overview_with_time_range(
|
|
self,
|
|
mock_analytics_service: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""Test overview with time range filter."""
|
|
sample_overview = create_sample_overview()
|
|
mock_analytics_service.get_overview = AsyncMock(return_value=sample_overview)
|
|
servicer = MockServicerHost(mock_analytics_service)
|
|
|
|
start_epoch = datetime(2026, 1, 1, tzinfo=UTC).timestamp()
|
|
end_epoch = datetime(2026, 1, 31, tzinfo=UTC).timestamp()
|
|
request = noteflow_pb2.GetAnalyticsOverviewRequest(
|
|
start_time=start_epoch,
|
|
end_time=end_epoch,
|
|
)
|
|
|
|
await servicer.GetAnalyticsOverview(request, mock_grpc_context)
|
|
|
|
mock_analytics_service.get_overview.assert_called_once()
|
|
call_kwargs = mock_analytics_service.get_overview.call_args.kwargs
|
|
assert call_kwargs["start_time"] is not None, "Start time should be passed"
|
|
assert call_kwargs["end_time"] is not None, "End time should be passed"
|
|
|
|
|
|
class TestListSpeakerStats:
|
|
"""Tests for ListSpeakerStats endpoint."""
|
|
|
|
async def test_list_speaker_stats_success(
|
|
self,
|
|
mock_analytics_service: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""Test successful speaker stats retrieval."""
|
|
sample_speaker_stats = create_sample_speaker_stats()
|
|
mock_analytics_service.get_speaker_stats = AsyncMock(return_value=sample_speaker_stats)
|
|
servicer = MockServicerHost(mock_analytics_service)
|
|
request = noteflow_pb2.ListSpeakerStatsRequest()
|
|
|
|
response = await servicer.ListSpeakerStats(request, mock_grpc_context)
|
|
|
|
assert len(response.speakers) == 2, "Should return 2 speakers"
|
|
assert response.speakers[0].display_name == "Alice", "First speaker should be Alice"
|
|
assert response.speakers[1].display_name == "Bob", "Second speaker should be Bob"
|
|
|
|
async def test_list_speaker_stats_with_filters(
|
|
self,
|
|
mock_analytics_service: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""Test speaker stats with project and time filters."""
|
|
sample_speaker_stats = create_sample_speaker_stats()
|
|
mock_analytics_service.get_speaker_stats = AsyncMock(return_value=sample_speaker_stats)
|
|
servicer = MockServicerHost(mock_analytics_service)
|
|
|
|
start_epoch = datetime(2026, 1, 15, tzinfo=UTC).timestamp()
|
|
end_epoch = datetime(2026, 1, 22, tzinfo=UTC).timestamp()
|
|
request = noteflow_pb2.ListSpeakerStatsRequest(
|
|
project_id=str(PROJECT_ID),
|
|
start_time=start_epoch,
|
|
end_time=end_epoch,
|
|
)
|
|
|
|
await servicer.ListSpeakerStats(request, mock_grpc_context)
|
|
|
|
mock_analytics_service.get_speaker_stats.assert_called_once()
|
|
call_kwargs = mock_analytics_service.get_speaker_stats.call_args.kwargs
|
|
assert call_kwargs["project_ids"] is not None, "Project IDs should be passed"
|
|
assert call_kwargs["start_time"] is not None, "Start time should be passed"
|
|
assert call_kwargs["end_time"] is not None, "End time should be passed"
|
|
|
|
async def test_list_speaker_stats_empty(
|
|
self,
|
|
mock_analytics_service: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""Test speaker stats when no speakers exist."""
|
|
mock_analytics_service.get_speaker_stats = AsyncMock(return_value=[])
|
|
servicer = MockServicerHost(mock_analytics_service)
|
|
request = noteflow_pb2.ListSpeakerStatsRequest()
|
|
|
|
response = await servicer.ListSpeakerStats(request, mock_grpc_context)
|
|
|
|
assert len(response.speakers) == 0, "Should return empty list"
|