parsedmarc/tests/test_elastic.py

"""Tests for parsedmarc.elastic

Mocks at the elasticsearch.dsl SDK boundary (connections.create_connection,
Index, Search, Document.save) so the tests verify the parsedmarc-side
transformation logic — document construction, index naming, deduplication
queries, error wrapping — without needing a running Elasticsearch cluster.
"""

import time
import unittest
from unittest.mock import MagicMock, call, patch

import parsedmarc.elastic as elastic_module
from parsedmarc import InvalidFailureReport
from parsedmarc.elastic import (
    AlreadySaved,
    ElasticsearchError,
    create_indexes,
    migrate_indexes,
    save_aggregate_report_to_elasticsearch,
    save_failure_report_to_elasticsearch,
    save_smtp_tls_report_to_elasticsearch,
    set_hosts,
)
from tests.tzutil import force_tz


# ---------------------------------------------------------------------------
# Sample report fixtures
# ---------------------------------------------------------------------------


def _aggregate_report(**overrides):
    base = {
        "xml_schema": "draft",
        "xml_namespace": None,
        "report_metadata": {
            "org_name": "TestOrg",
            "org_email": "dmarc@example.com",
            "org_extra_contact_info": None,
            "report_id": "agg-1",
            "begin_date": "2024-01-15 00:00:00",
            "end_date": "2024-01-16 00:00:00",
            "timespan_requires_normalization": False,
            "original_timespan_seconds": 86400,
            "errors": [],
            "generator": "TestGen/1.0",
        },
        "policy_published": {
            "domain": "example.com",
            "adkim": "r",
            "aspf": "r",
            "p": "none",
            "sp": "none",
            "pct": None,
            "fo": None,
            "np": "reject",
            "testing": "n",
            "discovery_method": "treewalk",
        },
        "records": [
            {
                "interval_begin": "2024-01-15 00:00:00",
                "interval_end": "2024-01-16 00:00:00",
                "normalized_timespan": False,
                "source": {
                    "ip_address": "192.0.2.1",
                    "country": "US",
                    "reverse_dns": None,
                    "base_domain": None,
                    "name": None,
                    "type": None,
                    "asn": 64496,
                    "as_name": "Example AS",
                    "as_domain": "example.net",
                },
                "count": 4,
                "alignment": {"spf": True, "dkim": True, "dmarc": True},
                "policy_evaluated": {
                    "disposition": "none",
                    "dkim": "pass",
                    "spf": "pass",
                    "policy_override_reasons": [
                        {"type": "local_policy", "comment": "approved"}
                    ],
                },
                "identifiers": {
                    "header_from": "example.com",
                    "envelope_from": "example.com",
                    "envelope_to": "rcpt@example.com",
                },
                "auth_results": {
                    "dkim": [
                        {
                            "domain": "example.com",
                            "selector": "s",
                            "result": "pass",
                            "human_result": None,
                        }
                    ],
                    "spf": [
                        {
                            "domain": "example.com",
                            "scope": "mfrom",
                            "result": "pass",
                            "human_result": None,
                        }
                    ],
                },
            }
        ],
    }
    base.update(overrides)
    return base


def _failure_report(**overrides):
    base = {
        "feedback_type": "auth-failure",
        "user_agent": "test/1.0",
        "version": "1",
        "original_envelope_id": None,
        "original_mail_from": "x@example.com",
        "original_rcpt_to": None,
        "arrival_date": "Thu, 1 Jan 2024 00:00:00 +0000",
        "arrival_date_utc": "2024-01-01 00:00:00",
        "authentication_results": None,
        "delivery_result": "other",
        "auth_failure": ["dmarc"],
        "authentication_mechanisms": [],
        "dkim_domain": None,
        "reported_domain": "example.com",
        "sample_headers_only": True,
        "source": {
            "ip_address": "192.0.2.5",
            "country": "US",
            "reverse_dns": None,
            "base_domain": None,
            "name": None,
            "type": None,
            "asn": 64496,
            "as_name": "Example AS",
            "as_domain": "example.net",
        },
        "sample": "raw",
        "parsed_sample": {
            "headers": {
                # mailparser emits headers as [[display_name, address]]
                # lists; an empty display becomes [["", address]].
                "From": [["Sender Name", "sender@example.com"]],
                "To": [["", "rcpt@example.com"]],
                "Subject": "Test",
            },
            "subject": "Test",
            "filename_safe_subject": "Test",
            "body": "body",
            "date": "Thu, 1 Jan 2024 00:00:00 +0000",
            "to": [{"display_name": None, "address": "rcpt@example.com"}],
            "reply_to": [],
            "cc": [],
            "bcc": [],
            "attachments": [],
        },
    }
    base.update(overrides)
    return base


def _smtp_tls_report(**overrides):
    base = {
        "organization_name": "TestOrg",
        "begin_date": "2024-02-03T00:00:00Z",
        "end_date": "2024-02-04T00:00:00Z",
        "contact_info": "tls@example.com",
        "report_id": "tls-1",
        "policies": [
            {
                "policy_domain": "example.com",
                "policy_type": "sts",
                "successful_session_count": 100,
                "failed_session_count": 1,
                "policy_strings": ["version: STSv1"],
                "mx_host_patterns": ["*.example.com"],
                "failure_details": [
                    {
                        "result_type": "certificate-expired",
                        "failed_session_count": 1,
                        "receiving_mx_hostname": "mx.example.com",
                        "sending_mta_ip": "10.0.0.1",
                    }
                ],
            }
        ],
    }
    base.update(overrides)
    return base


def _empty_search():
    """A Search() mock whose .execute() returns an empty hit list."""
    search = MagicMock()
    search.execute.return_value = []
    return search


def _populated_search():
    """A Search() mock whose .execute() returns a non-empty hit list."""
    search = MagicMock()
    search.execute.return_value = [MagicMock()]
    return search


# ---------------------------------------------------------------------------
# set_hosts: connection-parameter assembly
# ---------------------------------------------------------------------------


