[refactor] test utils: simplify signatures and remove from ruff exclusion list. (#12155)

Apply some small method-signature cleanups in `tests/utils.py`,
and remove it from the `ruff format` exclusion list.
This commit is contained in:
James Addison 2024-03-21 10:08:35 +00:00 committed by GitHub
parent b7f708dc63
commit 078a80a42d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 12 additions and 13 deletions

View File

@ -471,5 +471,4 @@ exclude = [
"tests/test_util/**/*",
"tests/test_versioning.py",
"tests/test_writers/**/*",
"tests/utils.py",
]

View File

@ -1,6 +1,6 @@
from __future__ import annotations
__all__ = ("http_server",)
__all__ = ('http_server',)
from contextlib import contextmanager
from http.server import ThreadingHTTPServer
@ -14,22 +14,22 @@ import filelock
if TYPE_CHECKING:
from collections.abc import Iterator
from socketserver import BaseRequestHandler
from typing import Any, Final
from typing import Final
# Generated with:
# $ openssl req -new -x509 -days 3650 -nodes -out cert.pem \
# -keyout cert.pem -addext "subjectAltName = DNS:localhost"
TESTS_ROOT: Final[Path] = Path(__file__).parent
CERT_FILE: Final[str] = str(TESTS_ROOT / "certs" / "cert.pem")
CERT_FILE: Final[str] = str(TESTS_ROOT / 'certs' / 'cert.pem')
# File lock for tests
LOCK_PATH: Final[str] = str(TESTS_ROOT / 'test-server.lock')
class HttpServerThread(Thread):
def __init__(self, handler: type[BaseRequestHandler], /, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.server = ThreadingHTTPServer(("localhost", 7777), handler)
def __init__(self, handler: type[BaseRequestHandler]) -> None:
super().__init__(daemon=True)
self.server = ThreadingHTTPServer(('localhost', 7777), handler)
def run(self) -> None:
self.server.serve_forever(poll_interval=0.001)
@ -41,20 +41,20 @@ class HttpServerThread(Thread):
class HttpsServerThread(HttpServerThread):
def __init__(
self, handler: type[BaseRequestHandler], /, *args: Any, **kwargs: Any,
) -> None:
super().__init__(handler, *args, **kwargs)
def __init__(self, handler: type[BaseRequestHandler]) -> None:
super().__init__(handler)
sslcontext = SSLContext(PROTOCOL_TLS_SERVER)
sslcontext.load_cert_chain(CERT_FILE)
self.server.socket = sslcontext.wrap_socket(self.server.socket, server_side=True)
@contextmanager
def http_server(handler: type[BaseRequestHandler], *, tls_enabled: bool = False) -> Iterator[HttpServerThread]:
def http_server(
handler: type[BaseRequestHandler], *, tls_enabled: bool = False
) -> Iterator[HttpServerThread]:
server_cls = HttpsServerThread if tls_enabled else HttpServerThread
with filelock.FileLock(LOCK_PATH):
server = server_cls(handler, daemon=True)
server = server_cls(handler)
server.start()
try:
yield server