"""Tests for activity feed selector.""" import base64 from datetime import UTC, datetime, timedelta import pytest from django.contrib.auth.models import User from accounts.models import Workspace, WorkspaceMembership from messages.models import Thread from messages.selectors import get_activity_feed from sources.models import Source @pytest.fixture def user(db): """Create a test user.""" return User.objects.create_user( username="testuser", email="test@example.com", password="testpass123", ) @pytest.fixture def workspace(db, user): """Create a test workspace.""" workspace = Workspace.objects.create( name="Test Workspace", slug="test-workspace", ) WorkspaceMembership.objects.create( user=user, workspace=workspace, role="owner", ) return workspace @pytest.fixture def source(db, workspace): """Create a test source.""" return Source.objects.create( name="Test Source", kind="github_repo", external_id="test/repo", workspace=workspace, ) @pytest.fixture def source2(db, workspace): """Create a second test source.""" return Source.objects.create( name="Test Source 2", kind="github_repo", external_id="test/repo2", workspace=workspace, ) @pytest.fixture def threads(db, workspace, source): """Create test threads for the workspace.""" now = datetime.now(UTC) threads = [] for i in range(5): thread = Thread.objects.create( title=f"Thread {i + 1}", external_id=f"thread-{i + 1}", workspace=workspace, source=source, activity_at=now - timedelta(hours=i), ) threads.append(thread) return threads class TestGetActivityFeed: """Tests for get_activity_feed selector.""" def test_returns_empty_result_for_no_threads(self, workspace): """Should return empty items when no threads exist.""" result = get_activity_feed(workspace_id=str(workspace.id)) assert result["items"] == [] assert result["total_count"] == 0 assert result["page_info"]["has_next_page"] is False assert result["page_info"]["has_previous_page"] is False def test_returns_threads_ordered_by_activity_at(self, workspace, threads): """Should return threads ordered by activity_at descending.""" result = get_activity_feed(workspace_id=str(workspace.id)) assert len(result["items"]) == 5 assert result["items"][0].title == "Thread 1" assert result["items"][4].title == "Thread 5" def test_respects_first_parameter(self, workspace, threads): """Should limit results to first parameter.""" result = get_activity_feed(workspace_id=str(workspace.id), first=3) assert len(result["items"]) == 3 assert result["page_info"]["has_next_page"] is True assert result["total_count"] == 5 def test_pagination_with_cursor(self, workspace, threads): """Should paginate using cursor.""" first_page = get_activity_feed(workspace_id=str(workspace.id), first=2) assert len(first_page["items"]) == 2 assert first_page["page_info"]["has_next_page"] is True end_cursor = first_page["page_info"]["end_cursor"] second_page = get_activity_feed( workspace_id=str(workspace.id), first=2, after=end_cursor, ) assert len(second_page["items"]) == 2 assert second_page["page_info"]["has_previous_page"] is True # First page has Thread 1 & 2, second page should have Thread 3 & 4 assert second_page["items"][0].title == "Thread 3" assert second_page["items"][1].title == "Thread 4" def test_filters_by_source_id(self, workspace, source, source2): """Should filter threads by source_id.""" now = datetime.now(UTC) # Create threads for different sources Thread.objects.create( title="Source 1 Thread", external_id="s1-thread", workspace=workspace, source=source, activity_at=now, ) Thread.objects.create( title="Source 2 Thread", external_id="s2-thread", workspace=workspace, source=source2, activity_at=now, ) result = get_activity_feed( workspace_id=str(workspace.id), source_id=str(source.id), ) assert len(result["items"]) == 1 assert result["items"][0].title == "Source 1 Thread" assert result["total_count"] == 1 def test_handles_invalid_cursor(self, workspace, threads): """Should handle invalid cursor gracefully.""" result = get_activity_feed( workspace_id=str(workspace.id), first=3, after="invalid-cursor", ) # Should return results from the start assert len(result["items"]) == 3 assert result["items"][0].title == "Thread 1" def test_provides_correct_cursors(self, workspace, threads): """Should provide correct start and end cursors.""" result = get_activity_feed(workspace_id=str(workspace.id), first=3) # Verify cursors can be decoded start_cursor = result["page_info"]["start_cursor"] end_cursor = result["page_info"]["end_cursor"] assert start_cursor is not None assert end_cursor is not None # Cursors should be valid base64-encoded timestamps start_decoded = base64.b64decode(start_cursor).decode() end_decoded = base64.b64decode(end_cursor).decode() # Should be valid ISO format timestamps datetime.fromisoformat(start_decoded) datetime.fromisoformat(end_decoded) def test_returns_correct_total_count(self, workspace, threads): """Should return total count regardless of pagination.""" result = get_activity_feed(workspace_id=str(workspace.id), first=2) assert result["total_count"] == 5 assert len(result["items"]) == 2 def test_filters_by_workspace(self, user, source): """Should only return threads from the specified workspace.""" now = datetime.now(UTC) # Create another workspace with threads other_workspace = Workspace.objects.create( name="Other Workspace", slug="other-workspace", ) other_source = Source.objects.create( name="Other Source", kind="github_repo", external_id="other/repo", workspace=other_workspace, ) Thread.objects.create( title="First Workspace Thread", external_id="first-ws-thread", workspace=source.workspace, source=source, activity_at=now, ) Thread.objects.create( title="Other Workspace Thread", external_id="other-ws-thread", workspace=other_workspace, source=other_source, activity_at=now, ) result = get_activity_feed(workspace_id=str(source.workspace.id)) assert len(result["items"]) == 1 assert result["items"][0].title == "First Workspace Thread"