1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
"""Shared helpers for Playwright integration tests against an in-process uvicorn server."""
from __future__ import annotations
import socket
import threading
import time
from collections.abc import Iterator
from contextlib import ExitStack, contextmanager
from pathlib import Path
from unittest.mock import patch
import uvicorn
def free_port() -> int:
with socket.socket() as sock:
sock.bind(("127.0.0.1", 0))
return int(sock.getsockname()[1])
@contextmanager
def running_server(
*,
tmp_path: Path,
projects: Path,
auth: bool = False,
auth_config_file: Path | None = None,
) -> Iterator[str]:
"""Serve ``pygitweb.main:app`` on a free localhost port in a daemon thread."""
from pygitweb.auth_config import init_auth_config
from pygitweb.config import settings
from pygitweb.main import app
if auth and auth_config_file is None:
raise ValueError("auth_config_file is required when auth is enabled")
port = free_port()
base = f"http://127.0.0.1:{port}/"
log_path = tmp_path / "smart_http.log"
with ExitStack() as stack:
stack.enter_context(patch.object(settings, "AUTH", auth))
stack.enter_context(patch.object(settings, "PROJECTROOT", str(projects)))
stack.enter_context(patch.object(settings, "PROJECTS_LIST", str(projects)))
stack.enter_context(patch("pygitweb.smart_http.smart_http_log_path", return_value=log_path))
if auth_config_file is not None:
stack.enter_context(patch.object(settings, "AUTH_CONFIG", str(auth_config_file)))
if auth:
cfg = init_auth_config(settings)
stack.enter_context(patch("pygitweb.auth_config.auth_config", cfg))
stack.enter_context(patch("pygitweb.auth.auth_config", cfg))
stack.enter_context(patch("pygitweb.main.auth_config", cfg))
stack.enter_context(patch("pygitweb.config.auth_config", cfg))
config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error", ws="none")
server = uvicorn.Server(config)
thread = threading.Thread(target=server.run, daemon=True)
thread.start()
deadline = time.monotonic() + 5.0
while not server.started and time.monotonic() < deadline:
time.sleep(0.01)
if not server.started:
server.should_exit = True
thread.join(timeout=5.0)
raise RuntimeError("integration server did not start")
try:
yield base
finally:
server.should_exit = True
thread.join(timeout=5.0)