From 285c7f84e48c4069566d1f451be5790920bae33a Mon Sep 17 00:00:00 2001
From: Carol <carol.ouellet@gmail.com>
Date: Mon, 3 Aug 2026 17:17:33 -0400
Subject: [PATCH 1/5] Extract duplicated literals, add dedicated SSH exceptions
 (group 8, part 1)

Clears 13 of the 19 open python maintainability findings: 7 python:S1192,
2 python:S108, 2 python:S1066, 2 python:S112. The 6 python:S3776 cognitive
complexity refactors are deliberately left for a separate PR so a
control-flow change can be reviewed and reverted on its own.

S1192: seven duplicated literals in api/routes.py become module-level
constants, following the placement and naming of the existing
SESSION_DURATION_SETTING_KEY and LOG_LEVEL_CHOICES. Sonar counted 3
occurrences each of "/login" and "/change-password"; the file actually
contains 5 of each, including the route decorators, and all were replaced.
Verified by AST that each literal now appears exactly once, as its constant
definition, and that no other string literal in the file changed.

S108: the two `async with db.execute(...): pass` blocks become
`await db.execute(...)`, the form used at every other write site in the
codebase. These two were the only outliers.

That change broke two tests, and the tests were at fault. Their hand-rolled
SettingsMockDB.execute returned a cursor supporting only `async with`,
while real aiosqlite execute() results support both protocols. The repo
already knows this: tests/test_sync_engine.py defines DualProtocolCM with a
docstring saying exactly that. Both mocks now use the same pattern, so the
fake no longer dictates which idiom the route uses.

S112: the two bare `raise Exception` in ssh_manager become
ServerConfigurationError and KeyDecryptionError, with tests covering both
plus __cause__ chaining. Both subclass Exception directly rather than
SSHCommandError, because SSHCommandError carries a remote exit status and
stderr for a command that actually ran, and these fail while assembling the
connection, before anything reaches the host. All existing handlers catch
Exception, so nothing downstream changes.

S1066: two nested ifs merged. `and` short-circuits, so _drop_if_stale's
side effect still fires only when the connection is cached, exactly as
before. The structurally identical re-check inside get_connection's lock
was merged too, though Sonar did not flag it, since leaving two adjacent
identical checks written differently is worse than either form.

Unit suite 939 to 942, the three new exception tests. Also re-run in random
order to check for contamination.
---
 api/routes.py                        | 94 +++++++++++++++-------------
 services/quadlet_parser.py           |  5 +-
 services/ssh_manager.py              | 37 ++++++++---
 tests/test_log_level.py              | 20 ++++--
 tests/test_session_duration.py       | 20 ++++--
 tests/test_ssh_manager_exceptions.py | 81 ++++++++++++++++++++++++
 6 files changed, 192 insertions(+), 65 deletions(-)
 create mode 100644 tests/test_ssh_manager_exceptions.py

diff --git a/api/routes.py b/api/routes.py
index b38435c..a480cf3 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -108,6 +108,14 @@ def _toast(request: Request, color: str, message: str, status_output=None) -> HT
 LOG_LEVEL_SETTING_KEY = "log_level"
 LOG_LEVEL_CHOICES = ("DEBUG", "INFO", "WARNING")
 
+LOGIN_PATH = "/login"
+CHANGE_PASSWORD_PATH = "/change-password"
+ADMIN_REQUIRED_DETAIL = "Admin access required."
+LOGIN_TEMPLATE = "login.html"
+PERMISSION_DENIED_HTML = "<p class='text-muted'>Permission denied.</p>"
+SETTINGS_ADMIN_TEMPLATE = "partials/settings_admin.html"
+SELECT_USERNAME_BY_ID_SQL = "SELECT username FROM users WHERE id = ?"
+
 # Mutable at runtime via PUT /api/settings/log-level; seeded the same way main.py seeds
 # its own startup logging.basicConfig level, until a value is persisted to the settings table.
 _log_level = os.getenv("LOG_LEVEL", "INFO").upper()
@@ -198,12 +206,11 @@ async def _persist_session_duration(seconds: int) -> None:
     """Persist the session duration to the settings table and update the in-memory value."""
     global _session_duration_seconds
     async with get_db_connection() as db:
-        async with db.execute(
+        await db.execute(
             "INSERT INTO settings (key, value) VALUES (?, ?) "
             "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
             (SESSION_DURATION_SETTING_KEY, str(seconds)),
-        ):
-            pass
+        )
         await db.commit()
     _session_duration_seconds = seconds
 
