48 lines
1.3 KiB
Python
48 lines
1.3 KiB
Python
import socket
|
|
import sys
|
|
from collections.abc import Iterator
|
|
|
|
import pytest
|
|
from xprocess import ProcessStarter
|
|
from xprocess import XProcess
|
|
|
|
SERVER_HOST = "127.0.0.1"
|
|
|
|
|
|
@pytest.fixture
|
|
def fastapi_server_url(xprocess: XProcess, unused_tcp_port: int) -> Iterator[str]:
|
|
class ServerStarter(ProcessStarter):
|
|
timeout = 10
|
|
terminate_on_interrupt = True
|
|
|
|
@property
|
|
def args(self) -> list[str]:
|
|
return [
|
|
sys.executable,
|
|
"-m",
|
|
"uvicorn",
|
|
"personal_mcp.app:create_app",
|
|
"--factory",
|
|
"--host",
|
|
SERVER_HOST,
|
|
"--port",
|
|
str(unused_tcp_port),
|
|
"--log-level",
|
|
"warning",
|
|
"--no-access-log",
|
|
]
|
|
|
|
def startup_check(self) -> bool:
|
|
try:
|
|
with socket.create_connection((SERVER_HOST, unused_tcp_port), timeout=0.1):
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
process_name = f"personal-mcp-http-{unused_tcp_port}"
|
|
xprocess.ensure(process_name, ServerStarter)
|
|
try:
|
|
yield f"http://{SERVER_HOST}:{unused_tcp_port}/mcp"
|
|
finally:
|
|
xprocess.getinfo(process_name).terminate()
|