060d4e00fd
validator/ holds mewslint (lint, css, fetch, DTD/CSS data); directory/ holds the mews.page service (app, db, check, pages, cli) and depends on validator/; quickstart/ holds mews-quickstart and md2mews.py for third parties who want to create Mews content without cloning the repo; spec/ consolidates doc/ and dtd/ with the canonical mews-0.1.css; site/ becomes a Pelican project with the forward-facing pages as Markdown, the check form kept as a raw HTML snippet. Root pyproject.toml is now a virtual uv workspace over validator, directory and quickstart, with cross-cutting tests that span more than one of them staying at the repo root. Makefile and flake.nix updated to match.
142 lines
4.2 KiB
Python
142 lines
4.2 KiB
Python
"""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 directory import db
|
|
|
|
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
|