class TestSetHosts(unittest.TestCase):
    """Verify the conn_params dict handed to the elasticsearch-py 8.x client
    matches each documented option. Each branch corresponds to a
    real-world deployment shape (TLS, basic auth, API key, custom CA).

    The 8.x client dropped the ``use_ssl`` / ``http_auth`` / ``timeout``
    connection kwargs: the scheme now has to be baked into each host URL,
    ``basic_auth`` replaces ``http_auth``, and ``request_timeout`` replaces
    ``timeout``.
    """

    def test_single_host_url_passed_through_unchanged(self):
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("https://es:9200")
        kwargs = mock_conn.call_args.kwargs
        self.assertEqual(kwargs["hosts"], ["https://es:9200"])

    def test_host_list_preserved(self):
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts(["http://es1:9200", "http://es2:9200"])
        kwargs = mock_conn.call_args.kwargs
        self.assertEqual(kwargs["hosts"], ["http://es1:9200", "http://es2:9200"])

    def test_bare_host_use_ssl_false_gets_http_prefix(self):
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("localhost", use_ssl=False)
        kwargs = mock_conn.call_args.kwargs
        self.assertEqual(kwargs["hosts"], ["http://localhost"])
        self.assertNotIn("use_ssl", kwargs)

    def test_bare_host_use_ssl_true_gets_https_prefix(self):
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("localhost", use_ssl=True)
        kwargs = mock_conn.call_args.kwargs
        self.assertEqual(kwargs["hosts"], ["https://localhost"])
        self.assertEqual(kwargs["verify_certs"], True)
        self.assertNotIn("use_ssl", kwargs)
        self.assertNotIn("ca_certs", kwargs)

    def test_explicit_url_passes_through_even_with_use_ssl_true(self):
        """A host that already carries a scheme is never re-prefixed,
        even when it disagrees with use_ssl."""
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("http://example.com:9200", use_ssl=True)
        kwargs = mock_conn.call_args.kwargs
        self.assertEqual(kwargs["hosts"], ["http://example.com:9200"])

    def test_timeout_default_60s_becomes_request_timeout(self):
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("es:9200")
        kwargs = mock_conn.call_args.kwargs
        self.assertEqual(kwargs["request_timeout"], 60.0)
        self.assertNotIn("timeout", kwargs)

    def test_timeout_custom(self):
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("es:9200", timeout=30.0)
        self.assertEqual(mock_conn.call_args.kwargs["request_timeout"], 30.0)

    def test_use_ssl_with_custom_ca(self):
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("es:9200", use_ssl=True, ssl_cert_path="/etc/ca.pem")
        kwargs = mock_conn.call_args.kwargs
        self.assertEqual(kwargs["ca_certs"], "/etc/ca.pem")

    def test_skip_certificate_verification_sets_verify_false(self):
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("es:9200", use_ssl=True, skip_certificate_verification=True)
        self.assertEqual(mock_conn.call_args.kwargs["verify_certs"], False)

    def test_username_password_sets_basic_auth(self):
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("es:9200", username="u", password="p")
        kwargs = mock_conn.call_args.kwargs
        self.assertEqual(kwargs["basic_auth"], ("u", "p"))
        self.assertNotIn("http_auth", kwargs)

    def test_username_without_password_not_set(self):
        """Half-configured auth is suspicious enough not to send."""
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("es:9200", username="u")
        self.assertNotIn("basic_auth", mock_conn.call_args.kwargs)

    def test_api_key_set(self):
        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
            set_hosts("es:9200", api_key="base64key==")
        self.assertEqual(mock_conn.call_args.kwargs["api_key"], "base64key==")


# ---------------------------------------------------------------------------
# create_indexes
# ---------------------------------------------------------------------------


class TestCreateIndexes(unittest.TestCase):
    def test_creates_missing_index_with_default_settings(self):
        with patch("parsedmarc.elastic.Index") as mock_index_cls:
            mock_index = mock_index_cls.return_value
            mock_index.exists.return_value = False
            create_indexes(["dmarc_aggregate-2024-01-15"])
        mock_index.settings.assert_called_once_with(
            number_of_shards=1, number_of_replicas=0
        )
        mock_index.create.assert_called_once()

    def test_creates_with_custom_settings(self):
        with patch("parsedmarc.elastic.Index") as mock_index_cls:
            mock_index = mock_index_cls.return_value
            mock_index.exists.return_value = False
            create_indexes(
                ["idx"], settings={"number_of_shards": 3, "refresh_interval": "5s"}
            )
        mock_index.settings.assert_called_once_with(
            number_of_shards=3, refresh_interval="5s"
        )

    def test_skips_existing_index(self):
        with patch("parsedmarc.elastic.Index") as mock_index_cls:
            mock_index = mock_index_cls.return_value
            mock_index.exists.return_value = True
            create_indexes(["idx"])
        mock_index.create.assert_not_called()

    def test_wraps_sdk_error(self):
        with patch("parsedmarc.elastic.Index") as mock_index_cls:
            mock_index_cls.return_value.exists.side_effect = RuntimeError(
                "cluster down"
            )
            with self.assertRaises(ElasticsearchError) as ctx:
                create_indexes(["idx"])
        self.assertIn("cluster down", str(ctx.exception))


class TestCreateIndexesServerless(unittest.TestCase):
    """Serverless mode strips shard/replica keys but keeps everything else.

    Elastic Cloud Serverless rejects ``number_of_shards`` and
    ``number_of_replicas`` with HTTP 400. Other settings like
    ``refresh_interval`` are accepted and must pass through unchanged.
    """

    def setUp(self):
        self._original = elastic_module._SERVERLESS
        elastic_module._SERVERLESS = True

    def tearDown(self):
        elastic_module._SERVERLESS = self._original

    def test_serverless_default_skips_settings_entirely(self):
        with patch("parsedmarc.elastic.Index") as mock_index_cls:
            mock_index = mock_index_cls.return_value
            mock_index.exists.return_value = False
            create_indexes(["idx"])
        mock_index.settings.assert_not_called()
        mock_index.create.assert_called_once()

    def test_serverless_filters_rejected_keys_and_passes_others_through(self):
        with patch("parsedmarc.elastic.Index") as mock_index_cls:
            mock_index = mock_index_cls.return_value
            mock_index.exists.return_value = False
            create_indexes(
                ["idx"],
                settings={
                    "number_of_shards": 3,
                    "number_of_replicas": 2,
                    "refresh_interval": "5s",
                },
            )
        mock_index.settings.assert_called_once_with(refresh_interval="5s")

    def test_serverless_skips_settings_when_only_rejected_keys(self):
        with patch("parsedmarc.elastic.Index") as mock_index_cls:
            mock_index = mock_index_cls.return_value
            mock_index.exists.return_value = False
            create_indexes(
                ["idx"], settings={"number_of_shards": 3, "number_of_replicas": 2}
            )
        mock_index.settings.assert_not_called()
        mock_index.create.assert_called_once()


# ---------------------------------------------------------------------------
# migrate_indexes
# ---------------------------------------------------------------------------


