tinyweb/conftest.py

"""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)