diff --git a/priorityweb/app.py b/priorityweb/app.py new file mode 100644 index 0000000..a712b7f --- /dev/null +++ b/priorityweb/app.py @@ -0,0 +1,920 @@ +#!/usr/bin/env python3 +""" +JonasSearch Vorschlagsverfeinerer Gamma 0.1 + +Lets users submit domains that are put at a very high priority in the +crawler's PostgreSQL queue (crawl_queue), so they get crawled next. + +Runs on PORT = statserver port + 1 (default 8082). + +How the queue jump works +------------------------ +The crawler claims work with: + + ORDER BY priority DESC, discovered_at ASC, id ASC + +Normal crawler-discovered URLs sit at priority 0..500. A suggestion is +upserted at SUGGEST_PRIORITY (1,000,000), which sorts it above the whole +existing queue WITHOUT touching the other ~13M pending rows. The table +is far too large for mass UPDATEs: such a query blocks the crawler for +minutes. Multiple suggestions all share SUGGEST_PRIORITY and are +crawled in submission order (discovered_at, id), so no demotion is +needed there either. +""" + +import os +import re +import threading +import time +from datetime import datetime, timezone + +import psycopg2 +from flask import Flask, jsonify, render_template_string, request +from psycopg2.extras import RealDictCursor + +DB_HOST = os.getenv("JONASSEARCH_DB_HOST", "127.0.0.1") +DB_PORT = int(os.getenv("JONASSEARCH_DB_PORT", "25432")) +DB_NAME = os.getenv("JONASSEARCH_DB_NAME", "jonassearch") +DB_USER = os.getenv("JONASSEARCH_DB_USER", "jonassearch") +DB_PASSWORD = os.getenv("JONASSEARCH_DB_PASSWORD", "") + +# Port of the statserver (jonassearch-stats.service, web/app.py). +STATSERVER_PORT = int(os.getenv("JONASSEARCH_STATSERVER_PORT", "8081")) +PORT = int(os.getenv("PORT", str(STATSERVER_PORT + 1))) + +# Very high priority for user suggestions. The current pending maximum +# in the queue is 500 (crawler-assigned), so this always sorts on top. +SUGGEST_PRIORITY = 1_000_000 + +# Hard safety net: this UI must never run away on the huge queue table. +STATEMENT_TIMEOUT_MS = int(os.getenv("JONASSEARCH_STATEMENT_TIMEOUT_MS", "15000")) + +# Max suggestions shown in the recent list. +MAX_RECENT = 50 + +# Very defensive domain validation (ASCII only). +DOMAIN_RE = re.compile( + r"^(?=.{1,253}\.)([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+" + r"[a-z]{2,63}$" +) + +app = Flask(__name__) + + +def db(): + return psycopg2.connect( + host=DB_HOST, + port=DB_PORT, + dbname=DB_NAME, + user=DB_USER, + password=DB_PASSWORD, + connect_timeout=5, + application_name="jonassearch-priority", + options=f"-c statement_timeout={STATEMENT_TIMEOUT_MS}", + ) + + +def query(sql, params=None): + conn = db() + try: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute(sql, params) + if cur.description: + return cur.fetchall() + return [] + finally: + conn.close() + + +def clean_domain(raw): + """ + Turn arbitrary user input into a registrable domain. + + Accepts "example.com", "https://www.example.com/pfad", + "www.example.com", strips scheme, path, port, "www." prefix. + Returns lowercase domain or None if invalid. + """ + if not raw: + return None + + value = raw.strip().lower() + + # Accept full URLs too. + if "://" in value: + value = value.split("://", 1)[1] + + # Cut off path, query, port and userinfo. + value = value.split("/", 1)[0] + value = value.split("?", 1)[0] + value = value.split("#", 1)[0] + value = value.split("@", 1)[-1] + value = value.rsplit(":", 1)[0] + + # Strip a leading www. so we prioritize the whole domain. + if value.startswith("www."): + value = value[4:] + + value = value.rstrip(".") + + if not DOMAIN_RE.match(value): + return None + + return value + + +def ensure_suggestions_table(conn): + """Bookkeeping table for the UI; the queue itself stays untouched.""" + with conn.cursor() as cur: + cur.execute(""" + CREATE TABLE IF NOT EXISTS domain_suggestions ( + id BIGSERIAL PRIMARY KEY, + domain TEXT NOT NULL UNIQUE, + requested_at TIMESTAMPTZ NOT NULL DEFAULT now(), + request_count INTEGER NOT NULL DEFAULT 1, + last_status TEXT + ) + """) + conn.commit() + + +def ensure_suggestions_table_standalone(): + """Create the bookkeeping table with its own short-lived connection.""" + conn = db() + try: + ensure_suggestions_table(conn) + finally: + conn.close() + + +def suggest_domain(conn, domain): + """ + Put the domain at the very front of the crawl queue. + + Uses the passed connection (so tests can run against temp tables) + and only touches rows by their unique url index — never the whole + table. + """ + url = f"https://{domain}/" + + with conn.cursor() as cur: + # The domain itself: insert at top priority. If it already + # exists, pull it back to pending at top priority — unless a + # worker is processing it right now, in which case we leave it + # alone and let the worker finish. + cur.execute(""" + INSERT INTO crawl_queue (url, source_url, priority, status) + VALUES (%s, %s, %s, 'pending') + ON CONFLICT (url) DO UPDATE SET + priority = EXCLUDED.priority, + status = 'pending', + last_attempt_at = NULL + WHERE crawl_queue.status IS DISTINCT FROM 'processing' + """, (url, "vorschlagsverfeinerer", SUGGEST_PRIORITY)) + + # UI bookkeeping. + cur.execute(""" + INSERT INTO domain_suggestions (domain, last_status) + VALUES (%s, 'queued') + ON CONFLICT (domain) DO UPDATE SET + requested_at = now(), + request_count = domain_suggestions.request_count + 1, + last_status = 'queued' + """, (domain,)) + + conn.commit() + + +def urls_ahead_of(domain): + """ + How many pending URLs are claimed before this suggestion. + + The crawler picks: priority DESC, then discovered_at ASC, then id + ASC. All suggestions share SUGGEST_PRIORITY, so only other + suggestions submitted earlier can be ahead. + """ + row = query(""" + SELECT COUNT(*) AS ahead + FROM crawl_queue + WHERE status = 'pending' + AND ( + priority > %(P)s + OR ( + priority = %(P)s + AND (discovered_at, id) < ( + SELECT discovered_at, id + FROM crawl_queue + WHERE url = %(url)s + ) + ) + ) + """, {"P": SUGGEST_PRIORITY, "url": f"https://{domain}/"}) + + return int(row[0]["ahead"] or 0) + + +# ============================================================ +# ROUTES +# ============================================================ + +@app.get("/") +def index(): + return render_template_string(HTML, port=PORT) + + +@app.get("/api/suggest") +def api_suggest(): + """Convenience: /api/suggest?domain=example.com also works.""" + return _handle_suggest(request.args.get("domain", "")) + + +@app.post("/api/suggest") +def api_suggest_post(): + body = request.get_json(silent=True) or {} + domain_input = body.get("domain") or request.form.get("domain", "") + return _handle_suggest(domain_input) + + +def _handle_suggest(raw_domain): + domain = clean_domain(raw_domain) + + if not domain: + return jsonify({ + "ok": False, + "error": ( + "Bitte eine gültige Domain eingeben, z. B. " + "example.com oder https://www.example.com/seite" + ), + }), 400 + + conn = None + try: + conn = db() + suggest_domain(conn, domain) + except Exception as exc: + if conn is not None: + conn.rollback() + conn.close() + conn = None + + # First run without the bookkeeping table: create it and retry + # once (42P01 = undefined_table). + if getattr(exc, "pgcode", None) == "42P01": + try: + ensure_suggestions_table_standalone() + conn = db() + suggest_domain(conn, domain) + except Exception as retry_exc: + if conn is not None: + conn.rollback() + return jsonify({ + "ok": False, + "error": f"Datenbankfehler: {retry_exc}", + }), 503 + else: + return jsonify({ + "ok": False, + "error": f"Datenbankfehler: {exc}", + }), 503 + finally: + if conn is not None: + conn.close() + + try: + ahead = urls_ahead_of(domain) + except Exception: + ahead = None + + return jsonify({ + "ok": True, + "domain": domain, + "priority": SUGGEST_PRIORITY, + "urls_ahead": ahead, + }) + + +@app.get("/api/suggestions") +def api_suggestions(): + """Recent suggestions for the UI list.""" + try: + rows = query(""" + SELECT domain, + requested_at, + request_count, + last_status, + EXISTS ( + SELECT 1 FROM crawl_queue q + WHERE q.url = 'https://' || domain_suggestions.domain || '/' + AND q.status = 'pending' + ) AS still_pending + FROM domain_suggestions + ORDER BY requested_at DESC + LIMIT %s + """, (MAX_RECENT,)) + except Exception: + rows = [] + + return jsonify([ + { + "domain": r["domain"], + "requested_at": r["requested_at"].isoformat(), + "request_count": int(r["request_count"]), + "last_status": r["last_status"], + "still_pending": bool(r["still_pending"]), + } + for r in rows + ]) + + +# ============================================================ +# QUEUE SNAPSHOT (background refresh) +# ============================================================ + +# The pending/processing counts scan a 13M-row queue (several seconds +# of index-only scan). They must never run inside a request: a +# background thread refreshes them every 60 s and /api/queue serves +# the cached snapshot. The "boosted" count only reads the tiny +# priority >= SUGGEST_PRIORITY index range and is always fast. + +QUEUE_REFRESH_SECONDS = int(os.getenv("JONASSEARCH_QUEUE_REFRESH_SECONDS", "60")) + +_queue_lock = threading.Lock() +_queue_snapshot = None + + +def compute_queue_snapshot(): + """Full queue stats; may take several seconds on the big table.""" + row = query(""" + SELECT + (SELECT COUNT(*) FROM crawl_queue + WHERE status = 'pending') AS pending, + (SELECT COUNT(*) FROM crawl_queue + WHERE status = 'processing') AS processing, + (SELECT COUNT(*) FROM crawl_queue + WHERE status = 'pending' + AND priority >= %s) AS boosted + """, (SUGGEST_PRIORITY,))[0] + + return { + "pending": int(row["pending"] or 0), + "processing": int(row["processing"] or 0), + "boosted": int(row["boosted"] or 0), + } + + +def _queue_worker(): + global _queue_snapshot + + while True: + try: + snapshot = compute_queue_snapshot() + with _queue_lock: + _queue_snapshot = snapshot + time.sleep(QUEUE_REFRESH_SECONDS) + except Exception: + # DB hiccup: retry shortly; the UI keeps the last snapshot. + time.sleep(15) + + +def start_queue_worker(): + thread = threading.Thread( + target=_queue_worker, + name="queue-stats", + daemon=True, + ) + thread.start() + + +@app.get("/api/queue") +def api_queue(): + """Small live status for the header widgets (cached, see above).""" + with _queue_lock: + snapshot = _queue_snapshot + + if snapshot is not None: + return jsonify({ + **snapshot, + "online": True, + "timestamp": datetime.now(timezone.utc).isoformat(), + }) + + # No snapshot yet (service just started): fall back to the fast + # boosted-only count instead of blocking the request for seconds. + try: + row = query(""" + SELECT COUNT(*) AS boosted + FROM crawl_queue + WHERE status = 'pending' + AND priority >= %s + """, (SUGGEST_PRIORITY,))[0] + except Exception: + return jsonify({"online": False}), 503 + + return jsonify({ + "online": True, + "pending": None, + "processing": None, + "boosted": int(row["boosted"] or 0), + "timestamp": datetime.now(timezone.utc).isoformat(), + }) + + +@app.get("/api/health") +def api_health(): + try: + query("SELECT 1") + return jsonify({"status": "ok"}) + except Exception as exc: + return jsonify({"status": "error", "error": str(exc)}), 503 + + +# ============================================================ +# HTML (inline, no extra static files needed) +# ============================================================ + +HTML = """ + + + + + + JonasSearch · Vorschlagsverfeinerer + + + + + +
+
+
SERVERJONASSEARCH
+

