Extract and test server configuration

requirements.txt

@@ -6,7 +6,7 @@ fastapi==0.78.0
 gunicorn==20.1.0
 punq==0.6.2
 pydantic[email]==1.9.0
-uvicorn[standard]==0.17.6
+-e git+https://github.com/bwhmather/uvicorn.git@server-lifecycle#egg=uvicorn [standard]
 sqlalchemy[asyncio,mypy]==1.4.39
 
 # Debug

server/infrastructure/server.py

@@ -0,0 +1,66 @@
+from typing import Callable, Union
+
+import uvicorn
+import uvicorn.supervisors
+
+from server.config.di import resolve
+from server.config.settings import Settings
+
+
+def get_server_config(
+    app: Union[str, Callable], settings: Settings = None
+) -> uvicorn.Config:
+    if settings is None:
+        settings = resolve(Settings)
+
+    kwargs = dict(
+        host=settings.host,
+        port=settings.port,
+    )
+
+    if settings.server_mode == "local":
+        kwargs.update(
+            # Enable hot reload.
+            reload=True,
+            reload_dirs=["server"],
+        )
+    elif settings.server_mode == "live":
+        kwargs.update(
+            # Pass any proxy headers, so that Uvicorn sees information about the
+            # connecting client, rather than the connecting Nginx proxy.
+            # See: https://www.uvicorn.org/deployment/#running-behind-nginx
+            proxy_headers=True,
+            # Match Nginx mount path.
+            root_path="/api",
+        )
+
+    return uvicorn.Config(app, **kwargs)
+
+
+class Server(uvicorn.Server):
+    pass
+
+
+def run(app: Union[str, Callable]) -> int:
+    """
+    Run the API server.
+
+    This is a simplified version of `uvicorn.run()`.
+    """
+    config = get_server_config(app)
+    server = Server(config)
+
+    if config.should_reload:
+        sock = config.bind_socket()
+        reloader = uvicorn.supervisors.ChangeReload(
+            config, target=server.run, sockets=[sock]
+        )
+        reloader.run()
+        return 0
+
+    server.run()
+
+    if not server.started:
+        return 3
+
+    return 0

server/main.py

@@ -1,37 +1,12 @@
+import sys
+
 from .api.app import create_app
 from .config.di import bootstrap
+from .infrastructure.server import run
 
 bootstrap()
 
 app = create_app()
 
 if __name__ == "__main__":
-    import uvicorn
-
-    from .config.di import resolve
-    from .config.settings import Settings
-
-    settings = resolve(Settings)
-
-    kwargs: dict = {
-        "host": settings.host,
-        "port": settings.port,
-    }
-
-    if settings.server_mode == "local":
-        kwargs.update(
-            # Enable hot reload.
-            reload=True,
-            reload_dirs=["server"],
-        )
-    elif settings.server_mode == "live":
-        kwargs.update(
-            # Pass any proxy headers, so that Uvicorn sees information about the
-            # connecting client, rather than the connecting Nginx proxy.
-            # See: https://www.uvicorn.org/deployment/#running-behind-nginx
-            proxy_headers=True,
-            # Match Nginx mount path.
-            root_path="/api",
-        )
-
-    uvicorn.run("server.main:app", **kwargs)
+    sys.exit(run("server.main:app"))

tests/api/test_config.py

@@ -0,0 +1,25 @@
+from server.config import Settings
+from server.config.di import resolve
+from server.infrastructure.server import get_server_config
+
+
+def test_server_config_local() -> None:
+    settings = resolve(Settings).copy(update={"server_mode": "local"})
+
+    config = get_server_config("server.main:app", settings)
+
+    assert config.host == "localhost"
+    assert config.port == 3579
+    assert config.should_reload
+    assert config.root_path == ""
+
+
+def test_server_config_live() -> None:
+    settings = resolve(Settings).copy(update={"server_mode": "live"})
+
+    config = get_server_config("server.main:app", settings)
+
+    assert config.host == "localhost"
+    assert config.port == 3579
+    assert config.proxy_headers
+    assert config.root_path == "/api"

tests/conftest.py

@@ -1,26 +1,23 @@
 import asyncio
 import os
-from typing import TYPE_CHECKING, AsyncIterator, Iterator
+from typing import AsyncIterator, Iterator
 
 import httpx
 import pytest
 import pytest_asyncio
 from alembic import command
 from alembic.config import Config
-from asgi_lifespan import LifespanManager
 from sqlalchemy_utils import create_database, database_exists, drop_database
 
 from server.application.datasets.queries import GetAllDatasets
 from server.config import Settings
 from server.config.di import bootstrap, resolve
 from server.domain.auth.entities import UserRole
 from server.infrastructure.database import Database
+from server.infrastructure.server import Server, get_server_config
 from server.seedwork.application.messages import MessageBus
 
-from .helpers import TestUser, create_client, create_test_user
-
-if TYPE_CHECKING:
-    from server.api.app import App
+from .helpers import TestUser, create_test_user
 
 os.environ["APP_TESTING"] = "True"
 
@@ -71,19 +68,34 @@ def event_loop() -> Iterator[asyncio.AbstractEventLoop]:
         loop.close()
 
 
+class TestServer(Server):
+    @property
+    def url(self) -> str:
+        return f"http://{self.config.host}:{self.config.port}"
+
+    def install_signal_handlers(self) -> None:
+        pass
+
+
 @pytest_asyncio.fixture(scope="session")
-async def app() -> AsyncIterator["App"]:
+async def server() -> AsyncIterator[TestServer]:
     from server.api.app import create_app
 
     app = create_app()
+    config = get_server_config(app)
+    server = TestServer(config)
 
-    async with LifespanManager(app):
-        yield app
+    await server.start_serving()
+    try:
+        yield server
+    finally:
+        server.close()
+        await server.wait_closed()
 
 
 @pytest_asyncio.fixture(scope="session")
-async def client(app: "App") -> AsyncIterator[httpx.AsyncClient]:
-    async with create_client(app) as client:
+async def client(server: TestServer) -> AsyncIterator[httpx.AsyncClient]:
+    async with httpx.AsyncClient(base_url=server.url) as client:
         yield client
 
 

tests/helpers.py

@@ -1,5 +1,4 @@
 import json
-from typing import Callable
 
 import httpx
 from pydantic import BaseModel
@@ -12,10 +11,6 @@
 from .factories import CreateUserFactory
 
 
-def create_client(app: Callable) -> httpx.AsyncClient:
-    return httpx.AsyncClient(app=app, base_url="http://testserver")
-
-
 def to_payload(obj: BaseModel) -> dict:
     """
     Convert a Pydantic model instance to a JSON-serializable dictionary.

tests/test_debugging.py

@@ -7,8 +7,6 @@
 from server.infrastructure.database import Database
 from server.seedwork.application.di import Container
 
-from .helpers import create_client
-
 
 @pytest.mark.asyncio
 async def test_debug_default_disabled(app: App, client: httpx.AsyncClient) -> None:
@@ -36,6 +34,6 @@ async def test_debug_enabled(monkeypatch: pytest.MonkeyPatch) -> None:
     db = container.resolve(Database)
     assert db.engine.echo
 
-    async with create_client(app) as client:
+    async with httpx.AsyncClient(app=app, base_url="http://testserver") as client:
         response = await client.get("/_debug_toolbar")
         assert response.status_code != 404