added pytest test suite (174 tests)

174 tests covering URL normalization, FTS5 query sanitization, SSRF/CSRF
guards, sharing-mode logic, DB schema and upsert paths, handler
end-to-end flows, and gateway body-size / mesh-whitelist guards. Each
recent bug-fix commit (6ffd38d, 1bc695f, 8dffd8c) has an explicit
regression test in test_regressions.py. One xfail documents a minor
latent bug in clean_url where port 80 is not stripped from upgraded
https URLs.
This commit is contained in:
lichenblankie 2026-04-24 15:03:29 -07:00
parent 55c6619ba3
commit 4d522ce62c
18 changed files with 1673 additions and 0 deletions

128
conftest.py Normal file
View file

@ -0,0 +1,128 @@
"""Shared pytest fixtures for TinyWeb tests.
Three fixtures cover most tests: `temp_db` swaps the SQLite path to a
per-test tempfile, `seeded_db` layers sample rows on top, and `csrf_session`
primes the thread-local CSRF token that handlers read.
"""
import socket
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent))
import db as db_module
import handlers as handlers_module
@pytest.fixture
def temp_db(tmp_path, monkeypatch):
"""Isolated SQLite DB per test.
Swaps `db.DATABASE` and `db.DATA_DIR` to a tempdir, clears the connection
pool before and after so state doesn't leak across tests, and calls
`init_db()` so every schema object exists.
"""
data_dir = tmp_path / "tinyweb"
data_dir.mkdir()
db_path = data_dir / "index.db"
monkeypatch.setattr(db_module, "DATA_DIR", str(data_dir))
monkeypatch.setattr(db_module, "DATABASE", str(db_path))
with db_module._pool_lock:
for conn in db_module._pool:
try:
conn.close()
except Exception:
pass
db_module._pool.clear()
db_module.init_db()
yield db_path
with db_module._pool_lock:
for conn in db_module._pool:
try:
conn.close()
except Exception:
pass
db_module._pool.clear()
@pytest.fixture
def seeded_db(temp_db):
"""A temp DB with a small, realistic set of pages/tags/links."""
db = db_module.get_db()
try:
rows = [
("https://example.com/rust-intro", "Rust Intro", "A gentle introduction to rust borrow checker.", "notes on ownership"),
("https://example.com/python-tips", "Python Tips", "Daily python tricks for readable code.", ""),
("https://example.com/ocaml-why", "Why OCaml", "Type systems and inference in ocaml.", "private thoughts"),
("https://news.example.org/mesh", "Mesh Networking", "Reticulum and LoRa for decentralized networks.", ""),
]
for url, title, body, note in rows:
db.execute(
"INSERT INTO pages (url, title, body, note, last_modified) "
"VALUES (?, ?, ?, ?, '2026-04-01T00:00:00')",
(url, title, body, note),
)
db.commit()
page_ids = {
row["url"]: row["id"]
for row in db.execute("SELECT id, url FROM pages").fetchall()
}
tag_rows = [
(page_ids["https://example.com/rust-intro"], ["rust", "public"]),
(page_ids["https://example.com/python-tips"], ["python"]),
(page_ids["https://example.com/ocaml-why"], ["ocaml", "private"]),
(page_ids["https://news.example.org/mesh"], ["mesh", "public"]),
]
for pid, tags in tag_rows:
for name in tags:
db.execute("INSERT OR IGNORE INTO tags (name) VALUES (?)", (name,))
tid = db.execute("SELECT id FROM tags WHERE name = ?", (name,)).fetchone()[0]
db.execute(
"INSERT OR IGNORE INTO page_tags (page_id, tag_id) VALUES (?, ?)",
(pid, tid),
)
db.execute(
"INSERT INTO links (page_id, url, label) VALUES (?, ?, ?)",
(page_ids["https://example.com/rust-intro"], "https://example.com/rust-advanced", "advanced rust guide"),
)
db.commit()
finally:
db_module.return_db(db)
return temp_db
@pytest.fixture
def csrf_session(monkeypatch):
"""Prime the CSRF thread-local so handler code that calls _get_csrf_token works."""
token = "test-csrf-token"
handlers_module._request_local.csrf_token = token
yield token
if hasattr(handlers_module._request_local, "csrf_token"):
del handlers_module._request_local.csrf_token
def patch_dns_fail(monkeypatch):
"""Make every socket.getaddrinfo call raise gaierror for the rest of this test."""
def boom(*args, **kwargs):
raise socket.gaierror("test: DNS disabled")
monkeypatch.setattr(socket, "getaddrinfo", boom)
def patch_dns_ok(monkeypatch, address="93.184.216.34"):
"""Make every getaddrinfo return a single public IP for the rest of this test."""
def ok(host, port, *args, **kwargs):
return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", (address, port or 80))]
monkeypatch.setattr(socket, "getaddrinfo", ok)
def patch_dns_private(monkeypatch, address="127.0.0.1"):
"""Make every getaddrinfo return a private/blocked IP for the rest of this test."""
def private(host, port, *args, **kwargs):
return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", (address, port or 80))]
monkeypatch.setattr(socket, "getaddrinfo", private)