move to test_server.py

tests/test_config.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import configparser
 import io
 import json
@@ -13,12 +12,11 @@
 from typing import Any, Literal
 from unittest.mock import MagicMock
 
-import httpx
 import pytest
 import yaml
 from pytest_mock import MockerFixture
 
-from tests.utils import as_cwd, run_server
+from tests.utils import as_cwd
 from uvicorn._types import (
     ASGIApplication,
     ASGIReceiveCallable,
@@ -31,9 +29,6 @@
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
 from uvicorn.middleware.wsgi import WSGIMiddleware
 from uvicorn.protocols.http.h11_impl import H11Protocol
-from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
-
-pytestmark = pytest.mark.anyio
 
 
 @pytest.fixture
@@ -59,12 +54,6 @@ def wsgi_app(environ: Environ, start_response: StartResponse) -> None:
     pass  # pragma: nocover
 
 
-async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
-    assert scope["type"] == "http"
-    await send({"type": "http.response.start", "status": 200, "headers": []})
-    await send({"type": "http.response.body", "body": b"", "more_body": False})
-
-
 @pytest.mark.parametrize(
     "app, expected_should_reload",
     [(asgi_app, False), ("tests.test_config:asgi_app", True)],
@@ -556,16 +545,3 @@ def test_warn_when_using_reload_and_workers(caplog: pytest.LogCaptureFixture) ->
     Config(app=asgi_app, reload=True, workers=2)
     assert len(caplog.records) == 1
     assert '"workers" flag is ignored when reloading is enabled.' in caplog.records[0].message
-
-
-async def test_request_than_limit_max_requests_warn_log(
-    unused_tcp_port: int, http_protocol_cls: type[H11Protocol | HttpToolsProtocol], caplog: pytest.LogCaptureFixture
-):
-    caplog.set_level(logging.WARNING, logger="uvicorn.error")
-    config = Config(app=app, limit_max_requests=1, port=unused_tcp_port, http=http_protocol_cls)
-    async with run_server(config):
-        async with httpx.AsyncClient() as client:
-            tasks = [client.get(f"http://127.0.0.1:{unused_tcp_port}") for _ in range(2)]
-            responses = await asyncio.gather(*tasks)
-            assert len(responses) == 2
-    assert f"Maximum request limit of {config.limit_max_requests} exceeded. Terminating process." in caplog.text

tests/test_server.py

@@ -2,15 +2,23 @@
 
 import asyncio
 import contextlib
+import logging
 import signal
 import sys
 from typing import Callable, ContextManager, Generator
 
+import httpx
 import pytest
 
+from tests.utils import run_server
+from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope
 from uvicorn.config import Config
+from uvicorn.protocols.http.h11_impl import H11Protocol
+from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
 from uvicorn.server import Server
 
+pytestmark = pytest.mark.anyio
+
 
 # asyncio does NOT allow raising in signal handlers, so to detect
 # raised signals raised a mutable `witness` receives the signal
@@ -37,6 +45,12 @@ async def dummy_app(scope, receive, send):  # pragma: py-win32
     pass
 
 
+async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
+    assert scope["type"] == "http"
+    await send({"type": "http.response.start", "status": 200, "headers": []})
+    await send({"type": "http.response.body", "body": b"", "more_body": False})
+
+
 if sys.platform == "win32":  # pragma: py-not-win32
     signals = [signal.SIGBREAK]
     signal_captures = [capture_signal_sync]
@@ -45,7 +59,6 @@ async def dummy_app(scope, receive, send):  # pragma: py-win32
     signal_captures = [capture_signal_sync, capture_signal_async]
 
 
-@pytest.mark.anyio
 @pytest.mark.parametrize("exception_signal", signals)
 @pytest.mark.parametrize("capture_signal", signal_captures)
 async def test_server_interrupt(
@@ -65,3 +78,16 @@ async def interrupt_running(srv: Server):
     assert witness
     # set by the server's graceful exit handler
     assert server.should_exit
+
+
+async def test_request_than_limit_max_requests_warn_log(
+    unused_tcp_port: int, http_protocol_cls: type[H11Protocol | HttpToolsProtocol], caplog: pytest.LogCaptureFixture
+):
+    caplog.set_level(logging.WARNING, logger="uvicorn.error")
+    config = Config(app=app, limit_max_requests=1, port=unused_tcp_port, http=http_protocol_cls)
+    async with run_server(config):
+        async with httpx.AsyncClient() as client:
+            tasks = [client.get(f"http://127.0.0.1:{unused_tcp_port}") for _ in range(2)]
+            responses = await asyncio.gather(*tasks)
+            assert len(responses) == 2
+    assert f"Maximum request limit of {config.limit_max_requests} exceeded. Terminating process." in caplog.text