class TestMigrateIndexes(unittest.TestCase):
    """migrate_indexes backfills dkim_results_combined/spf_results_combined
    (issue #169) on pre-existing aggregate documents as a non-blocking
    background task. It is guarded by a cheap count() query so repeated
    startups against an already-backfilled index are a no-op, and any SDK
    error is caught and logged rather than raised, so it never blocks
    parsedmarc startup."""

    def test_backfill_submitted_when_old_docs_exist(self):
        with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
            mock_client = MagicMock()
            mock_client.count.return_value = {"count": 42}
            mock_get_conn.return_value = mock_client
            migrate_indexes(aggregate_indexes=["dmarc_aggregate"])

        mock_client.update_by_query.assert_called_once()
        kwargs = mock_client.update_by_query.call_args.kwargs
        self.assertEqual(kwargs["index"], "dmarc_aggregate*")
        self.assertEqual(kwargs["conflicts"], "proceed")
        self.assertFalse(kwargs["wait_for_completion"])
        self.assertEqual(kwargs["query"], elastic_module._COMBINED_BACKFILL_QUERY)
        script_source = kwargs["script"]["source"]
        self.assertIn("ctx._source.dkim_results_combined", script_source)
        self.assertIn("ctx._source.spf_results_combined", script_source)

        # The count() guard query also targets the date-suffixed pattern.
        count_kwargs = mock_client.count.call_args.kwargs
        self.assertEqual(count_kwargs["index"], "dmarc_aggregate*")
        self.assertEqual(count_kwargs["query"], elastic_module._COMBINED_BACKFILL_QUERY)

    def test_backfill_skipped_when_no_old_docs(self):
        with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
            mock_client = MagicMock()
            mock_client.count.return_value = {"count": 0}
            mock_get_conn.return_value = mock_client
            migrate_indexes(aggregate_indexes=["dmarc_aggregate"])

        mock_client.update_by_query.assert_not_called()

    def test_backfill_skipped_when_no_aggregate_indexes(self):
        with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
            migrate_indexes()
            migrate_indexes(aggregate_indexes=None)

        mock_get_conn.assert_not_called()

    def test_backfill_failure_does_not_raise(self):
        with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
            mock_client = MagicMock()
            mock_client.count.side_effect = RuntimeError("cluster unreachable")
            mock_get_conn.return_value = mock_client
            with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
                migrate_indexes(aggregate_indexes=["dmarc_aggregate"])

        self.assertTrue(any("cluster unreachable" in msg for msg in cm.output))
        mock_client.update_by_query.assert_not_called()

    def test_get_connection_failure_does_not_raise(self):
        """connections.get_connection() itself sits outside the per-index
        try/except; if it raises (e.g. no Elasticsearch connection has been
        configured yet), migrate_indexes must still not propagate the
        exception, per its docstring's promise that any cluster error is
        caught and logged."""
        with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
            mock_get_conn.side_effect = RuntimeError("no connection")
            with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
                migrate_indexes(aggregate_indexes=["dmarc_aggregate"])

        self.assertTrue(
            any("Skipping the dkim_results_combined" in msg for msg in cm.output)
        )
        self.assertTrue(any("no connection" in msg for msg in cm.output))

    def test_smtp_tls_backfill_submitted_when_old_docs_exist(self):
        """SMTP TLS analogue of test_backfill_submitted_when_old_docs_exist:
        policies_combined/failure_details_combined backfill (also issue
        #169) is submitted with its own guard query and painless script."""
        with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
            mock_client = MagicMock()
            mock_client.count.return_value = {"count": 7}
            mock_get_conn.return_value = mock_client
            migrate_indexes(smtp_tls_indexes=["smtp_tls"])

        mock_client.update_by_query.assert_called_once()
        kwargs = mock_client.update_by_query.call_args.kwargs
        self.assertEqual(kwargs["index"], "smtp_tls*")
        self.assertEqual(kwargs["conflicts"], "proceed")
        self.assertFalse(kwargs["wait_for_completion"])
        self.assertEqual(
            kwargs["query"], elastic_module._SMTP_TLS_COMBINED_BACKFILL_QUERY
        )
        script_source = kwargs["script"]["source"]
        self.assertIn("ctx._source.policies_combined", script_source)
        self.assertIn("ctx._source.failure_details_combined", script_source)

        count_kwargs = mock_client.count.call_args.kwargs
        self.assertEqual(count_kwargs["index"], "smtp_tls*")
        self.assertEqual(
            count_kwargs["query"], elastic_module._SMTP_TLS_COMBINED_BACKFILL_QUERY
        )

    def test_smtp_tls_backfill_skipped_when_no_old_docs(self):
        with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
            mock_client = MagicMock()
            mock_client.count.return_value = {"count": 0}
            mock_get_conn.return_value = mock_client
            migrate_indexes(smtp_tls_indexes=["smtp_tls"])

        mock_client.update_by_query.assert_not_called()

    def test_smtp_tls_backfill_skipped_when_no_smtp_tls_indexes_or_aggregate(self):
        with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
            migrate_indexes()
            migrate_indexes(smtp_tls_indexes=None)

        mock_get_conn.assert_not_called()

    def test_smtp_tls_backfill_failure_does_not_raise(self):
        """SMTP TLS analogue of test_backfill_failure_does_not_raise: an
        error from the cluster during the smtp_tls_indexes loop is caught
        and logged rather than raised."""
        with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
            mock_client = MagicMock()
            mock_client.count.side_effect = RuntimeError("cluster unreachable")
            mock_get_conn.return_value = mock_client
            with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
                migrate_indexes(smtp_tls_indexes=["smtp_tls"])

        self.assertTrue(any("cluster unreachable" in msg for msg in cm.output))
        mock_client.update_by_query.assert_not_called()


def _typeless_fo_mapping(index_name, fo_type):
    """An indices.get_field_mapping response in the modern, typeless shape.

    Elasticsearch 8 has no mapping types, so the field sits directly under
    ``mappings``. This is the only shape a supported cluster returns; the
    ES 6-era type-keyed shape is covered separately.
    """
    return {
        index_name: {
            "mappings": {
                "published_policy.fo": {
                    "full_name": "published_policy.fo",
                    "mapping": {"fo": {"type": fo_type}},
                }
            }
        }
    }


