mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-18 03:29:11 +02:00
08d092be57* fix: skip TextIO log file test on Windows Avoids PytestUnraisableExceptionWarning caused by ProactorEventLoop cleanup timing issues with subprocess pipe transports. * Use WindowsSelectorEventLoopPolicy to fix Windows test warnings Re-add the SelectorEventLoop fix frombcd2e594that was inadvertently removed incf101c2a. This fixes ProactorEventLoop cleanup warnings on Windows CI without needing to skip individual tests.
83 lines
2.3 KiB
Python
83 lines
2.3 KiB
Python
import asyncio
|
|
import socket
|
|
import sys
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from fastmcp.utilities.tests import temporary_settings
|
|
|
|
# Use SelectorEventLoop on Windows to avoid ProactorEventLoop crashes
|
|
# See: https://github.com/python/cpython/issues/116773
|
|
if sys.platform == "win32":
|
|
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
|
|
|
|
|
|
def pytest_collection_modifyitems(items):
|
|
"""Automatically mark tests in integration_tests folder with 'integration' marker."""
|
|
for item in items:
|
|
# Check if the test is in the integration_tests folder
|
|
if "integration_tests" in str(item.fspath):
|
|
item.add_marker(pytest.mark.integration)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def import_rich_rule():
|
|
# What a hack
|
|
import rich.rule # noqa: F401
|
|
|
|
yield
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolate_settings_home(tmp_path: Path):
|
|
"""Ensure each test uses an isolated settings.home directory.
|
|
|
|
This prevents SQLite database locking issues on Windows when multiple
|
|
tests share the same DiskStore directory in settings.home / "oauth-proxy".
|
|
"""
|
|
test_home = tmp_path / "fastmcp-test-home"
|
|
test_home.mkdir(exist_ok=True)
|
|
|
|
with temporary_settings(home=test_home):
|
|
yield
|
|
|
|
|
|
def get_fn_name(fn: Callable[..., Any]) -> str:
|
|
return fn.__name__ # ty: ignore[unresolved-attribute]
|
|
|
|
|
|
@pytest.fixture
|
|
def worker_id(request):
|
|
"""Get the xdist worker ID, or 'master' if not using xdist."""
|
|
return getattr(request.config, "workerinput", {}).get("workerid", "master")
|
|
|
|
|
|
@pytest.fixture
|
|
def free_port():
|
|
"""Get a free port for the test to use."""
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
s.listen(1)
|
|
port = s.getsockname()[1]
|
|
return port
|
|
|
|
|
|
@pytest.fixture
|
|
def free_port_factory(worker_id):
|
|
"""Factory to get free ports that tracks used ports per test session."""
|
|
used_ports = set()
|
|
|
|
def get_port():
|
|
while True:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
s.listen(1)
|
|
port = s.getsockname()[1]
|
|
if port not in used_ports:
|
|
used_ports.add(port)
|
|
return port
|
|
|
|
return get_port
|