Merge pull request #80 from christianlouis/copilot/increase-test-coverage-80
Increase test coverage to >80% with CI enforcement
This commit is contained in:
Christian Krakau-Louis
@@ -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
|
||||
|
||||
@@ -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"""\
|
||||
<?xml version="1.0"?>
|
||||
<feedback>
|
||||
<report_metadata>
|
||||
<org_name>Test Org</org_name>
|
||||
<email>noreply@example.com</email>
|
||||
<report_id>abc-123</report_id>
|
||||
<date_range>
|
||||
<begin>1609459200</begin>
|
||||
<end>1609545600</end>
|
||||
</date_range>
|
||||
</report_metadata>
|
||||
<policy_published>
|
||||
<domain>example.com</domain>
|
||||
<adkim>r</adkim>
|
||||
<aspf>r</aspf>
|
||||
<p>none</p>
|
||||
<sp>none</sp>
|
||||
<pct>100</pct>
|
||||
</policy_published>
|
||||
<record>
|
||||
<row>
|
||||
<source_ip>1.2.3.4</source_ip>
|
||||
<count>1</count>
|
||||
<policy_evaluated>
|
||||
<disposition>none</disposition>
|
||||
<dkim>pass</dkim>
|
||||
<spf>pass</spf>
|
||||
</policy_evaluated>
|
||||
</row>
|
||||
<identifiers>
|
||||
<header_from>example.com</header_from>
|
||||
</identifiers>
|
||||
<auth_results>
|
||||
<dkim>
|
||||
<domain>example.com</domain>
|
||||
<result>pass</result>
|
||||
</dkim>
|
||||
<spf>
|
||||
<domain>example.com</domain>
|
||||
<result>pass</result>
|
||||
</spf>
|
||||
</auth_results>
|
||||
</record>
|
||||
</feedback>
|
||||
"""
|
||||
|
||||
|
||||
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"<xml/>", 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
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user