class TestMigrateIndexesFoMigration(unittest.TestCase):
    """parsedmarc releases before 5.0.0 declared published_policy.fo as an
    integer, so their indexes mapped it as `long`, which cannot hold the
    multi-value `fo` settings reports carry (`0:1`, `d:s`). migrate_indexes
    detects that and rebuilds the index with the text/keyword shape.

    Elasticsearch 8 refuses to open an index created before 7.0, so an
    affected index reaches a supported cluster only by being carried
    forward through a reindex — which keeps the old mapping whenever the
    destination is pre-created from it, as the standard reindex procedure
    does. Each test stubs the combined-field backfill that runs afterwards
    in the same call (count 0 → no-op)."""

    @staticmethod
    def _noop_backfill_client():
        client = MagicMock()
        client.count.return_value = {"count": 0}
        return client

    @staticmethod
    def _index_mocks(*, v2_exists):
        """Distinct Index() mocks per name, so the original and the -v2
        target can be told apart. A single shared mock cannot express
        "the original exists but its migration target does not", which is
        the ordinary case."""
        original = MagicMock(name="dmarc_aggregate")
        original.exists.return_value = True
        original.get_field_mapping.return_value = _typeless_fo_mapping(
            "dmarc_aggregate", "long"
        )
        v2 = MagicMock(name="dmarc_aggregate-v2")
        v2.exists.return_value = v2_exists
        return original, v2, lambda name: v2 if name.endswith("-v2") else original

    def test_skips_non_existent_index(self):
        with (
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn,
        ):
            mock_get_conn.return_value = self._noop_backfill_client()
            mock_index_cls.return_value.exists.return_value = False
            migrate_indexes(legacy_fo_indexes=["missing"])
        mock_index_cls.return_value.get_field_mapping.assert_not_called()

    def test_skips_when_field_is_unmapped(self):
        """An index that does not map published_policy.fo at all (e.g. an
        empty index with the default mapping) is left alone."""
        with (
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn,
            patch("parsedmarc.elastic.reindex") as mock_reindex,
        ):
            mock_get_conn.return_value = self._noop_backfill_client()
            idx = mock_index_cls.return_value
            idx.exists.return_value = True
            idx.get_field_mapping.return_value = {"dmarc_aggregate": {"mappings": {}}}
            migrate_indexes(legacy_fo_indexes=["dmarc_aggregate"])
        mock_reindex.assert_not_called()
        idx.create.assert_not_called()
        idx.delete.assert_not_called()

    def test_migrates_when_fo_is_long(self):
        """The actual migration path: when fo is mapped as 'long', a v2
        index is created with the corrected text/keyword mapping, data is
        reindexed into it, and the old index is deleted."""
        original, v2, factory = self._index_mocks(v2_exists=False)
        with (
            patch("parsedmarc.elastic.Index", side_effect=factory) as mock_index_cls,
            patch("parsedmarc.elastic.reindex") as mock_reindex,
            patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn,
        ):
            mock_client = self._noop_backfill_client()
            mock_get_conn.return_value = mock_client
            migrate_indexes(legacy_fo_indexes=["dmarc_aggregate"])

        self.assertIn(call("dmarc_aggregate-v2"), mock_index_cls.call_args_list)
        v2.create.assert_called_once_with()
        mapping_kwargs = v2.put_mapping.call_args.kwargs
        self.assertNotIn("body", mapping_kwargs)
        self.assertEqual(
            mapping_kwargs["properties"]["published_policy"]["properties"]["fo"],
            {
                "type": "text",
                "fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
            },
        )

        # reindex old -> new (v2) with the connection's client, and only
        # then is the original dropped. The v2 index is never deleted here:
        # nothing was left over to discard.
        mock_reindex.assert_called_once_with(
            mock_client, "dmarc_aggregate", "dmarc_aggregate-v2"
        )
        original.delete.assert_called_once_with()
        v2.delete.assert_not_called()

    def test_retries_after_an_interrupted_earlier_attempt(self):
        """A run that died between create() and delete() leaves a -v2 index
        behind. Reaching this code means the original still holds the data
        -- it is deleted only after the reindex succeeds -- so the leftover
        is debris. Without discarding it, create() raises "resource already
        exists" on every later startup and the index is never migrated;
        confirmed against a live cluster, where the unfixed code left the
        original in place and the debris document in the v2 index."""
        original, v2, factory = self._index_mocks(v2_exists=True)
        with (
            patch("parsedmarc.elastic.Index", side_effect=factory),
            patch("parsedmarc.elastic.reindex") as mock_reindex,
            patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn,
        ):
            mock_client = self._noop_backfill_client()
            mock_get_conn.return_value = mock_client
            with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
                migrate_indexes(legacy_fo_indexes=["dmarc_aggregate"])

        self.assertTrue(
            any("Discarding dmarc_aggregate-v2" in msg for msg in cm.output)
        )
        # The stale target is dropped, then recreated, and the migration
        # runs to completion instead of aborting on "already exists".
        v2.delete.assert_called_once_with()
        v2.create.assert_called_once_with()
        mock_reindex.assert_called_once_with(
            mock_client, "dmarc_aggregate", "dmarc_aggregate-v2"
        )
        original.delete.assert_called_once_with()

    def test_migrates_when_fo_is_long_under_a_mapping_type(self):
        """The Elasticsearch 6-era response nested the field under the
        mapping type name. No cluster either client can connect to still
        reports mappings that way, so this covers the fallback branch
        rather than a reachable deployment -- but the branch is what lets
        the type check stay a check on the mapped type instead of on the
        response shape, which is what broke this migration before."""
        with (
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch("parsedmarc.elastic.reindex") as mock_reindex,
            patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn,
        ):
            mock_get_conn.return_value = self._noop_backfill_client()
            idx = mock_index_cls.return_value
            idx.exists.return_value = True
            idx.get_field_mapping.return_value = {
                "dmarc_aggregate": {
                    "mappings": {
                        "doc": {
                            "published_policy.fo": {"mapping": {"fo": {"type": "long"}}}
                        }
                    }
                }
            }
            migrate_indexes(legacy_fo_indexes=["dmarc_aggregate"])
        mock_reindex.assert_called_once()

    def test_skips_when_fo_already_text(self):
        with (
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn,
            patch("parsedmarc.elastic.reindex") as mock_reindex,
        ):
            mock_get_conn.return_value = self._noop_backfill_client()
            idx = mock_index_cls.return_value
            idx.exists.return_value = True
            idx.get_field_mapping.return_value = _typeless_fo_mapping(
                "dmarc_aggregate", "text"
            )
            migrate_indexes(legacy_fo_indexes=["dmarc_aggregate"])
        mock_reindex.assert_not_called()
        idx.create.assert_not_called()
        idx.delete.assert_not_called()

    def test_index_exists_failure_does_not_raise(self):
        """A cluster error inside the per-index fo-migration loop (e.g.
        Index(...).exists() raising because the cluster is unreachable)
        must not abort startup: it is caught, logged, and the loop moves
        on to the combined-field backfill, which is exercised here with
        its own connection failure so both warnings are asserted."""
        with (
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn,
        ):
            mock_index_cls.return_value.exists.side_effect = ConnectionError(
                "cluster unreachable"
            )
            mock_get_conn.side_effect = RuntimeError("no connection")
            with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
                migrate_indexes(
                    aggregate_indexes=["dmarc_aggregate"],
                    legacy_fo_indexes=["dmarc_aggregate"],
                )

        self.assertTrue(
            any(
                "legacy published_policy.fo migration" in msg
                and "cluster unreachable" in msg
                for msg in cm.output
            )
        )
        self.assertTrue(
            any("Skipping the dkim_results_combined" in msg for msg in cm.output)
        )
        self.assertTrue(any("no connection" in msg for msg in cm.output))

    def test_no_legacy_indexes_means_no_lookup(self):
        """The combined-field backfill must not drag the fo migration
        along: passing only aggregate_indexes leaves the legacy loop
        untouched, so a modern deployment never probes for it."""
        with (
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn,
        ):
            mock_get_conn.return_value = self._noop_backfill_client()
            migrate_indexes(aggregate_indexes=["dmarc_aggregate"])
        mock_index_cls.return_value.exists.assert_not_called()


