36
36
SERVER_NAME = "test_race_condition_server"
37
37
38
38
39
- def check_server_logs_for_errors (process , test_name : str ):
39
+ def check_server_logs_for_errors (process : subprocess . Popen [ str ] , test_name : str ) -> None :
40
40
"""
41
41
Check server logs for ClosedResourceError and other race condition errors.
42
42
@@ -52,7 +52,7 @@ def check_server_logs_for_errors(process, test_name: str):
52
52
server_logs = ""
53
53
54
54
# Check for specific race condition errors
55
- errors_found = []
55
+ errors_found : list [ str ] = []
56
56
57
57
if "ClosedResourceError" in server_logs :
58
58
errors_found .append ("ClosedResourceError" )
@@ -93,7 +93,7 @@ async def on_list_tools(self) -> list[Tool]:
93
93
return []
94
94
95
95
96
- def run_server_with_logging (port : int ):
96
+ def run_server_with_logging (port : int ) -> None :
97
97
"""Run the StreamableHTTP server with logging to capture race condition errors."""
98
98
app = RaceConditionTestServer ()
99
99
@@ -122,7 +122,7 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
122
122
uvicorn .run (starlette_app , host = "127.0.0.1" , port = port , log_level = "debug" )
123
123
124
124
125
- def start_server_process (port : int ):
125
+ def start_server_process (port : int ) -> subprocess . Popen [ str ] :
126
126
"""Start server in a separate process."""
127
127
# Create a temporary script to run the server
128
128
import os
0 commit comments