add route moderation script

test/python/test_moderate_routes.py

@@ -0,0 +1,106 @@
+"""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_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_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_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)