Migrate Elasticsearch output to the elasticsearch-py 8.x client (#806)

The mandatory elasticsearch<7.14.0 + elasticsearch-dsl==7.4.0 pins
transitively forced urllib3<2 (EOL 1.26.x) onto every install. The old
<7.14.0 cap only existed to dodge the client product check that broke
OpenSearch users (#452, #653) — obsolete now that parsedmarc has a
dedicated [opensearch] backend on opensearch-py.

- Depend on elasticsearch>=8.18,<9 and drop elasticsearch-dsl entirely
  (the DSL ships inside the client as elasticsearch.dsl since 8.18.0).
  The 8.x client's elastic-transport allows urllib3>=1.26.2,<3, so
  installs can now resolve urllib3 2.x. The 8.x line supports both
  Elasticsearch 8.x and 9.x servers; ES 7.x servers are no longer
  supported, and OpenSearch users pointing [elasticsearch] at an
  OpenSearch cluster must switch to the [opensearch] section.
- set_hosts() now builds 8.x connection kwargs (scheme-qualified host
  URLs, request_timeout, basic_auth) while keeping the function
  signature and every INI option unchanged.
- migrate_indexes() is now a documented no-op kept for API
  compatibility: its only migration (re-typing published_policy.fo
  from long to text) applied exclusively to indices carrying the
  legacy ES 6-era "doc" mapping type, which cannot exist on any
  server the 8.x client can reach.
- The elasticsearch.dsl 8.x stubs use dataclass_transform and don't
  surface pre-8.x-style bare `name = Text()` fields as constructor
  parameters; each Document/InnerDoc class now carries a
  TYPE_CHECKING-only `__init__(*args, **kwargs)` declaration matching
  the real runtime signature, which also made nine pre-existing
  pyright ignores unnecessary.

Verified with ruff, pyright (0 errors/0 warnings), the full pytest
suite (718 passed), and a CLI run over the bundled samples; CI's live
elasticsearch:8.19.7 service exercises the new client end-to-end.

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

AGENTS.md

@@ -138,7 +138,7 @@ A test that mocks every dependency and asserts that the mocks were invoked is te
 
 Concrete patterns:
 
-- **Mock at SDK boundaries, not at internal helpers.** Patch `boto3.resource`, `kafka.KafkaProducer`, `requests.Session.post`, `elasticsearch_dsl.Document.save`, `azure.monitor.ingestion.LogsIngestionClient` — the seams where the project's code stops and an external system begins. Don't patch our own functions just to make a test "easier"; that hides bugs in the function instead of testing it.
+- **Mock at SDK boundaries, not at internal helpers.** Patch `boto3.resource`, `kafka.KafkaProducer`, `requests.Session.post`, `elasticsearch.dsl.Document.save`, `azure.monitor.ingestion.LogsIngestionClient` — the seams where the project's code stops and an external system begins. Don't patch our own functions just to make a test "easier"; that hides bugs in the function instead of testing it.
 - **Assert on what gets sent, not that something was sent.** For an output module, parse the body that was passed to the mocked transport (`json.loads(call.kwargs["data"])`, `kafka.send.call_args.args[1]`, `bucket.put_object.call_args.kwargs["Key"]`) and verify the *fields and values a dashboard or downstream consumer would actually filter on*. A test that only checks `mock.assert_called_once()` would pass even if the payload were `{}`.
 - **No trivial passthrough tests.** A test that calls a getter and asserts it returns the value just set isn't testing the code; it's testing Python's attribute machinery.
 - **No `# pragma: no cover`.** If a branch is unreachable, the right fix is to delete the branch, not to hide it.

CHANGELOG.md

@@ -1,8 +1,14 @@
 # Changelog
 
-## Unreleased
+## 10.2.3
 
-### 10.2.2
+### Changes
+
+- **Migrated the Elasticsearch output to the elasticsearch-py 8.x client** ([#806](https://github.com/domainaware/parsedmarc/issues/806)): the `elasticsearch-dsl` dependency is gone (the DSL is bundled in the client as `elasticsearch.dsl` since 8.18), and the new pin `elasticsearch>=8.18,<9` no longer forces `urllib3<2` — installs can now resolve urllib3 2.x. **Elasticsearch 7.x servers are no longer supported** (the 8.x client supports ES 8.x and 9.x servers). **OpenSearch users who were pointing the `[elasticsearch]` config section at an OpenSearch cluster must switch to the `[opensearch]` section** (the 8.x client's product check rejects OpenSearch). Also removed the dead ES 6-era `published_policy.fo` index migration in `migrate_indexes()` (unreachable via the 8.x client; the function remains as a no-op for API compatibility).
+
+## 10.2.2
+
+### Changes
 
 - Removed dead code found while extending test coverage: the unused `_SMTPTLSReportDoc.add_policy()` helpers in the Elasticsearch and OpenSearch outputs (the save paths construct policy documents directly), a no-op `failure_indexes` loop in both `migrate_indexes()` implementations (the parameter is still accepted; no failure-index migrations are currently needed), an unreachable `importlib.resources` ImportError fallback in `parsedmarc.utils` (it re-imported the same module, and `importlib.resources.files` always exists on the supported Python ≥3.10), and an unreachable "Invalid report content" guard in `extract_report()` (every input branch either assigns the file object or raises first, confirmed by pyright narrowing).
 

docs/source/elasticsearch.md

@@ -3,7 +3,10 @@
 To set up visual dashboards of DMARC data, install Elasticsearch and Kibana.
 
 :::{note}
-Elasticsearch and Kibana 6 or later are required
+Elasticsearch and Kibana 8 or later are required (parsedmarc's 8.x Python
+client also supports Elasticsearch 9). OpenSearch users must use the
+`[opensearch]` configuration section instead — the Elasticsearch 8.x client
+refuses to connect to non-Elasticsearch clusters.
 :::
 
 ## Installation

parsedmarc/constants.py

@@ -1,4 +1,4 @@
-__version__ = "10.2.2"
+__version__ = "10.2.3"
 
 USER_AGENT = f"parsedmarc/{__version__}"
 

parsedmarc/elastic.py

@@ -2,10 +2,9 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
-from elasticsearch.helpers import reindex
-from elasticsearch_dsl import (
+from elasticsearch.dsl import (
     Boolean,
     Date,
     Document,
@@ -16,11 +15,11 @@ from elasticsearch_dsl import (
     Keyword,
     Nested,
     Object,
+    Q,
     Search,
     Text,
     connections,
 )
-from elasticsearch_dsl.search import Q
 
 from parsedmarc import InvalidFailureReport
 from parsedmarc.log import logger
@@ -44,11 +43,26 @@ _SERVERLESS_REJECTED_SETTINGS = frozenset({"number_of_shards", "number_of_replic
 
 
 class _PolicyOverride(InnerDoc):
+    # The elasticsearch.dsl 8.x type stubs use dataclass_transform and only
+    # surface dataclass-style annotated fields (``name: M[...] =
+    # mapped_field(...)``) as constructor parameters. This module declares
+    # fields the pre-8.x way (``name = Text()``), which the runtime fully
+    # supports via ObjectBase.__init__(**kwargs), so this TYPE_CHECKING-only
+    # declaration restores the real runtime signature for the type checker.
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     type = Text()
     comment = Text()
 
 
 class _PublishedPolicy(InnerDoc):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     domain = Text()
     adkim = Text()
     aspf = Text()
@@ -62,6 +76,11 @@ class _PublishedPolicy(InnerDoc):
 
 
 class _DKIMResult(InnerDoc):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     domain = Text()
     selector = Text()
     result = Text()
@@ -69,6 +88,11 @@ class _DKIMResult(InnerDoc):
 
 
 class _SPFResult(InnerDoc):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     domain = Text()
     scope = Text()
     results = Text()
@@ -76,6 +100,11 @@ class _SPFResult(InnerDoc):
 
 
 class _AggregateReportDoc(Document):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     class Index:
         name = "dmarc_aggregate"
 
@@ -118,7 +147,7 @@ class _AggregateReportDoc(Document):
     generator = Text()
 
     def add_policy_override(self, type_: str, comment: str):
-        self.policy_overrides.append(_PolicyOverride(type=type_, comment=comment))  # pyright: ignore[reportCallIssue]
+        self.policy_overrides.append(_PolicyOverride(type=type_, comment=comment))
 
     def add_dkim_result(
         self,
@@ -134,7 +163,7 @@ class _AggregateReportDoc(Document):
                 result=result,
                 human_result=human_result,
             )
-        )  # pyright: ignore[reportCallIssue]
+        )
 
     def add_spf_result(
         self,
@@ -150,7 +179,7 @@ class _AggregateReportDoc(Document):
                 result=result,
                 human_result=human_result,
             )
-        )  # pyright: ignore[reportCallIssue]
+        )
 
     def save(self, **kwargs):  # pyright: ignore[reportIncompatibleMethodOverride]
         self.passed_dmarc = False
@@ -160,17 +189,32 @@ class _AggregateReportDoc(Document):
 
 
 class _EmailAddressDoc(InnerDoc):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     display_name = Text()
     address = Text()
 
 
 class _EmailAttachmentDoc(Document):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     filename = Text()
     content_type = Text()
     sha256 = Text()
 
 
 class _FailureSampleDoc(InnerDoc):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     raw = Text()
     headers = Object()
     headers_only = Boolean()
@@ -186,28 +230,33 @@ class _FailureSampleDoc(InnerDoc):
     attachments = Nested(_EmailAttachmentDoc)
 
     def add_to(self, display_name: str, address: str):
-        self.to.append(_EmailAddressDoc(display_name=display_name, address=address))  # pyright: ignore[reportCallIssue]
+        self.to.append(_EmailAddressDoc(display_name=display_name, address=address))
 
     def add_reply_to(self, display_name: str, address: str):
         self.reply_to.append(
             _EmailAddressDoc(display_name=display_name, address=address)
-        )  # pyright: ignore[reportCallIssue]
+        )
 
     def add_cc(self, display_name: str, address: str):
-        self.cc.append(_EmailAddressDoc(display_name=display_name, address=address))  # pyright: ignore[reportCallIssue]
+        self.cc.append(_EmailAddressDoc(display_name=display_name, address=address))
 
     def add_bcc(self, display_name: str, address: str):
-        self.bcc.append(_EmailAddressDoc(display_name=display_name, address=address))  # pyright: ignore[reportCallIssue]
+        self.bcc.append(_EmailAddressDoc(display_name=display_name, address=address))
 
     def add_attachment(self, filename: str, content_type: str, sha256: str):
         self.attachments.append(
             _EmailAttachmentDoc(
                 filename=filename, content_type=content_type, sha256=sha256
             )
-        )  # pyright: ignore[reportCallIssue]
+        )
 
 
 class _FailureReportDoc(Document):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     class Index:
         name = "dmarc_failure"
 
@@ -234,6 +283,11 @@ class _FailureReportDoc(Document):
 
 
 class _SMTPTLSFailureDetailsDoc(InnerDoc):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     result_type = Text()
     sending_mta_ip = Ip()
     receiving_mx_helo = Text()
@@ -244,6 +298,11 @@ class _SMTPTLSFailureDetailsDoc(InnerDoc):
 
 
 class _SMTPTLSPolicyDoc(InnerDoc):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     policy_domain = Text()
     policy_type = Text()
     policy_strings = Text()
@@ -275,10 +334,15 @@ class _SMTPTLSPolicyDoc(InnerDoc):
             additional_information=additional_information_uri,
             failure_reason_code=failure_reason_code,
         )
-        self.failure_details.append(_details)  # pyright: ignore[reportCallIssue]
+        self.failure_details.append(_details)
 
 
 class _SMTPTLSReportDoc(Document):
+    # TYPE_CHECKING __init__: see _PolicyOverride
+    if TYPE_CHECKING:
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
     class Index:
         name = "smtp_tls"
 
@@ -312,7 +376,9 @@ def set_hosts(
 
     Args:
         hosts (str | list[str]): A single hostname or URL, or list of hostnames or URLs
-        use_ssl (bool): Use an HTTPS connection to the server
+        use_ssl (bool): Controls the scheme prepended to any host that doesn't
+            already include one (``https://`` when True, ``http://`` when
+            False). Hosts that already carry a scheme are left unchanged.
         ssl_cert_path (str): Path to the certificate chain
         skip_certificate_verification (bool): Skip certificate verification
         username (str): The username to use for authentication
@@ -329,9 +395,12 @@ def set_hosts(
     _SERVERLESS = serverless
     if not isinstance(hosts, list):
         hosts = [hosts]
-    conn_params = {"hosts": hosts, "timeout": timeout}
+    scheme = "https://" if use_ssl else "http://"
+    normalized_hosts = [
+        host if "://" in host else "{0}{1}".format(scheme, host) for host in hosts
+    ]
+    conn_params = {"hosts": normalized_hosts, "request_timeout": timeout}
     if use_ssl:
-        conn_params["use_ssl"] = True
         if ssl_cert_path:
             conn_params["ca_certs"] = ssl_cert_path
         if skip_certificate_verification:
@@ -339,7 +408,7 @@ def set_hosts(
         else:
             conn_params["verify_certs"] = True
     if username and password:
-        conn_params["http_auth"] = (username, password)
+        conn_params["basic_auth"] = (username, password)
     if api_key:
         conn_params["api_key"] = api_key
     connections.create_connection(**conn_params)
@@ -385,43 +454,21 @@ def migrate_indexes(
     """
     Updates index mappings
 
+    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.
+
     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; no migrations are
-            currently needed for failure indexes)
+            (accepted for API compatibility; unused)
     """
-    version = 2
-    if aggregate_indexes is None:
-        aggregate_indexes = []
-    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
-
-        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)  # pyright: ignore[reportArgumentType]
-            Index(aggregate_index_name).delete()
 
 
 def save_aggregate_report_to_elasticsearch(
@@ -475,7 +522,10 @@ def save_aggregate_report_to_elasticsearch(
     search = Search(index=search_index)
     query = org_name_query & report_id_query & domain_query
     query = query & begin_date_query & end_date_query
-    search.query = query
+    # elasticsearch.dsl's own docs recommend ``search.query = Q(...)``, but
+    # the ProxyDescriptor.__set__ stub is typed to accept only
+    # Dict[str, Any], not the Query object the docs tell you to assign.
+    search.query = query  # pyright: ignore[reportAttributeAccessIssue]
     begin_date_human = begin_date.strftime("%Y-%m-%d %H:%M:%SZ")
     end_date_human = end_date.strftime("%Y-%m-%d %H:%M:%SZ")
 
@@ -701,7 +751,7 @@ def save_failure_report_to_elasticsearch(
         subject_query = {"match_phrase": {"sample.headers.subject": subject}}
         q = q & Q(subject_query)  # pyright: ignore[reportArgumentType]
 
-    search.query = q
+    search.query = q  # pyright: ignore[reportAttributeAccessIssue]
     existing = search.execute()
 
     if len(existing) > 0:
@@ -842,7 +892,7 @@ def save_smtp_tls_report_to_elasticsearch(
     search = Search(index=search_index)
     query = org_name_query & report_id_query
     query = query & begin_date_query & end_date_query
-    search.query = query
+    search.query = query  # pyright: ignore[reportAttributeAccessIssue]
 
     try:
         existing = search.execute()
@@ -932,7 +982,7 @@ def save_smtp_tls_report_to_elasticsearch(
                     additional_information_uri=additional_information_uri,
                     failure_reason_code=failure_reason_code,
                 )
-        smtp_tls_doc.policies.append(policy_doc)  # pyright: ignore[reportCallIssue]
+        smtp_tls_doc.policies.append(policy_doc)
 
     create_indexes([index], index_settings)
     smtp_tls_doc.meta.index = index  # pyright: ignore[reportOptionalMemberAccess, reportAttributeAccessIssue]

pyproject.toml

@@ -42,8 +42,7 @@ dependencies = [
     "boto3>=1.16.63",
     "dateparser>=1.1.1",
     "dnspython>=2.0.0",
-    "elasticsearch-dsl==7.4.0",
-    "elasticsearch<7.14.0",
+    "elasticsearch>=8.18,<9",
     "expiringdict>=1.1.4",
     "kafka-python>=2.3.2",
     "lxml>=4.4.0",

tests/test_elastic.py

@@ -1,6 +1,6 @@
 """Tests for parsedmarc.elastic
 
-Mocks at the elasticsearch-dsl SDK boundary (connections.create_connection,
+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.
@@ -16,7 +16,6 @@ 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,
@@ -216,11 +215,17 @@ def _populated_search():
 
 
 class TestSetHosts(unittest.TestCase):
-    """Verify the conn_params dict handed to elasticsearch-dsl
+    """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)."""
+    real-world deployment shape (TLS, basic auth, API key, custom CA).
 
-    def test_single_host_string_normalized_to_list(self):
+    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
@@ -228,27 +233,45 @@ class TestSetHosts(unittest.TestCase):
 
     def test_host_list_preserved(self):
         with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
-            set_hosts(["es1:9200", "es2:9200"])
+            set_hosts(["http://es1:9200", "http://es2:9200"])
         kwargs = mock_conn.call_args.kwargs
-        self.assertEqual(kwargs["hosts"], ["es1:9200", "es2:9200"])
+        self.assertEqual(kwargs["hosts"], ["http://es1:9200", "http://es2:9200"])
 
-    def test_timeout_default_60s(self):
+    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")
-        self.assertEqual(mock_conn.call_args.kwargs["timeout"], 60.0)
+        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["timeout"], 30.0)
-
-    def test_use_ssl_enables_verify_by_default(self):
-        with patch("parsedmarc.elastic.connections.create_connection") as mock_conn:
-            set_hosts("es:9200", use_ssl=True)
-        kwargs = mock_conn.call_args.kwargs
-        self.assertEqual(kwargs["use_ssl"], True)
-        self.assertEqual(kwargs["verify_certs"], True)
-        self.assertNotIn("ca_certs", kwargs)
+        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:
@@ -261,16 +284,18 @@ class TestSetHosts(unittest.TestCase):
             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_http_auth(self):
+    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")
-        self.assertEqual(mock_conn.call_args.kwargs["http_auth"], ("u", "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("http_auth", mock_conn.call_args.kwargs)
+        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:
@@ -370,84 +395,6 @@ class TestCreateIndexesServerless(unittest.TestCase):
         mock_index.create.assert_called_once()
 
 
-# ---------------------------------------------------------------------------
-# migrate_indexes
-# ---------------------------------------------------------------------------
-
-
-class TestMigrateIndexes(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."""
-
-    def test_no_indexes_is_noop(self):
-        migrate_indexes()  # Should not raise
-
-    def test_skips_non_existent_index(self):
-        with patch("parsedmarc.elastic.Index") as mock_index_cls:
-            mock_index_cls.return_value.exists.return_value = False
-            migrate_indexes(aggregate_indexes=["missing"])
-        # exists() returned False — no field_mapping fetch.
-        mock_index_cls.return_value.get_field_mapping.assert_not_called()
-
-    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.elastic.Index") as mock_index_cls:
-            idx = mock_index_cls.return_value
-            idx.exists.return_value = True
-            idx.get_field_mapping.return_value = {"some_key": {"mappings": {}}}
-            with patch("parsedmarc.elastic.reindex") as mock_reindex:
-                migrate_indexes(aggregate_indexes=["dmarc_aggregate-2023-01-01"])
-        mock_reindex.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 mapping, data is
-        reindexed, and the old index is deleted."""
-        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,
-        ):
-            idx = mock_index_cls.return_value
-            idx.exists.return_value = True
-            idx.get_field_mapping.return_value = {
-                "dmarc_aggregate-2023-01-01": {
-                    "mappings": {
-                        "doc": {
-                            "published_policy.fo": {"mapping": {"fo": {"type": "long"}}}
-                        }
-                    }
-                }
-            }
-            migrate_indexes(aggregate_indexes=["dmarc_aggregate-2023-01-01"])
-        # reindex called from old → new (v2) index.
-        mock_reindex.assert_called_once()
-        # connections.get_connection consulted to get the ES client.
-        mock_get_conn.assert_called_once()
-
-    def test_skips_when_fo_already_text(self):
-        with (
-            patch("parsedmarc.elastic.Index") as mock_index_cls,
-            patch("parsedmarc.elastic.reindex") as mock_reindex,
-        ):
-            idx = mock_index_cls.return_value
-            idx.exists.return_value = True
-            idx.get_field_mapping.return_value = {
-                "dmarc_aggregate-2024-01-01": {
-                    "mappings": {
-                        "doc": {
-                            "published_policy.fo": {"mapping": {"fo": {"type": "text"}}}
-                        }
-                    }
-                }
-            }
-            migrate_indexes(aggregate_indexes=["dmarc_aggregate-2024-01-01"])
-        mock_reindex.assert_not_called()
-
-
 # ---------------------------------------------------------------------------
 # save_aggregate_report_to_elasticsearch
 # ---------------------------------------------------------------------------
@@ -606,7 +553,7 @@ class TestSaveAggregateReport(unittest.TestCase):
 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
+    alignment. The SDK parent (elasticsearch.dsl.Document.save) is mocked so
     no cluster is needed."""
 
     def test_passed_dmarc_derived_from_alignment(self):