tests: add customize and data handler tests (35 new)
This commit is contained in:
parent
ae3f02fa31
commit
c20de93306
2 changed files with 303 additions and 0 deletions
161
tests/test_handlers_customize.py
Normal file
161
tests/test_handlers_customize.py
Normal file
|
|
@ -0,0 +1,161 @@
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from tinyweb.handlers import (
|
||||||
|
handle_style_form, handle_style_submit,
|
||||||
|
handle_style_template_submit, handle_field_save,
|
||||||
|
handle_about,
|
||||||
|
)
|
||||||
|
from tinyweb.templates import DEFAULT_TEMPLATE
|
||||||
|
from tinyweb.db import get_db, return_db, get_setting
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_form_renders_defaults(temp_db, csrf_session):
|
||||||
|
resp = handle_style_form()
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert "customize" in resp["body"].lower()
|
||||||
|
assert "tinyweb" in resp["body"]
|
||||||
|
assert "search" in resp["body"]
|
||||||
|
assert "template" in resp["body"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_form_shows_custom_site_name(temp_db, csrf_session):
|
||||||
|
from tinyweb.db import set_setting
|
||||||
|
set_setting("site_name", "my-mesh")
|
||||||
|
resp = handle_style_form()
|
||||||
|
assert "my-mesh" in resp["body"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_submit_name_saves(temp_db, csrf_session):
|
||||||
|
resp = handle_style_submit({
|
||||||
|
"_action": ["name"],
|
||||||
|
"site_name": ["My Network"],
|
||||||
|
})
|
||||||
|
assert resp["status"] in (302, 303)
|
||||||
|
assert get_setting("site_name", "") == "My Network"
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_submit_name_uses_default_when_empty(temp_db, csrf_session):
|
||||||
|
resp = handle_style_submit({
|
||||||
|
"_action": ["name"],
|
||||||
|
"site_name": [""],
|
||||||
|
})
|
||||||
|
assert resp["status"] in (302, 303)
|
||||||
|
assert get_setting("site_name", "") == "tinyweb"
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_submit_sharing_enables(temp_db, csrf_session):
|
||||||
|
handle_style_submit({
|
||||||
|
"_action": ["sharing"],
|
||||||
|
"sharing_enabled": ["1"],
|
||||||
|
"sharing_mode": ["exclude_private"],
|
||||||
|
})
|
||||||
|
assert get_setting("sharing_enabled", "0") == "1"
|
||||||
|
assert get_setting("sharing_mode", "") == "exclude_private"
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_submit_sharing_disables(temp_db, csrf_session):
|
||||||
|
handle_style_submit({
|
||||||
|
"_action": ["sharing"],
|
||||||
|
})
|
||||||
|
assert get_setting("sharing_enabled", "1") == "0"
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_submit_sharing_mode_require_public(temp_db, csrf_session):
|
||||||
|
handle_style_submit({
|
||||||
|
"_action": ["sharing"],
|
||||||
|
"sharing_enabled": ["1"],
|
||||||
|
"sharing_mode": ["require_public"],
|
||||||
|
})
|
||||||
|
assert get_setting("sharing_mode", "") == "require_public"
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_submit_search_saves(temp_db, csrf_session):
|
||||||
|
handle_style_submit({
|
||||||
|
"_action": ["search"],
|
||||||
|
"semantic_search": ["1"],
|
||||||
|
"use_reranker": ["1"],
|
||||||
|
"compress_embeddings": ["1"],
|
||||||
|
})
|
||||||
|
assert get_setting("semantic_search", "0") == "1"
|
||||||
|
assert get_setting("use_reranker", "0") == "1"
|
||||||
|
assert get_setting("compress_embeddings", "0") == "1"
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_submit_search_disables(temp_db, csrf_session):
|
||||||
|
handle_style_submit({"_action": ["search"]})
|
||||||
|
assert get_setting("semantic_search", "1") == "0"
|
||||||
|
assert get_setting("use_reranker", "1") == "0"
|
||||||
|
assert get_setting("compress_embeddings", "1") == "0"
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_submit_mesh_saves_tcp(temp_db, csrf_session):
|
||||||
|
handle_style_submit({
|
||||||
|
"_action": ["mesh"],
|
||||||
|
"tcp_enabled": ["1"],
|
||||||
|
"transport_host": ["mesh.example.org"],
|
||||||
|
"transport_port": ["9999"],
|
||||||
|
})
|
||||||
|
assert get_setting("tcp_enabled", "0") == "1"
|
||||||
|
assert get_setting("transport_host", "") == "mesh.example.org"
|
||||||
|
assert get_setting("transport_port", "") == "9999"
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_submit_mesh_disables_tcp(temp_db, csrf_session):
|
||||||
|
handle_style_submit({"_action": ["mesh"]})
|
||||||
|
assert get_setting("tcp_enabled", "1") == "0"
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_submit_mesh_saves_lora(temp_db, csrf_session):
|
||||||
|
handle_style_submit({
|
||||||
|
"_action": ["mesh"],
|
||||||
|
"lora_enabled": ["1"],
|
||||||
|
"lora_port": ["/dev/ttyUSB0"],
|
||||||
|
"lora_frequency": ["915000000"],
|
||||||
|
})
|
||||||
|
assert get_setting("lora_enabled", "0") == "1"
|
||||||
|
assert get_setting("lora_port", "") == "/dev/ttyUSB0"
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_template_submit_saves(temp_db, csrf_session):
|
||||||
|
custom = "<html><body>{{content}}</body></html>"
|
||||||
|
resp = handle_style_template_submit({"template": [custom]})
|
||||||
|
assert resp["status"] in (302, 303)
|
||||||
|
assert get_setting("custom_template") == custom
|
||||||
|
|
||||||
|
|
||||||
|
def test_style_template_submit_strips_default(temp_db, csrf_session):
|
||||||
|
"""Saving the default template should clear custom_template."""
|
||||||
|
resp = handle_style_template_submit({"template": [DEFAULT_TEMPLATE]})
|
||||||
|
assert resp["status"] in (302, 303)
|
||||||
|
assert get_setting("custom_template", None) == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_field_save_sets_setting(temp_db, csrf_session):
|
||||||
|
resp = handle_field_save({"key": ["site_name"], "value": ["my-site"]})
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert get_setting("site_name", "") == "my-site"
|
||||||
|
|
||||||
|
|
||||||
|
def test_field_save_rejects_empty_key(temp_db, csrf_session):
|
||||||
|
resp = handle_field_save({"key": [""], "value": ["x"]})
|
||||||
|
assert resp["status"] == 400
|
||||||
|
|
||||||
|
|
||||||
|
def test_about_renders_stats(seeded_db, csrf_session):
|
||||||
|
resp = handle_about()
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert "personal" in resp["body"].lower()
|
||||||
|
assert "4" in resp["body"]
|
||||||
|
assert "page(s) indexed" in resp["body"].lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_about_shows_dest_hash_when_available(seeded_db, csrf_session):
|
||||||
|
from tinyweb.db import set_setting
|
||||||
|
set_setting("dest_hash", "abc123")
|
||||||
|
resp = handle_about()
|
||||||
|
assert "abc123" in resp["body"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_about_no_subscribe_section_when_hash_unset(seeded_db, csrf_session):
|
||||||
|
resp = handle_about()
|
||||||
|
assert '<h2>subscribe</h2>' not in resp["body"].lower()
|
||||||
142
tests/test_handlers_data.py
Normal file
142
tests/test_handlers_data.py
Normal file
|
|
@ -0,0 +1,142 @@
|
||||||
|
import json
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from tinyweb.db import get_db, return_db
|
||||||
|
from tinyweb.handlers import (
|
||||||
|
handle_export, handle_import_form, handle_import_submit,
|
||||||
|
handle_reindex_form, handle_reindex_submit,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _all_urls():
|
||||||
|
db = get_db()
|
||||||
|
try:
|
||||||
|
return {r["url"] for r in db.execute("SELECT url FROM pages").fetchall()}
|
||||||
|
finally:
|
||||||
|
return_db(db)
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_returns_json(seeded_db, csrf_session):
|
||||||
|
resp = handle_export({})
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert resp["content_type"] == "application/json"
|
||||||
|
data = json.loads(resp["body"])
|
||||||
|
assert len(data) == 4
|
||||||
|
assert data[0]["url"] == "https://example.com/rust-intro"
|
||||||
|
assert data[0]["title"] == "Rust Intro"
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_paginates(seeded_db, csrf_session):
|
||||||
|
from tinyweb.handlers.data import MAX_EXPORT
|
||||||
|
resp = handle_export({"batch": ["0"]})
|
||||||
|
data = json.loads(resp["body"])
|
||||||
|
assert len(data) <= MAX_EXPORT
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_respects_batch_param(seeded_db, csrf_session):
|
||||||
|
resp = handle_export({"batch": ["99"]})
|
||||||
|
data = json.loads(resp["body"])
|
||||||
|
assert data == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_bad_batch_defaults_to_zero(seeded_db, csrf_session):
|
||||||
|
resp = handle_export({"batch": ["abc"]})
|
||||||
|
assert resp["status"] == 200
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_has_content_disposition(seeded_db, csrf_session):
|
||||||
|
resp = handle_export({})
|
||||||
|
assert any(
|
||||||
|
k.lower() == "content-disposition" and "tinyweb-export.json" in v
|
||||||
|
for k, v in resp.get("headers", {}).items()
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_import_form_renders(temp_db, csrf_session):
|
||||||
|
resp = handle_import_form()
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert "import" in resp["body"].lower()
|
||||||
|
assert "textarea" in resp["body"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_import_submit_empty_rejected(temp_db, csrf_session):
|
||||||
|
resp = handle_import_submit({"data": [""]})
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert "paste json" in resp["body"].lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_import_submit_invalid_json(temp_db, csrf_session):
|
||||||
|
resp = handle_import_submit({"data": ["not json"]})
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert "invalid json" in resp["body"].lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_import_submit_non_array_rejected(temp_db, csrf_session):
|
||||||
|
resp = handle_import_submit({"data": ['"string"']})
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert "expected a json array" in resp["body"].lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_import_submit_too_many_rejected(temp_db, csrf_session):
|
||||||
|
big = [{"url": "https://x.com/y"}] * 101
|
||||||
|
resp = handle_import_submit({"data": [json.dumps(big)]})
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert "too many" in resp["body"].lower()
|
||||||
|
|
||||||
|
|
||||||
|
@patch("tinyweb.handlers.data.index_url")
|
||||||
|
def test_import_submit_calls_index_url(mock_index, temp_db, csrf_session):
|
||||||
|
payload = [
|
||||||
|
{"url": "https://example.com/a", "note": "note a"},
|
||||||
|
{"url": "https://example.com/b", "note": "note b"},
|
||||||
|
]
|
||||||
|
resp = handle_import_submit({"data": [json.dumps(payload)]})
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert "imported 2" in resp["body"].lower()
|
||||||
|
assert mock_index.call_count == 2
|
||||||
|
mock_index.assert_any_call("https://example.com/a", "note a")
|
||||||
|
mock_index.assert_any_call("https://example.com/b", "note b")
|
||||||
|
|
||||||
|
|
||||||
|
@patch("tinyweb.handlers.data.index_url")
|
||||||
|
def test_import_submit_skips_empty_url(mock_index, temp_db, csrf_session):
|
||||||
|
payload = [
|
||||||
|
{"url": "", "note": ""},
|
||||||
|
{"url": "https://example.com/real", "note": ""},
|
||||||
|
]
|
||||||
|
resp = handle_import_submit({"data": [json.dumps(payload)]})
|
||||||
|
assert "imported 1" in resp["body"].lower()
|
||||||
|
assert mock_index.call_count == 1
|
||||||
|
|
||||||
|
|
||||||
|
@patch("tinyweb.handlers.data.index_url")
|
||||||
|
def test_import_submit_counts_errors(mock_index, temp_db, csrf_session):
|
||||||
|
mock_index.side_effect = [None, ValueError("boom")]
|
||||||
|
payload = [
|
||||||
|
{"url": "https://example.com/a", "note": ""},
|
||||||
|
{"url": "https://example.com/b", "note": ""},
|
||||||
|
]
|
||||||
|
resp = handle_import_submit({"data": [json.dumps(payload)]})
|
||||||
|
assert "imported 1" in resp["body"].lower()
|
||||||
|
assert "1 error" in resp["body"].lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_reindex_form_shows_disabled_message(temp_db, csrf_session):
|
||||||
|
resp = handle_reindex_form()
|
||||||
|
assert resp["status"] == 200
|
||||||
|
assert "disabled" in resp["body"].lower()
|
||||||
|
assert "settings" in resp["body"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_reindex_form_shows_progress_when_running(temp_db, csrf_session):
|
||||||
|
from tinyweb.db import set_setting
|
||||||
|
set_setting("semantic_search", "1")
|
||||||
|
set_setting("reindex_progress", "5/10")
|
||||||
|
resp = handle_reindex_form()
|
||||||
|
assert "reindex in progress" in resp["body"].lower()
|
||||||
|
assert "5/10" in resp["body"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_reindex_submit_redirects(temp_db, csrf_session):
|
||||||
|
resp = handle_reindex_submit({})
|
||||||
|
assert resp["status"] in (302, 303)
|
||||||
Loading…
Add table
Add a link
Reference in a new issue