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. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
8dffd8ccea
commit
44a16dea98
18 changed files with 1673 additions and 0 deletions
128
conftest.py
Normal file
128
conftest.py
Normal 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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue