chore: standardize security and quality foundations

backend/tests/test_api_models.py

@@ -0,0 +1,24 @@
+import unittest
+
+from pydantic import ValidationError
+
+from backend.app.api_models import PasswordResetRequest, SignupRequest
+
+
+class ApiRequestModelTests(unittest.TestCase):
+    def test_signup_rejects_unknown_fields(self) -> None:
+        with self.assertRaises(ValidationError):
+            SignupRequest(
+                invite_code="invite",
+                username="viewer",
+                password="strong password",
+                unexpected="value",
+            )
+
+    def test_password_reset_preserves_password_whitespace_for_policy_validation(self) -> None:
+        request = PasswordResetRequest(token="token", new_password=" leading and trailing ")
+        self.assertEqual(request.new_password, " leading and trailing ")
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_arr_helpers.py

@@ -0,0 +1,24 @@
+import unittest
+
+from backend.app.services.arr import RootFolderNotFoundError, resolve_root_folder_path
+
+
+class _ArrClient:
+    async def get_root_folders(self):
+        return [{"id": 7, "path": "/media/tv"}]
+
+
+class ArrHelperTests(unittest.IsolatedAsyncioTestCase):
+    async def test_resolves_numeric_root_folder_id(self) -> None:
+        self.assertEqual(await resolve_root_folder_path(_ArrClient(), "7", "Sonarr"), "/media/tv")
+
+    async def test_preserves_configured_path(self) -> None:
+        self.assertEqual(await resolve_root_folder_path(_ArrClient(), "/media/movies", "Radarr"), "/media/movies")
+
+    async def test_rejects_missing_root_folder_id(self) -> None:
+        with self.assertRaises(RootFolderNotFoundError):
+            await resolve_root_folder_path(_ArrClient(), "8", "Sonarr")
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_logging_config.py

@@ -0,0 +1,25 @@
+import json
+import logging
+import unittest
+
+from backend.app.logging_config import JsonLogFormatter, RequestContextFilter, bind_request_id, reset_request_id
+
+
+class JsonLoggingTests(unittest.TestCase):
+    def test_json_formatter_includes_request_context(self) -> None:
+        token = bind_request_id("request-123")
+        try:
+            record = logging.LogRecord("magent.test", logging.INFO, __file__, 1, "hello %s", ("world",), None)
+            RequestContextFilter().filter(record)
+            payload = json.loads(JsonLogFormatter().format(record))
+        finally:
+            reset_request_id(token)
+
+        self.assertEqual(payload["level"], "INFO")
+        self.assertEqual(payload["logger"], "magent.test")
+        self.assertEqual(payload["request_id"], "request-123")
+        self.assertEqual(payload["message"], "hello world")
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/test_schema_migrations.py

@@ -0,0 +1,36 @@
+import sqlite3
+import unittest
+
+from backend.app.schema_migrations import run_schema_migrations
+
+
+class SchemaMigrationTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.conn = sqlite3.connect(":memory:")
+        self.conn.execute(
+            "CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT NOT NULL UNIQUE, password_hash TEXT, role TEXT, created_at TEXT)"
+        )
+        self.conn.execute(
+            "CREATE TABLE signup_invites (id INTEGER PRIMARY KEY, code TEXT NOT NULL UNIQUE, created_at TEXT, updated_at TEXT)"
+        )
+        self.conn.execute("CREATE TABLE portal_items (id INTEGER PRIMARY KEY, kind TEXT, updated_at TEXT)")
+        self.conn.execute("CREATE TABLE requests_cache (request_id INTEGER PRIMARY KEY, created_at TEXT)")
+
+    def tearDown(self) -> None:
+        self.conn.close()
+
+    def test_migrations_are_versioned_and_idempotent(self) -> None:
+        self.assertEqual(run_schema_migrations(self.conn), [1])
+        self.assertEqual(run_schema_migrations(self.conn), [])
+
+        user_columns = {row[1] for row in self.conn.execute("PRAGMA table_info(users)")}
+        self.assertIn("auth_version", user_columns)
+        self.assertIn("email", user_columns)
+        request_columns = {row[1] for row in self.conn.execute("PRAGMA table_info(requests_cache)")}
+        self.assertIn("requested_by_id", request_columns)
+        applied = self.conn.execute("SELECT version, name FROM schema_migrations").fetchall()
+        self.assertEqual(applied, [(1, "legacy_columns_and_indexes")])
+
+
+if __name__ == "__main__":
+    unittest.main()