Extract duplicated literals, add dedicated SSH exceptions (group 8, part 1)

services/quadlet_parser.py

@@ -27,8 +27,7 @@ def validate_quadlet_syntax(content: str, quadlet_type: str = "container"):
         raise QuadletValidationError(f"Missing [{expected_section}] section for {quadlet_type} type.")
 
     # Validate specific requirements
-    if quadlet_type == 'container':
-        if 'Image' not in parser[expected_section]:
-            raise QuadletValidationError("[Container] section must define an 'Image'.")
+    if quadlet_type == 'container' and 'Image' not in parser[expected_section]:
+        raise QuadletValidationError("[Container] section must define an 'Image'.")
 
     return True

services/ssh_manager.py

@@ -27,6 +27,29 @@ class SSHTimeoutError(SSHCommandError):
     """Raised when a remote SSH command exceeds its timeout."""
 
 
+class ServerConfigurationError(Exception):
+    """Raised when a server row cannot be found, or is missing its SSH key
+    mapping, so a connection attempt cannot even be assembled.
+
+    Deliberately NOT a subclass of SSHCommandError: that type carries the
+    remote exit status and stderr of a command that actually ran, and this
+    failure happens while assembling the connection, before anything is
+    sent to the host. Typing it as a command failure would hand callers a
+    null exit status and imply a command failed when none was issued.
+    """
+
+
+class KeyDecryptionError(Exception):
+    """Raised when the stored SSH private key for a server cannot be
+    decrypted, typically because the master key changed since the server
+    was configured.
+
+    Deliberately NOT a subclass of SSHCommandError, for the same reason as
+    ServerConfigurationError: no remote command has run, so there is no
+    exit status or stderr to report.
+    """
+
+
 class HostKeyMismatchError(Exception):
     """Raised when a server's presented SSH host key does not match the
     previously pinned host key, or when strict host-key checking is enabled
@@ -65,9 +88,8 @@ def _drop_if_stale(self, server_id: int) -> bool:
         return False
 
     async def get_connection(self, server_id: int):
-        if server_id in self.connections:
-            if not self._drop_if_stale(server_id):
-                return self.connections[server_id]
+        if server_id in self.connections and not self._drop_if_stale(server_id):
+            return self.connections[server_id]
 
         # No live cached connection here means we need to connect. Serialize
         # via a per-server lock so concurrent callers don't each open their
@@ -80,9 +102,8 @@ async def get_connection(self, server_id: int):
             # for the lock. That connection could have died since it was
             # established/confirmed, so re-run the staleness check before
             # reusing it.
-            if server_id in self.connections:
-                if not self._drop_if_stale(server_id):
-                    return self.connections[server_id]
+            if server_id in self.connections and not self._drop_if_stale(server_id):
+                return self.connections[server_id]
 
             return await self.connect_to_server(server_id)
 
@@ -96,7 +117,7 @@ async def connect_to_server(self, server_id: int):
             """, (server_id,)) as cursor:
                 row = await cursor.fetchone()
             if not row:
-                raise Exception(f"Server {server_id} not found or missing SSH key mapping.")
+                raise ServerConfigurationError(f"Server {server_id} not found or missing SSH key mapping.")
 
             ip_address, ssh_user, encrypted_pk, stored_host_key = row
 
@@ -115,7 +136,7 @@ async def connect_to_server(self, server_id: int):
         try:
             private_key_str = decrypt_private_key(encrypted_pk)
         except (InvalidTag, ValueError) as exc:
