diff --git a/AGENTS.md b/AGENTS.md index 8e63f117..279841ae 100644 --- a/AGENTS.md +++ b/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. diff --git a/CHANGELOG.md b/CHANGELOG.md index 955d8917..5effba5e 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/docs/source/elasticsearch.md b/docs/source/elasticsearch.md index c4375279..ffdb972e 100644 --- a/docs/source/elasticsearch.md +++ b/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 diff --git a/parsedmarc/constants.py b/parsedmarc/constants.py index 37f667a8..9d84ca20 100644 --- a/parsedmarc/constants.py +++ b/parsedmarc/constants.py @@ -1,4 +1,4 @@ -__version__ = "10.2.2" +__version__ = "10.2.3" USER_AGENT = f"parsedmarc/{__version__}" diff --git a/parsedmarc/elastic.py b/parsedmarc/elastic.py index 3e4b6b1a..4d241eac 100644 --- a/parsedmarc/elastic.py +++ b/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] diff --git a/pyproject.toml b/pyproject.toml index 2c6e30cb..ab3f8951 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/test_elastic.py b/tests/test_elastic.py index eaa9fee8..be1c551a 100644 --- a/tests/test_elastic.py +++ b/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):