4d522ce62c
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.
112 lines
4 KiB
Python
112 lines
4 KiB
Python
"""Tests for subscription handlers.
|
|
|
|
Subscription add validates the destination hash (32-char hex) locally
|
|
before calling `fetch_remote_sites`; browse uses cached remote_pages when
|
|
available and falls back to a live fetch otherwise.
|
|
"""
|
|
from unittest.mock import patch
|
|
|
|
import handlers as handlers_module
|
|
from db import get_db, return_db
|
|
from handlers import handle_subscription_add, handle_subscription_browse
|
|
|
|
|
|
VALID_HASH = "a" * 32
|
|
|
|
|
|
def _subscription_count():
|
|
db = get_db()
|
|
try:
|
|
return db.execute("SELECT count(*) FROM subscriptions").fetchone()[0]
|
|
finally:
|
|
return_db(db)
|
|
|
|
|
|
def test_rejects_empty_dest_hash(temp_db, csrf_session):
|
|
resp = handle_subscription_add({"dest_hash": [""]})
|
|
assert "32-character" in resp["body"]
|
|
assert _subscription_count() == 0
|
|
|
|
|
|
def test_rejects_wrong_length(temp_db, csrf_session):
|
|
resp = handle_subscription_add({"dest_hash": ["abc123"]})
|
|
assert "32-character" in resp["body"]
|
|
assert _subscription_count() == 0
|
|
|
|
|
|
def test_rejects_non_hex(temp_db, csrf_session):
|
|
resp = handle_subscription_add({"dest_hash": ["z" * 32]})
|
|
assert "hex" in resp["body"].lower()
|
|
assert _subscription_count() == 0
|
|
|
|
|
|
def test_rejects_unreachable_peer(temp_db, csrf_session):
|
|
with patch.object(handlers_module, "fetch_remote_sites") as fetch:
|
|
fetch.side_effect = ConnectionError("unreachable")
|
|
resp = handle_subscription_add({"dest_hash": [VALID_HASH]})
|
|
assert "Could not reach" in resp["body"]
|
|
assert _subscription_count() == 0
|
|
|
|
|
|
def test_rejects_peer_with_sharing_disabled(temp_db, csrf_session):
|
|
with patch.object(handlers_module, "fetch_remote_sites") as fetch:
|
|
fetch.side_effect = PermissionError("sharing disabled")
|
|
resp = handle_subscription_add({"dest_hash": [VALID_HASH]})
|
|
assert "sharing disabled" in resp["body"]
|
|
assert _subscription_count() == 0
|
|
|
|
|
|
def test_successful_add_records_subscription(temp_db, csrf_session):
|
|
with patch.object(handlers_module, "fetch_remote_sites") as fetch:
|
|
fetch.return_value = {"name": "alice", "sites": []}
|
|
resp = handle_subscription_add({"dest_hash": [VALID_HASH]})
|
|
assert "Subscribed to alice" in resp["body"]
|
|
assert _subscription_count() == 1
|
|
|
|
|
|
def test_dest_hash_strips_angle_brackets(temp_db, csrf_session):
|
|
"""Users often paste hashes as `<aaa...>` from RNS log output; strip them."""
|
|
with patch.object(handlers_module, "fetch_remote_sites") as fetch:
|
|
fetch.return_value = {"name": "bob", "sites": []}
|
|
resp = handle_subscription_add({"dest_hash": [f"<{VALID_HASH}>"]})
|
|
assert _subscription_count() == 1
|
|
|
|
|
|
def test_browse_unknown_subscription_is_404(temp_db, csrf_session):
|
|
resp = handle_subscription_browse(99999)
|
|
assert resp["status"] == 404
|
|
|
|
|
|
def test_browse_marks_already_indexed_urls(seeded_db, csrf_session):
|
|
# Insert a subscription + some remote pages (one duplicate of local, one new).
|
|
db = get_db()
|
|
try:
|
|
db.execute(
|
|
"INSERT INTO subscriptions (dest_hash, name) VALUES (?, ?)",
|
|
(VALID_HASH, "alice"),
|
|
)
|
|
sub_id = db.execute("SELECT id FROM subscriptions").fetchone()["id"]
|
|
db.execute(
|
|
"INSERT INTO remote_pages (subscription_id, url, title, note, tags) "
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
(sub_id, "https://example.com/rust-intro", "Alice rust pick", "", ""),
|
|
)
|
|
db.execute(
|
|
"INSERT INTO remote_pages (subscription_id, url, title, note, tags) "
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
(sub_id, "https://new.example.com/shiny", "Shiny New Link", "note", "tag1"),
|
|
)
|
|
db.commit()
|
|
finally:
|
|
return_db(db)
|
|
|
|
resp = handle_subscription_browse(sub_id)
|
|
body = resp["body"]
|
|
assert resp["status"] == 200
|
|
assert "already indexed" in body
|
|
# The duplicate URL should appear in the "already indexed" section.
|
|
assert "Alice rust pick" in body
|
|
# The new URL should be in the selectable section.
|
|
assert "Shiny New Link" in body
|
|
# Count summary: "2 site(s) available, 1 new"
|
|
assert "1 new" in body
|