# ---------------------------------------------------------------------------
# save_aggregate_report_to_elasticsearch
# ---------------------------------------------------------------------------


class TestSaveAggregateReport(unittest.TestCase):
    """The aggregate-report save fans out across multiple SDK calls:
    Search (for dedup), Index.create (for the daily/monthly index),
    Document.save. Each test patches the boundary it needs and
    leaves the rest alone."""

    def _patches(self, search_factory=_empty_search):
        return [
            patch("parsedmarc.elastic.Search", return_value=search_factory()),
            patch(
                "parsedmarc.elastic.Index",
                return_value=MagicMock(exists=MagicMock(return_value=True)),
            ),
            patch.object(elastic_module._AggregateReportDoc, "save"),
        ]

    def test_save_emits_one_document_per_record(self):
        report = _aggregate_report()
        report["records"].append(report["records"][0].copy())
        patches = self._patches()
        with patches[0], patches[1], patches[2] as mock_save:
            save_aggregate_report_to_elasticsearch(report)
        # Two records → two saves.
        self.assertEqual(mock_save.call_count, 2)

    def test_already_saved_raises_when_search_returns_hit(self):
        """The dedup query is the only thing preventing
        double-indexing on re-run. A regression would silently
        re-save reports, inflating Kibana counts."""
        with (
            patch("parsedmarc.elastic.Search", return_value=_populated_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._AggregateReportDoc, "save") as mock_save,
        ):
            with self.assertRaises(AlreadySaved):
                save_aggregate_report_to_elasticsearch(_aggregate_report())
        mock_save.assert_not_called()

    def test_search_exception_wraps_to_elasticsearch_error(self):
        bad_search = MagicMock()
        bad_search.execute.side_effect = RuntimeError("network")
        with (
            patch("parsedmarc.elastic.Search", return_value=bad_search),
            patch("parsedmarc.elastic.Index"),
        ):
            with self.assertRaises(ElasticsearchError) as ctx:
                save_aggregate_report_to_elasticsearch(_aggregate_report())
        self.assertIn("network", str(ctx.exception))

    def test_save_exception_wraps_to_elasticsearch_error(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(
                elastic_module._AggregateReportDoc,
                "save",
                side_effect=RuntimeError("disk"),
            ),
        ):
            with self.assertRaises(ElasticsearchError) as ctx:
                save_aggregate_report_to_elasticsearch(_aggregate_report())
        self.assertIn("disk", str(ctx.exception))

    def test_index_name_uses_daily_format_by_default(self):
        """Index naming: dmarc_aggregate-YYYY-MM-DD by default."""
        index_calls = []
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch.object(elastic_module._AggregateReportDoc, "save"),
        ):
            mock_index_cls.return_value.exists.return_value = True
            save_aggregate_report_to_elasticsearch(_aggregate_report())
            index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
        self.assertIn("dmarc_aggregate-2024-01-15", index_calls)

    def test_index_name_uses_monthly_format_when_flag_set(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch.object(elastic_module._AggregateReportDoc, "save"),
        ):
            mock_index_cls.return_value.exists.return_value = True
            save_aggregate_report_to_elasticsearch(
                _aggregate_report(), monthly_indexes=True
            )
            index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
        self.assertIn("dmarc_aggregate-2024-01", index_calls)

    def test_index_name_honours_suffix_and_prefix(self):
        """Prefix/suffix support multi-tenant setups where one ES
        cluster serves several DMARC owners."""
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch.object(elastic_module._AggregateReportDoc, "save"),
        ):
            mock_index_cls.return_value.exists.return_value = True
            save_aggregate_report_to_elasticsearch(
                _aggregate_report(),
                index_suffix="tenant_a",
                index_prefix="customer1_",
            )
            index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
        self.assertIn("customer1_dmarc_aggregate_tenant_a-2024-01-15", index_calls)

    def test_dedup_search_pattern_uses_suffix_wildcard(self):
        """Existing-report search uses '*' so it matches both
        daily and monthly index buckets."""
        with (
            patch("parsedmarc.elastic.Search") as mock_search_cls,
            patch(
                "parsedmarc.elastic.Index",
                return_value=MagicMock(exists=MagicMock(return_value=True)),
            ),
            patch.object(elastic_module._AggregateReportDoc, "save"),
        ):
            mock_search_cls.return_value.execute.return_value = []
            save_aggregate_report_to_elasticsearch(
                _aggregate_report(), index_suffix="tenant_a", index_prefix="cust_"
            )
        # Search index pattern wraps prefix+name+suffix with trailing wildcard.
        search_index = mock_search_cls.call_args.kwargs["index"]
        self.assertIn("cust_dmarc_aggregate_tenant_a*", search_index)

    @unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
    def test_interval_dates_are_utc_regardless_of_host_timezone(self):
        """interval_begin/interval_end are UTC wall-clock strings (already
        converted to UTC at parse time in __init__.py); the index-date
        bucketing and stored date_begin/date_end must use their true UTC
        epoch on any host. Regression test for
        https://github.com/domainaware/parsedmarc/issues/819: the naive
        parse used to shift the stored epoch (and therefore the index
        date) by the host's UTC offset."""
        force_tz(self)
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch("parsedmarc.elastic._AggregateReportDoc") as mock_doc_cls,
        ):
            mock_index_cls.return_value.exists.return_value = True
            save_aggregate_report_to_elasticsearch(_aggregate_report())
            index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
        self.assertIn("dmarc_aggregate-2024-01-15", index_calls)
        # Fixture begin_date/interval_begin is 2024-01-15 00:00:00 UTC.
        self.assertEqual(
            mock_doc_cls.call_args.kwargs["date_begin"].timestamp(), 1705276800
        )

    def test_save_populates_combined_dkim_and_spf_fields(self):
        """Regression guard for issue #169: two DKIM signatures on one
        record must yield exactly two combined entries, not a 4-way
        cross-product. autospec=True is required on the save patch so
        mock_save.call_args captures the doc instance as ``self``."""
        report = _aggregate_report()
        report["records"][0]["auth_results"] = {
            "dkim": [
                {
                    "domain": "example.net",
                    "selector": "net1",
                    "result": "fail",
                    "human_result": None,
                },
                {
                    "domain": "example.org",
                    "selector": "org1",
                    "result": "pass",
                    "human_result": None,
                },
            ],
            "spf": [
                {
                    "domain": "example.org",
                    "scope": "mfrom",
                    "result": "pass",
                    "human_result": None,
                },
            ],
        }
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch(
                "parsedmarc.elastic.Index",
                return_value=MagicMock(exists=MagicMock(return_value=True)),
            ),
            patch.object(
                elastic_module._AggregateReportDoc, "save", autospec=True
            ) as mock_save,
        ):
            save_aggregate_report_to_elasticsearch(report)
        doc = mock_save.call_args[0][0]
        self.assertEqual(
            list(doc.dkim_results_combined),
            ["net1 / example.net / fail", "org1 / example.org / pass"],
        )
        self.assertEqual(list(doc.spf_results_combined), ["mfrom / example.org / pass"])


