mews.page/tests/conftest.py

"""Shared fixtures: fixture paths, a temporary database, and a local web server."""

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
import threading

import pytest

from mews import db

ROOT = Path(__file__).resolve().parent.parent
FIXTURES = Path(__file__).resolve().parent / "fixtures"


@pytest.fixture
def fail_page():
    """Read one of the non-conforming fixture pages."""

    def read(name: str) -> bytes:
        return (FIXTURES / "fail" / f"{name}.html").read_bytes()

    return read


@pytest.fixture
def connection(tmp_path):
    """An empty database."""
    handle = db.connect(str(tmp_path / "mews.db"))
    db.init(handle)
    yield handle
    handle.close()


class Site:
    """A web server on loopback whose responses the test writes."""

    def __init__(self) -> None:
        self.routes: dict[str, tuple[int, dict[str, str], bytes, bool]] = {}
        self.requests: list[tuple[str, dict[str, str]]] = []
        self.delay = 0.0
        self._server = ThreadingHTTPServer(("127.0.0.1", 0), _handler(self))
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
        self._thread.start()

    @property
    def base(self) -> str:
        """The address the server answers on."""
        host, port = self._server.server_address[:2]
        return f"http://{host}:{port}"

    def add(
        self,
        path: str,
        body: bytes = b"",
        status: int = 200,
        headers: dict[str, str] | None = None,
        declare_length: bool = True,
    ) -> str:
        """Serve body at path, and return its full address.

        With declare_length off the response carries no Content-Length and ends
        at the connection close, which is how a streaming body reaches the
        fetcher's byte cap.
        """
        self.routes[path] = (status, headers or {}, body, declare_length)
        return self.base + path

    def close(self) -> None:
        """Stop the server."""
        self._server.shutdown()
        self._server.server_close()


def _handler(site: "Site"):
    class Handler(BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"

        def do_GET(self):
            import time

            site.requests.append((self.path, dict(self.headers)))
            if site.delay:
                time.sleep(site.delay)
            status, headers, body, declare_length = site.routes.get(
                self.path, (404, {}, b"not here", True)
            )
            if not declare_length:
                self.protocol_version = "HTTP/1.0"
                self.close_connection = True
            self.send_response(status)
            for name, value in headers.items():
                self.send_header(name, value)
            if "Content-Type" not in headers and status == 200:
                self.send_header("Content-Type", "text/html; charset=utf-8")
            if declare_length:
                self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            if body:
                self.wfile.write(body)

        def log_message(self, *args):
            pass

        def handle_one_request(self):
            try:
                super().handle_one_request()
            except (BrokenPipeError, ConnectionResetError):
                self.close_connection = True

    return Handler


@pytest.fixture
def site():
    """A local web server for the checks that need a live page."""
    server = Site()
    yield server
    server.close()


@pytest.fixture
def conforming():
    """A page that follows the spec, with a stylesheet link the test can point."""

    def build(stylesheet: str = "mews-0.1.css", body: str = "<p>Hello.</p>") -> bytes:
        return f"""<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE html PUBLIC "-//WAPFORUM//DTD XHTML Mobile 1.2//EN"
  "http://www.openmobilealliance.org/tech/DTD/xhtml-mobile12.dtd">
<html xmlns="http://www.w3.org/1999/xhtml" xml:lang="en" lang="en">
  <head>
    <title>Test site</title>
    <meta name="mews-profile" content="0.1" />
    <meta name="viewport" content="width=device-width" />
    <meta name="description" content="A test page." />
    <link rel="stylesheet" type="text/css" href="{stylesheet}" />
  </head>
  <body>
    {body}
  </body>
</html>
""".encode()

    return build