@@ -231,12 +238,11 @@ async def _persist_log_level(level: str) -> None:
     """Persist the log level to the settings table, update the in-memory value, and apply it live."""
     global _log_level
     async with get_db_connection() as db:
-        async with db.execute(
+        await db.execute(
             "INSERT INTO settings (key, value) VALUES (?, ?) "
             "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
             (LOG_LEVEL_SETTING_KEY, level),
-        ):
-            pass
+        )
         await db.commit()
     _log_level = level
     logging.getLogger("quadlet-manager").setLevel(level)
@@ -252,18 +258,18 @@ async def _get_session(request: Request) -> dict:
 
     cookie = request.cookies.get(COOKIE_NAME)
     if not cookie:
-        raise HTTPException(status_code=303, headers={"Location": "/login"})
+        raise HTTPException(status_code=303, headers={"Location": LOGIN_PATH})
 
     session = _read_session_cookie(cookie)
     if not session:
-        raise HTTPException(status_code=303, headers={"Location": "/login"})
+        raise HTTPException(status_code=303, headers={"Location": LOGIN_PATH})
 
     if (
         session.get("must_change_password")
-        and request.url.path != "/change-password"
+        and request.url.path != CHANGE_PASSWORD_PATH
         and request.url.path != "/logout"
     ):
-        raise HTTPException(status_code=303, headers={"Location": "/change-password"})
+        raise HTTPException(status_code=303, headers={"Location": CHANGE_PASSWORD_PATH})
 
     return session
 
@@ -310,7 +316,7 @@ async def get_current_username(request: Request) -> str:
 async def require_admin(is_admin: bool = Depends(get_current_user_is_admin)) -> None:
     """Verify that the current user has admin privileges, raising 403 if not."""
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
 
 async def get_current_user_id(username: str = Depends(get_current_username)) -> int:
@@ -368,16 +374,16 @@ async def _ensure_default_theme(user_id: int) -> None:
 
 
 # ── Login / Logout ────────────────────────────────────────
-@router.get("/login", response_class=HTMLResponse)
+@router.get(LOGIN_PATH, response_class=HTMLResponse)
 async def login_page(request: Request):
     # If already logged in, redirect to dashboard
     role = await get_optional_user_role(request)
     if role:
         return RedirectResponse(url="/", status_code=303)
