1343 lines
48 KiB
Python
1343 lines
48 KiB
Python
import abc
|
|
import collections
|
|
import contextlib
|
|
import http
|
|
import http.server
|
|
import io
|
|
import logging
|
|
import tarfile
|
|
import typing
|
|
import unittest.mock
|
|
|
|
import requests
|
|
|
|
from stapler.handlers import BaseHandler, RequestHandler, UpgradeHandler
|
|
from stapler.page import Page
|
|
from stapler.params import Parameters
|
|
|
|
from . import BaseTestCase
|
|
|
|
|
|
class BaseHandlerTestCase(BaseTestCase, abc.ABC):
|
|
@abc.abstractmethod
|
|
def _get_handler(
|
|
self,
|
|
path: str = "/",
|
|
headers: dict[str, str | None] | None = None,
|
|
method: str = "GET",
|
|
rfile: io.BufferedIOBase | None = None,
|
|
) -> BaseHandler:
|
|
pass
|
|
|
|
@contextlib.contextmanager
|
|
def expects_status_only(
|
|
self,
|
|
handler: BaseHandler,
|
|
code: int,
|
|
message: str | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
) -> typing.Iterator[None]:
|
|
if headers is None:
|
|
headers = {}
|
|
send_response_mock = handler.send_response = unittest.mock.Mock() # ty:ignore[invalid-assignment]
|
|
send_header_mock = handler.send_header = unittest.mock.Mock() # ty:ignore[invalid-assignment]
|
|
end_headers_mock = handler.end_headers = unittest.mock.Mock() # ty:ignore[invalid-assignment]
|
|
yield
|
|
send_response_mock.assert_called_once_with(code, message)
|
|
send_header_mock.assert_has_calls(
|
|
[
|
|
unittest.mock.call("Content-Length", "0"),
|
|
]
|
|
+ [unittest.mock.call(header, value) for header, value in headers.items()],
|
|
any_order=True,
|
|
)
|
|
end_headers_mock.assert_called_once()
|
|
|
|
@contextlib.contextmanager
|
|
def expects_basic_body( # noqa: PLR0913
|
|
self,
|
|
handler: BaseHandler,
|
|
body: str,
|
|
content_type: str = "text/plain",
|
|
code: int = http.HTTPStatus.OK,
|
|
message: str | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
) -> typing.Iterator[None]:
|
|
if headers is None:
|
|
headers = {}
|
|
send_response_mock = handler.send_response = unittest.mock.Mock() # ty:ignore[invalid-assignment]
|
|
send_header_mock = handler.send_header = unittest.mock.Mock() # ty:ignore[invalid-assignment]
|
|
end_headers_mock = handler.end_headers = unittest.mock.Mock() # ty:ignore[invalid-assignment]
|
|
yield
|
|
send_response_mock.assert_called_once_with(code, message)
|
|
send_header_mock.assert_has_calls(
|
|
[
|
|
unittest.mock.call("Content-Length", str(len(body.encode()))),
|
|
unittest.mock.call("Content-Type", f"{content_type}; charset=UTF-8"),
|
|
]
|
|
+ [unittest.mock.call(header, value) for header, value in headers.items()],
|
|
any_order=True,
|
|
)
|
|
end_headers_mock.assert_called_once()
|
|
handler.wfile.seek(0)
|
|
self.assertEqual(handler.wfile.read(), body.encode())
|
|
|
|
@contextlib.contextmanager
|
|
def expects_error(
|
|
self,
|
|
handler: BaseHandler,
|
|
code: int,
|
|
message: str | None = None,
|
|
) -> typing.Iterator[None]:
|
|
shortmsg, _ = RequestHandler.responses[code]
|
|
if message is None:
|
|
message = shortmsg
|
|
with self.expects_status_only(handler, code, message):
|
|
yield
|
|
|
|
@contextlib.contextmanager
|
|
def expects_error_full(
|
|
self,
|
|
handler: BaseHandler,
|
|
code: int,
|
|
message: str | None = None,
|
|
explain: str | None = None,
|
|
) -> typing.Iterator[None]:
|
|
shortmsg, longmsg = http.server.BaseHTTPRequestHandler.responses[code]
|
|
if message is None:
|
|
message = shortmsg
|
|
if explain is None:
|
|
explain = longmsg
|
|
with self.expects_basic_body(
|
|
handler,
|
|
body=f"{code} {message}\n{explain}\n\n{handler.server_signature()}",
|
|
code=code,
|
|
message=message,
|
|
):
|
|
yield
|
|
|
|
|
|
class TestRequestHandler(BaseHandlerTestCase):
|
|
@typing.override
|
|
def setUp(self) -> None:
|
|
self.get_tmp_dir()
|
|
self.registry = self.new_mock()
|
|
self.token_manager = self.new_mock()
|
|
self.certbot_www = self.tmp_path / "certbot_www"
|
|
self.data_dir = self.new_mock()
|
|
super().setUp()
|
|
|
|
def _get_handler(
|
|
self,
|
|
path: str = "/",
|
|
headers: dict[str, str | None] | None = None,
|
|
method: str = "GET",
|
|
rfile: io.BufferedIOBase | None = None,
|
|
) -> RequestHandler:
|
|
if headers is None:
|
|
headers = {}
|
|
if "Accept" not in headers:
|
|
headers["Accept"] = "nothing"
|
|
with self.patch("http.server.BaseHTTPRequestHandler.__init__"):
|
|
handler = RequestHandler(
|
|
unittest.mock.MagicMock(),
|
|
"127.0.0.1",
|
|
unittest.mock.MagicMock(),
|
|
params=Parameters(
|
|
data_dir=self.get_tmp_dir(), certbot_www=str(self.certbot_www)
|
|
),
|
|
registry=self.registry,
|
|
token_manager=self.token_manager,
|
|
)
|
|
handler.address_string = lambda: "127.0.0.1" # ty:ignore[invalid-assignment]
|
|
handler.requestline = f"{method} {path}"
|
|
handler.path = path
|
|
handler.command = method
|
|
handler.request_version = "HTTP/0.9"
|
|
handler.client_address = ("127.0.0.1", 12345)
|
|
handler.headers = collections.defaultdict(lambda: None, headers) # ty:ignore[invalid-assignment]
|
|
handler.rfile = rfile if rfile is not None else io.BytesIO()
|
|
handler.wfile = io.BytesIO()
|
|
handler.logger = unittest.mock.Mock(logging.Logger)
|
|
handler.data_dir = self.data_dir
|
|
return handler
|
|
|
|
def test_handle_errors_silently(self) -> None:
|
|
with self.patch("http.server.BaseHTTPRequestHandler.__init__") as mock:
|
|
mock.side_effect = Exception
|
|
logging.basicConfig(level=logging.CRITICAL)
|
|
RequestHandler(
|
|
unittest.mock.MagicMock(),
|
|
"127.0.0.1",
|
|
unittest.mock.MagicMock(),
|
|
params=Parameters(
|
|
data_dir=self.get_tmp_dir(), certbot_www=str(self.certbot_www)
|
|
),
|
|
registry=self.registry,
|
|
token_manager=self.token_manager,
|
|
)
|
|
|
|
def test_handle_disconnect_silently(self) -> None:
|
|
with self.patch("http.server.BaseHTTPRequestHandler.__init__") as mock:
|
|
mock.side_effect = BrokenPipeError
|
|
logging.basicConfig(level=logging.CRITICAL)
|
|
RequestHandler(
|
|
unittest.mock.MagicMock(),
|
|
"127.0.0.1",
|
|
unittest.mock.MagicMock(),
|
|
params=Parameters(
|
|
data_dir=self.get_tmp_dir(), certbot_www=str(self.certbot_www)
|
|
),
|
|
registry=self.registry,
|
|
token_manager=self.token_manager,
|
|
)
|
|
|
|
def test_do_head_forward(self) -> None:
|
|
handler = self._get_handler()
|
|
with (
|
|
self.patch("http.server.SimpleHTTPRequestHandler.do_HEAD"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_HEAD()
|
|
|
|
def test_do_get_index(self) -> None:
|
|
handler = self._get_handler()
|
|
with (
|
|
self.expects_basic_body(handler, handler.server_signature()),
|
|
self.patch("http.server.SimpleHTTPRequestHandler.do_GET", count=0),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_GET()
|
|
|
|
def test_do_get_forward_on_other_path(self) -> None:
|
|
handler = self._get_handler("/file")
|
|
with (
|
|
self.mock_call(self.registry.get_from_path, ["file"], Page("file")),
|
|
self.patch("http.server.SimpleHTTPRequestHandler.do_GET"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_GET()
|
|
|
|
def test_do_get_forward_on_other_host(self) -> None:
|
|
handler = self._get_handler("/", {"Host": "other_host"})
|
|
with (
|
|
self.mock_call(
|
|
self.registry.get_from_host,
|
|
["other_host"],
|
|
),
|
|
self.patch("http.server.SimpleHTTPRequestHandler.do_GET"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_GET()
|
|
|
|
def test_do_put_no_token(self) -> None:
|
|
handler = self._get_handler("/path")
|
|
with (
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.BAD_REQUEST, "No X-Token header in request"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_post_is_do_put(self) -> None:
|
|
handler = self._get_handler("/path")
|
|
with (
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.BAD_REQUEST, "No X-Token header in request"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_POST()
|
|
|
|
def test_do_patch_is_do_put(self) -> None:
|
|
handler = self._get_handler("/path")
|
|
with (
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.BAD_REQUEST, "No X-Token header in request"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PATCH()
|
|
|
|
def test_do_put_invalid_token(self) -> None:
|
|
handler = self._get_handler("/path", {"X-Token": "secret"})
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], False), # noqa: FBT003
|
|
self.expects_error(handler, http.HTTPStatus.UNAUTHORIZED, "Invalid token"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_invalid_path(self) -> None:
|
|
handler = self._get_handler("/pa.th", {"X-Token": "secret"})
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.expects_error(handler, http.HTTPStatus.BAD_REQUEST, "Invalid path"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_invalid_token_for_path(self) -> None:
|
|
handler = self._get_handler("/path", {"X-Token": "secret"})
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
False, # noqa: FBT003
|
|
),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.FORBIDDEN, "Path forbidden for this token"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_invalid_host(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path", {"X-Token": "secret", "X-Host": "invalid_host"}
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.BAD_REQUEST, "Invalid requested host"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_invalid_host_only(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path", {"X-Token": "secret", "X-Host-Only": "invalid_host"}
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.BAD_REQUEST, "Invalid requested host"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_invalid_host_and_host_only(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path", {"X-Token": "secret", "X-Host": "host", "X-Host-Only": "host"}
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.expects_error(
|
|
handler,
|
|
http.HTTPStatus.BAD_REQUEST,
|
|
"Cannot use X-Host-Only with X-Host",
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_invalid_host_for_path(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path", {"X-Token": "secret", "X-Host": "example.com"}
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call(
|
|
self.registry.get_from_host, ["example.com"], Page("other_path")
|
|
),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.FORBIDDEN, "Host already taken"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_extract_no_content(self) -> None:
|
|
handler = self._get_handler("/path", {"X-Token": "secret"})
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.LENGTH_REQUIRED, "No body found"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_extract_content_too_large(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path", {"X-Token": "secret", "Content-Length": "999999999"}
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.expects_error(
|
|
handler,
|
|
http.HTTPStatus.CONTENT_TOO_LARGE,
|
|
"Archive too large",
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_extract_tar_error(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path", {"X-Token": "secret", "Content-Length": "1"}
|
|
)
|
|
handler.rfile.write(b"\0")
|
|
self.data_dir.extract_tar_bytes.side_effect = tarfile.TarError
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.BAD_REQUEST, "Invalid tar archive"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
self.data_dir.extract_tar_bytes.assert_called_once()
|
|
|
|
def test_do_put_extract_other_error(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path", {"X-Token": "secret", "Content-Length": "1"}
|
|
)
|
|
handler.rfile.write(b"\0")
|
|
self.data_dir.extract_tar_bytes.side_effect = Exception
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.expects_error(handler, http.HTTPStatus.INTERNAL_SERVER_ERROR, ""),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
self.data_dir.extract_tar_bytes.assert_called_once()
|
|
|
|
def test_do_put_extract_ok(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path", {"X-Token": "secret", "Content-Length": "1"}
|
|
)
|
|
handler.rfile.write(b"\0")
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call_unchecked(self.data_dir.extract_tar_bytes),
|
|
self.mock_call(self.registry.add, ["path"]),
|
|
self.mock_call(self.token_manager.set_token, ["path", "secret"]),
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_status_only(
|
|
handler, http.HTTPStatus.CREATED, "Resource updated"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_extract_with_host(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
{"X-Token": "secret", "Content-Length": "1", "X-Host": "example.com"},
|
|
)
|
|
handler.rfile.write(b"\0")
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call(self.registry.get_from_host, ["example.com"], Page("path")),
|
|
self.mock_call_unchecked(self.data_dir.extract_tar_bytes),
|
|
self.mock_call(self.registry.add, ["path"]),
|
|
self.mock_call(self.token_manager.set_token, ["path", "secret"]),
|
|
self.mock_call(self.registry.set_host, ["path", "example.com"]),
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_status_only(
|
|
handler, http.HTTPStatus.CREATED, "Resource updated"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_extract_with_spa(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path", {"X-Token": "secret", "X-SPA": "index.html", "Content-Length": "1"}
|
|
)
|
|
handler.rfile.write(b"\0")
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call_unchecked(self.data_dir.extract_tar_bytes),
|
|
self.mock_call(self.registry.add, ["path"]),
|
|
self.mock_call(self.token_manager.set_token, ["path", "secret"]),
|
|
self.mock_call(self.registry.set_spa, ["path", "index.html"]),
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_status_only(
|
|
handler, http.HTTPStatus.CREATED, "Resource updated"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_redirect_with_content(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
{
|
|
"X-Token": "secret",
|
|
"X-Redirect": "https://example.com",
|
|
"Content-Length": "1",
|
|
},
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.expects_error(
|
|
handler,
|
|
http.HTTPStatus.BAD_REQUEST,
|
|
"No content must be sent with X-Redirect",
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_redirect_ok(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
{
|
|
"X-Token": "secret",
|
|
"X-Redirect": "https://example.com",
|
|
},
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call(self.registry.set_redirect, ["path", "https://example.com"]),
|
|
self.mock_call(self.token_manager.set_token, ["path", "secret"]),
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_status_only(
|
|
handler, http.HTTPStatus.CREATED, "Resource updated"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_redirect_with_host(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
{
|
|
"X-Token": "secret",
|
|
"X-Redirect": "https://example.com",
|
|
"X-Host": "example.com",
|
|
},
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call(self.registry.get_from_host, ["example.com"], Page("path")),
|
|
self.mock_call(self.registry.set_redirect, ["path", "https://example.com"]),
|
|
self.mock_call(self.token_manager.set_token, ["path", "secret"]),
|
|
self.mock_call(self.registry.set_host, ["path", "example.com"]),
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_status_only(
|
|
handler, http.HTTPStatus.CREATED, "Resource updated"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_redirect_with_host_only(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
{
|
|
"X-Token": "secret",
|
|
"X-Redirect": "https://example.com",
|
|
"X-Host-Only": "example.com",
|
|
},
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call(self.registry.get_from_host, ["example.com"], Page("path")),
|
|
self.mock_call(self.registry.set_redirect, ["path", "https://example.com"]),
|
|
self.mock_call(self.token_manager.set_token, ["path", "secret"]),
|
|
self.mock_call(self.registry.set_host_only, ["path", "example.com"]),
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_status_only(
|
|
handler, http.HTTPStatus.CREATED, "Resource updated"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_proxy_with_content(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
{
|
|
"X-Token": "secret",
|
|
"X-Proxy": "https://example.com",
|
|
"Content-Length": "1",
|
|
},
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.expects_error(
|
|
handler,
|
|
http.HTTPStatus.BAD_REQUEST,
|
|
"No content must be sent with X-Proxy",
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_proxy_ok(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
{
|
|
"X-Token": "secret",
|
|
"X-Proxy": "https://example.com",
|
|
},
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call(self.registry.set_proxy, ["path", "https://example.com"]),
|
|
self.mock_call(self.token_manager.set_token, ["path", "secret"]),
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_status_only(
|
|
handler, http.HTTPStatus.CREATED, "Resource updated"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_proxy_with_host(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
{
|
|
"X-Token": "secret",
|
|
"X-Proxy": "https://example.com",
|
|
"X-Host": "example.com",
|
|
},
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call(self.registry.get_from_host, ["example.com"], Page("path")),
|
|
self.mock_call(self.registry.set_proxy, ["path", "https://example.com"]),
|
|
self.mock_call(self.token_manager.set_token, ["path", "secret"]),
|
|
self.mock_call(self.registry.set_host, ["path", "example.com"]),
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_status_only(
|
|
handler, http.HTTPStatus.CREATED, "Resource updated"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_put_proxy_and_redirect(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
{
|
|
"X-Token": "secret",
|
|
"X-Proxy": "https://example.com",
|
|
"X-Redirect": "https://example.com",
|
|
"X-Host": "example.com",
|
|
},
|
|
)
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.expects_status_only(
|
|
handler,
|
|
http.HTTPStatus.BAD_REQUEST,
|
|
"Cannot use X-Proxy with X-Redirect",
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_PUT()
|
|
|
|
def test_do_delete_no_token(self) -> None:
|
|
handler = self._get_handler("/path")
|
|
with (
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.BAD_REQUEST, "No X-Token header in request"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_DELETE()
|
|
|
|
def test_do_delete_invalid_token(self) -> None:
|
|
handler = self._get_handler("/path", {"X-Token": "secret"})
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], False), # noqa: FBT003
|
|
self.expects_error(handler, http.HTTPStatus.UNAUTHORIZED, "Invalid token"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_DELETE()
|
|
|
|
def test_do_delete_invalid_path(self) -> None:
|
|
handler = self._get_handler("/pa.th", {"X-Token": "secret"})
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.expects_error(handler, http.HTTPStatus.BAD_REQUEST, "Invalid path"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_DELETE()
|
|
|
|
def test_do_delete_invalid_token_for_path(self) -> None:
|
|
handler = self._get_handler("/path", {"X-Token": "secret"})
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
False, # noqa: FBT003
|
|
),
|
|
self.expects_error(
|
|
handler, http.HTTPStatus.FORBIDDEN, "Path forbidden for this token"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_DELETE()
|
|
|
|
def test_do_delete_not_found(self) -> None:
|
|
handler = self._get_handler("/path", {"X-Token": "secret"})
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call(self.data_dir.exists, ["path"], False), # noqa: FBT003
|
|
self.expects_error(handler, http.HTTPStatus.NOT_FOUND, "Not found"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_DELETE()
|
|
|
|
def test_do_delete_remove_error(self) -> None:
|
|
handler = self._get_handler("/path", {"X-Token": "secret"})
|
|
self.data_dir.remove.side_effect = Exception
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call(self.data_dir.exists, ["path"], True), # noqa: FBT003
|
|
self.mock_call(self.data_dir.exists, ["path"], True), # noqa: FBT003
|
|
self.expects_error(handler, http.HTTPStatus.INTERNAL_SERVER_ERROR, ""),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_DELETE()
|
|
self.data_dir.remove.assert_called_once_with("path")
|
|
|
|
def test_do_delete_ok(self) -> None:
|
|
handler = self._get_handler("/path", {"X-Token": "secret"})
|
|
with (
|
|
self.mock_call(self.token_manager.is_valid, ["secret"], True), # noqa: FBT003
|
|
self.mock_call(
|
|
self.token_manager.is_valid_for_path,
|
|
["secret", "path"],
|
|
True, # noqa: FBT003
|
|
),
|
|
self.mock_call(self.data_dir.exists, ["path"], True), # noqa: FBT003
|
|
self.mock_call(self.data_dir.remove, ["path"]),
|
|
self.mock_call(self.registry.remove, ["path"]),
|
|
self.expects_error(handler, http.HTTPStatus.OK, "Resource /path/ removed"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_DELETE()
|
|
|
|
def test_do_post_proxy_no_body(self) -> None:
|
|
handler = self._get_handler("/path", method="POST")
|
|
response = requests.Response()
|
|
response.status_code = 200
|
|
response.reason = "OK"
|
|
response.raw = io.BytesIO()
|
|
with (
|
|
self.mock_call(
|
|
self.registry.get_from_path,
|
|
["path"],
|
|
Page("path", proxy="https://example.com"),
|
|
),
|
|
self.patch_call(
|
|
"requests.request",
|
|
[
|
|
"POST",
|
|
"https://example.com",
|
|
],
|
|
response,
|
|
{
|
|
"data": None,
|
|
"headers": {
|
|
"Host": "example.com",
|
|
"Accept": "nothing",
|
|
"X-Real-IP": "127.0.0.1",
|
|
"X-Forwarded-Host": "localhost",
|
|
"X-Forwarded-For": "127.0.0.1",
|
|
"X-Forwarded-Proto": "https",
|
|
},
|
|
"allow_redirects": False,
|
|
"timeout": 480,
|
|
},
|
|
),
|
|
self.expects_status_only(handler, 200, "OK"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_POST()
|
|
|
|
def test_do_post_proxy_with_request_body(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
method="POST",
|
|
headers={"Content-Length": "5"},
|
|
rfile=io.BytesIO(b"hello"),
|
|
)
|
|
response = requests.Response()
|
|
response.status_code = 200
|
|
response.reason = "OK"
|
|
response.raw = io.BytesIO()
|
|
with (
|
|
self.mock_call(
|
|
self.registry.get_from_path,
|
|
["path"],
|
|
Page("path", proxy="https://example.com"),
|
|
),
|
|
self.patch_call(
|
|
"requests.request",
|
|
[
|
|
"POST",
|
|
"https://example.com",
|
|
],
|
|
response,
|
|
{
|
|
"data": b"hello",
|
|
"headers": {
|
|
"Host": "example.com",
|
|
"Accept": "nothing",
|
|
"X-Real-IP": "127.0.0.1",
|
|
"X-Forwarded-Host": "localhost",
|
|
"X-Forwarded-For": "127.0.0.1",
|
|
"X-Forwarded-Proto": "https",
|
|
"Content-Length": "5",
|
|
},
|
|
"allow_redirects": False,
|
|
"timeout": 480,
|
|
},
|
|
),
|
|
self.expects_status_only(handler, 200, "OK"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_POST()
|
|
|
|
def test_do_post_proxy_with_response_body(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path",
|
|
method="POST",
|
|
)
|
|
response = requests.Response()
|
|
response.status_code = 200
|
|
response.reason = "OK"
|
|
response.headers["Content-Type"] = "text/plain; charset=UTF-8"
|
|
response.raw = io.BytesIO(b"hello")
|
|
with (
|
|
self.mock_call(
|
|
self.registry.get_from_path,
|
|
["path"],
|
|
Page("path", proxy="https://example.com"),
|
|
),
|
|
self.patch_call(
|
|
"requests.request",
|
|
[
|
|
"POST",
|
|
"https://example.com",
|
|
],
|
|
response,
|
|
{
|
|
"data": None,
|
|
"headers": {
|
|
"Host": "example.com",
|
|
"Accept": "nothing",
|
|
"X-Real-IP": "127.0.0.1",
|
|
"X-Forwarded-Host": "localhost",
|
|
"X-Forwarded-For": "127.0.0.1",
|
|
"X-Forwarded-Proto": "https",
|
|
},
|
|
"allow_redirects": False,
|
|
"timeout": 480,
|
|
},
|
|
),
|
|
self.expects_basic_body(handler, "hello", message="OK"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_POST()
|
|
|
|
def test_do_post_proxy_fail(self) -> None:
|
|
handler = self._get_handler("/path", method="POST")
|
|
with (
|
|
self.mock_call(
|
|
self.registry.get_from_path,
|
|
["path"],
|
|
Page("path", proxy="https://example.com"),
|
|
),
|
|
self.patch_call(
|
|
"requests.request",
|
|
[
|
|
"POST",
|
|
"https://example.com",
|
|
],
|
|
None,
|
|
{
|
|
"data": None,
|
|
"headers": {
|
|
"Host": "example.com",
|
|
"Accept": "nothing",
|
|
"X-Real-IP": "127.0.0.1",
|
|
"X-Forwarded-Host": "localhost",
|
|
"X-Forwarded-For": "127.0.0.1",
|
|
"X-Forwarded-Proto": "https",
|
|
},
|
|
"allow_redirects": False,
|
|
"timeout": 480,
|
|
},
|
|
) as request_mock,
|
|
self.expects_status_only(
|
|
handler,
|
|
http.HTTPStatus.BAD_GATEWAY,
|
|
"Could not reach https://example.com",
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
request_mock.side_effect = Exception
|
|
handler.do_POST()
|
|
|
|
def test_do_post_proxy_sub_path(self) -> None:
|
|
handler = self._get_handler("/path/index.html", method="POST")
|
|
response = requests.Response()
|
|
response.status_code = 200
|
|
response.reason = "OK"
|
|
response.raw = io.BytesIO()
|
|
with (
|
|
self.mock_call(
|
|
self.registry.get_from_path,
|
|
["path"],
|
|
Page("path", proxy="https://example.com"),
|
|
),
|
|
self.patch_call(
|
|
"requests.request",
|
|
[
|
|
"POST",
|
|
"https://example.com/index.html",
|
|
],
|
|
response,
|
|
{
|
|
"data": None,
|
|
"headers": {
|
|
"Host": "example.com",
|
|
"Accept": "nothing",
|
|
"X-Real-IP": "127.0.0.1",
|
|
"X-Forwarded-Host": "localhost",
|
|
"X-Forwarded-For": "127.0.0.1",
|
|
"X-Forwarded-Proto": "https",
|
|
},
|
|
"allow_redirects": False,
|
|
"timeout": 480,
|
|
},
|
|
),
|
|
self.expects_status_only(handler, 200, "OK"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_POST()
|
|
|
|
def test_do_post_proxy_sub_path_for_host(self) -> None:
|
|
handler = self._get_handler(
|
|
"/path/index.html", method="POST", headers={"Host": "host"}
|
|
)
|
|
response = requests.Response()
|
|
response.status_code = 200
|
|
response.reason = "OK"
|
|
response.raw = io.BytesIO()
|
|
with (
|
|
self.mock_call(
|
|
self.registry.get_from_host,
|
|
["host"],
|
|
Page("path", proxy="https://example.com"),
|
|
),
|
|
self.patch_call(
|
|
"requests.request",
|
|
[
|
|
"POST",
|
|
"https://example.com/path/index.html",
|
|
],
|
|
response,
|
|
{
|
|
"data": None,
|
|
"headers": {
|
|
"Host": "example.com",
|
|
"Accept": "nothing",
|
|
"X-Real-IP": "127.0.0.1",
|
|
"X-Forwarded-Host": "host",
|
|
"X-Forwarded-For": "127.0.0.1",
|
|
"X-Forwarded-Proto": "https",
|
|
},
|
|
"allow_redirects": False,
|
|
"timeout": 480,
|
|
},
|
|
),
|
|
self.expects_status_only(handler, 200, "OK"),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_POST()
|
|
|
|
def test_do_method_not_supported(self) -> None:
|
|
handler = self._get_handler("/path")
|
|
for method in ["CONNECT", "TRACE", "OPTIONS"]:
|
|
with (
|
|
self.subTest(None, method=method),
|
|
self.mock_call(
|
|
self.registry.get_from_path,
|
|
["path"],
|
|
),
|
|
self.expects_status_only(
|
|
handler, http.HTTPStatus.METHOD_NOT_ALLOWED, "Method Not Allowed"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
getattr(handler, f"do_{method}")()
|
|
|
|
def test_do_redirect(self) -> None:
|
|
handler = self._get_handler("/path")
|
|
for method in [method.value for method in http.HTTPMethod]:
|
|
with (
|
|
self.subTest(None, method=method),
|
|
self.mock_call(
|
|
self.registry.get_from_path,
|
|
["path"],
|
|
Page("path", redirect="https://example.com"),
|
|
),
|
|
self.expects_status_only(
|
|
handler,
|
|
http.HTTPStatus.MOVED_PERMANENTLY,
|
|
headers={"Location": "https://example.com"},
|
|
),
|
|
self.patch(
|
|
f"http.server.SimpleHTTPRequestHandler.do_{method}", count=0
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
getattr(handler, f"do_{method}")()
|
|
|
|
def test_list_directory(self) -> None:
|
|
handler = self._get_handler("/path/", {"Accept": "text/html"})
|
|
with (
|
|
self.expects_error_full(
|
|
handler, http.HTTPStatus.NOT_FOUND, "File not found"
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.list_directory()
|
|
|
|
def test_translate_path_certbot(self) -> None:
|
|
handler = self._get_handler()
|
|
with (
|
|
self.patch("http.server.SimpleHTTPRequestHandler.translate_path", count=0),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/.well-known/acme-challenge/abcde"),
|
|
str(self.certbot_www / ".well-known" / "acme-challenge" / "abcde"),
|
|
)
|
|
|
|
def test_translate_path_host_not_found(self) -> None:
|
|
handler = self._get_handler(headers={"Host": "example.com"})
|
|
with (
|
|
self.mock_call(self.registry.get_from_host, ["example.com"]),
|
|
self.patch("http.server.SimpleHTTPRequestHandler.translate_path", count=0),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/"),
|
|
"",
|
|
)
|
|
|
|
def test_translate_path_invalid(self) -> None:
|
|
handler = self._get_handler()
|
|
with (
|
|
self.patch("http.server.SimpleHTTPRequestHandler.translate_path", count=0),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/invalid.path"),
|
|
"",
|
|
)
|
|
|
|
def test_translate_path_favicon(self) -> None:
|
|
handler = self._get_handler()
|
|
with (
|
|
self.patch_call(
|
|
"http.server.SimpleHTTPRequestHandler.translate_path",
|
|
["/favicon.ico"],
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/favicon.ico"),
|
|
None,
|
|
)
|
|
|
|
def test_translate_path_dotfile(self) -> None:
|
|
handler = self._get_handler()
|
|
with (
|
|
self.mock_call(self.registry.get_from_path, ["path"], Page("path")),
|
|
self.patch("http.server.SimpleHTTPRequestHandler.translate_path", count=0),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/path/.token"),
|
|
"",
|
|
)
|
|
|
|
def test_translate_path_with_host(self) -> None:
|
|
handler = self._get_handler(headers={"Host": "example.com"})
|
|
with (
|
|
self.mock_call(self.registry.get_from_host, ["example.com"], Page("path")),
|
|
self.patch_call(
|
|
"http.server.SimpleHTTPRequestHandler.translate_path",
|
|
["/path/index.html"],
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/index.html"),
|
|
None,
|
|
)
|
|
|
|
def test_translate_path_with_host_favicon(self) -> None:
|
|
handler = self._get_handler(headers={"Host": "example.com"})
|
|
with (
|
|
self.mock_call(self.registry.get_from_host, ["example.com"], Page("path")),
|
|
self.patch_call(
|
|
"http.server.SimpleHTTPRequestHandler.translate_path",
|
|
["/favicon.ico"],
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/favicon.ico"),
|
|
None,
|
|
)
|
|
|
|
def test_translate_path_default_host(self) -> None:
|
|
handler = self._get_handler()
|
|
with (
|
|
self.mock_call(self.registry.get_from_path, ["path"], Page("path")),
|
|
self.patch_call(
|
|
"http.server.SimpleHTTPRequestHandler.translate_path",
|
|
["/path/index.html"],
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/path/index.html"),
|
|
None,
|
|
)
|
|
|
|
def test_translate_path_default_host_only(self) -> None:
|
|
handler = self._get_handler()
|
|
with (
|
|
self.mock_call(
|
|
self.registry.get_from_path, ["path"], Page("path", host_only=True)
|
|
),
|
|
self.patch("http.server.SimpleHTTPRequestHandler.translate_path", count=0),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/path/index.html"),
|
|
"",
|
|
)
|
|
|
|
def test_translate_path_default_host_not_found(self) -> None:
|
|
handler = self._get_handler()
|
|
with (
|
|
self.mock_call(self.registry.get_from_path, ["path"]),
|
|
self.patch("http.server.SimpleHTTPRequestHandler.translate_path", count=0),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/path/index.html"),
|
|
"",
|
|
)
|
|
|
|
def test_translate_path_spa(self) -> None:
|
|
handler = self._get_handler()
|
|
with (
|
|
self.mock_call(
|
|
self.registry.get_from_path, ["path"], Page("path", spa="index.html")
|
|
),
|
|
self.patch_call(
|
|
"http.server.SimpleHTTPRequestHandler.translate_path",
|
|
["/path/index.html"],
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
self.assertEqual(
|
|
handler.translate_path("/path/to/thing"),
|
|
None,
|
|
)
|
|
|
|
|
|
class TestUpgradeHandler(BaseHandlerTestCase):
|
|
@typing.override
|
|
def setUp(self) -> None:
|
|
self.data_dir = self.new_mock()
|
|
super().setUp()
|
|
|
|
def _get_handler(
|
|
self,
|
|
path: str = "/",
|
|
headers: dict[str, str | None] | None = None,
|
|
method: str = "GET",
|
|
rfile: io.BufferedIOBase | None = None,
|
|
) -> UpgradeHandler:
|
|
if headers is None:
|
|
headers = {}
|
|
with self.patch("http.server.BaseHTTPRequestHandler.__init__"):
|
|
handler = UpgradeHandler(
|
|
unittest.mock.MagicMock(),
|
|
"127.0.0.1",
|
|
unittest.mock.MagicMock(),
|
|
params=Parameters(),
|
|
registry=self.new_mock(),
|
|
token_manager=self.new_mock(),
|
|
)
|
|
handler.address_string = lambda: "127.0.0.1" # ty:ignore[invalid-assignment]
|
|
handler.requestline = f"{method} {path}"
|
|
handler.path = path
|
|
handler.command = method
|
|
handler.request_version = "HTTP/0.9"
|
|
handler.headers = collections.defaultdict(lambda: None, headers) # ty:ignore[invalid-assignment]
|
|
handler.rfile = rfile if rfile is not None else io.BytesIO()
|
|
handler.wfile = io.BytesIO()
|
|
handler.client_address = ("127.0.0.1", 12345)
|
|
handler.logger = unittest.mock.Mock(logging.Logger)
|
|
handler.data_dir = self.data_dir
|
|
return handler
|
|
|
|
def test_do_get(self) -> None:
|
|
handler = self._get_handler("/file")
|
|
with (
|
|
self.expects_status_only(
|
|
handler,
|
|
http.HTTPStatus.MOVED_PERMANENTLY,
|
|
headers={"Location": "https://localhost/file"},
|
|
),
|
|
self.patch(
|
|
"http.server.SimpleHTTPRequestHandler.do_GET",
|
|
count=0,
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_GET()
|
|
|
|
def test_do_get_certbot(self) -> None:
|
|
handler = self._get_handler("/.well-known/acme-challenge/abcde")
|
|
with (
|
|
self.patch(
|
|
"http.server.SimpleHTTPRequestHandler.do_GET",
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_GET()
|
|
|
|
def test_do_head(self) -> None:
|
|
handler = self._get_handler("/file")
|
|
with (
|
|
self.expects_status_only(
|
|
handler,
|
|
http.HTTPStatus.MOVED_PERMANENTLY,
|
|
headers={"Location": "https://localhost/file"},
|
|
),
|
|
self.seal_mocks(),
|
|
):
|
|
handler.do_HEAD()
|