diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b52f2d6..1b86e95 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -92,7 +92,7 @@ jobs: - name: Run tests with coverage run: | cd backend - pytest --cov=app --cov-report=xml --cov-report=term-missing + pytest --cov=app --cov-report=xml --cov-report=term-missing --cov-fail-under=80 - name: Upload coverage to Codecov uses: codecov/codecov-action@v4 diff --git a/backend/app/tests/test_imap_client.py b/backend/app/tests/test_imap_client.py new file mode 100644 index 0000000..1a4ed35 --- /dev/null +++ b/backend/app/tests/test_imap_client.py @@ -0,0 +1,615 @@ +""" +Tests for IMAPClient service. + +Covers connection testing, mailbox listing, email processing, attachment parsing, +and report fetching with mocked IMAP connections. +""" + +import email +import imaplib +from email.mime.application import MIMEApplication +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText +from io import BytesIO +from unittest.mock import MagicMock, patch +from zipfile import ZipFile + +import pytest + +from app.services.imap_client import IMAPClient +from app.services.report_store import ReportStore + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +MINIMAL_DMARC_XML = b"""\ + + + + Test Org + noreply@example.com + abc-123 + + 1609459200 + 1609545600 + + + + example.com + r + r +

none

+ none + 100 + + + + 1.2.3.4 + 1 + + none + pass + pass + + + + example.com + + + + example.com + pass + + + example.com + pass + + + + +""" + + +def _make_zip_content(xml_bytes: bytes, filename: str = "report.xml") -> bytes: + buf = BytesIO() + with ZipFile(buf, "w") as zf: + zf.writestr(filename, xml_bytes) + return buf.getvalue() + + +def _make_email_with_attachment( + filename: str = "dmarc-report.xml", + content: bytes = MINIMAL_DMARC_XML, + content_type: str = "application/xml", + subject: str = "DMARC Report", + from_addr: str = "noreply@example.com", +) -> bytes: + msg = MIMEMultipart() + msg["Subject"] = subject + msg["From"] = from_addr + msg.attach(MIMEText("DMARC report attached.")) + part = MIMEApplication(content, Name=filename) + part["Content-Disposition"] = f'attachment; filename="{filename}"' + part.set_type(content_type) + msg.attach(part) + return msg.as_bytes() + + +# --------------------------------------------------------------------------- +# TestIMAPClientInit +# --------------------------------------------------------------------------- + + +class TestIMAPClientInit: + def test_default_construction_with_missing_credentials(self): + """Client can be instantiated even without configured settings.""" + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER=None, + IMAP_PORT=993, + IMAP_USERNAME=None, + IMAP_PASSWORD=None, + ) + client = IMAPClient() + assert client.server is None + assert client.username is None + + def test_explicit_credentials_override_settings(self): + """Explicit constructor arguments take precedence over settings.""" + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER="default.example.com", + IMAP_PORT=993, + IMAP_USERNAME="default@example.com", + IMAP_PASSWORD="default-password", + ) + client = IMAPClient( + server="custom.example.com", + port=143, + username="user@example.com", + password="secret", + delete_emails=True, + ) + assert client.server == "custom.example.com" + assert client.port == 143 + assert client.username == "user@example.com" + assert client.password == "secret" + assert client.delete_emails is True + + def test_report_store_assigned(self): + """IMAPClient stores a reference to the ReportStore singleton.""" + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER="imap.example.com", + IMAP_PORT=993, + IMAP_USERNAME="u", + IMAP_PASSWORD="p", + ) + client = IMAPClient() + assert client.report_store is ReportStore.get_instance() + + +# --------------------------------------------------------------------------- +# TestListMailboxes +# --------------------------------------------------------------------------- + + +class TestListMailboxes: + def _make_client(self): + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER="imap.example.com", + IMAP_PORT=993, + IMAP_USERNAME="u", + IMAP_PASSWORD="p", + ) + return IMAPClient() + + def test_parses_standard_mailbox_entry(self): + client = self._make_client() + raw = [b'(\\HasNoChildren) "/" INBOX'] + result = client._list_mailboxes(raw) + assert "INBOX" in result + + def test_skips_non_bytes_entries(self): + client = self._make_client() + result = client._list_mailboxes(["not bytes", None]) # type: ignore[list-item] + assert result == [] + + def test_handles_malformed_bytes(self): + client = self._make_client() + # bytes that can't be decoded normally should be silently skipped + result = client._list_mailboxes([b"short"]) + # Should not raise; may return empty or partial result + assert isinstance(result, list) + + def test_multiple_mailboxes(self): + client = self._make_client() + raw = [ + b'(\\HasNoChildren) "/" INBOX', + b'(\\HasNoChildren) "/" Sent', + b'(\\HasNoChildren) "/" Trash', + ] + result = client._list_mailboxes(raw) + assert len(result) == 3 + + +# --------------------------------------------------------------------------- +# TestTestConnection +# --------------------------------------------------------------------------- + + +class TestTestConnection: + def _make_client(self, server="imap.example.com", username="u", password="p"): + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER=server, + IMAP_PORT=993, + IMAP_USERNAME=username, + IMAP_PASSWORD=password, + ) + return IMAPClient() + + def test_returns_false_when_missing_credentials(self): + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER=None, + IMAP_PORT=993, + IMAP_USERNAME=None, + IMAP_PASSWORD=None, + ) + client = IMAPClient() + success, message, stats = client.test_connection() + assert success is False + assert "not fully configured" in message + assert stats == {} + + def test_successful_connection(self): + client = self._make_client() + mock_mail = MagicMock() + mock_mail.login.return_value = None + mock_mail.list.return_value = ("OK", [b'(\\HasNoChildren) "/" INBOX']) + mock_mail.select.return_value = ("OK", [b"10"]) + mock_mail.search.return_value = ("OK", [b"1 2 3"]) + + with patch("imaplib.IMAP4_SSL", return_value=mock_mail): + success, message, stats = client.test_connection() + + assert success is True + assert "successful" in message.lower() + assert stats["message_count"] == 10 + assert "INBOX" in stats["available_mailboxes"] + + def test_connection_exception_returns_false(self): + client = self._make_client() + with patch("imaplib.IMAP4_SSL", side_effect=ConnectionRefusedError("refused")): + success, message, stats = client.test_connection() + assert success is False + assert stats == {} + + def test_list_status_not_ok_returns_empty_mailboxes(self): + client = self._make_client() + mock_mail = MagicMock() + mock_mail.login.return_value = None + mock_mail.list.return_value = ("NO", []) + mock_mail.select.return_value = ("OK", [b"0"]) + mock_mail.search.return_value = ("OK", [b""]) + + with patch("imaplib.IMAP4_SSL", return_value=mock_mail): + success, message, stats = client.test_connection() + + assert success is True + assert stats["available_mailboxes"] == [] + + def test_select_not_ok_skips_message_count(self): + client = self._make_client() + mock_mail = MagicMock() + mock_mail.login.return_value = None + mock_mail.list.return_value = ("OK", []) + mock_mail.select.return_value = ("NO", []) + mock_mail.search.return_value = ("OK", [b""]) + + with patch("imaplib.IMAP4_SSL", return_value=mock_mail): + success, message, stats = client.test_connection() + + assert success is True + assert stats["message_count"] == 0 + + +# --------------------------------------------------------------------------- +# TestIsDmarcReportEmail +# --------------------------------------------------------------------------- + + +class TestIsDmarcReportEmail: + def _make_client(self): + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER="imap.example.com", + IMAP_PORT=993, + IMAP_USERNAME="u", + IMAP_PASSWORD="p", + ) + return IMAPClient() + + def _make_msg(self, subject="", from_addr="", has_xml_attachment=False): + msg = MIMEMultipart() + msg["Subject"] = subject + msg["From"] = from_addr + if has_xml_attachment: + part = MIMEApplication(b"", Name="report.xml") + part["Content-Disposition"] = 'attachment; filename="report.xml"' + msg.attach(part) + return msg + + def test_dmarc_keyword_in_subject(self): + client = self._make_client() + msg = self._make_msg(subject="DMARC Aggregate Report for example.com") + assert client._is_dmarc_report_email(msg) is True + + def test_no_keywords_no_attachments(self): + client = self._make_client() + msg = self._make_msg(subject="Hello World", from_addr="friend@example.com") + assert client._is_dmarc_report_email(msg) is False + + def test_dmarc_sender_matches(self): + client = self._make_client() + msg = self._make_msg( + subject="Weekly report", from_addr="noreply@google.com" + ) + assert client._is_dmarc_report_email(msg) is True + + def test_xml_attachment_matches(self): + client = self._make_client() + msg = self._make_msg(has_xml_attachment=True) + assert client._is_dmarc_report_email(msg) is True + + +# --------------------------------------------------------------------------- +# TestDecodeEmailHeader +# --------------------------------------------------------------------------- + + +class TestDecodeEmailHeader: + def _make_client(self): + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER="imap.example.com", + IMAP_PORT=993, + IMAP_USERNAME="u", + IMAP_PASSWORD="p", + ) + return IMAPClient() + + def test_plain_ascii_header(self): + client = self._make_client() + assert client._decode_email_header("Hello World") == "Hello World" + + def test_encoded_utf8_header(self): + client = self._make_client() + # "=?utf-8?b?..." encoded header + encoded = "=?utf-8?b?RFNIQVJDIG9yZyBuYW1l?=" + result = client._decode_email_header(encoded) + assert isinstance(result, str) + + +# --------------------------------------------------------------------------- +# TestHasDmarcAttachments +# --------------------------------------------------------------------------- + + +class TestHasDmarcAttachments: + def _make_client(self): + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER="imap.example.com", + IMAP_PORT=993, + IMAP_USERNAME="u", + IMAP_PASSWORD="p", + ) + return IMAPClient() + + @pytest.mark.parametrize( + "filename", + ["report.xml", "dmarc.zip", "report.gz", "report.gzip"], + ) + def test_dmarc_filename_extensions(self, filename): + client = self._make_client() + msg = MIMEMultipart() + part = MIMEApplication(b"data", Name=filename) + part["Content-Disposition"] = f'attachment; filename="{filename}"' + msg.attach(part) + assert client._has_dmarc_attachments(msg) is True + + @pytest.mark.parametrize( + "content_type", + [ + "application/zip", + "application/gzip", + "application/x-gzip", + "application/xml", + "text/xml", + ], + ) + def test_dmarc_content_types(self, content_type): + client = self._make_client() + msg = MIMEMultipart() + part = MIMEApplication(b"data") + part["Content-Disposition"] = "attachment" + part.set_type(content_type) + msg.attach(part) + assert client._has_dmarc_attachments(msg) is True + + def test_no_attachments_returns_false(self): + client = self._make_client() + msg = MIMEText("plain text body") + assert client._has_dmarc_attachments(msg) is False + + +# --------------------------------------------------------------------------- +# TestProcessAttachments +# --------------------------------------------------------------------------- + + +class TestProcessAttachments: + def _make_client(self): + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER="imap.example.com", + IMAP_PORT=993, + IMAP_USERNAME="u", + IMAP_PASSWORD="p", + ) + return IMAPClient() + + def test_processes_xml_attachment(self): + client = self._make_client() + msg = email.message_from_bytes( + _make_email_with_attachment("report.xml", MINIMAL_DMARC_XML, "application/xml") + ) + count = client._process_attachments(msg) + assert count == 1 + + def test_processes_zip_attachment(self): + client = self._make_client() + zip_content = _make_zip_content(MINIMAL_DMARC_XML, "report.xml") + msg = email.message_from_bytes( + _make_email_with_attachment("report.zip", zip_content, "application/zip") + ) + count = client._process_attachments(msg) + assert count == 1 + + def test_bad_attachment_does_not_raise(self): + client = self._make_client() + msg = email.message_from_bytes( + _make_email_with_attachment("report.xml", b"not xml at all") + ) + # Should not raise; just returns 0 + count = client._process_attachments(msg) + assert count == 0 + + def test_no_attachments_returns_zero(self): + client = self._make_client() + msg = MIMEText("Just text, no attachments.") + count = client._process_attachments(msg) + assert count == 0 + + +# --------------------------------------------------------------------------- +# TestProcessSingleEmail +# --------------------------------------------------------------------------- + + +class TestProcessSingleEmail: + def _make_client(self): + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER="imap.example.com", + IMAP_PORT=993, + IMAP_USERNAME="u", + IMAP_PASSWORD="p", + ) + return IMAPClient() + + def test_processes_valid_dmarc_email(self): + client = self._make_client() + raw = _make_email_with_attachment( + "report.xml", + MINIMAL_DMARC_XML, + "application/xml", + subject="DMARC Report", + ) + mock_mail = MagicMock() + mock_mail.fetch.return_value = ("OK", [(b"1", raw)]) + mock_mail.store.return_value = ("OK", None) + + stats = {"processed": 0, "reports_found": 0, "errors": []} + client._process_single_email(mock_mail, b"1", stats) + + assert stats["processed"] == 1 + assert stats["reports_found"] == 1 + + def test_fetch_error_skips_email(self): + client = self._make_client() + mock_mail = MagicMock() + mock_mail.fetch.return_value = ("NO", []) + + stats = {"processed": 0, "reports_found": 0, "errors": []} + client._process_single_email(mock_mail, b"1", stats) + + assert stats["processed"] == 0 + + def test_exception_adds_to_errors(self): + client = self._make_client() + mock_mail = MagicMock() + mock_mail.fetch.side_effect = RuntimeError("unexpected error") + + stats = {"processed": 0, "reports_found": 0, "errors": []} + client._process_single_email(mock_mail, b"1", stats) + + assert len(stats["errors"]) == 1 + + def test_marks_deleted_when_flag_set(self): + client = self._make_client() + client.delete_emails = True + raw = _make_email_with_attachment( + "report.xml", + MINIMAL_DMARC_XML, + "application/xml", + subject="DMARC Report", + ) + mock_mail = MagicMock() + mock_mail.fetch.return_value = ("OK", [(b"1", raw)]) + mock_mail.store.return_value = ("OK", None) + + stats = {"processed": 0, "reports_found": 0, "errors": []} + client._process_single_email(mock_mail, b"1", stats) + + # store should have been called twice: once for \\Seen, once for \\Deleted + assert mock_mail.store.call_count >= 2 + + +# --------------------------------------------------------------------------- +# TestFetchReports +# --------------------------------------------------------------------------- + + +class TestFetchReports: + def _make_client(self, server="imap.example.com", username="u", password="p"): + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER=server, + IMAP_PORT=993, + IMAP_USERNAME=username, + IMAP_PASSWORD=password, + ) + return IMAPClient() + + def test_returns_error_when_no_credentials(self): + with patch("app.services.imap_client.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + IMAP_SERVER=None, + IMAP_PORT=993, + IMAP_USERNAME=None, + IMAP_PASSWORD=None, + ) + client = IMAPClient() + result = client.fetch_reports() + assert result["success"] is False + + def test_search_failure_returns_error(self): + client = self._make_client() + mock_mail = MagicMock() + mock_mail.login.return_value = None + mock_mail.select.return_value = ("OK", [b"0"]) + mock_mail.search.return_value = ("NO", []) + + with patch("imaplib.IMAP4_SSL", return_value=mock_mail): + result = client.fetch_reports(days=7) + + assert result["success"] is False + + def test_successful_fetch_with_email(self): + client = self._make_client() + raw = _make_email_with_attachment( + "report.xml", + MINIMAL_DMARC_XML, + "application/xml", + subject="DMARC Report", + ) + mock_mail = MagicMock() + mock_mail.login.return_value = None + mock_mail.select.return_value = ("OK", [b"1"]) + mock_mail.search.return_value = ("OK", [b"1"]) + mock_mail.fetch.return_value = ("OK", [(b"1", raw)]) + mock_mail.store.return_value = ("OK", None) + mock_mail.logout.return_value = None + + with patch("imaplib.IMAP4_SSL", return_value=mock_mail): + result = client.fetch_reports(days=7) + + assert result["success"] is True + assert result["reports_found"] >= 1 + + def test_connection_error_returns_failure(self): + client = self._make_client() + with patch("imaplib.IMAP4_SSL", side_effect=imaplib.IMAP4.error("connection error")): + result = client.fetch_reports(days=7) + assert result["success"] is False + + def test_delete_emails_calls_expunge(self): + client = self._make_client() + client.delete_emails = True + mock_mail = MagicMock() + mock_mail.login.return_value = None + mock_mail.select.return_value = ("OK", [b"0"]) + mock_mail.search.return_value = ("OK", [b""]) + mock_mail.logout.return_value = None + + with patch("imaplib.IMAP4_SSL", return_value=mock_mail): + result = client.fetch_reports(days=3) + + mock_mail.expunge.assert_called_once() + assert result["success"] is True diff --git a/backend/app/tests/test_imap_endpoints.py b/backend/app/tests/test_imap_endpoints.py new file mode 100644 index 0000000..d868917 --- /dev/null +++ b/backend/app/tests/test_imap_endpoints.py @@ -0,0 +1,160 @@ +""" +Tests for the /api/v1/imap endpoints. + +Covers connection testing, report fetching (foreground and background), and status. +""" + +from unittest.mock import MagicMock, patch + +from fastapi.testclient import TestClient + + +class TestImapTestConnection: + """Tests for POST /api/v1/imap/test-connection""" + + def test_successful_connection(self, authed_client: TestClient): + mock_client = MagicMock() + mock_client.test_connection.return_value = ( + True, + "Connection successful", + { + "message_count": 5, + "unread_count": 2, + "dmarc_count": 1, + "available_mailboxes": ["INBOX", "Sent"], + "server": "imap.example.com", + "port": 993, + }, + ) + + with patch("app.api.api_v1.endpoints.imap.IMAPClient", return_value=mock_client): + response = authed_client.post( + "/api/v1/imap/test-connection", + json={ + "server": "imap.example.com", + "port": 993, + "username": "user@example.com", + "password": "secret", + "ssl": True, + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["success"] is True + assert data["message_count"] == 5 + assert data["unread_count"] == 2 + assert "INBOX" in data["available_mailboxes"] + + def test_failed_connection(self, authed_client: TestClient): + mock_client = MagicMock() + mock_client.test_connection.return_value = (False, "Connection failed", {}) + + with patch("app.api.api_v1.endpoints.imap.IMAPClient", return_value=mock_client): + response = authed_client.post( + "/api/v1/imap/test-connection", + json={"server": "bad.example.com", "port": 993}, + ) + + assert response.status_code == 200 + data = response.json() + assert data["success"] is False + assert data["message_count"] == 0 + + def test_requires_auth(self, client: TestClient): + """Without auth, the endpoint should return 401.""" + response = client.post( + "/api/v1/imap/test-connection", + json={"server": "imap.example.com", "port": 993}, + ) + assert response.status_code == 401 + + +class TestImapFetchReports: + """Tests for POST /api/v1/imap/fetch-reports""" + + def test_fetch_foreground_success(self, authed_client: TestClient): + mock_client = MagicMock() + mock_client.fetch_reports.return_value = { + "success": True, + "processed": 3, + "reports_found": 2, + "new_domains": ["example.com"], + "errors": [], + } + + with patch("app.api.api_v1.endpoints.imap.IMAPClient", return_value=mock_client): + response = authed_client.post("/api/v1/imap/fetch-reports?days=7") + + assert response.status_code == 200 + data = response.json() + assert data["success"] is True + assert data["processed_emails"] == 3 + assert data["reports_found"] == 2 + + def test_fetch_days_too_low(self, authed_client: TestClient): + response = authed_client.post("/api/v1/imap/fetch-reports?days=0") + assert response.status_code == 400 + + def test_fetch_days_too_high(self, authed_client: TestClient): + response = authed_client.post("/api/v1/imap/fetch-reports?days=400") + assert response.status_code == 400 + + def test_fetch_background_for_long_range(self, authed_client: TestClient): + """Days > 14 should queue a background task.""" + mock_client = MagicMock() + with patch("app.api.api_v1.endpoints.imap.IMAPClient", return_value=mock_client): + response = authed_client.post("/api/v1/imap/fetch-reports?days=30") + + assert response.status_code == 200 + data = response.json() + assert data["success"] is True + assert "background" in data["message"].lower() + + def test_fetch_with_errors_in_result(self, authed_client: TestClient): + mock_client = MagicMock() + mock_client.fetch_reports.return_value = { + "success": True, + "processed": 1, + "reports_found": 0, + "new_domains": [], + "errors": ["Could not parse email 1"], + } + + with patch("app.api.api_v1.endpoints.imap.IMAPClient", return_value=mock_client): + response = authed_client.post("/api/v1/imap/fetch-reports?days=3") + + assert response.status_code == 200 + data = response.json() + assert data["errors"] is not None + + def test_fetch_exception_returns_500(self, authed_client: TestClient): + mock_client = MagicMock() + mock_client.fetch_reports.side_effect = RuntimeError("unexpected") + + with patch("app.api.api_v1.endpoints.imap.IMAPClient", return_value=mock_client): + response = authed_client.post("/api/v1/imap/fetch-reports?days=5") + + assert response.status_code == 500 + + def test_requires_auth(self, client: TestClient): + response = client.post("/api/v1/imap/fetch-reports?days=7") + assert response.status_code == 401 + + +class TestImapStatus: + """Tests for GET /api/v1/imap/status""" + + def test_status_returns_200(self, authed_client: TestClient): + response = authed_client.get("/api/v1/imap/status") + assert response.status_code == 200 + + def test_status_response_structure(self, authed_client: TestClient): + response = authed_client.get("/api/v1/imap/status") + data = response.json() + assert "is_running" in data + assert "timestamp" in data + + def test_requires_auth(self, client: TestClient): + response = client.get("/api/v1/imap/status") + assert response.status_code == 401 diff --git a/backend/app/tests/test_security_extra.py b/backend/app/tests/test_security_extra.py new file mode 100644 index 0000000..48183d1 --- /dev/null +++ b/backend/app/tests/test_security_extra.py @@ -0,0 +1,174 @@ +""" +Additional tests for app.core.security covering JWT, password utilities, +create_access_token, and require_admin_auth branches not yet exercised. +""" + +from datetime import timedelta + +import pytest +from jose import jwt + +from app.core.security import ( + add_api_key, + create_access_token, + generate_api_key, + verify_token, +) + + +# --------------------------------------------------------------------------- +# create_access_token +# --------------------------------------------------------------------------- + + +class TestCreateAccessToken: + def test_returns_decodable_token(self): + from app.core.config import get_settings + + settings = get_settings() + token = create_access_token("test-subject") + payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) + assert payload["sub"] == "test-subject" + + def test_custom_expiry_is_respected(self): + import time + + from app.core.config import get_settings + + settings = get_settings() + delta = timedelta(seconds=60) + token = create_access_token("user@example.com", expires_delta=delta) + payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) + # exp should be roughly `now + 60 seconds` (within a 2-second tolerance) + assert abs(payload["exp"] - (int(time.time()) + 60)) <= 2 + + +# --------------------------------------------------------------------------- +# verify_token (JWT bearer dependency) +# --------------------------------------------------------------------------- + + +class TestVerifyToken: + @pytest.mark.asyncio + async def test_valid_token_returns_payload(self): + token = create_access_token("unit-test-user") + + from fastapi.security import HTTPAuthorizationCredentials + + creds = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token) + payload = await verify_token(creds) + assert payload["sub"] == "unit-test-user" + + @pytest.mark.asyncio + async def test_no_credentials_raises_401(self): + from fastapi import HTTPException + + with pytest.raises(HTTPException) as exc_info: + await verify_token(None) + assert exc_info.value.status_code == 401 + + @pytest.mark.asyncio + async def test_invalid_token_raises_401(self): + from fastapi import HTTPException + from fastapi.security import HTTPAuthorizationCredentials + + creds = HTTPAuthorizationCredentials(scheme="Bearer", credentials="invalid.token.here") + with pytest.raises(HTTPException) as exc_info: + await verify_token(creds) + assert exc_info.value.status_code == 401 + + +# --------------------------------------------------------------------------- +# require_admin_auth – branches: valid API key, valid JWT, no auth +# --------------------------------------------------------------------------- + + +class TestRequireAdminAuth: + @pytest.mark.asyncio + async def test_valid_api_key_returns_auth_context(self): + from app.core.security import require_admin_auth + + key = generate_api_key() + add_api_key(key) + try: + result = await require_admin_auth(api_key=key, bearer=None) + assert result["auth_type"] == "api_key" + finally: + from app.core.security import _api_keys + + _api_keys.discard(key) + + @pytest.mark.asyncio + async def test_valid_jwt_returns_auth_context(self): + from app.core.security import require_admin_auth + + token = create_access_token("admin-user") + from fastapi.security import HTTPAuthorizationCredentials + + creds = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token) + result = await require_admin_auth(api_key=None, bearer=creds) + assert result["auth_type"] == "jwt" + assert result["payload"]["sub"] == "admin-user" + + @pytest.mark.asyncio + async def test_invalid_jwt_and_no_api_key_raises_401(self): + from fastapi import HTTPException + from fastapi.security import HTTPAuthorizationCredentials + + from app.core.security import require_admin_auth + + creds = HTTPAuthorizationCredentials( + scheme="Bearer", credentials="bad.token.value" + ) + with pytest.raises(HTTPException) as exc_info: + await require_admin_auth(api_key=None, bearer=creds) + assert exc_info.value.status_code == 401 + + @pytest.mark.asyncio + async def test_no_auth_at_all_raises_401(self): + from fastapi import HTTPException + + from app.core.security import require_admin_auth + + with pytest.raises(HTTPException) as exc_info: + await require_admin_auth(api_key=None, bearer=None) + assert exc_info.value.status_code == 401 + + +# --------------------------------------------------------------------------- +# get_api_key dependency +# --------------------------------------------------------------------------- + + +class TestGetApiKeyDependency: + @pytest.mark.asyncio + async def test_missing_key_raises_401(self): + from fastapi import HTTPException + + from app.core.security import get_api_key + + with pytest.raises(HTTPException) as exc_info: + await get_api_key(None) + assert exc_info.value.status_code == 401 + + @pytest.mark.asyncio + async def test_invalid_key_raises_401(self): + from fastapi import HTTPException + + from app.core.security import get_api_key + + with pytest.raises(HTTPException) as exc_info: + await get_api_key("this-key-does-not-exist") + assert exc_info.value.status_code == 401 + + @pytest.mark.asyncio + async def test_valid_key_returns_key(self): + from app.core.security import _api_keys, get_api_key + + key = generate_api_key() + add_api_key(key) + try: + result = await get_api_key(key) + assert result == key + finally: + _api_keys.discard(key) diff --git a/backend/app/tests/test_setup_endpoints.py b/backend/app/tests/test_setup_endpoints.py new file mode 100644 index 0000000..e66f8b9 --- /dev/null +++ b/backend/app/tests/test_setup_endpoints.py @@ -0,0 +1,106 @@ +""" +Tests for the /api/v1/setup endpoints. + +Covers initial setup status, admin user setup, and system configuration. +""" + +import pytest +from fastapi.testclient import TestClient + +from app.api.api_v1.endpoints.setup import setup_status + + +@pytest.fixture(autouse=True) +def reset_setup_status(): + """Reset in-memory setup_status before each test to avoid state leakage.""" + original = dict(setup_status) + setup_status["is_setup_complete"] = False + setup_status["admin_email"] = None + setup_status["app_name"] = "DMARQ" + yield + setup_status.update(original) + + +class TestSetupStatus: + """Tests for GET /api/v1/setup/status""" + + def test_initial_status_not_complete(self, client: TestClient): + response = client.get("/api/v1/setup/status") + assert response.status_code == 200 + data = response.json() + assert data["is_setup_complete"] is False + assert data["app_name"] == "DMARQ" + + def test_status_after_system_setup(self, client: TestClient): + client.post( + "/api/v1/setup/system", + json={"app_name": "MyApp", "base_url": "https://example.com"}, + ) + response = client.get("/api/v1/setup/status") + data = response.json() + assert data["is_setup_complete"] is True + assert data["app_name"] == "MyApp" + + +class TestSetupAdmin: + """Tests for POST /api/v1/setup/admin""" + + def test_admin_setup_succeeds_on_first_call(self, client: TestClient): + response = client.post( + "/api/v1/setup/admin", + json={ + "email": "admin@example.com", + "username": "admin", + "password": "test-placeholder-password", + }, + ) + assert response.status_code == 201 + assert "message" in response.json() + + def test_admin_setup_fails_if_already_complete(self, client: TestClient): + # Mark setup as complete + setup_status["is_setup_complete"] = True + + response = client.post( + "/api/v1/setup/admin", + json={ + "email": "admin2@example.com", + "username": "admin2", + "password": "pass", + }, + ) + assert response.status_code == 400 + assert "already completed" in response.json()["detail"] + + def test_admin_setup_rejects_invalid_email(self, client: TestClient): + response = client.post( + "/api/v1/setup/admin", + json={ + "email": "not-an-email", + "username": "admin", + "password": "pass", + }, + ) + assert response.status_code == 422 + + +class TestSetupSystem: + """Tests for POST /api/v1/setup/system""" + + def test_system_setup_saves_app_name(self, client: TestClient): + response = client.post( + "/api/v1/setup/system", + json={"app_name": "DMARQ Custom", "base_url": "https://dmarq.example.com"}, + ) + assert response.status_code == 200 + assert "message" in response.json() + assert setup_status["app_name"] == "DMARQ Custom" + assert setup_status["is_setup_complete"] is True + + def test_system_setup_marks_complete(self, client: TestClient): + assert setup_status["is_setup_complete"] is False + client.post( + "/api/v1/setup/system", + json={"app_name": "Test", "base_url": "https://test.example.com"}, + ) + assert setup_status["is_setup_complete"] is True diff --git a/backend/app/tests/test_stats_endpoints.py b/backend/app/tests/test_stats_endpoints.py new file mode 100644 index 0000000..bdf5e35 --- /dev/null +++ b/backend/app/tests/test_stats_endpoints.py @@ -0,0 +1,115 @@ +""" +Tests for the /api/v1/stats endpoints. + +Covers dashboard statistics and per-domain statistics with cache refresh. +""" + +from unittest.mock import MagicMock, patch + +from fastapi.testclient import TestClient + + +class TestDashboardStatistics: + """Tests for GET /api/v1/stats/dashboard""" + + def test_dashboard_returns_200(self, client: TestClient): + response = client.get("/api/v1/stats/dashboard") + assert response.status_code == 200 + + def test_dashboard_response_contains_api_version(self, client: TestClient): + response = client.get("/api/v1/stats/dashboard") + data = response.json() + assert data["api_version"] == "1.0" + + def test_dashboard_response_contains_period_days(self, client: TestClient): + response = client.get("/api/v1/stats/dashboard") + data = response.json() + assert data["period_days"] == 30 + + def test_dashboard_period_days_query_param(self, client: TestClient): + response = client.get("/api/v1/stats/dashboard?period_days=7") + assert response.status_code == 200 + data = response.json() + assert data["period_days"] == 7 + + def test_dashboard_force_refresh(self, client: TestClient): + """force_refresh=true should trigger cache invalidation without error.""" + response = client.get("/api/v1/stats/dashboard?force_refresh=true") + assert response.status_code == 200 + data = response.json() + assert "api_version" in data + + def test_dashboard_force_refresh_calls_invalidate_cache(self, client: TestClient): + """Verify StatsSummarizer.invalidate_cache is called when force_refresh is set.""" + with patch("app.api.api_v1.endpoints.stats.StatsSummarizer") as MockSummarizer: + mock_instance = MagicMock() + mock_instance.calculate_summary_statistics.return_value = {"total": 0} + MockSummarizer.return_value = mock_instance + + response = client.get("/api/v1/stats/dashboard?force_refresh=true") + assert response.status_code == 200 + mock_instance.invalidate_cache.assert_called_once() + + def test_dashboard_no_force_refresh_skips_invalidate(self, client: TestClient): + """Without force_refresh, invalidate_cache should NOT be called.""" + with patch("app.api.api_v1.endpoints.stats.StatsSummarizer") as MockSummarizer: + mock_instance = MagicMock() + mock_instance.calculate_summary_statistics.return_value = {"total": 0} + MockSummarizer.return_value = mock_instance + + response = client.get("/api/v1/stats/dashboard") + assert response.status_code == 200 + mock_instance.invalidate_cache.assert_not_called() + + +class TestDomainStatistics: + """Tests for GET /api/v1/stats/domain/{domain_id}""" + + def test_domain_stats_returns_200(self, client: TestClient): + response = client.get("/api/v1/stats/domain/example.com") + assert response.status_code == 200 + + def test_domain_stats_contains_api_version(self, client: TestClient): + response = client.get("/api/v1/stats/domain/example.com") + data = response.json() + assert data["api_version"] == "1.0" + + def test_domain_stats_contains_period_days(self, client: TestClient): + response = client.get("/api/v1/stats/domain/example.com") + data = response.json() + assert data["period_days"] == 30 + + def test_domain_stats_custom_period(self, client: TestClient): + response = client.get("/api/v1/stats/domain/example.com?period_days=14") + assert response.status_code == 200 + data = response.json() + assert data["period_days"] == 14 + + def test_domain_stats_force_refresh(self, client: TestClient): + response = client.get("/api/v1/stats/domain/example.com?force_refresh=true") + assert response.status_code == 200 + + def test_domain_stats_force_refresh_calls_invalidate_with_domain( + self, client: TestClient + ): + """Verify invalidate_cache is called with the domain ID.""" + with patch("app.api.api_v1.endpoints.stats.StatsSummarizer") as MockSummarizer: + mock_instance = MagicMock() + mock_instance.calculate_summary_statistics.return_value = {"total": 0} + MockSummarizer.return_value = mock_instance + + response = client.get( + "/api/v1/stats/domain/example.com?force_refresh=true" + ) + assert response.status_code == 200 + mock_instance.invalidate_cache.assert_called_once_with("example.com") + + def test_domain_stats_no_force_refresh_skips_invalidate(self, client: TestClient): + with patch("app.api.api_v1.endpoints.stats.StatsSummarizer") as MockSummarizer: + mock_instance = MagicMock() + mock_instance.calculate_summary_statistics.return_value = {"total": 0} + MockSummarizer.return_value = mock_instance + + response = client.get("/api/v1/stats/domain/example.com") + assert response.status_code == 200 + mock_instance.invalidate_cache.assert_not_called()