360 lines
16 KiB
Python
360 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Tests für den Statserver (web/app.py), ohne Datenbank.
|
||
|
||
Geprüft werden die DB-freien Teile: Parsen der psql-Zeile, der
|
||
2-Stunden-Rhythmus für den exakten Bild-Count, die Wahl zwischen
|
||
exakter Zählung und Stichprobe sowie die Einbindung der neuen
|
||
Images-Anzeige in die HTML-Seite.
|
||
"""
|
||
|
||
import importlib.util
|
||
import os
|
||
import subprocess
|
||
import unittest
|
||
from datetime import datetime, timedelta
|
||
from unittest import mock
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
|
||
# web/app.py unter einem eindeutigen Modulnamen laden, damit es nicht mit
|
||
# priorityweb/app.py (gleicher Dateiname) kollidiert.
|
||
_spec = importlib.util.spec_from_file_location(
|
||
"jonassearch_stats_app", os.path.join(HERE, "app.py")
|
||
)
|
||
stats_app = importlib.util.module_from_spec(_spec)
|
||
_spec.loader.exec_module(stats_app)
|
||
|
||
|
||
class ParseStatsRowTest(unittest.TestCase):
|
||
|
||
def test_count_and_images(self):
|
||
self.assertEqual(
|
||
stats_app._parse_stats_row("981981|12345"), (981981, 12345)
|
||
)
|
||
|
||
def test_empty_sample_is_none(self):
|
||
self.assertEqual(stats_app._parse_stats_row("10|"), (10, None))
|
||
self.assertEqual(stats_app._parse_stats_row("10|\n"), (10, None))
|
||
|
||
|
||
class ImagesExactDueTest(unittest.TestCase):
|
||
|
||
NOW = datetime(2026, 10, 17, 12, 0, 0)
|
||
|
||
def _entry(self, age, exact):
|
||
return {
|
||
"timestamp": (self.NOW - age).isoformat(),
|
||
"count": 1,
|
||
"images": 2,
|
||
"images_exact": exact,
|
||
}
|
||
|
||
def test_no_data_means_due(self):
|
||
self.assertTrue(stats_app.images_exact_due([], now=self.NOW))
|
||
|
||
def test_recent_exact_is_not_due(self):
|
||
data = [self._entry(timedelta(hours=1), True)]
|
||
self.assertFalse(stats_app.images_exact_due(data, now=self.NOW))
|
||
|
||
def test_two_hours_old_is_due(self):
|
||
data = [self._entry(timedelta(hours=2), True)]
|
||
self.assertTrue(stats_app.images_exact_due(data, now=self.NOW))
|
||
|
||
def test_estimates_in_between_do_not_reset_schedule(self):
|
||
data = [
|
||
self._entry(timedelta(hours=1, minutes=30), True),
|
||
self._entry(timedelta(minutes=30), False),
|
||
]
|
||
self.assertFalse(stats_app.images_exact_due(data, now=self.NOW))
|
||
|
||
def test_old_entries_without_flag_are_ignored(self):
|
||
data = [{"timestamp": self.NOW.isoformat(), "count": 1}]
|
||
self.assertTrue(stats_app.images_exact_due(data, now=self.NOW))
|
||
|
||
def test_malformed_exact_timestamp_is_ignored(self):
|
||
data = [{"timestamp": "kaputt", "images_exact": True}]
|
||
self.assertTrue(stats_app.images_exact_due(data, now=self.NOW))
|
||
|
||
|
||
class GetStatsTest(unittest.TestCase):
|
||
|
||
@mock.patch.object(stats_app, "_psql_scalar")
|
||
def test_exact_query(self, psql):
|
||
psql.return_value = "100|250"
|
||
self.assertEqual(stats_app.get_stats(True), (100, 250, True))
|
||
self.assertEqual(psql.call_args[0][0], stats_app.EXACT_IMAGE_SQL)
|
||
self.assertNotIn("TABLESAMPLE", stats_app.EXACT_IMAGE_SQL)
|
||
|
||
@mock.patch.object(stats_app, "_psql_scalar")
|
||
def test_estimate_query(self, psql):
|
||
psql.return_value = "100|250"
|
||
self.assertEqual(stats_app.get_stats(False), (100, 250, False))
|
||
self.assertEqual(psql.call_args[0][0], stats_app.ESTIMATE_IMAGE_SQL)
|
||
self.assertIn("TABLESAMPLE SYSTEM (1)", stats_app.ESTIMATE_IMAGE_SQL)
|
||
|
||
@mock.patch.object(stats_app, "_psql_scalar")
|
||
def test_empty_sample_on_small_table_falls_back_to_exact(self, psql):
|
||
psql.side_effect = ["10|", "10|20"]
|
||
self.assertEqual(stats_app.get_stats(False), (10, 20, True))
|
||
self.assertEqual(psql.call_count, 2)
|
||
self.assertEqual(psql.call_args_list[1][0][0],
|
||
stats_app.EXACT_IMAGE_SQL)
|
||
|
||
@mock.patch.object(stats_app, "_psql_scalar")
|
||
def test_empty_sample_on_large_table_stays_unknown(self, psql):
|
||
psql.return_value = "200000|"
|
||
self.assertEqual(stats_app.get_stats(False), (200000, None, False))
|
||
|
||
@mock.patch.object(stats_app, "_psql_scalar")
|
||
def test_missing_pictures_column_falls_back_to_page_count(self, psql):
|
||
error = subprocess.CalledProcessError(
|
||
1, ["psql"], stderr='ERROR: column "pictures" does not exist'
|
||
)
|
||
psql.side_effect = [error, "10"]
|
||
self.assertEqual(stats_app.get_stats(False), (10, None, False))
|
||
self.assertEqual(psql.call_args_list[1][0][0],
|
||
"SELECT COUNT(*) FROM pages;")
|
||
|
||
|
||
class GetPendingTest(unittest.TestCase):
|
||
|
||
@mock.patch.object(stats_app, "_psql_scalar")
|
||
def test_counts_pending_queue_rows(self, psql):
|
||
psql.return_value = "1384711"
|
||
self.assertEqual(stats_app.get_pending(), 1384711)
|
||
self.assertIn("FROM crawl_queue", stats_app.PENDING_QUEUE_SQL)
|
||
self.assertIn("status = 'pending'", stats_app.PENDING_QUEUE_SQL)
|
||
|
||
@mock.patch.object(stats_app, "_psql_scalar")
|
||
def test_unavailable_queue_returns_none(self, psql):
|
||
psql.side_effect = subprocess.CalledProcessError(
|
||
1, ["psql"], stderr='ERROR: relation "crawl_queue" does not exist'
|
||
)
|
||
self.assertIsNone(stats_app.get_pending())
|
||
|
||
@mock.patch.object(stats_app, "_psql_scalar")
|
||
def test_unexpected_output_returns_none(self, psql):
|
||
psql.return_value = ""
|
||
self.assertIsNone(stats_app.get_pending())
|
||
|
||
|
||
class CollectTest(unittest.TestCase):
|
||
|
||
def _run_collect(self, get_stats, get_pending):
|
||
with mock.patch.object(stats_app, "load_data", return_value=[]), \
|
||
mock.patch.object(stats_app, "save_data") as save_data, \
|
||
mock.patch.object(stats_app, "get_stats", get_stats), \
|
||
mock.patch.object(stats_app, "get_pending", get_pending), \
|
||
mock.patch.object(stats_app, "time") as time_mock:
|
||
time_mock.sleep.side_effect = KeyboardInterrupt
|
||
with self.assertRaises(KeyboardInterrupt):
|
||
stats_app.collect()
|
||
return save_data.call_args[0][0][0]
|
||
|
||
def test_pending_is_stored_with_every_sample(self):
|
||
entry = self._run_collect(
|
||
mock.Mock(return_value=(1000, 2000, True)),
|
||
mock.Mock(return_value=4711),
|
||
)
|
||
self.assertEqual(entry["pending"], 4711)
|
||
self.assertEqual(entry["count"], 1000)
|
||
self.assertEqual(entry["images"], 2000)
|
||
|
||
def test_missing_queue_does_not_break_the_sample(self):
|
||
entry = self._run_collect(
|
||
mock.Mock(return_value=(1000, 2000, False)),
|
||
mock.Mock(return_value=None),
|
||
)
|
||
self.assertIsNone(entry["pending"])
|
||
self.assertEqual(entry["count"], 1000)
|
||
|
||
|
||
class StatsPageTest(unittest.TestCase):
|
||
|
||
def test_page_contains_image_counter(self):
|
||
client = stats_app.app.test_client()
|
||
response = client.get("/")
|
||
self.assertEqual(response.status_code, 200)
|
||
html = response.get_data(as_text=True)
|
||
self.assertIn('id="current-images"', html)
|
||
self.assertIn('id="current-images-meta"', html)
|
||
self.assertIn("Estimated · exact every 2 h", html)
|
||
self.assertIn("imagesExact", html)
|
||
|
||
def test_page_contains_6h_range_and_card(self):
|
||
client = stats_app.app.test_client()
|
||
response = client.get("/")
|
||
self.assertEqual(response.status_code, 200)
|
||
html = response.get_data(as_text=True)
|
||
self.assertIn('data-range="6h"', html)
|
||
self.assertIn('id="stat-delta-6h"', html)
|
||
self.assertIn('id="stat-delta-6h-hint"', html)
|
||
self.assertIn("'6h': 6 * 3600000", html)
|
||
self.assertIn("closestBefore(all, last.x - 6 * 3600000", html)
|
||
|
||
def test_page_contains_all_image_graphs(self):
|
||
client = stats_app.app.test_client()
|
||
response = client.get("/")
|
||
self.assertEqual(response.status_code, 200)
|
||
html = response.get_data(as_text=True)
|
||
|
||
self.assertIn("Images – all graphs", html)
|
||
for canvas in (
|
||
"imagesMainChart",
|
||
"imagesDeltaChart",
|
||
"imagesCumChart",
|
||
"imagesHourlyChart",
|
||
"imagesHodChart",
|
||
):
|
||
self.assertIn('id="' + canvas + '"', html)
|
||
self.assertIn('id="records-images"', html)
|
||
self.assertIn('id="heatmap-images"', html)
|
||
self.assertIn('id="images-empty"', html)
|
||
|
||
# Image estimates are excluded by default and can be included with the persistent toggle.
|
||
self.assertIn('id="show-estimated-images"', html)
|
||
self.assertIn('type="checkbox" id="show-estimated-images"', html)
|
||
self.assertIn("Include estimated image counts in graphs", html)
|
||
self.assertIn("function toImagePoints(points, includeEstimated)", html)
|
||
self.assertIn("p.images !== null && p.images !== undefined", html)
|
||
self.assertIn("includeEstimated || p.imagesExact", html)
|
||
self.assertIn("localStorage.getItem('js-show-estimated-images')", html)
|
||
self.assertIn("localStorage.setItem('js-show-estimated-images'", html)
|
||
self.assertIn("toImagePoints(all, showEstimatedImages)", html)
|
||
self.assertIn("Only exact measurements (counted every 2 h) are plotted", html)
|
||
self.assertIn("Exact and estimated measurements are plotted", html)
|
||
self.assertIn("Only estimated image counts so far", html)
|
||
self.assertIn("No image counts in the selected range", html)
|
||
# The headline counter stays the latest collected value; only graphs toggle estimates.
|
||
self.assertIn("elImages.textContent = fmt(last.images)", html)
|
||
# Images estimates affect graphs only; other series keep their own points.
|
||
self.assertIn("renderCharts({\n prefix: '',\n points: fpoints", html)
|
||
self.assertIn("renderCharts({\n prefix: 'pending',\n points: pendingFpoints", html)
|
||
self.assertIn("renderHeatmap(imageAll, 'heatmap-images', 'Images')", html)
|
||
self.assertIn("renderRecords(imageDeltas, 'records-images', 'Images')", html)
|
||
# existing page chart ids must stay lowercase (mainChart, not MainChart)
|
||
self.assertIn("function chartId(prefix, suffix)", html)
|
||
self.assertIn("chartId(prefix, 'MainChart')", html)
|
||
# both series use the same time window (anchored to the latest page sample)
|
||
self.assertIn("filterRange(imageAll, last.x)", html)
|
||
|
||
# both metrics go through the same renderers, with the right units
|
||
self.assertIn("unit: 'Images'", html)
|
||
self.assertIn("unit: 'Pages'", html)
|
||
self.assertIn("renderRecords(deltas, 'records', 'Pages')", html)
|
||
self.assertIn("renderRecords(imageDeltas, 'records-images', 'Images')", html)
|
||
self.assertIn("renderHeatmap(all, 'heatmap', 'Pages')", html)
|
||
self.assertIn("renderHeatmap(imageAll, 'heatmap-images', 'Images')", html)
|
||
|
||
def test_page_has_metric_tabs_with_one_panel_each(self):
|
||
client = stats_app.app.test_client()
|
||
response = client.get("/")
|
||
self.assertEqual(response.status_code, 200)
|
||
html = response.get_data(as_text=True)
|
||
|
||
# three equally styled tab buttons (pages / images / queue) with icons
|
||
self.assertIn('class="metric-btn', html)
|
||
for metric in ("pages", "images", "queue"):
|
||
self.assertIn('data-metric="' + metric + '"', html)
|
||
self.assertIn("metric-icon", html)
|
||
self.assertIn("📄", html)
|
||
self.assertIn("🖼️", html)
|
||
self.assertIn("⏳", html)
|
||
|
||
# every metric has its own panel; only the initial one is active
|
||
self.assertEqual(html.count('class="metric-panel active" data-metric="pages"'), 1)
|
||
for metric in ("images", "queue"):
|
||
self.assertIn('class="metric-panel" data-metric="' + metric + '"', html)
|
||
self.assertIn('id="summary-images"', html)
|
||
self.assertIn('id="summary-queue"', html)
|
||
|
||
# switching tabs hides the other panels and is remembered
|
||
self.assertIn("function setMetric(next, persist)", html)
|
||
self.assertIn("panel.classList.toggle('active', panel.dataset.metric === activeMetric)", html)
|
||
self.assertIn("localStorage.setItem('js-metric', activeMetric)", html)
|
||
self.assertIn("localStorage.getItem('js-metric')", html)
|
||
|
||
# hidden panels must not create Chart.js instances (canvas has 0 px)
|
||
self.assertIn("function isVisible(el)", html)
|
||
self.assertIn("if (!target || !isVisible(target)) return;", html)
|
||
|
||
# summary cards are generated for the non-page tabs
|
||
self.assertIn("function renderSummary(containerId, all, points)", html)
|
||
self.assertIn("renderSummary('summary-images', imageAll, imageFpoints)", html)
|
||
self.assertIn("renderSummary('summary-queue', pendingAll, pendingFpoints)", html)
|
||
|
||
# exports follow the selected tab
|
||
self.assertIn("function exportSeries()", html)
|
||
self.assertIn("return { name: 'jonassearch_images'", html)
|
||
self.assertIn("return { name: 'jonassearch_queue'", html)
|
||
|
||
def test_page_has_target_eta_field_for_active_tab(self):
|
||
client = stats_app.app.test_client()
|
||
response = client.get("/")
|
||
self.assertEqual(response.status_code, 200)
|
||
html = response.get_data(as_text=True)
|
||
|
||
# a plain number field at the top, plus the computed duration
|
||
self.assertIn('id="goal-input"', html)
|
||
self.assertIn('type="number" inputmode="numeric"', html)
|
||
self.assertIn('id="goal-eta"', html)
|
||
self.assertIn('id="goal-hint"', html)
|
||
self.assertIn('id="goal-unit"', html)
|
||
self.assertIn("Reach target", html)
|
||
|
||
# the estimate uses the trend of the selected range (last 2 h first)
|
||
self.assertIn("function recentTrend(points)", html)
|
||
self.assertIn("const reg = recentTrend(fpoints);", html)
|
||
self.assertIn("function fmtEta(ms)", html)
|
||
self.assertIn("const speed = trend ? trend.slope : null;", html)
|
||
self.assertIn("const etaMs = remaining / speed * 3600000;", html)
|
||
self.assertIn("if (remaining === 0) {", html)
|
||
self.assertIn("Moving away from the target", html)
|
||
|
||
# one target per tab, restored on tab switch and persisted per browser
|
||
self.assertIn("function goalSeries()", html)
|
||
self.assertIn("goalTargets[activeMetric] || ''", html)
|
||
self.assertIn("localStorage.setItem('js-goal-' + activeMetric, raw)", html)
|
||
self.assertIn("localStorage.getItem('js-goal-' + metric)", html)
|
||
self.assertIn("toImagePoints(all, showEstimatedImages)", html)
|
||
|
||
# recomputed on refresh, tab switch and range switch
|
||
self.assertIn("renderSummary('summary-queue', pendingAll, pendingFpoints);\n renderGoal();", html)
|
||
self.assertIn("goalInput.value = goalTargets[activeMetric] || '';\n renderGoal();", html)
|
||
|
||
def test_page_contains_pending_queue_counter_and_graphs(self):
|
||
client = stats_app.app.test_client()
|
||
response = client.get("/")
|
||
self.assertEqual(response.status_code, 200)
|
||
html = response.get_data(as_text=True)
|
||
|
||
self.assertIn('id="current-pending"', html)
|
||
self.assertIn('id="current-pending-meta"', html)
|
||
self.assertIn("Queue – all graphs", html)
|
||
for canvas in (
|
||
"pendingMainChart",
|
||
"pendingDeltaChart",
|
||
"pendingCumChart",
|
||
"pendingHourlyChart",
|
||
"pendingHodChart",
|
||
):
|
||
self.assertIn('id="' + canvas + '"', html)
|
||
self.assertIn('id="records-pending"', html)
|
||
self.assertIn('id="heatmap-pending"', html)
|
||
self.assertIn('id="pending-empty"', html)
|
||
|
||
# pending values are parsed and plotted without sampling/filtering
|
||
self.assertIn("function toPendingPoints(points)", html)
|
||
self.assertIn("p.pending !== null && p.pending !== undefined", html)
|
||
self.assertIn("filterRange(pendingAll, last.x)", html)
|
||
self.assertIn("unit: 'Pending'", html)
|
||
self.assertIn("renderRecords(pendingDeltas, 'records-pending', 'Pending')", html)
|
||
self.assertIn("renderHeatmap(pendingAll, 'heatmap-pending', 'Pending')", html)
|
||
self.assertIn("No queue counts yet", html)
|
||
self.assertIn("No queue counts in the selected range.", html)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|