Address Copilot findings: harden migrate_indexes, normalize panel titles

Three unresolved review threads, all verified against cli.py's
re-raising init handler before fixing:

- elastic.py/opensearch.py: connections.get_connection() sat outside
  migrate_indexes()'s try/except, so a connection-registration failure
  would abort startup despite the docstring's promise that migration
  errors are caught and logged. Now caught, logged, and skipped until
  the next startup.
- opensearch.py: the legacy published_policy.fo migration loop did
  unguarded network I/O (exists/get_field_mapping/reindex/delete), so a
  transient cluster error aborted startup on the OpenSearch path while
  the identical situation on the Elasticsearch path was logged and
  survived. Each index's migration attempt is now wrapped, warns, and
  moves on.
- opensearch_dashboards.ndjson: normalized two pre-existing panel
  titles in the aggregate dashboard's panelsJSON ("Reporting
  organizations " trailing space, "Map  of message sources by country"
  double space).

Regression tests assert migrate_indexes never propagates connection or
per-index cluster errors on either backend.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

parsedmarc/elastic.py

@@ -599,7 +599,15 @@ def migrate_indexes(
     if not aggregate_indexes:
         return
 
-    client = connections.get_connection()
+    try:
+        client = connections.get_connection()
+    except Exception as e:
+        logger.warning(
+            "Skipping the dkim_results_combined/spf_results_combined "
+            f"backfill: could not get an Elasticsearch connection: {e}. "
+            "This will be retried at the next startup."
+        )
+        return
     for name in aggregate_indexes:
         pattern = f"{name}*"
         try:

parsedmarc/opensearch.py

@@ -527,35 +527,54 @@ def migrate_indexes(
 
     version = 2
     for aggregate_index_name in aggregate_indexes:
-        if not Index(aggregate_index_name).exists():
-            continue
-        aggregate_index = Index(aggregate_index_name)
-        doc = "doc"
-        fo_field = "published_policy.fo"
-        fo = "fo"
-        fo_mapping = aggregate_index.get_field_mapping(fields=[fo_field])
-        fo_mapping = fo_mapping[list(fo_mapping.keys())[0]]["mappings"]
-        if doc not in fo_mapping:
-            continue
+        try:
+            if not Index(aggregate_index_name).exists():
+                continue
+            aggregate_index = Index(aggregate_index_name)
+            doc = "doc"
+            fo_field = "published_policy.fo"
+            fo = "fo"
+            fo_mapping = aggregate_index.get_field_mapping(fields=[fo_field])
+            fo_mapping = fo_mapping[list(fo_mapping.keys())[0]]["mappings"]
+            if doc not in fo_mapping:
+                continue
 
-        fo_mapping = fo_mapping[doc][fo_field]["mapping"][fo]
-        fo_type = fo_mapping["type"]
-        if fo_type == "long":
-            new_index_name = "{0}-v{1}".format(aggregate_index_name, version)
-            body = {
-                "properties": {
-                    "published_policy.fo": {
-                        "type": "text",
-                        "fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
+            fo_mapping = fo_mapping[doc][fo_field]["mapping"][fo]
+            fo_type = fo_mapping["type"]
+            if fo_type == "long":
+                new_index_name = "{0}-v{1}".format(aggregate_index_name, version)
+                body = {
+                    "properties": {
+                        "published_policy.fo": {
+                            "type": "text",
+                            "fields": {
+                                "keyword": {"type": "keyword", "ignore_above": 256}
+                            },
+                        }
                     }
                 }
-            }
-            Index(new_index_name).create()
-            Index(new_index_name).put_mapping(doc_type=doc, body=body)
-            reindex(connections.get_connection(), aggregate_index_name, new_index_name)
-            Index(aggregate_index_name).delete()
+                Index(new_index_name).create()
+                Index(new_index_name).put_mapping(doc_type=doc, body=body)
+                reindex(
+                    connections.get_connection(), aggregate_index_name, new_index_name
+                )
+                Index(aggregate_index_name).delete()
+        except Exception as e:
+            logger.warning(
+                "Failed the legacy published_policy.fo migration for "
+                f"{aggregate_index_name}: {e}. This will be retried at the "
+                "next startup."
+            )
 
-    client = connections.get_connection()
+    try:
+        client = connections.get_connection()
+    except Exception as e:
+        logger.warning(
+            "Skipping the dkim_results_combined/spf_results_combined "
+            f"backfill: could not get an OpenSearch connection: {e}. "
+            "This will be retried at the next startup."
+        )
+        return
     for name in aggregate_indexes:
         pattern = f"{name}*"
         try:

tests/test_elastic.py

@@ -458,6 +458,22 @@ class TestMigrateIndexes(unittest.TestCase):
         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))
+
 
 # ---------------------------------------------------------------------------
 # save_aggregate_report_to_elasticsearch

tests/test_opensearch.py

@@ -449,6 +449,27 @@ class TestMigrateIndexes(unittest.TestCase):
         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 for the combined-field backfill; if it raises (e.g. no
+        OpenSearch 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.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_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))
+
 
 class TestMigrateIndexesFoMigration(unittest.TestCase):
     """The legacy `published_policy.fo` field was mapped as `long` in
@@ -539,6 +560,35 @@ class TestMigrateIndexesFoMigration(unittest.TestCase):
             migrate_indexes(aggregate_indexes=["dmarc_aggregate-2024-01-01"])
         mock_reindex.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.opensearch.Index") as mock_index_cls,
+            patch("parsedmarc.opensearch.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"])
+
+        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))
+
 
 # ---------------------------------------------------------------------------
 # save_aggregate_report_to_opensearch