-    return templates.TemplateResponse(request, "login.html", {"error": None})
+    return templates.TemplateResponse(request, LOGIN_TEMPLATE, {"error": None})
 
 
-@router.post("/login", response_class=HTMLResponse)
+@router.post(LOGIN_PATH, response_class=HTMLResponse)
 async def login_submit(request: Request, username: str = Form(...), password: str = Form(...)):
     async with get_db_connection() as db:
         async with db.execute(
@@ -387,7 +393,7 @@ async def login_submit(request: Request, username: str = Form(...), password: st
             row = await cursor.fetchone()
 
     if not row:
-        return templates.TemplateResponse(request, "login.html", {
+        return templates.TemplateResponse(request, LOGIN_TEMPLATE, {
             "error": "Invalid username or password"
         }, status_code=401)
 
@@ -406,13 +412,13 @@ async def login_submit(request: Request, username: str = Form(...), password: st
         pass
 
     if not credentials_valid:
-        return templates.TemplateResponse(request, "login.html", {
+        return templates.TemplateResponse(request, LOGIN_TEMPLATE, {
             "error": "Invalid username or password"
         }, status_code=401)
 
     # Credentials valid – set session cookie and redirect to dashboard
     # (or to the forced password-change page, if flagged)
-    redirect_url = "/change-password" if must_change_password else "/"
+    redirect_url = CHANGE_PASSWORD_PATH if must_change_password else "/"
     response = RedirectResponse(url=redirect_url, status_code=303)
     response.set_cookie(
         key=COOKIE_NAME,
@@ -426,18 +432,18 @@ async def login_submit(request: Request, username: str = Form(...), password: st
 
 @router.get("/logout")
 async def logout():
-    response = RedirectResponse(url="/login", status_code=303)
+    response = RedirectResponse(url=LOGIN_PATH, status_code=303)
     response.delete_cookie(COOKIE_NAME)
     return response
 
 
-@router.get("/change-password", response_class=HTMLResponse)
+@router.get(CHANGE_PASSWORD_PATH, response_class=HTMLResponse)
 async def change_password_page(request: Request):
     await _get_session(request)
     return templates.TemplateResponse(request, "change_password.html", {"error": None})
 
 
-@router.post("/change-password", response_class=HTMLResponse)
+@router.post(CHANGE_PASSWORD_PATH, response_class=HTMLResponse)
 async def change_password_submit(
     request: Request,
     new_password: str = Form(...),
@@ -910,7 +916,7 @@ async def settings_add_server(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
     if scope_filter not in VALID_SCOPE_FILTERS:
         raise HTTPException(status_code=422, detail="scope_filter must be 'user', 'global', or 'both'.")
 
@@ -944,7 +950,7 @@ async def settings_update_server(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
     if scope_filter not in VALID_SCOPE_FILTERS:
         raise HTTPException(status_code=422, detail="scope_filter must be 'user', 'global', or 'both'.")
 
@@ -968,7 +974,7 @@ async def settings_delete_server(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     # Close cached SSH connection if present
     pool.connections.pop(server_id, None)
@@ -1002,7 +1008,7 @@ async def settings_repin_server_host_key(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     # Drop the cached SSH connection so the next connect performs a fresh
     # handshake against whatever key the server presents.
@@ -1029,7 +1035,7 @@ async def settings_reorder_servers(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     body = await request.json()
     order = body.get("order", [])
@@ -1060,9 +1066,9 @@ async def settings_get_session_duration(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        return HTMLResponse("<p class='text-muted'>Permission denied.</p>", status_code=403)
+        return HTMLResponse(PERMISSION_DENIED_HTML, status_code=403)
 
-    return templates.TemplateResponse(request, "partials/settings_admin.html", {
+    return templates.TemplateResponse(request, SETTINGS_ADMIN_TEMPLATE, {
         "current_seconds": _session_duration_seconds,
         "duration_choices": SESSION_DURATION_LABELS,
         "current_log_level": _log_level,
@@ -1078,13 +1084,13 @@ async def settings_update_session_duration(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     if session_duration_seconds not in SESSION_DURATION_CHOICES:
         raise HTTPException(status_code=400, detail="Invalid session duration.")
 
     await _persist_session_duration(session_duration_seconds)
-    return templates.TemplateResponse(request, "partials/settings_admin.html", {
+    return templates.TemplateResponse(request, SETTINGS_ADMIN_TEMPLATE, {
         "current_seconds": _session_duration_seconds,
         "duration_choices": SESSION_DURATION_LABELS,
         "current_log_level": _log_level,
@@ -1099,9 +1105,9 @@ async def settings_get_log_level(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        return HTMLResponse("<p class='text-muted'>Permission denied.</p>", status_code=403)
+        return HTMLResponse(PERMISSION_DENIED_HTML, status_code=403)
 
-    return templates.TemplateResponse(request, "partials/settings_admin.html", {
+    return templates.TemplateResponse(request, SETTINGS_ADMIN_TEMPLATE, {
         "current_seconds": _session_duration_seconds,
         "duration_choices": SESSION_DURATION_LABELS,
         "current_log_level": _log_level,
@@ -1117,13 +1123,13 @@ async def settings_update_log_level(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     if log_level not in LOG_LEVEL_CHOICES:
         raise HTTPException(status_code=400, detail="Invalid log level.")
 
     await _persist_log_level(log_level)
-    return templates.TemplateResponse(request, "partials/settings_admin.html", {
+    return templates.TemplateResponse(request, SETTINGS_ADMIN_TEMPLATE, {
         "current_seconds": _session_duration_seconds,
         "duration_choices": SESSION_DURATION_LABELS,
         "current_log_level": _log_level,
@@ -1140,7 +1146,7 @@ async def settings_list_users(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        return HTMLResponse("<p class='text-muted'>Permission denied.</p>", status_code=403)
+        return HTMLResponse(PERMISSION_DENIED_HTML, status_code=403)
 
     session = await _get_session(request)
     async with get_db_connection() as db:
@@ -1165,7 +1171,7 @@ async def settings_add_user(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     if user_role not in ("viewer", "editor"):
         raise HTTPException(status_code=400, detail="Invalid role.")
@@ -1196,7 +1202,7 @@ async def settings_update_user_role(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     if user_role not in ("viewer", "editor"):
         raise HTTPException(status_code=400, detail="Invalid role.")
@@ -1204,7 +1210,7 @@ async def settings_update_user_role(
     session = await _get_session(request)
     async with get_db_connection() as db:
         # Prevent demoting yourself
-        async with db.execute("SELECT username FROM users WHERE id = ?", (user_id,)) as cursor:
+        async with db.execute(SELECT_USERNAME_BY_ID_SQL, (user_id,)) as cursor:
             row = await cursor.fetchone()
         if row and row[0] == session["username"]:
             return HTMLResponse(
@@ -1237,11 +1243,11 @@ async def settings_toggle_admin(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     session = await _get_session(request)
     async with get_db_connection() as db:
-        async with db.execute("SELECT username FROM users WHERE id = ?", (user_id,)) as cursor:
+        async with db.execute(SELECT_USERNAME_BY_ID_SQL, (user_id,)) as cursor:
             row = await cursor.fetchone()
         if row and row[0] == session["username"]:
             return HTMLResponse(
@@ -1276,12 +1282,12 @@ async def settings_delete_user(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     session = await _get_session(request)
     async with get_db_connection() as db:
         # Prevent self-deletion
-        async with db.execute("SELECT username FROM users WHERE id = ?", (user_id,)) as cursor:
+        async with db.execute(SELECT_USERNAME_BY_ID_SQL, (user_id,)) as cursor:
             row = await cursor.fetchone()
         if row and row[0] == session["username"]:
             return HTMLResponse(
@@ -1631,7 +1637,7 @@ async def api_list_keys(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        return HTMLResponse("<p class='text-muted'>Permission denied.</p>", status_code=403)
+        return HTMLResponse(PERMISSION_DENIED_HTML, status_code=403)
 
     async with get_db_connection() as db:
         async with db.execute("SELECT id, key_name FROM ssh_keys ORDER BY key_name") as cursor:
@@ -1651,7 +1657,7 @@ async def api_add_key(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     if key_file and key_file.filename:
         raw = await key_file.read()
@@ -1681,7 +1687,7 @@ async def api_delete_key(
     is_admin: bool = Depends(get_current_user_is_admin),
 ):
     if not is_admin:
-        raise HTTPException(status_code=403, detail="Admin access required.")
+        raise HTTPException(status_code=403, detail=ADMIN_REQUIRED_DETAIL)
 
     async with get_db_connection() as db:
         async with db.execute(
diff --git a/services/quadlet_parser.py b/services/quadlet_parser.py
index 8b43397..3635410 100644
--- a/services/quadlet_parser.py
+++ b/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
diff --git a/services/ssh_manager.py b/services/ssh_manager.py
index 674b64a..c557696 100644
--- a/services/ssh_manager.py
+++ b/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."
diff --git a/tests/test_log_level.py b/tests/test_log_level.py
index adb0f34..b23fc8c 100644
--- a/tests/test_log_level.py
+++ b/tests/test_log_level.py
@@ -80,11 +80,21 @@ def execute(self, query, params=()):
         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
+        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()
 
 
 @pytest.mark.unit
diff --git a/tests/test_session_duration.py b/tests/test_session_duration.py
index 67829f2..6f083f6 100644
--- a/tests/test_session_duration.py
+++ b/tests/test_session_duration.py
@@ -78,11 +78,21 @@ def execute(self, query, params=()):
         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
+        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()
 
 
 @pytest.mark.unit
diff --git a/tests/test_ssh_manager_exceptions.py b/tests/test_ssh_manager_exceptions.py
new file mode 100644
index 0000000..560c6b6
--- /dev/null
+++ b/tests/test_ssh_manager_exceptions.py
@@ -0,0 +1,81 @@
+import pytest
+from unittest.mock import AsyncMock, patch
+from contextlib import asynccontextmanager
+from cryptography.exceptions import InvalidTag
+from services.ssh_manager import (
+    SSHConnectionPool,
+    ServerConfigurationError,
+    KeyDecryptionError,
+)
+
+
+@pytest.fixture
+def pool():
+    return SSHConnectionPool()
+
+
+@pytest.fixture
+def mock_db_ctx_factory():
+    """Build a mock get_db_connection() context manager whose cursor.fetchone()
+    returns the given row (None to simulate "server not found").
+    """
+    def _factory(row):
+        @asynccontextmanager
+        async def _mock_db():
+            db_mock = AsyncMock()
+            cursor_mock = AsyncMock()
+            cursor_mock.fetchone.return_value = row
+
+            @asynccontextmanager
+            async def _mock_execute(*args, **kwargs):
+                yield cursor_mock
+
+            db_mock.execute = _mock_execute
+            yield db_mock
+        return _mock_db
+    return _factory
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_connect_to_server_missing_row_raises_server_configuration_error(pool, mock_db_ctx_factory):
+    mock_db_ctx = mock_db_ctx_factory(None)
+
+    with patch("services.ssh_manager.get_db_connection", side_effect=mock_db_ctx):
+        with pytest.raises(ServerConfigurationError) as excinfo:
+            await pool.connect_to_server(42)
+
+    assert str(excinfo.value) == "Server 42 not found or missing SSH key mapping."
+    # Still catchable as a plain Exception so existing broad handlers work.
+    assert isinstance(excinfo.value, Exception)
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_connect_to_server_decryption_failure_raises_key_decryption_error(pool, mock_db_ctx_factory):
+    mock_db_ctx = mock_db_ctx_factory(("127.0.0.1:22", "user", "enc", None))
+    original_exc = InvalidTag()
+
+    with patch("services.ssh_manager.get_db_connection", side_effect=mock_db_ctx), \
+         patch("services.ssh_manager.decrypt_private_key", side_effect=original_exc):
+        with pytest.raises(KeyDecryptionError) as excinfo:
+            await pool.connect_to_server(7)
+
+    assert "Failed to decrypt SSH key for server 7." in str(excinfo.value)
+    assert excinfo.value.__cause__ is original_exc
+    # Still catchable as a plain Exception so existing broad handlers work.
+    assert isinstance(excinfo.value, Exception)
+
+
+@pytest.mark.asyncio
+@pytest.mark.unit
+async def test_connect_to_server_decryption_value_error_is_chained(pool, mock_db_ctx_factory):
+    mock_db_ctx = mock_db_ctx_factory(("127.0.0.1:22", "user", "enc", None))
+    original_exc = ValueError("bad padding")
+
+    with patch("services.ssh_manager.get_db_connection", side_effect=mock_db_ctx), \
+         patch("services.ssh_manager.decrypt_private_key", side_effect=original_exc):
+        with pytest.raises(KeyDecryptionError) as excinfo:
+            await pool.connect_to_server(7)
+
+    assert excinfo.value.__cause__ is original_exc

From e88fee8564bbeece7c554d4a583b0f7ee2b35867 Mon Sep 17 00:00:00 2001
From: Carol <carol.ouellet@gmail.com>
Date: Mon, 3 Aug 2026 17:28:49 -0400
Subject: [PATCH 2/5] Extract the shared settings-endpoint mocks into
 tests/settings_mocks.py

Fixes the quality gate on this PR. Adding DualProtocolCM to both
test_log_level.py and test_session_duration.py pushed new-code duplication
to 14% against a 3% threshold.

The duplication was mostly pre-existing: the two files carried byte-identical
copies of the login mock, the login helper and SettingsMockDB, 75 duplicated
lines each. Touching those lines made them count as new code, which is the
same inherit-the-debt effect seen on PR #304.

Extracting the shared scaffolding fixes the cause rather than the symptom.
The two files drop from 147 and 127 lines to 79 and 56, and shared blocks
fall from 75 lines each to 22, which is now just imports and fixtures.

Follows the existing tests/sudo_permissions.py precedent for a non-test
helper module under tests/. test_sync_engine.py has its own DualProtocolCM
copy, deliberately left alone here since it is modified on PR #309 and
touching it would conflict.

Unit 942 and unmarked 47, both unchanged.
---
 tests/settings_mocks.py        | 94 ++++++++++++++++++++++++++++++++++
 tests/test_log_level.py        | 89 ++------------------------------
 tests/test_session_duration.py | 84 ++----------------------------
 3 files changed, 101 insertions(+), 166 deletions(-)
 create mode 100644 tests/settings_mocks.py

diff --git a/tests/settings_mocks.py b/tests/settings_mocks.py
new file mode 100644
index 0000000..4098ae4
--- /dev/null
+++ b/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
diff --git a/tests/test_log_level.py b/tests/test_log_level.py
index b23fc8c..a68faef 100644
--- a/tests/test_log_level.py
+++ b/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,94 +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)
-
-        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()
-
-
 @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
@@ -122,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"},
diff --git a/tests/test_session_duration.py b/tests/test_session_duration.py
index 6f083f6..0e09c9d 100644
--- a/tests/test_session_duration.py
+++ b/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,80 +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)
-
-        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()
-
-
 @pytest.mark.unit
 def test_session_duration_requires_admin(client):
     cookie = _login(client, is_admin=False)
@@ -133,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"},

From f65b5d9b51f80e7e0f039ddf887af28a56aac580 Mon Sep 17 00:00:00 2001
From: Carol <carol.ouellet@gmail.com>
Date: Mon, 3 Aug 2026 18:14:48 -0400
Subject: [PATCH 3/5] fix(auth): enforce session dependency in change password
 routes

---
 api/routes.py                | 8 ++++----
 tests/test_api_auth_sweep.py | 1 +
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/api/routes.py b/api/routes.py
index a480cf3..fd44b11 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -437,19 +437,19 @@ async def logout():
     return response
 
 
-@router.get(CHANGE_PASSWORD_PATH, response_class=HTMLResponse)
-async def change_password_page(request: Request):
+@router.get(CHANGE_PASSWORD_PATH, response_class=HTMLResponse, responses=AUTH_REDIRECT_RESPONSES)
+async def change_password_page(request: Request, session: dict = Depends(_get_session)):
     await _get_session(request)
     return templates.TemplateResponse(request, "change_password.html", {"error": None})
 
 
-@router.post(CHANGE_PASSWORD_PATH, response_class=HTMLResponse)
+@router.post(CHANGE_PASSWORD_PATH, response_class=HTMLResponse, responses=AUTH_REDIRECT_RESPONSES)
 async def change_password_submit(
     request: Request,
     new_password: str = Form(...),
     confirm_password: str = Form(...),
+    session: dict = Depends(_get_session),
 ):
-    session = await _get_session(request)
 
     if not new_password or new_password != confirm_password:
         return templates.TemplateResponse(request, "change_password.html", {
diff --git a/tests/test_api_auth_sweep.py b/tests/test_api_auth_sweep.py
index bfa4371..044eb01 100644
--- a/tests/test_api_auth_sweep.py
+++ b/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,

From 42b6a8d1385ea4a0bd5616938dbbfda8e0a84870 Mon Sep 17 00:00:00 2001
From: Carol <carol.ouellet@gmail.com>
Date: Mon, 3 Aug 2026 19:05:04 -0400
Subject: [PATCH 4/5] fix(routes): use Annotated for session and form
 parameters in change password routes

---
 api/routes.py | 14 ++++++++------
 1 file changed, 8 insertions(+), 6 deletions(-)

diff --git a/api/routes.py b/api/routes.py
index fd44b11..6cdd947 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -1,6 +1,6 @@
 from fastapi import APIRouter, Request, Depends, Form, File, HTTPException, UploadFile, WebSocket
 from fastapi.responses import HTMLResponse, StreamingResponse, RedirectResponse, JSONResponse, Response
-from typing import Any, Optional
+from typing import Annotated, Any, Optional
 from fastapi.templating import Jinja2Templates
 import asyncio
 import hashlib
@@ -438,17 +438,19 @@ async def logout():
 
 
 @router.get(CHANGE_PASSWORD_PATH, response_class=HTMLResponse, responses=AUTH_REDIRECT_RESPONSES)
-async def change_password_page(request: Request, session: dict = Depends(_get_session)):
-    await _get_session(request)
+async def change_password_page(
+    request: Request, 
+    session: Annotated[dict, Depends(_get_session)],
+):
     return templates.TemplateResponse(request, "change_password.html", {"error": None})
 
 
 @router.post(CHANGE_PASSWORD_PATH, response_class=HTMLResponse, responses=AUTH_REDIRECT_RESPONSES)
 async def change_password_submit(
     request: Request,
-    new_password: str = Form(...),
-    confirm_password: str = Form(...),
-    session: dict = Depends(_get_session),
+    new_password: Annotated[str, Form(...)],
+    confirm_password: Annotated[str, Form(...)],
+    session: Annotated[dict, Depends(_get_session)],
 ):
 
     if not new_password or new_password != confirm_password:

From 79b6b587cf7730a31bb065325e7ff5540b471bf5 Mon Sep 17 00:00:00 2001
From: Carol <carol.ouellet@gmail.com>
Date: Mon, 3 Aug 2026 19:08:56 -0400
Subject: [PATCH 5/5] fix(routes): handle case when session_secret is not found
 in database

---
 api/routes.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/api/routes.py b/api/routes.py
index 6cdd947..58c5a52 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -151,7 +151,7 @@ async def ensure_session_secret() -> None:
             "SELECT value FROM settings WHERE key = 'session_secret'"
         ) as select_cursor:
             row = await select_cursor.fetchone()
-        resolved = row[0]
+        resolved = row[0] if row else candidate
 
         if insert_cursor.rowcount == 0:
             logger.warning(