161 lines
8.2 KiB
Python
161 lines
8.2 KiB
Python
"""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)
|