#!/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_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)