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,
        content_length: int = 0,
    ) -> 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(content_length)),
            ]
            + [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_index(self) -> None:
        handler = self._get_handler()
        with (
            self.expects_status_only(
                handler, 200, content_length=len(handler.server_signature())
            ),
            self.patch("http.server.SimpleHTTPRequestHandler.do_HEAD", count=0),
            self.seal_mocks(),
        ):
            handler.do_HEAD()

    def test_do_head_forward(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_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,
                    "stream": False,
                },
            ),
            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,
                    "stream": False,
                },
            ),
            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,
                    "stream": False,
                },
            ),
            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,
                    "stream": False,
                },
            ) 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,
                    "stream": False,
                },
            ),
            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,
                    "stream": False,
                },
            ),
            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()