class TestAggregateDocPassedDmarc(unittest.TestCase):
    """The _AggregateReportDoc.save() override derives passed_dmarc — the
    field dashboards filter on for DMARC pass/fail — from SPF/DKIM
    alignment. The SDK parent (elasticsearch.dsl.Document.save) is mocked so
    no cluster is needed."""

    def test_passed_dmarc_derived_from_alignment(self):
        cases = [
            (True, False, True),
            (False, True, True),
            (True, True, True),
            (False, False, False),
        ]
        for spf_aligned, dkim_aligned, expected in cases:
            with self.subTest(spf=spf_aligned, dkim=dkim_aligned):
                with patch.object(
                    elastic_module.Document, "save", return_value=None
                ) as mock_super_save:
                    doc = elastic_module._AggregateReportDoc(
                        spf_aligned=spf_aligned, dkim_aligned=dkim_aligned
                    )
                    doc.save()
                mock_super_save.assert_called_once()
                self.assertEqual(bool(doc.passed_dmarc), expected)


class TestAggregateDocCombinedResults(unittest.TestCase):
    """add_dkim_result/add_spf_result never touch the network, so these
    construct _AggregateReportDoc directly rather than going through the
    save_* entry point."""

    def test_add_dkim_result_appends_combined_string(self):
        """Regression guard for issue #169: dkim_results/spf_results are
        arrays of objects that the engine dynamic-maps as plain ``object``
        (not ``nested``) and flattens, so Kibana/Grafana tables cannot
        terms-aggregate their subfields without producing a cross-product
        of selector/domain/result values. The composed
        "selector / domain / result" string preserves the per-signature
        pairing that the flattened array loses."""
        doc = elastic_module._AggregateReportDoc()
        doc.add_dkim_result(
            domain="example.net", selector="net1", result="fail", human_result=None
        )
        doc.add_dkim_result(
            domain="example.org", selector="org1", result="pass", human_result=None
        )
        expected = ["net1 / example.net / fail", "org1 / example.org / pass"]
        # dkim_results_combined is declared as Text(multi=True, ...); the SDK
        # stub types the class attribute as Text (no Iterable protocol),
        # even though the runtime value is an AttrList once multi=True is
        # set. Same category of stub gap as the Q()/meta.index ignores in
        # elastic.py.
        self.assertEqual(list(doc.dkim_results_combined), expected)  # pyright: ignore[reportArgumentType]
        self.assertEqual(doc.to_dict()["dkim_results_combined"], expected)

    def test_add_spf_result_appends_combined_string(self):
        doc = elastic_module._AggregateReportDoc()
        doc.add_spf_result(
            domain="example.org", scope="mfrom", result="pass", human_result=None
        )
        expected = ["mfrom / example.org / pass"]
        self.assertEqual(list(doc.spf_results_combined), expected)  # pyright: ignore[reportArgumentType]
        self.assertEqual(doc.to_dict()["spf_results_combined"], expected)

    def test_spf_result_serializes_under_singular_result_key(self):
        """The _SPFResult class previously declared a dead ``results``
        (plural) field while the save path wrote ``result``; verify the
        serialized inner doc actually uses the singular key."""
        doc = elastic_module._AggregateReportDoc()
        doc.add_spf_result(
            domain="example.org", scope="mfrom", result="pass", human_result=None
        )
        d = doc.to_dict()["spf_results"][0]
        self.assertEqual(d["result"], "pass")
        self.assertNotIn("results", d)


# ---------------------------------------------------------------------------
# save_failure_report_to_elasticsearch
# ---------------------------------------------------------------------------


