randogoth/mews.page
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: split the repo into self-contained validator, directory, quickstart, spec and site folders

Browse source 
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.
This commit is contained in:
randogoth 2026-10-13 12:29:23 +03:00
parent 5fa8708bd1
commit 060d4e00fd
90 changed files with 1290 additions and 287 deletions
Download patch file Download diff file Expand all files Collapse all files

142
directory/tests/conftest.py Normal file
View file

@ -0,0 +1,142 @@
"""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

14
directory/tests/fixtures/fail/div.html vendored Normal file
View file

@ -0,0 +1,14 @@
<?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</title>
<meta name="mews-profile" content="0.1" />
<meta name="viewport" content="width=device-width" />
<link rel="stylesheet" type="text/css" href="mews-0.1.css" />
</head>
<body>
<div><p>Hello.</p></div>
</body>
</html>

395
directory/tests/test_service.py Normal file
View file

@ -0,0 +1,395 @@
"""Submitting, listing, the endpoint, rechecks and the directory page."""
from datetime import timedelta
import functools
import pytest
from directory import app, check, db, pages
from mews.fetch import Fetcher, registered_domain
@pytest.fixture
def loopback(monkeypatch):
"""Let the service read the test server, and give it a registrable domain.
The test server answers on 127.0.0.1, which has no public suffix, so the
name it would be listed under is supplied here rather than relaxing the
rule in the service.
"""
monkeypatch.setattr(app, "Fetcher", functools.partial(Fetcher, allow_loopback=True))
monkeypatch.setattr(
check,
"registered_domain",
lambda host: "example.test" if host == "127.0.0.1" else registered_domain(host),
)
@pytest.fixture
def service(tmp_path, monkeypatch, loopback):
"""The app, wired to an empty database and a temporary directory page."""
database = tmp_path / "mews.db"
monkeypatch.setenv("MEWS_DB", str(database))
monkeypatch.setenv("MEWS_DIRECTORY", str(tmp_path / "directory.html"))
handle = db.connect(str(database))
db.init(handle)
handle.close()
return tmp_path
def post(url: str = "", *, listing: bool = False, client: str = "203.0.113.5"):
"""Send one form submission to the WSGI app and return status and body."""
from io import BytesIO
from urllib.parse import urlencode
fields = {"url": url, "list" if listing else "check": "x"}
data = urlencode(fields).encode()
captured: dict = {}
def start_response(status, headers):
captured["status"] = status
captured["headers"] = headers
body = b"".join(
app.application(
{
"PATH_INFO": "/result",
"REQUEST_METHOD": "POST",
"CONTENT_LENGTH": str(len(data)),
"wsgi.input": BytesIO(data),
"HTTP_X_REAL_IP": client,
},
start_response,
)
)
return captured, body.decode()
def test_a_conforming_page_is_listed(service, site, conforming):
url = site.add("/index.html", conforming())
captured, body = post(url, listing=True)
assert captured["status"] == "200 OK"
assert "Your site is listed" in body
handle = db.connect()
assert [row["domain"] for row in db.listed(handle)] == ["example.test"]
assert (service / "directory.html").exists()
assert "example.test" in (service / "directory.html").read_text()
def test_checking_does_not_list(service, site, conforming):
url = site.add("/index.html", conforming())
_, body = post(url, listing=False)
assert "This page conforms" in body
assert db.listed(db.connect()) == []
def test_a_failing_page_is_not_listed_and_says_why(service, site, fail_page):
url = site.add("/index.html", fail_page("div"))
_, body = post(url, listing=True)
assert "doesn't conform" in body
assert "Section 4.1" in body
assert db.listed(db.connect()) == []
def test_every_response_is_a_mews_page(service, site, conforming):
from mews.lint import validate_bytes
url = site.add("/index.html", conforming())
for listing in (False, True):
_, body = post(url, listing=listing)
assert validate_bytes(body.encode()).conforms
def test_no_response_sets_a_cookie(service, site, conforming):
url = site.add("/index.html", conforming())
captured, _ = post(url, listing=True)
assert not [name for name, _ in captured["headers"] if name.lower() == "set-cookie"]
def test_responses_are_not_cached(service, site, conforming):
url = site.add("/index.html", conforming())
captured, _ = post(url, listing=True)
assert ("Cache-Control", "no-store") in captured["headers"]
def test_an_address_that_is_not_a_page_is_refused(service):
_, body = post("http://10.0.0.1/", listing=True)
assert "domain name rather than an IP address" in body
def test_an_unreachable_page_is_reported(service, site):
_, body = post(site.base + "/missing.html", listing=True)
assert "That page returned 404" in body
def test_resubmitting_a_site_updates_the_same_row(
service, site, conforming, monkeypatch
):
"""One row per registrable domain, however many pages of it are submitted."""
first = site.add("/a.html", conforming())
post(first, listing=True)
# Past the per-domain cooldown, which would otherwise turn this away.
later = db.now() + timedelta(hours=2)
monkeypatch.setattr(db, "now", lambda: later)
second = site.add("/b.html", conforming())
post(second, listing=True, client="203.0.113.6")
rows = db.listed(db.connect())
assert len(rows) == 1
assert rows[0]["url"] == second
def test_a_blocked_domain_is_refused(service, site, conforming):
handle = db.connect()
db.block(handle, "example.test", "spam")
url = site.add("/index.html", conforming())
_, body = post(url, listing=True)
assert "taken out of the directory" in body
def test_the_endpoint_answers_other_paths_and_methods_politely(service):
from io import BytesIO
captured: dict = {}
def start_response(status, headers):
captured["status"] = status
body = b"".join(
app.application(
{
"PATH_INFO": "/result",
"REQUEST_METHOD": "GET",
"wsgi.input": BytesIO(b""),
},
start_response,
)
).decode()
assert captured["status"].startswith("405")
assert "Nothing to show" in body
def test_rate_limits_stop_a_flood(connection):
client = db.ip_hash(connection, "203.0.113.9")
for _ in range(db.IP_LISTINGS_PER_HOUR):
assert (
db.rate_limited(connection, client=client, domain="a.example", listing=True)
is None
)
db.record_submission(
connection,
url="https://a.example/",
domain="other.example",
client=client,
outcome="listed",
)
assert "Try again in an hour" in db.rate_limited(
connection, client=client, domain="a.example", listing=True
)
def test_checking_a_page_does_not_block_listing_it(connection):
"""Checking first is the sensible move; it must not cost the listing."""
client = db.ip_hash(connection, "203.0.113.9")
db.record_submission(
connection,
url="https://a.example/",
domain="a.example",
client=client,
outcome="checked",
)
assert (
db.rate_limited(connection, client=client, domain="a.example", listing=True)
is None
)
def test_a_domain_has_to_wait_between_submissions(connection):
client = db.ip_hash(connection, "203.0.113.9")
db.record_submission(
connection,
url="https://a.example/",
domain="a.example",
client=client,
outcome="rejected",
)
assert "ten minutes" in db.rate_limited(
connection, client=client, domain="a.example", listing=True
)
def test_a_domain_that_keeps_failing_waits_a_day(connection, monkeypatch):
client = db.ip_hash(connection, "203.0.113.9")
base = db.now()
for index in range(db.DOMAIN_REJECTS_BEFORE_SLOWDOWN):
monkeypatch.setattr(db, "now", lambda i=index: base - timedelta(hours=i + 1))
db.record_submission(
connection,
url="https://a.example/",
domain="a.example",
client=client,
outcome="rejected",
)
monkeypatch.setattr(db, "now", lambda: base)
assert "try again tomorrow" in db.rate_limited(
connection, client=client, domain="a.example", listing=True
)
# --- rechecks ----------------------------------------------------------
def _list_site(connection, url):
return db.upsert_site(
connection,
domain="example.test",
url=url,
title="Test site",
description="",
language="en",
etag=None,
last_modified=None,
)[0]
def test_a_page_that_still_conforms_stays(connection, site, conforming):
url = site.add("/index.html", conforming(), headers={"ETag": '"v1"'})
site_id = _list_site(connection, url)
with Fetcher(allow_loopback=True) as fetcher:
assert (
check.recheck(
connection, db.site(connection, "example.test"), fetcher=fetcher
)
== "pass"
)
assert db.site(connection, "example.test")["state"] == "listed"
assert db.site(connection, "example.test")["etag"] == '"v1"'
assert site_id
def test_a_page_that_stopped_conforming_is_dropped(connection, site, fail_page):
url = site.add("/index.html", fail_page("div"))
_list_site(connection, url)
with Fetcher(allow_loopback=True) as fetcher:
assert (
check.recheck(
connection, db.site(connection, "example.test"), fetcher=fetcher
)
== "fail"
)
row = db.site(connection, "example.test")
assert row["state"] == "removed"
assert "Section 4.1" in row["reason"]
assert db.listed(connection) == []
def test_a_not_modified_answer_passes_without_rereading(connection, site, conforming):
url = site.add("/index.html", conforming(), headers={"ETag": '"v1"'})
_list_site(connection, url)
site.add("/index.html", b"", status=304, headers={"ETag": '"v1"'})
connection.execute("UPDATE sites SET etag = '\"v1\"'")
with Fetcher(allow_loopback=True) as fetcher:
assert (
check.recheck(
connection, db.site(connection, "example.test"), fetcher=fetcher
)
== "pass"
)
assert db.site(connection, "example.test")["state"] == "listed"
assert "If-None-Match" in site.requests[-1][1]
def test_downtime_takes_several_tries_to_drop_a_site(connection, site, conforming):
url = site.add("/index.html", conforming())
_list_site(connection, url)
site.routes.clear()
site.add("/index.html", b"broken", status=503)
for _ in range(db.TRANSIENT_LIMIT - 1):
with Fetcher(allow_loopback=True) as fetcher:
assert (
check.recheck(
connection, db.site(connection, "example.test"), fetcher=fetcher
)
== "unreachable"
)
assert db.site(connection, "example.test")["state"] == "listed"
with Fetcher(allow_loopback=True) as fetcher:
assert (
check.recheck(
connection, db.site(connection, "example.test"), fetcher=fetcher
)
== "fail"
)
assert db.site(connection, "example.test")["state"] == "removed"
def test_only_sites_that_are_due_are_rechecked(connection, monkeypatch):
_list_site(connection, "https://example.test/")
assert db.due(connection) == []
later = db.now() + timedelta(days=db.RECHECK_DAYS + 1)
monkeypatch.setattr(db, "now", lambda: later)
assert len(db.due(connection)) == 1
# --- the directory page ------------------------------------------------
def test_the_directory_page_conforms(connection, tmp_path):
from mews.lint import validate_bytes
_list_site(connection, "https://example.test/")
target = tmp_path / "directory.html"
pages.write_directory(str(target), db.listed(connection))
assert validate_bytes(target.read_bytes()).conforms
def test_a_hostile_title_is_escaped_and_stripped(connection, tmp_path):
from mews.lint import validate_bytes
connection.execute(
"INSERT INTO sites (domain, url, title, state, listed_at) "
"VALUES ('x.test', 'https://x.test/', '<script>‮oops</script>', "
"'listed', '2026-01-01T00:00:00+00:00')"
)
connection.commit()
target = tmp_path / "directory.html"
pages.write_directory(str(target), db.listed(connection))
text = target.read_text()
assert "<script>" not in text
assert "&lt;script&gt;" in text
assert validate_bytes(target.read_bytes()).conforms
def test_the_directory_page_is_readable_by_the_web_server(connection, tmp_path):
"""It is served off disk by another user, whatever umask the service has."""
import os
import stat
target = tmp_path / "directory.html"
old_umask = os.umask(0o077)
try:
pages.write_directory(str(target), db.listed(connection))
finally:
os.umask(old_umask)
assert stat.S_IMODE(target.stat().st_mode) == 0o644
def test_an_empty_directory_still_conforms(connection, tmp_path):
from mews.lint import validate_bytes
target = tmp_path / "directory.html"
pages.write_directory(str(target), db.listed(connection))
assert validate_bytes(target.read_bytes()).conforms
assert "Nothing here yet" in target.read_text()
def test_a_page_that_would_not_conform_leaves_the_old_one_alone(
connection, tmp_path, monkeypatch
):
target = tmp_path / "directory.html"
pages.write_directory(str(target), db.listed(connection))
before = target.read_text()
monkeypatch.setattr(pages, "directory", lambda rows: "<p>not a page</p>")
with pytest.raises(ValueError, match="doesn't conform"):
pages.write_directory(str(target), db.listed(connection))
assert target.read_text() == before