Warn on unmatched CLI paths; send Kafka failure/SMTP TLS reports per-report (#890)

Fixes the two remaining bugs and the nits uncovered during the review
session:

- The CLI logs a warning for each file_path argument that matches no
  files, instead of silently succeeding with empty results — a typo'd
  path in a cron job previously 'worked' forever while processing
  nothing (_expand_file_path_args dropped it as a zero-match glob).
  Deliberately a warning, not an error: a glob legitimately matching
  nothing must not break existing workflows, and mailbox-only runs
  stay silent.

- save_failure_reports_to_kafka and save_smtp_tls_reports_to_kafka now
  send one message per report, mirroring the aggregate saver's
  per-slice sends. Every released version documented per-record sends,
  but the code sent the whole list as one message, which a large batch
  — failure reports carry message samples — could push past Kafka's
  default 1MB message limit. Consumer-visible: these topics now carry
  individual report objects, not one JSON array per batch.

- The CLI module docstring and argparse description now mention SMTP
  TLS reports; usage.md's CLI-help mirror regenerated.

- The elastic/opensearch to_header display-name tests now assert the
  joined 'RT <rcpt@example.com>' string reaches the saved document
  (autospec save), plus the exactly-once call, instead of only that
  save ran.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

CHANGELOG.md

@@ -6,10 +6,12 @@
 
 - **Breaking: the output and mailbox integrations are now optional extras, so `pip install parsedmarc` installs the parsing core and a working core CLI instead of every SDK** ([#883](https://github.com/domainaware/parsedmarc/issues/883)). A 10.x install pulled in the Elasticsearch, OpenSearch, Kafka, AWS (boto3), Azure, Gmail, and Microsoft Graph client libraries whether or not a deployment used any of them — roughly 1 GB of site-packages against roughly 300 MB without — which is a lot to ask of the motivating case, running parsedmarc as a parsing library on a small mail host. The base install now covers the parsing library plus the CLI reading from files, IMAP, Maildir, and mbox, and writing CSV/JSON, Splunk HEC, webhook, and syslog output; those outputs need only `httpx` and the standard library, so they deliberately have no extra of their own. Everything else moves behind an extra: `elastic` (Elasticsearch), `opensearch` (OpenSearch, including the boto3 SigV4 signer), `kafka`, `s3`, `gelf`, `loganalytics` (Azure Monitor), `msgraph` (Microsoft 365 mailboxes), and `gmail` (Gmail API mailboxes), joining the `postgresql` extra that already existed. **CLI users upgrading from 10.x should switch their upgrade command to `pip install -U "parsedmarc[all]"`** — the new umbrella extra — to keep every integration available; add `postgresql` to the list (`parsedmarc[all,postgresql]`) if the PostgreSQL backend is in use. `all` deliberately excludes `postgresql` because `psycopg`'s prebuilt binary wheels do not exist for every platform, and `pip install parsedmarc[all]` must not fail on a platform they do not cover. Users of the prebuilt Docker image (`ghcr.io/domainaware/parsedmarc`) see no change at all: the image now installs `[all,postgresql]`, so it still bundles every integration. A configuration section whose extra is missing no longer fails with an `ImportError` traceback at startup; it fails fast with a `ConfigurationError` naming both the section and the command that fixes it, e.g. `The [elasticsearch] configuration section requires the elastic extra: pip install parsedmarc[elastic]`. Finally, the never-imported `dateparser` dependency is dropped in favor of declaring `python-dateutil`, which `parsedmarc.utils` actually imports and which used to arrive only transitively through `dateparser`.
 - The CLI now accepts `--dns-timeout` as an alias of `--dns_timeout`, which is kept for backward compatibility (public since 6.0.0); `--dns-retries` already used the hyphenated form.
+- **Failure and SMTP TLS reports are now sent to Kafka as one message per report, matching the aggregate saver's long-documented per-record behavior.** `save_failure_reports_to_kafka` and `save_smtp_tls_reports_to_kafka` documented per-record sends in every released version, but the code actually sent the entire report list as a single Kafka message (an unreleased docstring pass in [#888](https://github.com/domainaware/parsedmarc/pull/888) had briefly aligned the wording to the buggy code); a large batch could exceed Kafka's default 1MB message limit, and failure reports in particular carry message samples that make that more likely. Both savers now send/flush one message per report, mirroring `save_aggregate_reports_to_kafka`'s existing per-slice shape. This is consumer-visible: consumers now receive individual report objects on these topics rather than one JSON array per batch.
 
 ### Bug fixes
 
 - **`[elasticsearch]`/`[opensearch]` `number_of_replicas` is no longer ignored when `number_of_shards` is not also set** — the parser only read `number_of_replicas` inside the `number_of_shards` branch (accidental nesting dating to the 6.4.0-era code), while the documentation lists the two options independently and the client code accepts them independently.
+- **The CLI now logs a warning for each file path argument that matches no files**, instead of silently succeeding with empty results. `_expand_file_path_args` glob-expands each `file_path` argument, so a non-existent plain path (e.g. a typo in a cron job) previously vanished as a zero-match glob with no message of any kind; a mailbox-only run that passes no file arguments still logs nothing.
 
 ## 10.5.0
 

docs/source/usage.md

@@ -11,7 +11,7 @@ usage: parsedmarc [-h] [-c CONFIG_FILE] [-r] [--strip-attachment-payloads] [-o O
                   [-w] [--verbose] [--debug] [--log-file LOG_FILE] [--no-prettify-json] [-v]
                   [file_path ...]
 
-Parses DMARC reports
+Parses DMARC and SMTP TLS reports
 
 positional arguments:
   file_path             one or more paths to aggregate, failure, or SMTP TLS report files, emails, mbox files, or

parsedmarc/cli.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-"""A CLI for parsing DMARC reports"""
+"""A CLI for parsing DMARC and SMTP TLS reports"""
 
 import atexit
 import functools
@@ -286,9 +286,17 @@ def _expand_file_path_args(paths: list[str], recursive: bool = False) -> list[st
     ``recursive`` also enables ``**`` to match any number of directories
     (including none) in glob patterns supplied directly as arguments, per
     the same stdlib glob semantics.
+
+    A ``path`` argument that matches nothing (a plain path that does
+    not exist and is not a glob match, or a directory/glob pattern
+    with no matches) is logged as a WARNING naming that argument,
+    so a typo'd path in a cron job doesn't silently "succeed" while
+    processing nothing forever. An empty ``paths`` list (e.g. a
+    mailbox-only run with no file arguments) logs nothing.
     """
     expanded: list[str] = []
     for path in paths:
+        before = len(expanded)
         if os.path.isdir(path):
             pattern = os.path.join(glob_escape(path), "**" if recursive else "*")
             for match in sorted(glob(pattern, recursive=recursive)):
@@ -303,6 +311,8 @@ def _expand_file_path_args(paths: list[str], recursive: bool = False) -> list[st
             expanded.append(path)
         else:
             expanded += glob(path, recursive=recursive)
+        if len(expanded) == before:
+            logger.warning("No files matched %s", path)
     return expanded
 
 
@@ -2449,7 +2459,7 @@ def _main():
 
         return output_errors
 
-    arg_parser = ArgumentParser(description="Parses DMARC reports")
+    arg_parser = ArgumentParser(description="Parses DMARC and SMTP TLS reports")
     arg_parser.add_argument(
         "-c",
         "--config-file",

parsedmarc/kafkaclient.py

@@ -160,7 +160,15 @@ class KafkaClient(object):
         failure_topic: str,
     ):
         """
-        Saves failure DMARC reports to Kafka as a single message
+        Saves failure DMARC reports to Kafka, one message per report, so a
+        large batch is far less likely to exceed Kafka's default 1MB
+        message limit
+
+        Mirrors ``save_aggregate_reports_to_kafka``'s per-slice send/flush
+        shape (minus the aggregate-only ``strip_metadata``/date-range
+        logic): failure reports carry message samples, so a batch sent as
+        a single JSON array could easily exceed the broker's default max
+        message size.
 
         Args:
             failure_reports (list):  A list of failure report dicts
@@ -174,17 +182,18 @@ class KafkaClient(object):
         if len(failure_reports) < 1:
             return
 
-        try:
-            logger.debug("Saving failure reports to Kafka")
-            self.producer.send(failure_topic, failure_reports)
-        except UnknownTopicOrPartitionError:
-            raise KafkaError("Kafka error: Unknown topic or partition on broker")
-        except Exception as e:
-            raise KafkaError(f"Kafka error: {e.__str__()}")
-        try:
-            self.producer.flush()
-        except Exception as e:
-            raise KafkaError(f"Kafka error: {e.__str__()}")
+        for report in failure_reports:
+            try:
+                logger.debug("Saving failure report to Kafka")
+                self.producer.send(failure_topic, report)
+            except UnknownTopicOrPartitionError:
+                raise KafkaError("Kafka error: Unknown topic or partition on broker")
+            except Exception as e:
+                raise KafkaError(f"Kafka error: {e.__str__()}")
+            try:
+                self.producer.flush()
+            except Exception as e:
+                raise KafkaError(f"Kafka error: {e.__str__()}")
 
     # Backward-compatible alias
     save_forensic_reports_to_kafka = save_failure_reports_to_kafka
@@ -195,7 +204,13 @@ class KafkaClient(object):
         smtp_tls_topic: str,
     ):
         """
-        Saves SMTP TLS reports to Kafka as a single message
+        Saves SMTP TLS reports to Kafka, one message per report, so a
+        large batch is far less likely to exceed Kafka's default 1MB
+        message limit
+
+        Mirrors ``save_aggregate_reports_to_kafka``'s per-slice send/flush
+        shape (minus the aggregate-only ``strip_metadata``/date-range
+        logic).
 
         Args:
             smtp_tls_reports (list):  A list of SMTP TLS report dicts
@@ -209,14 +224,15 @@ class KafkaClient(object):
         if len(smtp_tls_reports) < 1:
             return
 
-        try:
-            logger.debug("Saving SMTP TLS reports to Kafka")
-            self.producer.send(smtp_tls_topic, smtp_tls_reports)
-        except UnknownTopicOrPartitionError:
-            raise KafkaError("Kafka error: Unknown topic or partition on broker")
-        except Exception as e:
-            raise KafkaError(f"Kafka error: {e.__str__()}")
-        try:
-            self.producer.flush()
-        except Exception as e:
-            raise KafkaError(f"Kafka error: {e.__str__()}")
+        for report in smtp_tls_reports:
+            try:
+                logger.debug("Saving SMTP TLS report to Kafka")
+                self.producer.send(smtp_tls_topic, report)
+            except UnknownTopicOrPartitionError:
+                raise KafkaError("Kafka error: Unknown topic or partition on broker")
+            except Exception as e:
+                raise KafkaError(f"Kafka error: {e.__str__()}")
+            try:
+                self.producer.flush()
+            except Exception as e:
+                raise KafkaError(f"Kafka error: {e.__str__()}")

tests/test_cli.py

@@ -570,6 +570,78 @@ hosts = localhost
                 [nested_xml],
             )
 
+    def test_expand_file_path_args_warns_on_nonexistent_path(self):
+        """A file_path argument that matches no files at all -- a plain
+        path that doesn't exist and isn't a glob match -- logs a WARNING
+        naming that argument, instead of silently vanishing as a
+        zero-match glob. The argument still contributes no paths to the
+        returned list.
+
+        Regression test: ``parsedmarc --offline -o out /nonexistent.xml``
+        used to exit 0 with empty results and no message of any kind, so
+        a typo'd path in a cron job "worked" forever while processing
+        nothing.
+        """
+        from parsedmarc.cli import _expand_file_path_args
+
+        with tempfile.TemporaryDirectory() as d:
+            missing = os.path.join(d, "nonexistent.xml")
+            with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
+                result = _expand_file_path_args([missing])
+            self.assertEqual(result, [])
+            self.assertTrue(
+                any(missing in line for line in cm.output),
+                f"expected a warning naming {missing}, got: {cm.output}",
+            )
+
+    def test_expand_file_path_args_no_warning_for_matching_path(self):
+        """A file_path argument that matches at least one file logs no
+        warning at all."""
+        from parsedmarc.cli import _expand_file_path_args
+
+        with tempfile.TemporaryDirectory() as d:
+            report = os.path.join(d, "report.xml")
+            with open(report, "w") as f:
+                f.write("x")
+
+            with self.assertNoLogs("parsedmarc.log", level="WARNING"):
+                result = _expand_file_path_args([report])
+            self.assertEqual(result, [report])
+
+            wildcard = os.path.join(d, "*.xml")
+            with self.assertNoLogs("parsedmarc.log", level="WARNING"):
+                result = _expand_file_path_args([wildcard])
+            self.assertEqual(result, [report])
+
+    def test_expand_file_path_args_empty_list_is_silent(self):
+        """An empty ``paths`` list (e.g. a mailbox-only run with no file
+        arguments) logs nothing."""
+        from parsedmarc.cli import _expand_file_path_args
+
+        with self.assertNoLogs("parsedmarc.log", level="WARNING"):
+            result = _expand_file_path_args([])
+        self.assertEqual(result, [])
+
+    def test_expand_file_path_args_mixed_warns_only_for_unmatched(self):
+        """When some arguments match and others don't, only the
+        unmatched argument produces a warning."""
+        from parsedmarc.cli import _expand_file_path_args
+
+        with tempfile.TemporaryDirectory() as d:
+            report = os.path.join(d, "report.xml")
+            with open(report, "w") as f:
+                f.write("x")
+            missing = os.path.join(d, "nonexistent.xml")
+
+            with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
+                result = _expand_file_path_args([report, missing])
+
+            self.assertEqual(result, [report])
+            warning_lines = [line for line in cm.output if "WARNING" in line]
+            self.assertEqual(len(warning_lines), 1)
+            self.assertIn(missing, warning_lines[0])
+            self.assertNotIn(report, warning_lines[0])
+
     def test_apply_env_overrides_injects_values(self):
         """Env vars are injected into an existing ConfigParser."""
         from configparser import ConfigParser

tests/test_elastic.py

@@ -1200,16 +1200,23 @@ class TestSaveFailureReport(unittest.TestCase):
 
     def test_to_header_with_non_empty_display_joins_with_brackets(self):
         """The other branch: non-empty display joins display+addr
-        with " <" and appends ">", e.g. 'RT <rcpt@example.com>'."""
+        with " <" and appends ">", e.g. 'RT <rcpt@example.com>'.
+        Asserts the joined string actually reaches the saved document
+        (see test_reply_to_header_flattened_and_indexed for the
+        autospec-save pattern), not merely that save ran."""
         report = _failure_report()
         report["parsed_sample"]["headers"]["To"] = [["RT", "rcpt@example.com"]]
         with (
             patch("parsedmarc.elastic.Search", return_value=_empty_search()),
             patch("parsedmarc.elastic.Index"),
-            patch.object(elastic_module._FailureReportDoc, "save") as mock_save,
+            patch.object(
+                elastic_module._FailureReportDoc, "save", autospec=True
+            ) as mock_save,
         ):
             save_failure_report_to_elasticsearch(report)
         mock_save.assert_called_once()
+        doc = mock_save.call_args.args[0]
+        self.assertEqual(doc.sample.headers["to"], "RT <rcpt@example.com>")
 
     def test_sample_address_lists_indexed_for_reply_to_cc_bcc_attachments(self):
         """A failure report sample can carry reply_to / cc / bcc /

tests/test_kafkaclient.py

@@ -186,22 +186,33 @@ class TestSaveFailureReportsToKafka(unittest.TestCase):
         with patch("parsedmarc.kafkaclient.KafkaProducer"):
             return KafkaClient(kafka_hosts=["b:9092"])
 
-    def test_sends_full_list_in_one_message(self):
-        """Failure reports are sent as one Kafka message carrying the
-        whole list — unlike aggregate records, which are sent as
-        individual slices to stay under Kafka's default 1MB message
-        cap."""
+    def test_sends_one_message_per_report(self):
+        """Failure reports are sent as one Kafka message per report,
+        mirroring the aggregate saver's per-slice sends, so a large batch
+        is far less likely to exceed Kafka's default 1MB message limit
+        — failure
+        reports carry message samples, so a whole-list send is
+        particularly likely to blow past that cap."""
         client = self._client()
         reports = [{"id": "f1"}, {"id": "f2"}]
         client.save_failure_reports_to_kafka(reports, "dmarc-failure")
-        _producer(client).send.assert_called_once_with("dmarc-failure", reports)
+        producer = _producer(client)
+        self.assertEqual(producer.send.call_count, 2)
+        sent = [call.args for call in producer.send.call_args_list]
+        self.assertEqual(
+            sent, [("dmarc-failure", reports[0]), ("dmarc-failure", reports[1])]
+        )
+        self.assertEqual(producer.flush.call_count, 2)
 
     def test_dict_input_normalized_to_list(self):
+        """A single-report dict is wrapped to a one-element list
+        internally, then sent as that one unwrapped report (not the
+        list itself)."""
         client = self._client()
         client.save_failure_reports_to_kafka({"id": "single"}, "topic")
-        # The send payload is wrapped to a single-element list.
+        self.assertEqual(_producer(client).send.call_count, 1)
         args = _producer(client).send.call_args.args
-        self.assertEqual(args[1], [{"id": "single"}])
+        self.assertEqual(args[1], {"id": "single"})
 
     def test_empty_list_is_a_noop(self):
         client = self._client()
@@ -232,17 +243,29 @@ class TestSaveSmtpTlsReportsToKafka(unittest.TestCase):
         with patch("parsedmarc.kafkaclient.KafkaProducer"):
             return KafkaClient(kafka_hosts=["b:9092"])
 
-    def test_sends_full_list_in_one_message(self):
+    def test_sends_one_message_per_report(self):
+        """SMTP TLS reports are sent as one Kafka message per report,
+        mirroring the aggregate saver's per-slice sends, so a large batch
+        is far less likely to exceed Kafka's default 1MB message
+        limit."""
         client = self._client()
-        reports = [{"organization_name": "x"}]
+        reports = [{"organization_name": "x"}, {"organization_name": "y"}]
         client.save_smtp_tls_reports_to_kafka(reports, "smtp-tls")
-        _producer(client).send.assert_called_once_with("smtp-tls", reports)
+        producer = _producer(client)
+        self.assertEqual(producer.send.call_count, 2)
+        sent = [call.args for call in producer.send.call_args_list]
+        self.assertEqual(sent, [("smtp-tls", reports[0]), ("smtp-tls", reports[1])])
+        self.assertEqual(producer.flush.call_count, 2)
 
     def test_dict_input_normalized_to_list(self):
+        """A single-report dict is wrapped to a one-element list
+        internally, then sent as that one unwrapped report (not the
+        list itself)."""
         client = self._client()
         client.save_smtp_tls_reports_to_kafka({"organization_name": "x"}, "topic")
+        self.assertEqual(_producer(client).send.call_count, 1)
         args = _producer(client).send.call_args.args
-        self.assertEqual(args[1], [{"organization_name": "x"}])
+        self.assertEqual(args[1], {"organization_name": "x"})
 
     def test_empty_list_is_a_noop(self):
         client = self._client()

tests/test_opensearch.py

@@ -1193,16 +1193,23 @@ class TestSaveFailureReport(unittest.TestCase):
 
     def test_to_header_with_non_empty_display_joins_with_brackets(self):
         """The other branch: non-empty display joins display+addr
-        with " <" and appends ">", e.g. 'RT <rcpt@example.com>'."""
+        with " <" and appends ">", e.g. 'RT <rcpt@example.com>'.
+        Asserts the joined string actually reaches the saved document
+        (see test_reply_to_header_flattened_and_indexed for the
+        autospec-save pattern), not merely that save ran."""
         report = _failure_report()
         report["parsed_sample"]["headers"]["To"] = [["RT", "rcpt@example.com"]]
         with (
             patch("parsedmarc.opensearch.Search", return_value=_empty_search()),
             patch("parsedmarc.opensearch.Index"),
-            patch.object(opensearch_module._FailureReportDoc, "save") as mock_save,
+            patch.object(
+                opensearch_module._FailureReportDoc, "save", autospec=True
+            ) as mock_save,
         ):
             save_failure_report_to_opensearch(report)
         mock_save.assert_called_once()
+        doc = mock_save.call_args.args[0]
+        self.assertEqual(doc.sample.headers["to"], "RT <rcpt@example.com>")
 
     def test_sample_address_lists_indexed_for_reply_to_cc_bcc_attachments(self):
         """A failure report sample can carry reply_to / cc / bcc /