class TestSaveFailureReport(unittest.TestCase):
    def test_save_emits_one_document(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._FailureReportDoc, "save") as mock_save,
        ):
            save_failure_report_to_elasticsearch(_failure_report())
        mock_save.assert_called_once()

    def test_already_saved_raises_on_dedup_hit(self):
        """Failure-report dedup uses arrival_date + From/To/Subject
        from the parsed sample. A hit means we've already indexed
        this exact failure sample."""
        with (
            patch("parsedmarc.elastic.Search", return_value=_populated_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._FailureReportDoc, "save") as mock_save,
        ):
            with self.assertRaises(AlreadySaved):
                save_failure_report_to_elasticsearch(_failure_report())
        mock_save.assert_not_called()

    def test_save_exception_wraps_to_elasticsearch_error(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(
                elastic_module._FailureReportDoc,
                "save",
                side_effect=RuntimeError("disk"),
            ),
        ):
            with self.assertRaises(ElasticsearchError) as ctx:
                save_failure_report_to_elasticsearch(_failure_report())
        self.assertIn("disk", str(ctx.exception))

    def test_keyerror_wraps_to_invalid_failure_report(self):
        """A malformed failure report (missing a required field) is
        surfaced as InvalidFailureReport so the caller can route it
        differently from infra errors."""
        report = _failure_report()
        del report["feedback_type"]
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._FailureReportDoc, "save"),
        ):
            with self.assertRaises(InvalidFailureReport):
                save_failure_report_to_elasticsearch(report)

    def test_index_dedup_pattern_searches_both_old_and_new_names(self):
        """The split-PR rename forensic→failure left existing data
        in dmarc_forensic*; the dedup search must check both names
        so re-runs don't double-index."""
        with (
            patch("parsedmarc.elastic.Search") as mock_search_cls,
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._FailureReportDoc, "save"),
        ):
            mock_search_cls.return_value.execute.return_value = []
            save_failure_report_to_elasticsearch(_failure_report())
        search_index = mock_search_cls.call_args.kwargs["index"]
        self.assertIn("dmarc_failure*", search_index)
        self.assertIn("dmarc_forensic*", search_index)

    def test_index_name_uses_arrival_date_for_monthly_partition(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch.object(elastic_module._FailureReportDoc, "save"),
        ):
            save_failure_report_to_elasticsearch(
                _failure_report(), monthly_indexes=True
            )
            index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
        self.assertIn("dmarc_failure-2024-01", index_calls)

    @unittest.skipUnless(hasattr(time, "tzset"), "requires POSIX time.tzset()")
    def test_arrival_date_epoch_is_utc_regardless_of_host_timezone(self):
        """arrival_date_utc is a UTC wall-clock string; the epoch-ms
        value stored in the document (and used in the dedup query) must
        be its true UTC epoch on any host. Regression test for
        https://github.com/domainaware/parsedmarc/issues/811 (bug 1):
        the naive parse used to shift the stored epoch by the host's
        UTC offset."""
        force_tz(self)

        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch("parsedmarc.elastic._FailureReportDoc") as mock_doc_cls,
        ):
            save_failure_report_to_elasticsearch(_failure_report())
        # Fixture arrival_date_utc is 2024-01-01 00:00:00 UTC.
        self.assertEqual(mock_doc_cls.call_args.kwargs["arrival_date"], 1704067200000)

    def test_failure_search_index_with_suffix_and_prefix(self):
        """When both suffix and prefix are set, the dedup search
        pattern joins them onto BOTH dmarc_failure* and
        dmarc_forensic* (the rename back-compat)."""
        with (
            patch("parsedmarc.elastic.Search") as mock_search_cls,
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._FailureReportDoc, "save"),
        ):
            mock_search_cls.return_value.execute.return_value = []
            save_failure_report_to_elasticsearch(
                _failure_report(),
                index_suffix="tenant_a",
                index_prefix="cust_",
            )
        search_index = mock_search_cls.call_args.kwargs["index"]
        self.assertIn("cust_dmarc_failure_tenant_a*", search_index)
        self.assertIn("cust_dmarc_forensic_tenant_a*", search_index)

    def test_failure_index_honours_suffix_and_prefix(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch.object(elastic_module._FailureReportDoc, "save"),
        ):
            save_failure_report_to_elasticsearch(
                _failure_report(),
                index_suffix="tenant_a",
                index_prefix="cust_",
            )
            index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
        self.assertIn("cust_dmarc_failure_tenant_a-2024-01-01", index_calls)

    def test_from_header_with_empty_display_name(self):
        """When the From display name is empty, the code uses the
        address alone (covers the early-return branch in the
        display-name handling)."""
        report = _failure_report()
        report["parsed_sample"]["headers"]["From"] = [["", "sender@example.com"]]
        report["parsed_sample"]["headers"]["To"] = [["", "rcpt@example.com"]]
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._FailureReportDoc, "save") as mock_save,
        ):
            save_failure_report_to_elasticsearch(report)
        mock_save.assert_called_once()

    def test_to_header_with_non_empty_display_joins_with_brackets(self):
        """The other branch: non-empty display joins display+addr
        with " <" and appends ">", e.g. 'RT <rcpt@example.com>'."""
        report = _failure_report()
        report["parsed_sample"]["headers"]["To"] = [["RT", "rcpt@example.com"]]
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._FailureReportDoc, "save") as mock_save,
        ):
            save_failure_report_to_elasticsearch(report)
        mock_save.assert_called_once()

    def test_sample_address_lists_indexed_for_reply_to_cc_bcc_attachments(self):
        """A failure report sample can carry reply_to / cc / bcc /
        attachments. Each populates a nested InnerDoc on the sample —
        if the add_* helpers regress, those nested docs would be
        silently empty in Elasticsearch."""
        report = _failure_report()
        report["parsed_sample"]["reply_to"] = [
            {"display_name": "RT", "address": "rt@example.com"}
        ]
        report["parsed_sample"]["cc"] = [
            {"display_name": "CC", "address": "cc@example.com"}
        ]
        report["parsed_sample"]["bcc"] = [
            {"display_name": "", "address": "bcc@example.com"}
        ]
        report["parsed_sample"]["attachments"] = [
            {
                "filename": "a.pdf",
                "mail_content_type": "application/pdf",
                "sha256": "deadbeef",
            }
        ]
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._FailureReportDoc, "save") as mock_save,
        ):
            save_failure_report_to_elasticsearch(report)
        mock_save.assert_called_once()

    def test_reply_to_header_flattened_and_indexed(self):
        """A Reply-To header is flattened to a display string on
        ``sample.headers["reply-to"]`` — so the failure dashboard's
        ``sample.headers.reply-to.keyword`` column resolves — and each
        Reply-To address also populates the nested ``sample.reply_to``
        docs. Asserts on the document handed to .save(), not merely
        that save ran."""
        report = _failure_report()
        report["parsed_sample"]["headers"]["Reply-To"] = [
            ["Real One", "real@phish.example"]
        ]
        report["parsed_sample"]["reply_to"] = [
            {"display_name": "Real One", "address": "real@phish.example"}
        ]
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(
                elastic_module._FailureReportDoc, "save", autospec=True
            ) as mock_save,
        ):
            save_failure_report_to_elasticsearch(report)
        doc = mock_save.call_args.args[0]
        self.assertEqual(
            doc.sample.headers["reply-to"], "Real One <real@phish.example>"
        )
        self.assertEqual(
            [a.address for a in doc.sample.reply_to], ["real@phish.example"]
        )

    def test_reply_to_header_without_display_name_flattens_to_address(self):
        """A Reply-To header with no display name flattens to the bare
        address — the empty-display branch of the header flattening,
        matching the From/To handling."""
        report = _failure_report()
        report["parsed_sample"]["headers"]["Reply-To"] = [["", "noname@phish.example"]]
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(
                elastic_module._FailureReportDoc, "save", autospec=True
            ) as mock_save,
        ):
            save_failure_report_to_elasticsearch(report)
        doc = mock_save.call_args.args[0]
        self.assertEqual(doc.sample.headers["reply-to"], "noname@phish.example")


# ---------------------------------------------------------------------------
# save_smtp_tls_report_to_elasticsearch
# ---------------------------------------------------------------------------


