"""Tests für lokale Freigaben mit echter SQLite-Datenbank.""" from contextlib import redirect_stdout, redirect_stderr from datetime import datetime, timezone import io from pathlib import Path import sqlite3 import sys import tempfile import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT / "tools" / "python")) import moderate_routes class ModerationTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) self.environment = patch.dict("os.environ", {}, clear=True) self.environment.start() self.addCleanup(self.environment.stop) self.path = self.root / "data" / "wegwichtel.sqlite" self.path.parent.mkdir() with sqlite3.connect(self.path) as database: database.executescript((ROOT / "database" / "schema.sql").read_text()) database.execute("INSERT INTO routes (name, slug, gpx_path) VALUES (?, ?, ?)", ("Teststrecke", "test", "test.gpx")) database.execute("INSERT INTO routes (name, slug, gpx_path, status) VALUES (?, ?, ?, ?)", ("Gelöschte Strecke", "deleted", "deleted.gpx", "deleted")) self.out = io.StringIO() self.error = io.StringIO() def run_cli(self, args): with redirect_stdout(self.out), redirect_stderr(self.error): return moderate_routes.main(args, self.root) def activation(self, identifier=1): with sqlite3.connect(self.path) as database: return database.execute("SELECT activated_at FROM routes WHERE id = ?", (identifier,)).fetchone()[0] def test_listing_enable_and_disable(self): self.assertEqual(self.run_cli([]), 0) self.assertIn("1\tactive\tnicht freigegeben\tTeststrecke", self.out.getvalue()) before = datetime.now(timezone.utc) self.assertEqual(self.run_cli(["-e", "1"]), 0) stamp = self.activation() self.assertTrue(stamp.endswith("Z")) self.assertGreaterEqual(datetime.fromisoformat(stamp.replace("Z", "+00:00")), before.replace(microsecond=before.microsecond // 1000 * 1000)) self.assertEqual(self.run_cli([]), 0) self.assertIn(stamp, self.out.getvalue()) self.assertEqual(self.run_cli(["--disable", "1"]), 0) self.assertIsNone(self.activation()) self.assertIsNone(self.activation(2)) def test_creator_id_is_shown_in_listing(self): identity = 'a' * 64 with sqlite3.connect(self.path) as database: database.execute('UPDATE routes SET creator_id=? WHERE id=1', (identity,)) self.assertEqual(self.run_cli([]), 0) self.assertIn('Ersteller-ID', self.out.getvalue()) self.assertIn(identity, self.out.getvalue()) def test_aliases(self): for flag in ["--enable", "--enanble"]: self.assertEqual(self.run_cli([flag, "1"]), 0) self.assertIsNotNone(self.activation()) self.assertEqual(self.run_cli(["-d", "1"]), 0) self.assertIsNone(self.activation()) def test_missing_and_deleted_routes_do_not_change_database(self): self.assertEqual(self.run_cli(["-e", "999"]), 1) self.assertEqual(self.run_cli(["-e", "2"]), 1) self.assertIsNone(self.activation()) self.assertIsNone(self.activation(2)) self.assertIn("existiert nicht", self.error.getvalue()) self.assertIn("gelöscht", self.error.getvalue()) def test_env_and_environment_precedence(self): (self.root / ".env").write_text('# Kommentar\nexport DATA_DIR="anderer Ordner" # Kommentar\n', encoding="utf-8") self.assertEqual(moderate_routes.database_path(self.root), self.root / "anderer Ordner" / "wegwichtel.sqlite") with patch.dict("os.environ", {"DATA_DIR": str(self.path.parent)}): self.assertEqual(self.run_cli([]), 0) (self.root / ".env").write_text("DATA_DIR=\n", encoding="utf-8") self.assertEqual(moderate_routes.database_path(self.root), self.path) def test_missing_database_is_not_created(self): self.path.unlink() for args in [[], ["-e", "1"]]: self.assertEqual(self.run_cli(args), 1) self.assertFalse(self.path.exists()) def test_export_prefix_accepts_whitespace_and_preserves_variable_names(self): env_file = self.root / ".env" for prefix in ["export ", " export\t", "\texport ", ""]: env_file.write_text(f'{prefix}DATA_DIR="anderer Ordner"\n', encoding="utf-8") self.assertEqual(moderate_routes.database_path(self.root), self.root / "anderer Ordner" / "wegwichtel.sqlite") env_file.write_text("exportDATA_DIR=anders\nexport_path=anders\n", encoding="utf-8") self.assertEqual(moderate_routes.database_path(self.root), self.path) def test_env_read_errors_are_reported_without_changing_database(self): with patch.object(Path, "read_text", side_effect=PermissionError("Lesen nicht erlaubt")): self.assertEqual(self.run_cli(["-e", "1"]), 1) self.assertIsNone(self.activation()) self.assertIn("Lesen nicht erlaubt", self.error.getvalue()) def test_bad_env_reports_line_without_secrets(self): for value in ["secret-without-equals", 'DATA_DIR="secret']: (self.root / ".env").write_text(value, encoding="utf-8") self.assertEqual(self.run_cli([]), 1) self.assertNotIn("secret", self.error.getvalue()) self.assertIn("Zeile 1", self.error.getvalue()) def test_empty_database(self): with sqlite3.connect(self.path) as database: database.execute("DELETE FROM routes") self.assertEqual(self.run_cli([]), 0) self.assertIn("Keine Strecken", self.out.getvalue()) def test_station_table_has_ordered_media_links_without_database_writes(self): with sqlite3.connect(self.path) as database: database.execute("INSERT INTO pois (route_id, title, description, sequence, audio_path) VALUES (1, 'Ton', 'Erste\nStation', 9, 'active/routes/1/audio/a.mp3')") database.execute("INSERT INTO pois (route_id, title, description, sequence) VALUES (1, 'Bild', 'Zweite', 3)") database.execute("INSERT INTO poi_images (poi_id, path) VALUES (2, 'active/routes/1/images/a.png')") database.execute("INSERT INTO pois (route_id, title, sequence) VALUES (1, 'Ohne Medien', 12)") database.execute("INSERT INTO pois (route_id, title, audio_path) VALUES (2, 'Gelöscht', 'trash/routes/2/audio/a.mp3')") (self.root / '.env').write_text('MODERATION_REVIEW_URL=https://wegwichtel.de/moderation/api/\n') before = self.path.read_bytes() self.assertEqual(self.run_cli([]), 0) output = self.out.getvalue() self.assertIn('Beschreibung', output) self.assertIn('Erste Station', output) self.assertRegex(output, r'2\s+\| 1\s+\| Zweite') self.assertIn('https://wegwichtel.de/moderation/api/routes/1/pois/2/pictures/1', output) self.assertIn('https://wegwichtel.de/moderation/api/routes/1/pois/1/audio', output) self.assertNotIn('/routes/2/pois/', output) self.assertEqual(self.path.read_bytes(), before) self.out.seek(0) self.out.truncate() self.assertEqual(self.run_cli([]), 0) self.assertEqual(output, self.out.getvalue()) def test_invalid_review_configuration(self): for setting in ['MODERATION_REVIEW_URL=http://example.com/review', 'MODERATION_REVIEW_URL=https://example.com/review?token=x', 'MODERATION_REVIEW_URL=https://user:password@example.com/review', 'MODERATION_REVIEW_URL="https://example.com/review#fragment"']: (self.root / '.env').write_text(setting) self.assertEqual(self.run_cli([]), 1) def test_invalid_arguments(self): for args in [["-e", "0"], ["-d", "abc"], ["-e", "1", "-d", "1"], ["-e"]]: with redirect_stderr(self.error), self.assertRaises(SystemExit) as result: moderate_routes.main(args, self.root) self.assertEqual(result.exception.code, 2)