fix: resolve CodeQL alerts, improve test coverage, fix StaticPool for HTTP tests

Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/07b58025-2aeb-4e81-a168-7a6fccdc3569 Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
2026-03-29 17:08:39 +00:00
parent 883d54a428
commit 84117abd13
4 changed files with 405 additions and 19 deletions
@@ -2,6 +2,9 @@
 Tests for MailSource model and mail-sources API endpoints.
 """

+from unittest.mock import MagicMock, patch
+
+import pytest
 from fastapi.testclient import TestClient
 from sqlalchemy.orm import Session

@@ -71,24 +74,19 @@ class TestMailSourceModel:


 class TestMailSourcesAPI:
-    """Integration tests for /api/v1/mail-sources endpoints."""
+    """Integration tests for /api/v1/mail-sources endpoints (no auth)."""

-    API_KEY_HEADER = {"X-API-Key": "test-key"}  # will be set in fixture
-
-    def _get_headers(self, client: TestClient) -> dict:
-        """Return auth headers using the running app's in-memory API key."""
-        # The test app generates its own key in-memory; we can't predict it.
-        # Use the TestClient without auth to check 401, and skip auth-required tests
-        # by injecting the dependency override instead.
-        return {}
-
-    def test_list_empty(self, client: TestClient):
+    def test_list_requires_auth(self, client: TestClient):
        resp = client.get("/api/v1/mail-sources")
        # Without auth, expect 401 or 403
        assert resp.status_code in (401, 403)