-            raise Exception(
+            raise KeyDecryptionError(
                 f"Failed to decrypt SSH key for server {server_id}. "
                 "The master key may have changed since this server was configured. "
                 "Set QUADLET_MASTER_KEY to a stable value and re-add the server if needed."

tests/settings_mocks.py

@@ -0,0 +1,94 @@
+"""Shared mocks for the admin settings endpoints.
+
+`test_log_level.py` and `test_session_duration.py` both drive an admin-only
+settings endpoint through the real app, so both need the same two fakes: a
+login that yields an admin (or non-admin) session cookie, and an in-memory
+stand-in for the `settings` key/value table. They previously carried
+byte-identical copies of both.
+"""
+from contextlib import asynccontextmanager
+from unittest.mock import patch, AsyncMock, MagicMock
+
+
+def mock_login_db(is_admin: bool):
+    """Mock get_db_connection so /login authenticates as a user with the given admin flag."""
+    from argon2 import PasswordHasher
+    pwd_hash = PasswordHasher().hash("password123")
+
+    class MockCursor:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            pass
+
+        async def fetchone(self):
+            return (pwd_hash, "editor", int(is_admin), 0)
+
+    mock_db = MagicMock()
+    mock_db.execute = MagicMock(return_value=MockCursor())
+    mock_db.commit = AsyncMock()
+
+    @asynccontextmanager
+    async def _mock_conn():
+        yield mock_db
+
+    return _mock_conn, mock_db
+
+
+def login(client, is_admin: bool):
+    """Log in through the real /login route and return the session cookie."""
+    conn_factory, _ = mock_login_db(is_admin)
+    with patch("api.routes.get_db_connection", side_effect=conn_factory):
+        response = client.post(
+            "/login",
+            data={"username": "someone", "password": "password123"},
+            follow_redirects=False,
+        )
+    return response.cookies["qm_session"]
+
+
+class SettingsMockDB:
+    """Minimal in-memory stand-in for the `settings` key/value table."""
+
+    def __init__(self):
+        self.store = {}
+
+    async def commit(self):
+        pass
+
+    def execute(self, query, params=()):
+        cursor = MagicMock()
+        if query.strip().startswith("SELECT value FROM settings"):
+            key = params[0]
+            value = self.store.get(key)
+            cursor.fetchone = AsyncMock(return_value=(value,) if value is not None else None)
+        elif query.strip().startswith("INSERT INTO settings"):
+            key, value = params
+            self.store[key] = value
+            cursor.fetchone = AsyncMock(return_value=None)
+        else:
+            cursor.fetchone = AsyncMock(return_value=None)
+
+        class DualProtocolCM:
+            """Objects returned by aiosqlite execute() support both
+            `async with obj` and `await obj`."""
+
+            async def __aenter__(self):
+                return cursor
+
+            async def __aexit__(self, *args):
+                return False
+
+            def __await__(self):
+                async def _resolve():
+                    return cursor
+                return _resolve().__await__()
+
+        return DualProtocolCM()
+
+
+@asynccontextmanager
+async def settings_conn(settings_db):
+    """get_db_connection() replacement yielding the given SettingsMockDB."""
+    yield settings_db

tests/test_api_auth_sweep.py

@@ -25,6 +25,7 @@
 # Dependencies that reject cookieless requests (all raise a 303 redirect to
 # /login via _get_session when no valid session cookie is present).
 AUTH_ENFORCING_DEPS = {
+    routes_module._get_session,
     routes_module.get_current_user_role,
     routes_module.get_current_user_is_admin,
     routes_module.get_current_user_id,

tests/test_log_level.py

@@ -1,11 +1,11 @@
 import logging
 
 import pytest
-from unittest.mock import patch, AsyncMock, MagicMock
-from contextlib import asynccontextmanager
+from unittest.mock import patch
 from fastapi.testclient import TestClient
 from main import app
 import api.routes as api_routes
+from tests.settings_mocks import SettingsMockDB, login as _login, mock_login_db as _mock_login_db, settings_conn
 
 
 @pytest.fixture
@@ -23,84 +23,17 @@ def reset_log_level():
     api_routes._log_level = original
 
 
-def _mock_login_db(is_admin: bool):
-    """Mock get_db_connection so /login authenticates as a user with the given admin flag."""
-    from argon2 import PasswordHasher
-    pwd_hash = PasswordHasher().hash("password123")
-
-    class MockCursor:
-        async def __aenter__(self):
-            return self
-        async def __aexit__(self, exc_type, exc, tb):
-            pass
-        async def fetchone(self):
-            return (pwd_hash, "editor", int(is_admin), 0)
-
-    mock_db = MagicMock()
-    mock_db.execute = MagicMock(return_value=MockCursor())
-    mock_db.commit = AsyncMock()
-
-    @asynccontextmanager
-    async def _mock_conn():
-        yield mock_db
-
-    return _mock_conn, mock_db
-
-
-def _login(client, is_admin: bool):
-    conn_factory, _ = _mock_login_db(is_admin)
-    with patch("api.routes.get_db_connection", side_effect=conn_factory):
-        response = client.post(
-            "/login",
-            data={"username": "someone", "password": "password123"},
-            follow_redirects=False,
-        )
-    return response.cookies["qm_session"]
-
-
-class SettingsMockDB:
-    """Minimal in-memory stand-in for the `settings` key/value table."""
-
-    def __init__(self):
-        self.store = {}
-
-    async def commit(self):
-        pass
-
-    def execute(self, query, params=()):
-        cursor = MagicMock()
-        if query.strip().startswith("SELECT value FROM settings"):
-            key = params[0]
-            value = self.store.get(key)
-            cursor.fetchone = AsyncMock(return_value=(value,) if value is not None else None)
-        elif query.strip().startswith("INSERT INTO settings"):
-            key, value = params
-            self.store[key] = value
-            cursor.fetchone = AsyncMock(return_value=None)
-        else:
-            cursor.fetchone = AsyncMock(return_value=None)
-
-        async def _aenter():
-            return cursor
-        cursor.__aenter__ = AsyncMock(side_effect=_aenter)
-        cursor.__aexit__ = AsyncMock(return_value=False)
-        return cursor
-
-
 @pytest.mark.unit
 @pytest.mark.asyncio
 async def test_load_log_level_ignores_invalid_stored_value():
     settings_db = SettingsMockDB()
     settings_db.store["log_level"] = "VERBOSE"
 
-    @asynccontextmanager
-    async def _mock_settings_conn():
-        yield settings_db
 
     logger = logging.getLogger("quadlet-manager")
     original_level = logger.level
 
-    with patch("api.routes.get_db_connection", side_effect=_mock_settings_conn):
+    with patch("api.routes.get_db_connection", side_effect=lambda: settings_conn(settings_db)):
         await api_routes._load_log_level_from_db()
 
     assert logger.level == original_level
@@ -112,11 +45,7 @@ def test_log_level_persists_and_applies_live(client):
 
     settings_db = SettingsMockDB()
 
-    @asynccontextmanager
-    async def _mock_settings_conn():
-        yield settings_db
-
-    with patch("api.routes.get_db_connection", side_effect=_mock_settings_conn):
+    with patch("api.routes.get_db_connection", side_effect=lambda: settings_conn(settings_db)):
         response = client.put(
             "/api/settings/log-level",
             data={"log_level": "DEBUG"},

tests/test_session_duration.py

@@ -1,9 +1,9 @@
 import pytest
-from unittest.mock import patch, AsyncMock, MagicMock
-from contextlib import asynccontextmanager
+from unittest.mock import patch
 from fastapi.testclient import TestClient
 from main import app
 import api.routes as api_routes
+from tests.settings_mocks import SettingsMockDB, login as _login, mock_login_db as _mock_login_db, settings_conn
 
 
 @pytest.fixture
@@ -21,70 +21,6 @@ def reset_session_duration():
     api_routes._session_duration_seconds = original
 
 
-def _mock_login_db(is_admin: bool):
-    """Mock get_db_connection so /login authenticates as a user with the given admin flag."""
-    from argon2 import PasswordHasher
-    pwd_hash = PasswordHasher().hash("password123")
-
-    class MockCursor:
-        async def __aenter__(self):
-            return self
-        async def __aexit__(self, exc_type, exc, tb):
-            pass
-        async def fetchone(self):
-            return (pwd_hash, "editor", int(is_admin), 0)
-
-    mock_db = MagicMock()
-    mock_db.execute = MagicMock(return_value=MockCursor())
-    mock_db.commit = AsyncMock()
-
-    @asynccontextmanager
-    async def _mock_conn():
-        yield mock_db
-
-    return _mock_conn, mock_db
-
-
-def _login(client, is_admin: bool):
-    conn_factory, _ = _mock_login_db(is_admin)
-    with patch("api.routes.get_db_connection", side_effect=conn_factory):
-        response = client.post(
-            "/login",
-            data={"username": "someone", "password": "password123"},
-            follow_redirects=False,
-        )
-    return response.cookies["qm_session"]
-
-
-class SettingsMockDB:
-    """Minimal in-memory stand-in for the `settings` key/value table."""
-
-    def __init__(self):
-        self.store = {}
-
-    async def commit(self):
-        pass
-
-    def execute(self, query, params=()):
-        cursor = MagicMock()
-        if query.strip().startswith("SELECT value FROM settings"):
-            key = params[0]
-            value = self.store.get(key)
-            cursor.fetchone = AsyncMock(return_value=(value,) if value is not None else None)
-        elif query.strip().startswith("INSERT INTO settings"):
-            key, value = params
-            self.store[key] = value
-            cursor.fetchone = AsyncMock(return_value=None)
-        else:
-            cursor.fetchone = AsyncMock(return_value=None)
-
-        async def _aenter():
-            return cursor
-        cursor.__aenter__ = AsyncMock(side_effect=_aenter)
-        cursor.__aexit__ = AsyncMock(return_value=False)
-        return cursor
-
-
 @pytest.mark.unit
 def test_session_duration_requires_admin(client):
     cookie = _login(client, is_admin=False)
@@ -123,11 +59,7 @@ def test_session_duration_persists_and_applies_to_new_logins(client):
 
     settings_db = SettingsMockDB()
 
-    @asynccontextmanager
-    async def _mock_settings_conn():
-        yield settings_db
-
-    with patch("api.routes.get_db_connection", side_effect=_mock_settings_conn):
+    with patch("api.routes.get_db_connection", side_effect=lambda: settings_conn(settings_db)):
         response = client.put(
             "/api/settings/session-duration",
             data={"session_duration_seconds": "604800"},