diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ca9b0aa..d2bbbc67 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ - **The Elasticsearch/OpenSearch aggregate dashboards' over-time charts (and the Grafana ES dashboard's summary pies and time series) bucketed on the multi-valued `date_range` field**; a date histogram counts a report once per value, double-counting any report whose begin and end dates fall in different buckets. All date histograms and time-range filters now use the single-valued `date_begin`, matching the report-begin semantics of the PostgreSQL (`begin_date`) and Splunk (`_time` = interval begin) dashboards. - **Aggregate-report policy and authentication result words are now normalized to lowercase** ([#288](https://github.com/domainaware/parsedmarc/issues/288)): reporters that emit mixed-case values such as `Pass` no longer create duplicate result categories in outputs and dashboards. - **The results email (SMTP and Microsoft Graph) is no longer sent when no reports were parsed** ([#200](https://github.com/domainaware/parsedmarc/issues/200)): previously an empty run — an empty inbox, or one where every message was invalid — still emailed a zip of headers-only CSVs. The email step is now skipped with an INFO log when the run produced no aggregate, failure, or SMTP TLS reports. -- **The DKIM/SPF alignment-detail tables on the Kibana/OpenSearch Dashboards, Grafana (Elasticsearch), and Splunk aggregate dashboards no longer show a selector × domain × result cross-product** ([#169](https://github.com/domainaware/parsedmarc/issues/169)). Elasticsearch/OpenSearch flatten the `dkim_results`/`spf_results` object arrays (they are dynamic-mapped as `object`, not `nested`), so stacking terms aggregations on their subfields produced every combination of selector/domain/result across a report's signatures, each repeating the full message count. Aggregate documents now also carry `dkim_results_combined` and `spf_results_combined` — one `"selector / domain / result"` (`"scope / domain / result"`) string per auth result — and the dashboards aggregate those instead; the Splunk detail panels now pair the values with `mvzip`/`mvexpand`. Documents saved by older versions can be backfilled with a documented `_update_by_query` command (see the Elasticsearch docs page). The Grafana "DKIM Alignment Details" panel's dmarcian.com DKIM-checker data link was removed because it required the separate domain/selector columns. The SMTP TLS visualizations have the same class of defect and are tracked separately. +- **The DKIM/SPF alignment-detail tables on the Kibana/OpenSearch Dashboards, Grafana (Elasticsearch), and Splunk aggregate dashboards no longer show a selector × domain × result cross-product** ([#169](https://github.com/domainaware/parsedmarc/issues/169)). Elasticsearch/OpenSearch flatten the `dkim_results`/`spf_results` object arrays (they are dynamic-mapped as `object`, not `nested`), so stacking terms aggregations on their subfields produced every combination of selector/domain/result across a report's signatures, each repeating the full message count. Aggregate documents now also carry `dkim_results_combined` and `spf_results_combined` — one `"selector / domain / result"` (`"scope / domain / result"`) string per auth result — and the dashboards aggregate those instead; the Splunk detail panels now pair the values with `mvzip`/`mvexpand`. Documents saved by older versions are now backfilled automatically at startup (a non-blocking, idempotent background task); the documented `_update_by_query` command (see the Elasticsearch docs page) remains available for running the backfill manually. The Grafana "DKIM Alignment Details" panel's dmarcian.com DKIM-checker data link was removed because it required the separate domain/selector columns. The SMTP TLS visualizations have the same class of defect and are tracked separately. - **Corrected the dead `_SPFResult.results` (plural) field declaration to `result`**, matching what was always written to it. ## 10.2.4 diff --git a/docs/source/elasticsearch.md b/docs/source/elasticsearch.md index 5e08b983..4408f79f 100644 --- a/docs/source/elasticsearch.md +++ b/docs/source/elasticsearch.md @@ -235,14 +235,28 @@ result paired, which the dashboards' alignment-detail tables aggregate on. Reports saved by older versions lack these fields and will not appear in those tables. -Running the following once per cluster backfills the fields on existing -documents. It is idempotent (documents that already have the fields are -skipped), so it is safe to re-run. It works identically on OpenSearch; -just adjust the URL and credentials. The query matches only documents -that have at least one DKIM or SPF auth result and lack the corresponding -combined field; documents with no auth results are skipped, because an -`exists` query cannot see an empty array, and for search purposes an -empty `dkim_results_combined` is identical to an absent one. +parsedmarc now backfills this automatically. On startup, it runs a cheap +count query against each configured aggregate index pattern to check for +documents that have DKIM or SPF results but are missing the corresponding +combined field. If any are found, it submits the backfill as a background +`_update_by_query` task (`wait_for_completion=false`), so startup is never +blocked on it; progress is logged, including the task ID. The check itself +is idempotent — once an index is fully backfilled, later startups see a +count of 0 and log nothing further — and it works the same way on +OpenSearch. Any error talking to the cluster (for example, no indexes yet +on a fresh install) is logged as a warning and retried on the next startup, +rather than aborting parsedmarc. + +If you upgrade the dashboards without pointing the new parsedmarc version +at the cluster, or you'd rather control when the write load happens, you +can still run the backfill manually. It is idempotent (documents that +already have the fields are skipped), so it is safe to re-run. It works +identically on OpenSearch; just adjust the URL and credentials. The query +matches only documents that have at least one DKIM or SPF auth result and +lack the corresponding combined field; documents with no auth results are +skipped, because an `exists` query cannot see an empty array, and for +search purposes an empty `dkim_results_combined` is identical to an +absent one. ```bash curl -X POST "http://localhost:9200/dmarc_aggregate*/_update_by_query?conflicts=proceed&wait_for_completion=false" \ diff --git a/parsedmarc/elastic.py b/parsedmarc/elastic.py index bf100428..2f5dccc8 100644 --- a/parsedmarc/elastic.py +++ b/parsedmarc/elastic.py @@ -41,6 +41,64 @@ _SERVERLESS = False # settings (e.g. ``refresh_interval``) are accepted and pass through. _SERVERLESS_REJECTED_SETTINGS = frozenset({"number_of_shards", "number_of_replicas"}) +# Guard query for the dkim_results_combined/spf_results_combined backfill +# (see ``migrate_indexes``). Matches only documents that have at least one +# DKIM or SPF auth result and are missing the corresponding combined field. +# Empty arrays are invisible to ``exists``, so documents with zero DKIM/SPF +# results are correctly skipped, making this idempotent: once a document is +# backfilled it no longer matches, and re-running against an already +# up-to-date index counts 0. +_COMBINED_BACKFILL_QUERY: dict[str, Any] = { + "bool": { + "minimum_should_match": 1, + "should": [ + { + "bool": { + "must": [{"exists": {"field": "dkim_results.domain"}}], + "must_not": [{"exists": {"field": "dkim_results_combined"}}], + } + }, + { + "bool": { + "must": [{"exists": {"field": "spf_results.domain"}}], + "must_not": [{"exists": {"field": "spf_results_combined"}}], + } + }, + ], + } +} + +# Painless script that (re)derives dkim_results_combined/spf_results_combined +# from dkim_results/spf_results, matching the format written by +# save_aggregate_report_to_elasticsearch(): "{selector} / {domain} / {result}" +# per DKIM result and "{scope} / {domain} / {result}" per SPF result. +_COMBINED_BACKFILL_SCRIPT = ( + "List dk = new ArrayList(); " + "def dr = ctx._source.dkim_results; " + "if (dr != null) { " + "if (!(dr instanceof List)) { dr = [dr]; } " + "for (e in dr) { " + "if (e == null) { continue; } " + 'def sel = e.selector != null ? e.selector : "none"; ' + 'def dom = e.domain != null ? e.domain : "none"; ' + 'def res = e.result != null ? e.result : "none"; ' + 'dk.add(sel + " / " + dom + " / " + res); ' + "} } " + "ctx._source.dkim_results_combined = dk; " + "List sp = new ArrayList(); " + "def sr = ctx._source.spf_results; " + "if (sr != null) { " + "if (!(sr instanceof List)) { sr = [sr]; } " + "for (e in sr) { " + "if (e == null) { continue; } " + 'def sc = e.scope != null ? e.scope : "mfrom"; ' + 'def dom = e.domain != null ? e.domain : "none"; ' + 'def res = e.result != null ? e.result : (e.results != null ? e.results : "none"); ' + 'sp.add(sc + " / " + dom + " / " + res); ' + "} } " + "ctx._source.spf_results_combined = sp;" +) + class _PolicyOverride(InnerDoc): # The elasticsearch.dsl 8.x type stubs use dataclass_transform and only @@ -487,23 +545,66 @@ def migrate_indexes( failure_indexes: list[str] | None = None, ): """ - Updates index mappings + Backfills the ``dkim_results_combined``/``spf_results_combined`` fields + (added for issue #169) on aggregate report documents that were saved + before those fields existed. - This is a no-op kept for API compatibility (``cli.py`` calls it on - startup). The only migration this function ever performed was - re-typing ``published_policy.fo`` from ``long`` to ``text``, which - applied exclusively to indices still carrying the legacy - Elasticsearch 6-era ``"doc"`` mapping type. The 8.x client can only - reach servers (Elasticsearch 8.x/9.x) whose indices were created on - Elasticsearch 7.x or later and are therefore typeless, so that - migration path is unreachable and has been removed. + For each name in ``aggregate_indexes``, this submits an + ``update_by_query`` against the ``f"{name}*"`` index pattern (the real + indexes are date-suffixed) as a non-blocking background task + (``wait_for_completion=False``), so it never delays parsedmarc startup. + Submission is guarded by a cheap ``count`` query that only matches + documents with DKIM/SPF results but no combined field, so once an index + is fully backfilled, later calls are a fast no-op. Any error talking to + the cluster (e.g. no indexes yet on a fresh install, or a transient + connection issue) is caught and logged as a warning rather than raised; + the backfill is simply retried on the next startup, and the manual + ``_update_by_query`` command documented in + ``docs/source/elasticsearch.md`` remains available in the meantime. Args: aggregate_indexes (list): A list of aggregate index names - (accepted for API compatibility; unused) failure_indexes (list): A list of failure index names (accepted for API compatibility; unused) """ + if not aggregate_indexes: + return + + client = connections.get_connection() + for name in aggregate_indexes: + pattern = f"{name}*" + try: + count_response = client.count( + index=pattern, + query=_COMBINED_BACKFILL_QUERY, + ignore_unavailable=True, + allow_no_indices=True, + ) + count = count_response["count"] + if not count: + continue + update_response = client.update_by_query( + index=pattern, + query=_COMBINED_BACKFILL_QUERY, + script={"source": _COMBINED_BACKFILL_SCRIPT, "lang": "painless"}, + conflicts="proceed", + wait_for_completion=False, + ignore_unavailable=True, + allow_no_indices=True, + ) + task_id = update_response.get("task") + logger.info( + "Backfilling dkim_results_combined/spf_results_combined on " + f"{count} existing documents in {pattern} (task {task_id})" + ) + except Exception as e: + logger.warning( + "Failed to check/submit the dkim_results_combined/" + f"spf_results_combined backfill for {pattern}: {e}. This " + "will be retried at the next startup; the manual " + "_update_by_query command in the documentation remains " + "available in the meantime." + ) def save_aggregate_report_to_elasticsearch( diff --git a/parsedmarc/opensearch.py b/parsedmarc/opensearch.py index 828358d0..160b2b93 100644 --- a/parsedmarc/opensearch.py +++ b/parsedmarc/opensearch.py @@ -34,6 +34,65 @@ class OpenSearchError(Exception): """Raised when an OpenSearch error occurs""" +# Guard query for the dkim_results_combined/spf_results_combined backfill +# (see ``migrate_indexes``). Matches only documents that have at least one +# DKIM or SPF auth result and are missing the corresponding combined field. +# Empty arrays are invisible to ``exists``, so documents with zero DKIM/SPF +# results are correctly skipped, making this idempotent: once a document is +# backfilled it no longer matches, and re-running against an already +# up-to-date index counts 0. +_COMBINED_BACKFILL_QUERY: dict[str, Any] = { + "bool": { + "minimum_should_match": 1, + "should": [ + { + "bool": { + "must": [{"exists": {"field": "dkim_results.domain"}}], + "must_not": [{"exists": {"field": "dkim_results_combined"}}], + } + }, + { + "bool": { + "must": [{"exists": {"field": "spf_results.domain"}}], + "must_not": [{"exists": {"field": "spf_results_combined"}}], + } + }, + ], + } +} + +# Painless script that (re)derives dkim_results_combined/spf_results_combined +# from dkim_results/spf_results, matching the format written by +# save_aggregate_report_to_opensearch(): "{selector} / {domain} / {result}" +# per DKIM result and "{scope} / {domain} / {result}" per SPF result. +_COMBINED_BACKFILL_SCRIPT = ( + "List dk = new ArrayList(); " + "def dr = ctx._source.dkim_results; " + "if (dr != null) { " + "if (!(dr instanceof List)) { dr = [dr]; } " + "for (e in dr) { " + "if (e == null) { continue; } " + 'def sel = e.selector != null ? e.selector : "none"; ' + 'def dom = e.domain != null ? e.domain : "none"; ' + 'def res = e.result != null ? e.result : "none"; ' + 'dk.add(sel + " / " + dom + " / " + res); ' + "} } " + "ctx._source.dkim_results_combined = dk; " + "List sp = new ArrayList(); " + "def sr = ctx._source.spf_results; " + "if (sr != null) { " + "if (!(sr instanceof List)) { sr = [sr]; } " + "for (e in sr) { " + "if (e == null) { continue; } " + 'def sc = e.scope != null ? e.scope : "mfrom"; ' + 'def dom = e.domain != null ? e.domain : "none"; ' + 'def res = e.result != null ? e.result : (e.results != null ? e.results : "none"); ' + 'sp.add(sc + " / " + dom + " / " + res); ' + "} } " + "ctx._source.spf_results_combined = sp;" +) + + class _PolicyOverride(InnerDoc): type = Text() comment = Text() @@ -406,7 +465,27 @@ def migrate_indexes( failure_indexes: list[str] | None = None, ): """ - Updates index mappings + Runs index migrations and backfills. + + First, the legacy ``published_policy.fo`` migration: indexes where that + field was mapped as ``long`` (data indexed by very old parsedmarc + releases under the Elasticsearch 6-era ``doc`` mapping type) are rebuilt + as a ``-v2`` index with the text/keyword shape. + + Second, the ``dkim_results_combined``/``spf_results_combined`` backfill + (added for issue #169) for aggregate report documents that were saved + before those fields existed. For each name in ``aggregate_indexes``, + this submits an ``update_by_query`` against the ``f"{name}*"`` index + pattern (the real indexes are date-suffixed) as a non-blocking + background task (``wait_for_completion=False``), so it never delays + parsedmarc startup. Submission is guarded by a cheap ``count`` query + that only matches documents with DKIM/SPF results but no combined + field, so once an index is fully backfilled, later calls are a fast + no-op. Any error talking to the cluster (e.g. no indexes yet on a + fresh install, or a transient connection issue) is caught and logged + as a warning rather than raised; the backfill is simply retried on the + next startup, and the manual ``_update_by_query`` command documented + in ``docs/source/elasticsearch.md`` remains available in the meantime. Args: aggregate_indexes (list): A list of aggregate index names @@ -414,9 +493,10 @@ def migrate_indexes( (accepted for API compatibility; no migrations are currently needed for failure indexes) """ + if not aggregate_indexes: + return + version = 2 - if aggregate_indexes is None: - aggregate_indexes = [] for aggregate_index_name in aggregate_indexes: if not Index(aggregate_index_name).exists(): continue @@ -446,6 +526,47 @@ def migrate_indexes( reindex(connections.get_connection(), aggregate_index_name, new_index_name) Index(aggregate_index_name).delete() + client = connections.get_connection() + for name in aggregate_indexes: + pattern = f"{name}*" + try: + count_response = client.count( + index=pattern, + body={"query": _COMBINED_BACKFILL_QUERY}, + ignore_unavailable=True, + allow_no_indices=True, + ) + count = count_response["count"] + if not count: + continue + update_response = client.update_by_query( + index=pattern, + body={ + "query": _COMBINED_BACKFILL_QUERY, + "script": { + "source": _COMBINED_BACKFILL_SCRIPT, + "lang": "painless", + }, + }, + conflicts="proceed", + wait_for_completion=False, + ignore_unavailable=True, + allow_no_indices=True, + ) + task_id = update_response.get("task") + logger.info( + "Backfilling dkim_results_combined/spf_results_combined on " + f"{count} existing documents in {pattern} (task {task_id})" + ) + except Exception as e: + logger.warning( + "Failed to check/submit the dkim_results_combined/" + f"spf_results_combined backfill for {pattern}: {e}. This " + "will be retried at the next startup; the manual " + "_update_by_query command in the documentation remains " + "available in the meantime." + ) + def save_aggregate_report_to_opensearch( aggregate_report: dict[str, Any], diff --git a/tests/test_elastic.py b/tests/test_elastic.py index 1740d091..f8eb9e79 100644 --- a/tests/test_elastic.py +++ b/tests/test_elastic.py @@ -16,6 +16,7 @@ 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, @@ -395,6 +396,69 @@ class TestCreateIndexesServerless(unittest.TestCase): 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() + + # --------------------------------------------------------------------------- # save_aggregate_report_to_elasticsearch # --------------------------------------------------------------------------- diff --git a/tests/test_opensearch.py b/tests/test_opensearch.py index 1c3e4996..a3374f5e 100644 --- a/tests/test_opensearch.py +++ b/tests/test_opensearch.py @@ -376,16 +376,100 @@ class TestCreateIndexes(unittest.TestCase): 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.opensearch.Index") as mock_index_cls, + patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn, + ): + # The legacy fo migration that runs first sees no base index. + mock_index_cls.return_value.exists.return_value = False + 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["body"]["query"], opensearch_module._COMBINED_BACKFILL_QUERY + ) + script_source = kwargs["body"]["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["body"]["query"], opensearch_module._COMBINED_BACKFILL_QUERY + ) + + def test_backfill_skipped_when_no_old_docs(self): + with ( + patch("parsedmarc.opensearch.Index") as mock_index_cls, + patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn, + ): + mock_index_cls.return_value.exists.return_value = False + 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.opensearch.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.opensearch.Index") as mock_index_cls, + patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn, + ): + mock_index_cls.return_value.exists.return_value = False + 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() + + +class TestMigrateIndexesFoMigration(unittest.TestCase): """The legacy `published_policy.fo` field was mapped as `long` in older indexes. migrate_indexes detects that and rebuilds the index with the text/keyword shape. The branch is gnarly; a regression - would silently leave old data un-migrated.""" + would silently leave old data un-migrated. Each test stubs the + combined-field backfill that now runs afterwards in the same call + (count 0 → no-op).""" - def test_no_indexes_is_noop(self): - migrate_indexes() # Should not raise + @staticmethod + def _noop_backfill_client(): + client = MagicMock() + client.count.return_value = {"count": 0} + return client def test_skips_non_existent_index(self): - with patch("parsedmarc.opensearch.Index") as mock_index_cls: + with ( + patch("parsedmarc.opensearch.Index") as mock_index_cls, + patch("parsedmarc.opensearch.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(aggregate_indexes=["missing"]) # exists() returned False — no field_mapping fetch. @@ -394,12 +478,16 @@ class TestMigrateIndexes(unittest.TestCase): def test_skips_when_doc_mapping_absent(self): """An index that has 'fo' but not under the 'doc' type (e.g., empty index with default mapping) is left alone.""" - with patch("parsedmarc.opensearch.Index") as mock_index_cls: + with ( + patch("parsedmarc.opensearch.Index") as mock_index_cls, + patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn, + patch("parsedmarc.opensearch.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 = {"some_key": {"mappings": {}}} - with patch("parsedmarc.opensearch.reindex") as mock_reindex: - migrate_indexes(aggregate_indexes=["dmarc_aggregate-2023-01-01"]) + migrate_indexes(aggregate_indexes=["dmarc_aggregate-2023-01-01"]) mock_reindex.assert_not_called() def test_migrates_when_fo_is_long(self): @@ -411,6 +499,8 @@ class TestMigrateIndexes(unittest.TestCase): patch("parsedmarc.opensearch.reindex") as mock_reindex, patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn, ): + mock_client = self._noop_backfill_client() + mock_get_conn.return_value = mock_client idx = mock_index_cls.return_value idx.exists.return_value = True idx.get_field_mapping.return_value = { @@ -423,16 +513,18 @@ class TestMigrateIndexes(unittest.TestCase): } } migrate_indexes(aggregate_indexes=["dmarc_aggregate-2023-01-01"]) - # reindex called from old → new (v2) index. + # reindex called from old → new (v2) index, with the client from + # connections.get_connection(). mock_reindex.assert_called_once() - # connections.get_connection consulted to get the ES client. - mock_get_conn.assert_called_once() + self.assertIs(mock_reindex.call_args.args[0], mock_client) def test_skips_when_fo_already_text(self): with ( patch("parsedmarc.opensearch.Index") as mock_index_cls, + patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn, patch("parsedmarc.opensearch.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 = {