-    def test_create_and_list(self, client: TestClient, db_session: Session):
-        """Create a mail source directly in DB and list via authenticated-less read."""
+    def test_create_requires_auth(self, client: TestClient):
+        resp = client.post("/api/v1/mail-sources", json={"name": "x", "method": "IMAP"})
+        assert resp.status_code in (401, 403)
+
+    def test_model_create_and_list(self, client: TestClient, db_session: Session):
+        """Create a mail source directly in DB and verify it's retrievable."""
        source = MailSource(
            name="Direct DB Source",
            method="IMAP",
@@ -153,3 +151,350 @@ class TestMailSourcesAPI:
        assert "Enabled A" in names
        assert "Enabled B" in names
        assert "Disabled" not in names
+
+
+# ---------------------------------------------------------------------------
+# Authenticated HTTP API tests (uses authed_client fixture from conftest)
+# ---------------------------------------------------------------------------
+
+
+class TestMailSourcesAPIAuthed:
+    """HTTP-level tests using the authed_client fixture (auth dependency bypassed)."""
+
+    # ------------------------------------------------------------------
+    # List
+    # ------------------------------------------------------------------
+
+    def test_list_empty(self, authed_client: TestClient):
+        resp = authed_client.get("/api/v1/mail-sources")
+        assert resp.status_code == 200
+        assert resp.json() == []
+
+    # ------------------------------------------------------------------
+    # Create
+    # ------------------------------------------------------------------
+
+    def test_create_imap_source(self, authed_client: TestClient):
+        payload = {
+            "name": "My IMAP",
+            "method": "IMAP",
+            "server": "imap.example.com",
+            "port": 993,
+            "username": "user@example.com",
+            "password": "s3cr3t",
+            "use_ssl": True,
+            "folder": "INBOX",
+            "polling_interval": 60,
+            "enabled": True,
+        }
+        resp = authed_client.post("/api/v1/mail-sources", json=payload)
+        assert resp.status_code == 201
+        data = resp.json()
+        assert data["name"] == "My IMAP"
+        assert data["method"] == "IMAP"
+        assert data["server"] == "imap.example.com"
+        assert data["password"] == "**redacted**"
+        assert data["id"] is not None
+
+    def test_create_normalizes_method_to_uppercase(self, authed_client: TestClient):
+        payload = {"name": "lowercase method", "method": "imap"}
+        resp = authed_client.post("/api/v1/mail-sources", json=payload)
+        assert resp.status_code == 201
+        assert resp.json()["method"] == "IMAP"
+
+    # ------------------------------------------------------------------
+    # Get single
+    # ------------------------------------------------------------------
+
+    def test_get_existing_source(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources", json={"name": "Get Test", "method": "IMAP"}
+        )
+        source_id = create_resp.json()["id"]
+
+        resp = authed_client.get(f"/api/v1/mail-sources/{source_id}")
+        assert resp.status_code == 200
+        assert resp.json()["id"] == source_id
+        assert resp.json()["name"] == "Get Test"
+
+    def test_get_nonexistent_source_returns_404(self, authed_client: TestClient):
+        resp = authed_client.get("/api/v1/mail-sources/99999")
+        assert resp.status_code == 404
+
+    # ------------------------------------------------------------------
+    # Update
+    # ------------------------------------------------------------------
+
+    def test_update_name(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources", json={"name": "Original", "method": "IMAP"}
+        )
+        source_id = create_resp.json()["id"]
+
+        update_resp = authed_client.put(
+            f"/api/v1/mail-sources/{source_id}", json={"name": "Updated"}
+        )
+        assert update_resp.status_code == 200
+        assert update_resp.json()["name"] == "Updated"
+
+    def test_update_method_normalizes_uppercase(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources", json={"name": "MethodTest", "method": "IMAP"}
+        )
+        source_id = create_resp.json()["id"]
+
+        update_resp = authed_client.put(
+            f"/api/v1/mail-sources/{source_id}", json={"method": "pop3"}
+        )
+        assert update_resp.status_code == 200
+        assert update_resp.json()["method"] == "POP3"
+
+    def test_update_nonexistent_source_returns_404(self, authed_client: TestClient):
+        resp = authed_client.put("/api/v1/mail-sources/99999", json={"name": "x"})
+        assert resp.status_code == 404
+
+    # ------------------------------------------------------------------
+    # Delete
+    # ------------------------------------------------------------------
+
+    def test_delete_source(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources", json={"name": "Delete Me", "method": "IMAP"}
+        )
+        source_id = create_resp.json()["id"]
+
+        del_resp = authed_client.delete(f"/api/v1/mail-sources/{source_id}")
+        assert del_resp.status_code == 204
+
+        # Verify gone
+        get_resp = authed_client.get(f"/api/v1/mail-sources/{source_id}")
+        assert get_resp.status_code == 404
+
+    def test_delete_nonexistent_source_returns_404(self, authed_client: TestClient):
+        resp = authed_client.delete("/api/v1/mail-sources/99999")
+        assert resp.status_code == 404
+
+    # ------------------------------------------------------------------
+    # List after creates
+    # ------------------------------------------------------------------
+
+    def test_list_multiple_sources(self, authed_client: TestClient):
+        for i in range(3):
+            authed_client.post(
+                "/api/v1/mail-sources", json={"name": f"Source {i}", "method": "IMAP"}
+            )
+        resp = authed_client.get("/api/v1/mail-sources")
+        assert resp.status_code == 200
+        assert len(resp.json()) == 3
+
+    # ------------------------------------------------------------------
+    # Toggle
+    # ------------------------------------------------------------------
+
+    def test_toggle_disables_then_enables(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources", json={"name": "Toggle", "method": "IMAP", "enabled": True}
+        )
+        source_id = create_resp.json()["id"]
+
+        toggle_resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/toggle")
+        assert toggle_resp.status_code == 200
+        assert toggle_resp.json()["enabled"] is False
+
+        toggle_resp2 = authed_client.post(f"/api/v1/mail-sources/{source_id}/toggle")
+        assert toggle_resp2.status_code == 200
+        assert toggle_resp2.json()["enabled"] is True
+
+    def test_toggle_nonexistent_returns_404(self, authed_client: TestClient):
+        resp = authed_client.post("/api/v1/mail-sources/99999/toggle")
+        assert resp.status_code == 404
+
+    # ------------------------------------------------------------------
+    # Test stored source
+    # ------------------------------------------------------------------
+
+    def test_test_stored_imap_source_success(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources",
+            json={
+                "name": "IMAP Test",
+                "method": "IMAP",
+                "server": "imap.example.com",
+                "username": "u",
+                "password": "p",
+            },
+        )
+        source_id = create_resp.json()["id"]
+
+        mock_stats = {
+            "message_count": 10,
+            "unread_count": 2,
+            "dmarc_count": 1,
+            "available_mailboxes": ["INBOX"],
+        }
+        mock_client = MagicMock()
+        mock_client.test_connection.return_value = (True, "Connection successful", mock_stats)
+
+        with patch("app.api.api_v1.endpoints.mail_sources.IMAPClient", return_value=mock_client):
+            resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/test")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is True
+        assert data["message"] == "Connection successful"
+        assert data["message_count"] == 10
+
+    def test_test_stored_imap_source_failure(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources",
+            json={"name": "IMAP Fail", "method": "IMAP", "server": "bad.host"},
+        )
+        source_id = create_resp.json()["id"]
+
+        mock_client = MagicMock()
+        mock_client.test_connection.return_value = (
+            False,
+            "Connection failed. Check server address and credentials.",
+            {},
+        )
+
+        with patch("app.api.api_v1.endpoints.mail_sources.IMAPClient", return_value=mock_client):
+            resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/test")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+
+    def test_test_stored_non_imap_source(self, authed_client: TestClient):
+        create_resp = authed_client.post(
+            "/api/v1/mail-sources", json={"name": "POP3 Source", "method": "POP3"}
+        )
+        source_id = create_resp.json()["id"]
+
+        resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/test")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "not yet implemented" in data["message"]
+
+    def test_test_stored_nonexistent_returns_404(self, authed_client: TestClient):
+        resp = authed_client.post("/api/v1/mail-sources/99999/test")
+        assert resp.status_code == 404
+
+    # ------------------------------------------------------------------
+    # Ad-hoc test connection
+    # ------------------------------------------------------------------
+
+    def test_adhoc_imap_success(self, authed_client: TestClient):
+        payload = {
+            "server": "imap.example.com",
+            "port": 993,
+            "username": "user@example.com",
+            "password": "secret",
+            "ssl": True,
+            "method": "IMAP",
+        }
+        mock_stats = {
+            "message_count": 5,
+            "unread_count": 1,
+            "dmarc_count": 0,
+            "available_mailboxes": ["INBOX"],
+        }
+        mock_client = MagicMock()
+        mock_client.test_connection.return_value = (True, "Connection successful", mock_stats)
+
+        with patch("app.api.api_v1.endpoints.mail_sources.IMAPClient", return_value=mock_client):
+            resp = authed_client.post("/api/v1/mail-sources/test-connection", json=payload)
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is True
+        assert data["message_count"] == 5
+
+    def test_adhoc_non_imap_returns_not_implemented(self, authed_client: TestClient):
+        payload = {
+            "server": "pop3.example.com",
+            "port": 110,
+            "username": "u",
+            "password": "p",
+            "ssl": False,
+            "method": "POP3",
+        }
+        resp = authed_client.post("/api/v1/mail-sources/test-connection", json=payload)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "not yet implemented" in data["message"]
+
+    def test_adhoc_gmail_api_returns_not_implemented(self, authed_client: TestClient):
+        payload = {"method": "GMAIL_API"}
+        resp = authed_client.post("/api/v1/mail-sources/test-connection", json=payload)
+        assert resp.status_code == 200
+        assert resp.json()["success"] is False
+        assert "not yet implemented" in resp.json()["message"]
+
+
+# ---------------------------------------------------------------------------
+# _sanitize_for_log helper
+# ---------------------------------------------------------------------------
+
+
+class TestSanitizeForLog:
+    """Unit tests for the _sanitize_for_log helper."""
+
+    def test_strips_newline(self):
+        from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log
+
+        assert "\n" not in _sanitize_for_log("hello\nworld")
+
+    def test_strips_carriage_return(self):
+        from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log
+
+        assert "\r" not in _sanitize_for_log("foo\rbar")
+
+    def test_integer_is_safe(self):
+        from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log
+
+        assert _sanitize_for_log(42) == "42"
+
+    def test_normal_string_unchanged(self):
+        from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log
+
+        assert _sanitize_for_log("example.com") == "example.com"
+
+
+# ---------------------------------------------------------------------------
+# Source-to-response helper (password masking)
+# ---------------------------------------------------------------------------
+
+
+class TestSourceToResponse:
+    """Tests for the _source_to_response password-masking helper."""
+
+    def test_password_is_redacted_when_set(self, db_session: Session):
+        from app.api.api_v1.endpoints.mail_sources import _source_to_response
+
+        source = MailSource(name="Redact Test", method="IMAP", password="plaintext")
+        db_session.add(source)
+        db_session.commit()
+        db_session.refresh(source)
+
+        response = _source_to_response(source)
+        assert response.password == "**redacted**"
+
+    def test_password_is_none_when_not_set(self, db_session: Session):
+        from app.api.api_v1.endpoints.mail_sources import _source_to_response
+
+        source = MailSource(name="No Password", method="IMAP")
+        db_session.add(source)
+        db_session.commit()
+        db_session.refresh(source)
+
+        response = _source_to_response(source)
+        assert response.password is None
+
+
+# ---------------------------------------------------------------------------
+# Pytest marker to avoid warnings for test methods without assertions
+# ---------------------------------------------------------------------------
+pytestmark = pytest.mark.usefixtures("_reset_report_store")