Final SSE test and CI workflow setup

Author: Vinisha-projects

.github/workflows/test.yml

@@ -1,30 +1,23 @@
 name: Run SSE Test
 
 on:
-  push:
-    branches:
-      - sse-test-final
   pull_request:
-    branches:
-      - main
 
 jobs:
   test:
     runs-on: ubuntu-latest
     steps:
-      - name: Checkout code
-        uses: actions/checkout@v3
+      - uses: actions/checkout@v3
 
       - name: Set up Python
         uses: actions/setup-python@v4
         with:
-          python-version: '3.10'
+          python-version: '3.11'
 
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
           pip install -r tests/requirements-dev.txt
 
-      - name: Run tests
-        run: |
-          pytest tests/test_sse_client_server_hardened.py
+      - name: Run test file
+        run: pytest tests/test_sse_client_server_hardened.py

tests/requirements-dev.txt

@@ -1,6 +1,5 @@
 fastapi
-uvicorn
 httpx
-httpx-sse
+uvicorn
 sse-starlette
-anyio
+anyio

tests/test_sse_client_server_hardened.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import AsyncGenerator, List
+from typing import AsyncGenerator
 
 from fastapi import FastAPI
 from starlette.responses import StreamingResponse
@@ -8,37 +8,36 @@
 import httpx
 from mcp.client.sse import aconnect_sse
 
+
 app = FastAPI()
 
+
 @app.get("/sse")
 async def sse_endpoint() -> StreamingResponse:
     async def event_stream() -> AsyncGenerator[str, None]:
         for i in range(3):
-            yield f"data: Hello {i + 1}\n\n"
+            yield f"data: Hello {i+1}\n\n"
             await asyncio.sleep(0.1)
     return StreamingResponse(event_stream(), media_type="text/event-stream")
 
+
 def run_mock_server() -> None:
     uvicorn.run(app, host="127.0.0.1", port=8012, log_level="warning")
 
-async def run_test() -> None:
+
+async def test_aconnect_sse_server_response() -> None:
     server_thread = Thread(target=run_mock_server, daemon=True)
     server_thread.start()
     await asyncio.sleep(1)
 
-    messages: List[str] = []
+    messages = []
 
     async with httpx.AsyncClient() as client:
         async with aconnect_sse(client, "GET", "http://127.0.0.1:8012/sse") as event_source:
             async for event in event_source.aiter_sse():
                 if event.data:
-                    print("Event received:", event.data)
                     messages.append(event.data)
                 if len(messages) == 3:
                     break
 
-    assert messages == ["Hello 1", "Hello 2", "Hello 3"]
-    print("\nTest passed! SSE connection via aconnect_sse worked correctly.")
-
-def test_aconnect_sse_server_response() -> None:
-    asyncio.run(run_test())
+    assert messages == ["Hello 1", "Hello 2", "Hello 3"]