bdlg2023-sms-back/app/tests/test_projects_stream.py

import json
from unittest.mock import AsyncMock, patch

import pytest
from httpx import AsyncClient

# Adjust this import to match where you instantiated your broadcast object!
from app.main import app
from app.models import OrgRole, Project
from app.tests.conftest import default_project_id

pytestmark = pytest.mark.asyncio


class TestProjectStream:
    @pytest.mark.parametrize(
        "role, expected_status",
        [
            (OrgRole.ORG_ADMIN, 200),
            (OrgRole.RESPO_BENEVOLE, 200),
            (OrgRole.RESPO_COMMISSION, 200),
            (None, 403),
        ],
    )
    async def test_role_access(
        self, client: AsyncClient, default_project: Project, make_org_user, role, expected_status
    ):
        """Test that only authorized roles can connect to the stream."""
        _, headers = make_org_user(role=role)
        url = app.url_path_for("project_stream", project_id=default_project_id)

        # We mock 'broadcast.subscribe' so it immediately returns an empty async iterator
        # This prevents the endpoint from hanging forever in the test
        with patch("app.core.events.broadcast.subscribe") as mock_subscribe:
            # Create an async generator that yields nothing and closes
            async def mock_generator():
                return
                yield

            mock_subscriber = AsyncMock()
            mock_subscriber.__aenter__.return_value = mock_generator()
            mock_subscribe.return_value = mock_subscriber

            async with client.stream(
                "GET", f"{url}?client_id=test-auth", headers=headers
            ) as response:
                assert response.status_code == expected_status

    async def test_receives_sse_events(
        self, client: AsyncClient, default_project: Project, make_org_user
    ):
        """Test that events from the broadcaster are correctly formatted as SSE."""
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        url = app.url_path_for("project_stream", project_id=default_project_id)

        # Create a fake Redis event
        class FakeEvent:
            def __init__(self):
                self.message = json.dumps(
                    {
                        "client_id": "someone-else",
                        "event": "slot_updated",
                        "item": {"id": "123", "title": "Installation"},
                    }
                )

        with patch("app.core.events.broadcast.subscribe") as mock_subscribe:
            # Make the mocked subscribe yield our FakeEvent, then stop
            async def mock_generator():
                yield FakeEvent()

            mock_subscriber = AsyncMock()
            mock_subscriber.__aenter__.return_value = mock_generator()
            mock_subscribe.return_value = mock_subscriber

            async with client.stream(
                "GET", f"{url}?client_id=listener-client", headers=headers
            ) as response:
                assert response.status_code == 200

                # Read the response
                lines = [line async for line in response.aiter_lines() if line.strip()]

                # Verify the formatting is standard SSE format
                assert lines[0] == "event: slot_updated"
                assert "Installation" in lines[1]