Vorschlagsverfeinerer

+

Gamma 0.1 · Domain vorschlagen und die Warteschlange überholen

+
+ +
+ + Verbinde… +
+
+ +
+ +
+

Domain priorisieren

+

+ Eingabe als Domain oder komplette URL — z. B. + example.com oder + https://www.example.com/seite. + Der Vorschlag landet mit sehr hoher Priorität direkt vorne in der + Crawl-Queue. +

+ +
+
+ + +
+
+ +
+
+ +
+ +
+ QUEUE PENDING + — +
+ +
+ IN BEARBEITUNG + — +
+ +
+ PRIORISIERTE DOMAINS + — +
+ +
+ +
+

Letzte Vorschläge

+

Die zuletzt priorisierten Domains und ihr Queue-Status

+
+
Lade…
+
+
+ +
+ + + + + + + +""" + + +if __name__ == "__main__": + host = os.getenv("HOST", "0.0.0.0") + + # Bookkeeping table for first run. + try: + ensure_suggestions_table_standalone() + except Exception as exc: + print(f"WARN: could not ensure domain_suggestions table: {exc}") + + start_queue_worker() + + print("=" * 70) + print(" JonasSearch Vorschlagsverfeinerer Gamma 0.1") + print("=" * 70) + print(f"Statserver : Port {STATSERVER_PORT}") + print(f"HTTP : {host}:{PORT} (Statserver + 1)") + print(f"DB : {DB_HOST}:{DB_PORT}/{DB_NAME}") + print(f"Priorität : {SUGGEST_PRIORITY}") + print("=" * 70) + + app.run( + host=host, + port=PORT, + debug=False, + threaded=True, + ) diff --git a/priorityweb/jonassearch-priority.service b/priorityweb/jonassearch-priority.service new file mode 100644 index 0000000..e81f734 --- /dev/null +++ b/priorityweb/jonassearch-priority.service @@ -0,0 +1,36 @@ +# JonasSearch Vorschlagsverfeinerer (priority suggestion web UI) +# +# Installieren (als root): +# cp /srv/serverjonas/jonassearch/priorityweb/jonassearch-priority.service \ +# /etc/systemd/system/ +# systemctl daemon-reload +# systemctl enable --now jonassearch-priority.service +# +# Port: Statserver (8081) + 1 = 8082 + +[Unit] +Description=JonasSearch Vorschlagsverfeinerer (priority domain suggestions) +After=network-online.target postgresql@17-jonassearch.service +Wants=network-online.target + +[Service] +Type=simple + +User=jonassearch +Group=jonassearch + +WorkingDirectory=/srv/serverjonas/jonassearch + +EnvironmentFile=/srv/serverjonas/search/.env + +ExecStart=/usr/bin/python3 /srv/serverjonas/jonassearch/priorityweb/app.py + +Restart=on-failure +RestartSec=10 + +# Sauberer Shutdown +KillSignal=SIGTERM +TimeoutStopSec=30 + +[Install] +WantedBy=multi-user.target diff --git a/priorityweb/test_app.py b/priorityweb/test_app.py new file mode 100644 index 0000000..5b848a9 --- /dev/null +++ b/priorityweb/test_app.py @@ -0,0 +1,474 @@ +#!/usr/bin/env python3 +""" +Simple local test suite for the Vorschlagsverfeinerer. + +DB-backed tests use temp tables shadowing the real ones, so the +production crawl_queue is never touched. +""" + +import os +import sys +import unittest +from unittest import mock + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + + +def _load_env_file(): + """ + Load ../.env like the systemd EnvironmentFile does, so the tests + also run outside systemd (local manual runs). Must happen before + app.py is imported, which reads the DB variables at import time. + """ + env_path = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + os.pardir, + ".env", + ) + + try: + with open(env_path) as handle: + for line in handle: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip()) + except OSError: + pass + + +_load_env_file() + +import app as priority_app # noqa: E402 + + +class CleanDomainTest(unittest.TestCase): + + def test_plain_domain(self): + self.assertEqual( + priority_app.clean_domain("example.com"), "example.com" + ) + + def test_uppercase_and_spaces(self): + self.assertEqual( + priority_app.clean_domain(" EXAMPLE.COM "), "example.com" + ) + + def test_www_prefix(self): + self.assertEqual( + priority_app.clean_domain("www.example.com"), "example.com" + ) + + def test_full_url(self): + self.assertEqual( + priority_app.clean_domain("https://www.example.com/pfad?x=1"), + "example.com", + ) + + def test_http_url_with_port(self): + self.assertEqual( + priority_app.clean_domain("http://example.com:8080/index.html"), + "example.com", + ) + + def test_trailing_dot(self): + self.assertEqual( + priority_app.clean_domain("example.com."), "example.com" + ) + + def test_subdomain(self): + self.assertEqual( + priority_app.clean_domain("docs.example.co.uk"), + "docs.example.co.uk", + ) + + def test_invalid_empty(self): + self.assertIsNone(priority_app.clean_domain("")) + + def test_invalid_no_tld(self): + self.assertIsNone(priority_app.clean_domain("localhost")) + + def test_invalid_injection(self): + self.assertIsNone( + priority_app.clean_domain("example.com'; DROP TABLE pages; --") + ) + + def test_invalid_javascript_scheme(self): + self.assertIsNone( + priority_app.clean_domain("javascript:alert(1)") + ) + + +class QueuePriorityTest(unittest.TestCase): + """ + Test the queue-jump SQL against a real database connection whose + temp tables shadow the production tables. + """ + + def setUp(self): + self.conn = priority_app.db() + + with self.conn.cursor() as cur: + cur.execute(""" + CREATE TEMP TABLE crawl_queue ( + id BIGSERIAL PRIMARY KEY, + url TEXT NOT NULL UNIQUE, + source_url TEXT, + discovered_at TIMESTAMPTZ NOT NULL DEFAULT now(), + priority INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN + ('pending','processing','done','failed')), + attempts INTEGER NOT NULL DEFAULT 0, + last_attempt_at TIMESTAMPTZ + ) + """) + cur.execute(""" + CREATE TEMP TABLE domain_suggestions ( + id BIGSERIAL PRIMARY KEY, + domain TEXT NOT NULL UNIQUE, + requested_at TIMESTAMPTZ NOT NULL DEFAULT now(), + request_count INTEGER NOT NULL DEFAULT 1, + last_status TEXT + ) + """) + + self.conn.commit() + + def tearDown(self): + self.conn.rollback() + self.conn.close() + + def _rows(self, sql, params=None): + with self.conn.cursor() as cur: + cur.execute(sql, params) + return cur.fetchall() + + def test_suggestion_jumps_queue(self): + with self.conn.cursor() as cur: + # Simulate queue state: normal pending URLs, current max + # priority 500 like in production. + cur.execute(""" + INSERT INTO crawl_queue (url, priority, status) VALUES + ('https://a.example/', 0, 'pending'), + ('https://b.example/', 0, 'pending'), + ('https://canva.example/', 500, 'pending') + """) + self.conn.commit() + + priority_app.suggest_domain(self.conn, "cool.example") + + # The crawler's exact claim order must put the suggestion first. + order = self._rows(""" + SELECT url FROM crawl_queue + WHERE status = 'pending' + ORDER BY priority DESC, discovered_at ASC, id ASC + """) + self.assertEqual(order[0][0], "https://cool.example/") + + row = self._rows( + "SELECT priority, status FROM crawl_queue " + "WHERE url = 'https://cool.example/'" + )[0] + self.assertEqual(row, (priority_app.SUGGEST_PRIORITY, "pending")) + + # Other rows were not touched. + self.assertEqual( + self._rows( + "SELECT priority FROM crawl_queue " + "WHERE url = 'https://canva.example/'" + )[0][0], + 500, + ) + + # Suggestion bookkeeping was written. + self.assertEqual( + self._rows("SELECT domain FROM domain_suggestions"), + [("cool.example",)], + ) + + def test_existing_failed_url_is_requeued(self): + with self.conn.cursor() as cur: + cur.execute(""" + INSERT INTO crawl_queue (url, priority, status) VALUES + ('https://old.example/', 0, 'failed'), + ('https://x.example/', 0, 'pending') + """) + self.conn.commit() + + priority_app.suggest_domain(self.conn, "old.example") + + row = self._rows( + "SELECT priority, status FROM crawl_queue " + "WHERE url = 'https://old.example/'" + )[0] + self.assertEqual(row, (priority_app.SUGGEST_PRIORITY, "pending")) + + def test_processing_row_is_left_alone(self): + with self.conn.cursor() as cur: + cur.execute(""" + INSERT INTO crawl_queue (url, priority, status) VALUES + ('https://busy.example/', 123, 'processing') + """) + self.conn.commit() + + priority_app.suggest_domain(self.conn, "busy.example") + + # A worker is crawling it right now: do not interfere. + row = self._rows( + "SELECT priority, status FROM crawl_queue " + "WHERE url = 'https://busy.example/'" + )[0] + self.assertEqual(row, (123, "processing")) + + # But it is still recorded as a suggestion. + self.assertEqual( + self._rows( + "SELECT request_count FROM domain_suggestions " + "WHERE domain = 'busy.example'" + )[0][0], + 1, + ) + + def test_multiple_suggestions_crawl_in_order(self): + priority_app.suggest_domain(self.conn, "first.example") + priority_app.suggest_domain(self.conn, "second.example") + + order = self._rows(""" + SELECT url FROM crawl_queue + WHERE status = 'pending' + ORDER BY priority DESC, discovered_at ASC, id ASC + """) + + self.assertEqual( + [row[0] for row in order[:2]], + ["https://first.example/", "https://second.example/"], + ) + + +class UrlsAheadTest(unittest.TestCase): + """urls_ahead_of must count only earlier suggestions.""" + + def setUp(self): + self.conn = priority_app.db() + + with self.conn.cursor() as cur: + cur.execute(""" + CREATE TEMP TABLE crawl_queue ( + id BIGSERIAL PRIMARY KEY, + url TEXT NOT NULL UNIQUE, + source_url TEXT, + discovered_at TIMESTAMPTZ NOT NULL DEFAULT now(), + priority INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, + last_attempt_at TIMESTAMPTZ + ) + """) + cur.execute(""" + INSERT INTO crawl_queue (url, priority, status) VALUES + ('https://earlier.example/', 1000000, 'pending'), + ('https://me.example/', 1000000, 'pending'), + ('https://later.example/', 1000000, 'pending'), + ('https://normal.example/', 0, 'pending'), + ('https://gone.example/', 1000000, 'done') + """) + # Force a strict discovered_at ordering. + cur.execute(""" + UPDATE crawl_queue + SET discovered_at = now() - ( + CASE url + WHEN 'https://earlier.example/' THEN interval '30 s' + WHEN 'https://me.example/' THEN interval '20 s' + ELSE interval '10 s' + END + ) + WHERE url LIKE 'https://%.example/' + """) + + self.conn.commit() + + def tearDown(self): + self.conn.rollback() + self.conn.close() + + def test_only_earlier_suggestions_count(self): + def fake_query(sql, params=None): + with self.conn.cursor() as cur: + cur.execute(sql, params) + rows = cur.fetchall() + cols = [d.name for d in cur.description] + return [dict(zip(cols, row)) for row in rows] + + with mock.patch.object(priority_app, "query", fake_query): + ahead = priority_app.urls_ahead_of("me.example") + + # Only 'earlier.example' is pending, same priority, older. + self.assertEqual(ahead, 1) + + +class ApiTest(unittest.TestCase): + """HTTP-layer tests with mocked database access.""" + + def setUp(self): + priority_app.app.config["TESTING"] = True + self.client = priority_app.app.test_client() + + def test_index_serves_html(self): + response = self.client.get("/") + self.assertEqual(response.status_code, 200) + self.assertIn(b"Vorschlagsverfeinerer", response.data) + self.assertIn(b"domainInput", response.data) + + def test_suggest_valid_domain(self): + with mock.patch.object(priority_app, "db"), \ + mock.patch.object(priority_app, "suggest_domain"), \ + mock.patch.object(priority_app, "urls_ahead_of", + return_value=2): + response = self.client.post( + "/api/suggest", json={"domain": "example.com"} + ) + + self.assertEqual(response.status_code, 200) + data = response.get_json() + self.assertTrue(data["ok"]) + self.assertEqual(data["domain"], "example.com") + self.assertEqual(data["urls_ahead"], 2) + self.assertEqual(data["priority"], priority_app.SUGGEST_PRIORITY) + + def test_suggest_invalid_domain(self): + response = self.client.post( + "/api/suggest", json={"domain": "not a domain!"} + ) + + self.assertEqual(response.status_code, 400) + self.assertFalse(response.get_json()["ok"]) + + def test_suggest_get_variant(self): + with mock.patch.object(priority_app, "db"), \ + mock.patch.object(priority_app, "suggest_domain"), \ + mock.patch.object(priority_app, "urls_ahead_of", + side_effect=RuntimeError): + response = self.client.get("/api/suggest?domain=example.org") + + self.assertEqual(response.status_code, 200) + data = response.get_json() + self.assertTrue(data["ok"]) + self.assertEqual(data["urls_ahead"], None) + + def test_suggest_creates_missing_table_and_retries(self): + """ + First run without domain_suggestions: the insert fails with + 42P01, the table gets created, and the suggestion succeeds. + """ + calls = {"suggest": 0} + + class FakeUndefinedTable(Exception): + pgcode = "42P01" + + def fake_db(): + return mock.MagicMock() + + def fake_suggest(conn, domain): + calls["suggest"] += 1 + if calls["suggest"] == 1: + raise FakeUndefinedTable( + "relation \"domain_suggestions\" does not exist" + ) + + with mock.patch.object(priority_app, "db", side_effect=fake_db), \ + mock.patch.object(priority_app, "suggest_domain", + side_effect=fake_suggest), \ + mock.patch.object(priority_app, + "ensure_suggestions_table_standalone") as \ + ensure_mock, \ + mock.patch.object(priority_app, "urls_ahead_of", return_value=0): + response = self.client.post( + "/api/suggest", json={"domain": "example.com"} + ) + + self.assertEqual(response.status_code, 200) + self.assertTrue(response.get_json()["ok"]) + self.assertEqual(calls["suggest"], 2) + ensure_mock.assert_called_once() + + def test_suggest_db_error_returns_503(self): + with mock.patch.object(priority_app, "db", + side_effect=RuntimeError("db down")): + response = self.client.post( + "/api/suggest", json={"domain": "example.com"} + ) + + self.assertEqual(response.status_code, 503) + self.assertFalse(response.get_json()["ok"]) + + def test_queue_endpoint_serves_snapshot(self): + snapshot = {"pending": 3, "processing": 1, "boosted": 2} + with mock.patch.object(priority_app, "_queue_lock"): + priority_app._queue_snapshot = dict(snapshot) + response = self.client.get("/api/queue") + + self.assertEqual(response.status_code, 200) + data = response.get_json() + self.assertTrue(data["online"]) + self.assertEqual(data["pending"], 3) + self.assertEqual(data["processing"], 1) + self.assertEqual(data["boosted"], 2) + + def test_queue_endpoint_fallback_counts_boosted_only(self): + old = priority_app._queue_snapshot + priority_app._queue_snapshot = None + try: + with mock.patch.object( + priority_app, "query", + return_value=[{"boosted": 7}], + ): + response = self.client.get("/api/queue") + finally: + priority_app._queue_snapshot = old + + self.assertEqual(response.status_code, 200) + data = response.get_json() + self.assertTrue(data["online"]) + self.assertIsNone(data["pending"]) + self.assertEqual(data["boosted"], 7) + + def test_queue_endpoint_db_down(self): + old = priority_app._queue_snapshot + priority_app._queue_snapshot = None + try: + with mock.patch.object(priority_app, "query", + side_effect=RuntimeError("no db")): + response = self.client.get("/api/queue") + finally: + priority_app._queue_snapshot = old + + self.assertEqual(response.status_code, 503) + self.assertFalse(response.get_json()["online"]) + + def test_suggestions_endpoint_empty(self): + with mock.patch.object(priority_app, "query", + side_effect=RuntimeError("no table")): + response = self.client.get("/api/suggestions") + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.get_json(), []) + + def test_health_ok(self): + with mock.patch.object(priority_app, "query", return_value=[]): + response = self.client.get("/api/health") + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.get_json()["status"], "ok") + + def test_health_db_down(self): + with mock.patch.object(priority_app, "query", + side_effect=RuntimeError("down")): + response = self.client.get("/api/health") + + self.assertEqual(response.status_code, 503) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/web/app.py b/web/app.py index 8fec781..0c88876 100644 --- a/web/app.py +++ b/web/app.py @@ -10,7 +10,7 @@ from pathlib import Path from flask import Flask, jsonify, render_template_string, Response DB_DATA = Path(__file__).parent / "page_counts.json" -INTERVAL = 15 * 60 # 15 Minuten +INTERVAL = 30 * 60 # 30 Minuten app = Flask(__name__)