diff --git a/backend/app/tests/test_forensic_parser.py b/backend/app/tests/test_forensic_parser.py index 21cec73..54eca27 100644 --- a/backend/app/tests/test_forensic_parser.py +++ b/backend/app/tests/test_forensic_parser.py @@ -1,8 +1,17 @@ import email +from unittest.mock import MagicMock import pytest -from app.services.forensic_parser import ForensicParser, MAX_FORENSIC_REPORT_SIZE +from app.services import forensic_parser as forensic_parser_module +from app.services.forensic_parser import ( + ForensicParser, + MAX_FORENSIC_REPORT_SIZE, + _coerce_text, + _domain_from_address, + _message_part_payload, + _payload_text, +) SAMPLE_FORENSIC_EMAIL = b"""\ @@ -177,3 +186,64 @@ def test_parse_forensic_email_rejects_empty_and_oversized_payloads(): with pytest.raises(ValueError, match="too large"): ForensicParser.parse_bytes(b"x" * (MAX_FORENSIC_REPORT_SIZE + 1)) + + +def test_forensic_parser_helper_fallbacks(monkeypatch): + multipart = email.message_from_bytes( + b"Subject: container\r\nContent-Type: multipart/mixed; boundary=x\r\n\r\n--x--" + ) + message = email.message_from_bytes(b"Subject: plain\r\n\r\nbody") + + assert _coerce_text(b"hello") == "hello" + assert _payload_text(multipart) == "" + assert _message_part_payload(message) is None + assert _domain_from_address("not-an-address") == "" + + part = MagicMock() + part.get_payload.side_effect = [None, ["nested"]] + assert _payload_text(part) == "" + + part = MagicMock() + part.get_payload.side_effect = [None, "plain text"] + assert _payload_text(part) == "plain text" + + monkeypatch.setattr(forensic_parser_module, "parsedate_to_datetime", lambda _value: None) + assert forensic_parser_module._parse_datetime("Fri, 22 May 2026 10:15:00 +0000") is None + + +def test_is_forensic_report_detects_feedback_part_without_report_container(): + msg = email.message_from_bytes( + b"""\ +Subject: Delivery notice +MIME-Version: 1.0 +Content-Type: multipart/mixed; boundary="b" + +--b +Content-Type: message/feedback-report + +Feedback-Type: auth-failure + +--b-- +""" + ) + + assert ForensicParser.is_forensic_report(msg) is True + + +def test_is_forensic_report_ignores_non_dmarc_header_parts(): + msg = email.message_from_bytes( + b"""\ +Subject: DMARC forensic notice +MIME-Version: 1.0 +Content-Type: multipart/mixed; boundary="b" + +--b +Content-Type: text/rfc822-headers + +Authentication-Results: mx; spf=pass + +--b-- +""" + ) + + assert ForensicParser.is_forensic_report(msg) is True diff --git a/backend/app/tests/test_forensics_api.py b/backend/app/tests/test_forensics_api.py index 2957278..d968dc6 100644 --- a/backend/app/tests/test_forensics_api.py +++ b/backend/app/tests/test_forensics_api.py @@ -1,5 +1,8 @@ import pytest +from fastapi import HTTPException +from sqlalchemy.exc import IntegrityError +from app.api.api_v1.endpoints import forensics as forensics_endpoint from app.models.report import ForensicReport from app.services.forensic_parser import ForensicParser from app.services.forensic_persistence import ( @@ -77,6 +80,22 @@ def test_upload_forensic_report_rejects_empty_file(authed_client): assert response.status_code == 400 +def test_validate_upload_rejects_missing_name_and_large_file(): + missing_name = type("Upload", (), {"filename": ""})() + too_large = type("Upload", (), {"filename": "report.eml"})() + + with pytest.raises(HTTPException) as missing: + forensics_endpoint._validate_upload(missing_name, b"content") + assert missing.value.status_code == 400 + + with pytest.raises(HTTPException) as large: + forensics_endpoint._validate_upload( + too_large, + b"x" * (forensics_endpoint.MAX_FORENSIC_REPORT_SIZE + 1), + ) + assert large.value.status_code == 413 + + def test_upload_forensic_report_rejects_invalid_email(authed_client): response = authed_client.post( "/api/v1/forensics/upload", @@ -86,6 +105,39 @@ def test_upload_forensic_report_rejects_invalid_email(authed_client): assert response.status_code == 400 +def test_upload_forensic_report_handles_save_duplicate_race(authed_client, monkeypatch): + parsed = ForensicParser.parse_bytes(SAMPLE_FORENSIC_EMAIL) + row = ForensicReport(report_id=parsed["report_id"], reported_domain=parsed["reported_domain"]) + monkeypatch.setattr(forensics_endpoint, "forensic_report_exists", lambda *_args: False) + monkeypatch.setattr( + forensics_endpoint, + "save_forensic_report", + lambda *_args: (row, False), + ) + + response = authed_client.post( + "/api/v1/forensics/upload", + files={"file": ("report.eml", SAMPLE_FORENSIC_EMAIL, "message/rfc822")}, + ) + + assert response.status_code == 409 + + +def test_upload_forensic_report_unexpected_error_returns_500(authed_client, monkeypatch): + monkeypatch.setattr( + forensics_endpoint.ForensicParser, + "parse_bytes", + lambda _content: (_ for _ in ()).throw(RuntimeError("boom")), + ) + + response = authed_client.post( + "/api/v1/forensics/upload", + files={"file": ("report.eml", SAMPLE_FORENSIC_EMAIL, "message/rfc822")}, + ) + + assert response.status_code == 500 + + def test_forensic_detail_returns_404(authed_client): response = authed_client.get("/api/v1/forensics/999") @@ -119,3 +171,17 @@ def test_save_forensic_report_duplicate_and_invalid_domain_paths(db_session): row, invalid_created = save_forensic_report(db_session, invalid) assert invalid_created is True assert row.domain_id is None + + +def test_save_forensic_report_reraises_unexpected_integrity_errors(db_session, monkeypatch): + parsed = ForensicParser.parse_bytes(SAMPLE_FORENSIC_EMAIL) + parsed["report_id"] = "ruf-race-without-existing-row" + parsed["reported_domain"] = "" + + def raise_integrity_error(): + raise IntegrityError("insert", {}, Exception("unique")) + + monkeypatch.setattr(db_session, "flush", raise_integrity_error) + + with pytest.raises(IntegrityError): + save_forensic_report(db_session, parsed) diff --git a/backend/app/tests/test_gmail_client.py b/backend/app/tests/test_gmail_client.py index 51be7a4..a274e06 100644 --- a/backend/app/tests/test_gmail_client.py +++ b/backend/app/tests/test_gmail_client.py @@ -493,6 +493,24 @@ class TestProcessMessage: assert stats["duplicate_forensic_reports"] == 1 assert stats["details"][1]["status"] == "duplicate" + def test_forensic_save_duplicate_result_is_skipped(self, db_session): + client = _make_client(db=db_session) + stats = {"forensic_reports_found": 0, "duplicate_forensic_reports": 0, "errors": []} + + with ( + patch("app.services.gmail_client.forensic_report_exists", return_value=False), + patch("app.services.gmail_client.save_forensic_report", return_value=(None, False)), + ): + imported = client._process_forensic_message( + SAMPLE_FORENSIC_EMAIL, + stats, + message_id="msg-1", + ) + + assert imported == 0 + assert stats["duplicate_forensic_reports"] == 1 + assert stats["details"][0]["status"] == "duplicate" + def test_forensic_parse_error_adds_error_detail(self, db_session): client = _make_client(db=db_session) stats = {"errors": []} @@ -795,11 +813,17 @@ class TestFetchReports: class TestIngestedIdHelpers: + def test_load_empty_json_text_returns_empty_list(self): + assert GmailClient.load_ingested_ids("") == [] + def test_load_non_list_json_returns_no_error(self): # Valid JSON but not a list – should gracefully not raise result = GmailClient.load_ingested_ids('{"key": "value"}') assert result is not None # no crash + def test_load_invalid_json_returns_empty_list(self): + assert GmailClient.load_ingested_ids("{not-json") == [] + def test_dump_preserves_order(self): ids = ["z", "a", "m"] dumped = GmailClient.dump_ingested_ids(ids) diff --git a/backend/app/tests/test_imap_client.py b/backend/app/tests/test_imap_client.py index c4dd146..a0bc6b9 100644 --- a/backend/app/tests/test_imap_client.py +++ b/backend/app/tests/test_imap_client.py @@ -667,6 +667,24 @@ class TestProcessSingleEmail: assert stats["duplicate_forensic_reports"] == 1 assert stats["details"][1]["status"] == "duplicate" + def test_forensic_save_duplicate_result_adds_detail(self, db_session): + client = self._make_client(db=db_session) + stats = {"forensic_reports_found": 0, "duplicate_forensic_reports": 0, "errors": []} + + with ( + patch("app.services.imap_client.forensic_report_exists", return_value=False), + patch("app.services.imap_client.save_forensic_report", return_value=(None, False)), + ): + imported = client._process_forensic_email( + SAMPLE_FORENSIC_EMAIL, + stats=stats, + message_id="msg-1", + ) + + assert imported is False + assert stats["duplicate_forensic_reports"] == 1 + assert stats["details"][0]["status"] == "duplicate" + def test_forensic_parse_error_adds_error_detail(self, db_session): client = self._make_client(db=db_session) stats = {"errors": []}