class TestSaveSmtpTlsReport(unittest.TestCase):
    def test_save_emits_one_document(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._SMTPTLSReportDoc, "save") as mock_save,
        ):
            save_smtp_tls_report_to_elasticsearch(_smtp_tls_report())
        mock_save.assert_called_once()

    def test_already_saved_raises_on_dedup_hit(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_populated_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._SMTPTLSReportDoc, "save") as mock_save,
        ):
            with self.assertRaises(AlreadySaved):
                save_smtp_tls_report_to_elasticsearch(_smtp_tls_report())
        mock_save.assert_not_called()

    def test_search_exception_wraps_to_elasticsearch_error(self):
        bad = MagicMock()
        bad.execute.side_effect = RuntimeError("network")
        with (
            patch("parsedmarc.elastic.Search", return_value=bad),
            patch("parsedmarc.elastic.Index"),
        ):
            with self.assertRaises(ElasticsearchError):
                save_smtp_tls_report_to_elasticsearch(_smtp_tls_report())

    def test_save_exception_wraps_to_elasticsearch_error(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(
                elastic_module._SMTPTLSReportDoc,
                "save",
                side_effect=RuntimeError("disk"),
            ),
        ):
            with self.assertRaises(ElasticsearchError):
                save_smtp_tls_report_to_elasticsearch(_smtp_tls_report())

    def test_index_name_uses_begin_date_for_monthly_partition(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch.object(elastic_module._SMTPTLSReportDoc, "save"),
        ):
            save_smtp_tls_report_to_elasticsearch(
                _smtp_tls_report(), monthly_indexes=True
            )
            index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
        self.assertIn("smtp_tls-2024-02", index_calls)

    def test_index_name_honours_suffix_and_prefix(self):
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index") as mock_index_cls,
            patch.object(elastic_module._SMTPTLSReportDoc, "save"),
        ):
            save_smtp_tls_report_to_elasticsearch(
                _smtp_tls_report(), index_suffix="t1", index_prefix="cust_"
            )
            index_calls = [c.args[0] for c in mock_index_cls.call_args_list]
        self.assertIn("cust_smtp_tls_t1-2024-02-03", index_calls)

    def test_policy_without_strings_or_mx_patterns(self):
        """policy_strings / mx_host_patterns are optional in the
        report shape — verify the branch where they're absent."""
        report = _smtp_tls_report()
        for policy in report["policies"]:
            policy.pop("policy_strings", None)
            policy.pop("mx_host_patterns", None)
            policy.pop("failure_details", None)
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._SMTPTLSReportDoc, "save") as mock_save,
        ):
            save_smtp_tls_report_to_elasticsearch(report)
        mock_save.assert_called_once()

    def test_failure_details_all_optional_fields_populated(self):
        """Exercise every optional field in failure_details so the
        full set of `if "x" in failure_detail` branches runs."""
        report = _smtp_tls_report()
        report["policies"][0]["failure_details"] = [
            {
                "result_type": "certificate-expired",
                "failed_session_count": 1,
                "receiving_mx_hostname": "mx.example.com",
                "additional_information_uri": "https://example.com/why",
                "failure_reason_code": "ERR_CERT",
                "ip_address": "10.0.0.5",
                "receiving_ip": "10.0.0.2",
                "receiving_mx_helo": "mx.helo.example.com",
                "sending_mta_ip": "10.0.0.1",
            }
        ]
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(elastic_module._SMTPTLSReportDoc, "save") as mock_save,
        ):
            save_smtp_tls_report_to_elasticsearch(report)
        mock_save.assert_called_once()

    def test_save_populates_combined_policy_and_failure_detail_fields(self):
        """Regression guard for the SMTP TLS analogue of issue #169:
        policies and their failure_details are object arrays, so stacked
        terms aggregations on their subfields cross-product just like
        dkim_results/spf_results did. Two policies (one with two failure
        details, one with none) must yield exactly two policies_combined
        entries and two failure_details_combined entries, not a
        cross-product. autospec=True is required on the save patch so
        mock_save.call_args captures the doc instance as ``self``."""
        report = _smtp_tls_report(
            policies=[
                {
                    "policy_domain": "example.com",
                    "policy_type": "sts",
                    "successful_session_count": 100,
                    "failed_session_count": 2,
                    "failure_details": [
                        {
                            "result_type": "certificate-expired",
                            "failed_session_count": 1,
                            "sending_mta_ip": "192.0.2.1",
                            "receiving_ip": "203.0.113.1",
                            "receiving_mx_hostname": "mx1.example.com",
                            "additional_info_uri": (
                                "https://reports.example.com/tls-help"
                            ),
                        },
                        {
                            "result_type": "starttls-not-supported",
                            "failed_session_count": 1,
                            "sending_mta_ip": "192.0.2.2",
                            "receiving_ip": "203.0.113.2",
                            "receiving_mx_hostname": "mx2.example.com",
                        },
                    ],
                },
                {
                    "policy_domain": "example.net",
                    "policy_type": "tlsa",
                    "successful_session_count": 50,
                    "failed_session_count": 0,
                },
            ]
        )
        with (
            patch("parsedmarc.elastic.Search", return_value=_empty_search()),
            patch("parsedmarc.elastic.Index"),
            patch.object(
                elastic_module._SMTPTLSReportDoc, "save", autospec=True
            ) as mock_save,
        ):
            save_smtp_tls_report_to_elasticsearch(report)
        doc = mock_save.call_args[0][0]
        self.assertEqual(
            list(doc.policies_combined), ["example.com / sts", "example.net / tlsa"]
        )
        expected_detail_expired = (
            "example.com / sts / certificate-expired / 192.0.2.1 / "
            "203.0.113.1 / mx1.example.com"
        )
        expected_detail_starttls = (
            "example.com / sts / starttls-not-supported / 192.0.2.2 / "
            "203.0.113.2 / mx2.example.com"
        )
        self.assertEqual(
            list(doc.failure_details_combined),
            [expected_detail_expired, expected_detail_starttls],
        )
        # The parser emits additional_info_uri (SMTPTLSFailureDetailsOptional
        # in types.py); the saver must persist it on the declared
        # additional_information_uri field rather than dropping it.
        self.assertEqual(
            doc.policies[0].failure_details[0].additional_information_uri,
            "https://reports.example.com/tls-help",
        )


class TestBackwardCompatAlias(unittest.TestCase):
    def test_save_forensic_alias_points_to_save_failure(self):
        self.assertIs(
            elastic_module.save_forensic_report_to_elasticsearch,
            elastic_module.save_failure_report_to_elasticsearch,
        )

    def test_forensic_doc_alias_points_to_failure_doc(self):
        self.assertIs(
            elastic_module._ForensicReportDoc, elastic_module._FailureReportDoc
        )
        self.assertIs(
            elastic_module._ForensicSampleDoc, elastic_module._FailureSampleDoc
        )


# Silence unused-import lint in the test module preamble.
_ = call


if __name__ == "__main__":
    unittest.main(verbosity=2)