tests: remove no covers
This commit is contained in:
@@ -187,6 +187,7 @@ class CertManager:
|
|||||||
if not self.exists(default_host):
|
if not self.exists(default_host):
|
||||||
self.logger.warning("Cannot create HTTPS context for %s", default_host)
|
self.logger.warning("Cannot create HTTPS context for %s", default_host)
|
||||||
return None
|
return None
|
||||||
|
self.logger.debug("Creating HTTP context...")
|
||||||
cert_file = self.get_cert(default_host)
|
cert_file = self.get_cert(default_host)
|
||||||
key_file = self.get_key(default_host)
|
key_file = self.get_key(default_host)
|
||||||
self.last_file_change = cert_file.stat().st_mtime
|
self.last_file_change = cert_file.stat().st_mtime
|
||||||
|
|||||||
@@ -97,17 +97,12 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
|||||||
content_type: str = "text/plain",
|
content_type: str = "text/plain",
|
||||||
code: int = http.HTTPStatus.OK,
|
code: int = http.HTTPStatus.OK,
|
||||||
message: str | None = None,
|
message: str | None = None,
|
||||||
headers: dict[str, str] | None = None,
|
|
||||||
) -> None:
|
) -> None:
|
||||||
if headers is None:
|
|
||||||
headers = {}
|
|
||||||
encoded: bytes = body.encode()
|
encoded: bytes = body.encode()
|
||||||
self.out_size = len(encoded)
|
self.out_size = len(encoded)
|
||||||
self.send_response(code, message)
|
self.send_response(code, message)
|
||||||
self.send_header("Content-Type", f"{content_type}; charset=UTF-8")
|
self.send_header("Content-Type", f"{content_type}; charset=UTF-8")
|
||||||
self.send_header("Content-Length", str(len(encoded)))
|
self.send_header("Content-Length", str(len(encoded)))
|
||||||
for header, value in headers.items():
|
|
||||||
self.send_header(header, value) # pragma: no cover
|
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
self.wfile.write(encoded)
|
self.wfile.write(encoded)
|
||||||
self.close_connection = True
|
self.close_connection = True
|
||||||
|
|||||||
+4
-4
@@ -55,7 +55,7 @@ class StaplerServer:
|
|||||||
self.data_dir.init()
|
self.data_dir.init()
|
||||||
self.token_manager.init()
|
self.token_manager.init()
|
||||||
|
|
||||||
def __request_handler( # pragma: no cover
|
def __request_handler(
|
||||||
self, *args: typing.Any
|
self, *args: typing.Any
|
||||||
) -> http.server.BaseHTTPRequestHandler:
|
) -> http.server.BaseHTTPRequestHandler:
|
||||||
return RequestHandler(
|
return RequestHandler(
|
||||||
@@ -96,7 +96,7 @@ class StaplerServer:
|
|||||||
)
|
)
|
||||||
return server, context is not None
|
return server, context is not None
|
||||||
|
|
||||||
def __upgrade_handler( # pragma: no cover
|
def __upgrade_handler(
|
||||||
self, *args: typing.Any
|
self, *args: typing.Any
|
||||||
) -> http.server.BaseHTTPRequestHandler:
|
) -> http.server.BaseHTTPRequestHandler:
|
||||||
return UpgradeHandler(
|
return UpgradeHandler(
|
||||||
@@ -120,14 +120,14 @@ class StaplerServer:
|
|||||||
threading.Thread(target=server.serve_forever, daemon=True).start()
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||||||
return server
|
return server
|
||||||
|
|
||||||
def __token_manager_background(self) -> None: # pragma: no cover
|
def __token_manager_background(self) -> None:
|
||||||
with contextlib.suppress(KeyboardInterrupt):
|
with contextlib.suppress(KeyboardInterrupt):
|
||||||
while True:
|
while True:
|
||||||
if self.token_manager.detect_file_change():
|
if self.token_manager.detect_file_change():
|
||||||
self.token_manager.init()
|
self.token_manager.init()
|
||||||
time.sleep(1)
|
time.sleep(1)
|
||||||
|
|
||||||
def __cert_manager_background(self) -> None: # pragma: no cover
|
def __cert_manager_background(self) -> None:
|
||||||
with contextlib.suppress(KeyboardInterrupt):
|
with contextlib.suppress(KeyboardInterrupt):
|
||||||
while True:
|
while True:
|
||||||
if (
|
if (
|
||||||
|
|||||||
Reference in New Issue
Block a user