amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-07-31 12:45:58 +00:00
Code Issues Projects Releases Wiki Activity

Add [general] archive_directory to archive processed local files (#570) (#856)

Browse Source 
* Ignore Claude Code agent worktrees under .claude/worktrees/

Untracked repo snapshots from agent sessions were making repo-root
ruff check . fail on stale code and cluttering git status. ruff
respects .gitignore, so ignoring the directory fixes both.

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

* Add [general] archive_directory to archive processed local files (#570)

Move successfully processed report files given as local path arguments
into <archive_directory>/<year>/<month>/<Aggregate|Failure|SMTP-TLS>/,
dated from the parsed report's own metadata (aggregate begin_date,
failure arrival_date_utc, SMTP TLS begin_date) rather than the
filename. Files that fail to parse as a report (ParserError) go to
<archive_directory>/Invalid/; other failures (e.g. transient I/O
errors) leave the file in place so a later run can retry it.

Existing destination files are never overwritten: each candidate name
is claimed atomically (O_CREAT | O_EXCL) and collisions get a numeric
suffix before the extension. Files already inside the archive
directory are excluded from processing (compared via realpath so
symlinked spellings still match), so the archive can safely live
inside an input directory, as the issue requests. mbox files and
mailbox modes are unaffected; mailbox modes keep [mailbox]
archive_folder.

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

* Address review feedback on docstrings and the placeholder-cleanup except

- _move_file_to_archive's docstring no longer claims the copy2 fallback
  replaces the placeholder atomically; only the same-filesystem os.rename
  path is atomic. The fallback is a plain copy-and-overwrite, which is
  still collision-safe because the placeholder already claimed the name.
- The empty except OSError in the placeholder cleanup now carries a
  comment explaining that it is deliberate best-effort cleanup and the
  re-raised move failure is the actionable error.
- test_general_archive_directory_unset_leaves_attribute_absent's
  docstring now describes what the assertion actually tests (the
  attribute staying absent from a bare Namespace) and moves the
  real-CLI None-default behavior to a parenthetical.

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

* Cover the two defensive exception branches Codecov flagged

Codecov's patch report flagged four uncovered lines, all in the two
platform-dependent exception branches of the archive helpers:

- _exclude_archived_paths's except ValueError branch. Per the Python
  docs for os.path.commonpath, ValueError is raised when paths "are on
  the different drives" (Windows) or mix absolute and relative
  pathnames; both inputs are realpath()-resolved so only the
  different-drives case remains, which Linux CI can't produce
  naturally. The new test simulates the raise and asserts the
  non-comparable path is kept for parsing rather than excluded.

- _move_file_to_archive's except OSError placeholder-cleanup branch.
  The new test fails the move with a non-OSError type and the cleanup
  with OSError, then asserts the move error is what propagates (the
  cleanup error is swallowed, not allowed to mask it) and that the
  zero-byte placeholder survives its failed cleanup.

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

* Tag arrival_date_utc as UTC when parsing the archive date

human_timestamp_to_datetime's docstring names arrival_date_utc as
exactly the kind of known-UTC naive string that should be parsed with
assume_utc=True; _archive_subdir_for_result was parsing it naive.

The flag is scoped to the failure branch because the shared call also
handles the other two report types: aggregate begin_date is a
local-time wall-clock string (timestamp_to_human uses
datetime.fromtimestamp), so tagging it UTC would be wrong, and SMTP TLS
begin_date carries an RFC 3339 offset, making assume_utc a no-op.

No observable behavior change: only the wall-clock year/month fields
are read and assume_utc never shifts wall-clock time, so no new test
can honestly distinguish the two versions — this aligns the call with
its dependency's documented contract and hardens against a future edit
adding a real timezone conversion.

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

* Create the archive placeholder with mode 0o600

os.open's mode parameter defaults to 0o777 (masked by the umask), so
the O_CREAT|O_EXCL placeholder in _move_file_to_archive was created
executable and group-accessible on typical umasks (0o775 under umask
002). Normally it's replaced immediately, but a placeholder that
outlives a failed move+cleanup persisted with those permissions.

Pass 0o600 explicitly. The mode never reaches the real archived file:
os.rename replaces the placeholder's inode outright, and the copy2
fallback's copystat overwrites the mode with the source file's. The
leftover-placeholder regression test now also asserts the surviving
placeholder has no owner-exec or group/other bits (umask-independent,
since the umask only clears bits); the assertion fails against the
unfixed default-mode call.

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

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Sean Whalen
2026-07-26 11:56:01 -04:00
committed by GitHub
co-authored by Claude Fable 5
parent f3689c3125
commit 264b9a4556
5 changed files with 955 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+4
View File
@@ -152,3 +152,7 @@ parsedmarc/resources/maps/domain_info.tsv
coverage.json
junit.xml
dashboard-screenshots/
# Claude Code agent worktrees — untracked repo snapshots; ignoring them
# keeps repo-root ruff runs and git status clean
.claude/worktrees/
CHANGELOG.md
+1
View File
@@ -7,6 +7,7 @@
- **`n_procs` parallel parsing now covers messages from mbox files and mailbox connections** ([#147](https://github.com/domainaware/parsedmarc/issues/147)), not just report files passed directly as CLI arguments: IMAP, Microsoft Graph, Gmail API, and Maildir connections, including watch mode. `get_dmarc_reports_from_mbox()`, `get_dmarc_reports_from_mailbox()`, and `watch_inbox()` all gained an `n_procs` keyword argument. Parallel workers are now a reused process pool for the whole run, with a bounded submission window that keeps at most roughly `2 * n_procs` messages in flight at a time so memory stays bounded even for huge mboxes; the main process sends periodic IMAP keepalives while workers parse.
- **Added `ParserConfig`, a single frozen dataclass carrying every parsing/enrichment option** ([#503](https://github.com/domainaware/parsedmarc/issues/503)): offline mode, IP database path, reverse DNS map and PSL overrides paths/URLs, DNS nameservers/timeout/retries, `strip_attachment_payloads`, `normalize_timespan_threshold_hours`, and the three caches the parser and enrichment code share across calls (IP address info, seen aggregate report IDs, and the reverse DNS map). `parse_aggregate_report_xml()`, `parse_aggregate_report_file()`, `parse_failure_report()`, `parse_report_email()`, `parse_report_file()`, `get_dmarc_reports_from_mbox()`, `get_dmarc_reports_from_mailbox()`, and `watch_inbox()` all accept a keyword-only `config=` argument; when it's provided, the individual option keyword arguments are ignored in favor of the config's values, and every existing per-option keyword argument continues to work unchanged when `config=` is omitted. Every explicitly constructed `ParserConfig` owns fresh, isolated caches; omitting `config=` falls back to the existing process-wide shared default caches, unchanged and identity-preserved (`parsedmarc.IP_ADDRESS_CACHE`, `parsedmarc.SEEN_AGGREGATE_REPORT_IDS`, `parsedmarc.REVERSE_DNS_MAP`). Caches never cross multiprocessing worker boundaries: `ParserConfig.__getstate__` drops the three cache fields when a config is pickled for a worker, and each worker accumulates its own from that point on. `keep_alive` and `n_procs` remain separate keyword arguments — they control process/worker orchestration, not parsing or enrichment behavior, so they're deliberately not `ParserConfig` fields. See the new "Using parsedmarc as a library" section of the usage docs for an example.
- **Added a "Domain policy" dropdown filter to the Splunk aggregate DMARC dashboard** ([#854](https://github.com/domainaware/parsedmarc/issues/854)): filters every panel on the published DMARC policy (`published_policy.p`) or subdomain policy (`published_policy.sp`), making it easy to see which domains with a `none` policy are ready to move to `quarantine` or `reject`. It sits immediately before the "Message disposition" dropdown and offers the same choices (`any`/`none`/`quarantine`/`reject`, defaulting to `any`).
- **New `[general]` option `archive_directory`**: move successfully processed local report files into a dated archive tree (`<archive_directory>/<year>/<month>/<Aggregate|Failure|SMTP-TLS>/`); files that fail to parse as a report go to `<archive_directory>/Invalid/`, while files that fail for other reasons (e.g. transient I/O errors) are left in place so a later run can retry them; existing files are never overwritten (numeric suffix); files already under the archive directory are excluded from later runs. Applies only to report files passed directly as local path arguments ([#570](https://github.com/domainaware/parsedmarc/issues/570)).
### Bug fixes
docs/source/usage.md
+18
View File
@@ -132,6 +132,24 @@ The full set of configuration options are:
payloads from results
- `silent` - bool: Set this to `False` to output results to STDOUT
- `output` - str: Directory to place JSON and CSV files in. This is required if you set either of the JSON output file options.
- `archive_directory` - str: Optional. When set, successfully
processed report files given as local file/directory path
arguments are moved into
`<archive_directory>/<year>/<month>/<Aggregate|Failure|SMTP-TLS>/`
(year and month come from the report's own begin/arrival date, with
the month zero-padded). A successfully parsed report whose archive
date can't be determined is left in place with a logged warning.
Files that fail to parse as a report are moved to
`<archive_directory>/Invalid/`; files that fail for other reasons,
such as transient I/O errors, are left in place so a later run can
retry them. An existing destination file is never overwritten; a
numeric suffix is appended before the extension (e.g.
`report-1.xml`). This applies only to direct local file input —
reports fetched from mailboxes (IMAP, Microsoft Graph, Gmail API,
Maildir) use `[mailbox] archive_folder` instead, and mbox files are
never moved. Files already inside `archive_directory` are excluded
from processing, so the archive may safely live inside an input
directory. A failed move is logged and does not stop the run.
- `aggregate_json_filename` - str: filename for the aggregate
JSON output file
- `failure_json_filename` - str: filename for the failure
parsedmarc/cli.py
+196 -1
View File
@@ -9,6 +9,7 @@ import http.client
import json
import logging
import os
import shutil
import signal
import sys
import time
@@ -58,12 +59,13 @@ from parsedmarc.mail import (
MSGraphConnection,
)
from parsedmarc.parallel import _parse_report_file_job, parallel_map
from parsedmarc.types import ParsingResults
from parsedmarc.types import ParsedReport, ParsingResults
from parsedmarc.utils import (
InvalidIPinfoAPIKey,
configure_ipinfo_api,
get_base_domain,
get_reverse_dns,
human_timestamp_to_datetime,
is_mbox,
load_ip_db,
load_psl_overrides,
@@ -225,6 +227,192 @@ def _expand_file_path_args(paths: list[str], recursive: bool = False) -> list[st
return expanded
def _exclude_archived_paths(file_paths: list[str], archive_directory: str) -> list[str]:
"""Filter *file_paths* down to paths that are not already inside
*archive_directory*.
The archive directory may live inside an input directory (e.g.
``<input>/archive``), so without this filter a file already moved
into the archive on a previous run would be picked up again by a
later ``file_path`` directory expansion, re-parsed, and re-archived
(colliding with itself and accumulating numeric suffixes forever).
Paths are resolved with ``os.path.realpath`` (not just
``os.path.abspath``) so a symlinked spelling of either the archive
directory or an input path still matches: e.g. ``archive_directory``
configured via a ``/data`` symlink while the input directory is
passed as the real ``/mnt/...`` path would otherwise never compare
equal, and every run would re-archive the same files with a new
numeric suffix forever.
"""
archive_root = os.path.normcase(os.path.realpath(archive_directory))
kept: list[str] = []
for path in file_paths:
abs_path = os.path.normcase(os.path.realpath(path))
try:
inside_archive = (
os.path.commonpath([archive_root, abs_path]) == archive_root
)
except ValueError:
# Paths are on different drives (Windows) or otherwise not
# comparable, so the path can't be inside the archive.
inside_archive = False
if inside_archive:
logger.debug(f"Excluding already-archived file {path}")
continue
kept.append(path)
return kept
def _archive_subdir_for_result(result: ParsedReport) -> str | None:
"""Return the ``<year>/<month>/<type folder>`` subdirectory a parsed
report's source file should be archived under, or ``None`` when the
report type is unrecognized or its date can't be determined.
The date comes from the parsed report itself, not the source
filename or file mtime: aggregate reports use
``report_metadata.begin_date``, failure reports use
``arrival_date_utc``, and SMTP TLS reports use ``begin_date``.
"""
report_type = result["report_type"]
# Only the wall-clock year/month fields are read from the parsed
# datetime, so no timezone conversion ever happens here — but tag
# the strings whose zone is known, per human_timestamp_to_datetime's
# contract. Aggregate begin_date is a local-time string
# (timestamp_to_human uses datetime.fromtimestamp) and must stay
# naive; arrival_date_utc is UTC wall-clock; SMTP TLS begin_date is
# RFC 3339 with an offset, so assume_utc would be a no-op anyway.
assume_utc = False
try:
if result["report_type"] == "aggregate":
type_folder = "Aggregate"
date_string = result["report"]["report_metadata"]["begin_date"]
elif result["report_type"] == "failure":
type_folder = "Failure"
date_string = result["report"]["arrival_date_utc"]
assume_utc = True
elif result["report_type"] == "smtp_tls":
type_folder = "SMTP-TLS"
date_string = result["report"]["begin_date"]
else:
logger.warning(f"Cannot archive unknown report type: {report_type}")
return None
dt = human_timestamp_to_datetime(date_string, assume_utc=assume_utc)
except (KeyError, TypeError, ValueError, OverflowError) as e:
logger.warning(f"Cannot determine archive date for {report_type} report: {e}")
return None
return os.path.join(f"{dt.year:04d}", f"{dt.month:02d}", type_folder)
def _move_file_to_archive(file_path: str, dest_dir: str) -> str:
"""Move *file_path* into *dest_dir*, creating it if needed, and return
the final destination path.
An existing file at the destination is never overwritten: a numeric
suffix is appended before the extension (``name-1.xml``,
``name-2.xml``, ...) until a free name is found. For multi-suffix
names like ``report.xml.gz`` the numeric suffix lands before the
last suffix only (``report.xml-1.gz``); this is acceptable.
The free-name claim is atomic (``os.open`` with
``O_CREAT | O_EXCL``) rather than an exists-check-then-move: a plain
``os.path.exists()`` check followed by ``shutil.move()`` is a
TOCTOU race between concurrent ``parsedmarc`` invocations sharing an
archive directory, and ``shutil.move()`` silently overwrites an
existing destination on POSIX, which would violate the
never-overwrite guarantee. Instead, each candidate name is staked
out with a zero-byte placeholder file before the real move happens;
``shutil.move()`` then replaces that placeholder with the real file
— atomically via ``os.rename`` when source and destination are on
the same POSIX filesystem, otherwise (Windows, or a cross-device
move) via a ``copy2``-and-overwrite that is not atomic but still
cannot collide with a concurrent invocation, since the placeholder
already claimed the name.
"""
os.makedirs(dest_dir, exist_ok=True)
base, ext = os.path.splitext(os.path.basename(file_path))
candidate = os.path.basename(file_path)
n = 1
while True:
dest_path = os.path.join(dest_dir, candidate)
try:
# 0o600 (not os.open's 0o777 default) so a placeholder that
# outlives a failed move+cleanup is never executable or
# group/other-accessible. The mode never reaches the real
# archived file: os.rename replaces the placeholder's inode
# outright, and the copy2 fallback's copystat overwrites the
# mode with the source file's.
fd = os.open(dest_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
except FileExistsError:
candidate = f"{base}-{n}{ext}"
n += 1
continue
os.close(fd)
break
try:
shutil.move(file_path, dest_path)
except Exception:
try:
os.remove(dest_path)
except OSError:
# Best-effort cleanup of the just-created placeholder; the
# move failure re-raised below is the error that matters.
pass
raise
return dest_path
def _archive_processed_file(
file_path: str, archive_directory: str, result: ParsedReport | Exception
) -> None:
"""Move *file_path* into *archive_directory* after processing.
Files that failed to parse as a report (*result* is a
``ParserError`` — every parse-failure exception, including
``InvalidSMTPTLSReport``, subclasses it) go to
``<archive_directory>/Invalid/``. Files that failed for some other
reason (a transient ``OSError``/``PermissionError`` from the parse
job's broad catch, or an unexpected parser bug) are left in place so
a later run can retry them — renaming a valid-but-currently-unreadable
report into ``Invalid/`` would permanently sideline it, since moving
a file needs no read permission on its contents, and
``_exclude_archived_paths`` would then hide it from every future run
too. The parse loop already logged the error either way.
Successfully parsed files go to the dated ``<year>/<month>/<type>``
subdirectory returned by ``_archive_subdir_for_result``; if that
returns ``None`` (unknown report type or unparseable date), the file
is left in place — a warning was already logged by that helper.
A move failure is logged and never allowed to abort the run: the
file has already been successfully parsed (or definitively failed
to parse), so a filesystem error while archiving it should not
cause the caller to lose that work.
"""
if isinstance(result, ParserError):
subdir = "Invalid"
elif isinstance(result, Exception):
logger.debug(
f"Leaving {file_path} in place: {result.__class__.__name__} is not "
"a report-parsing failure, so it may be retryable"
)
return
else:
subdir = _archive_subdir_for_result(result)
if subdir is None:
return
dest_dir = os.path.join(archive_directory, subdir)
try:
dest_path = _move_file_to_archive(file_path, dest_dir)
except Exception as e:
logger.error(f"Error moving {file_path} to the archive: {e}")
return
logger.debug(f"Archived {file_path} to {dest_path}")
# All known INI config section names, used for env var resolution.
_KNOWN_SECTIONS = frozenset(
{
@@ -483,6 +671,8 @@ def _parse_config(config: ConfigParser, opts):
)
if "output" in general_config:
opts.output = _expand_path(general_config["output"])
if "archive_directory" in general_config:
opts.archive_directory = _expand_path(general_config["archive_directory"])
if "aggregate_json_filename" in general_config:
opts.aggregate_json_filename = general_config["aggregate_json_filename"]
if "failure_json_filename" in general_config:
@@ -2131,6 +2321,7 @@ def _main():
maildir_create=False,
log_file=args.log_file,
n_procs=1,
archive_directory=None,
ip_db_path=None,
ipinfo_url=None,
ipinfo_api_token=None,
@@ -2307,6 +2498,8 @@ def _main():
signal.signal(signal.SIGINT, _handle_sigint)
file_paths = _expand_file_path_args(args.file_path, recursive=args.recursive)
if opts.archive_directory:
file_paths = _exclude_archived_paths(file_paths, opts.archive_directory)
mbox_paths = []
for file_path in file_paths:
@@ -2354,6 +2547,8 @@ def _main():
failure_reports.append(result["report"])
elif result["report_type"] == "smtp_tls":
smtp_tls_reports.append(result["report"])
if opts.archive_directory:
_archive_processed_file(file_path, opts.archive_directory, result)
if pbar is not None:
pbar.close()
tests/test_cli.py
+736 -1
View File
@@ -6,6 +6,7 @@ import json
import logging
import os
import signal
import stat
import sys
import tempfile
import unittest
@@ -26,7 +27,7 @@ import parsedmarc
import parsedmarc.cli
import parsedmarc.elastic
import parsedmarc.opensearch as opensearch_module
from parsedmarc.types import AggregateReport
from parsedmarc.types import AggregateReport, ParsedReport
SAMPLE_AGGREGATE_REPORT_PATH = (
"samples/aggregate/!example.com!1538204542!1538463818.xml"
@@ -1086,6 +1087,321 @@ class TestDirectoryFilePaths(unittest.TestCase):
self.assertEqual(report_ids, {top_level_id, nested_id})
class TestArchiveDirectory(unittest.TestCase):
"""End-to-end coverage of issue #570: ``[general] archive_directory``
moves successfully processed local report files into a dated
``<year>/<month>/<Aggregate|Failure|SMTP-TLS>/`` tree, and unparseable
files into ``<archive_directory>/Invalid/``. Runs the real ``_main()``
entry point via ``patch.object(sys, "argv", ...)`` against on-disk
sample reports, per AGENTS.md's mock-at-SDK-boundary rule (there is no
external SDK boundary in this code path to mock)."""
AGGREGATE_SAMPLE_1 = "samples/aggregate/!example.com!1538204542!1538463818.xml"
AGGREGATE_SAMPLE_2 = (
"samples/aggregate/!large-example.com!1711897200!1711983600.xml"
)
# Not dmarc_ruf_report_linkedin.eml: that sample begins with a
# "From dmarc-noreply@linkedin.com ..." mbox envelope line, which is
# why Python's mailbox module reads it as a (single-message) mbox
# file and is_mbox() classifies it as an mbox — routing it through
# _main()'s mbox_paths branch instead of the direct-file archiving
# path this test exercises (mbox files are intentionally never
# archived — see the archive_directory docs in docs/source/usage.md).
# The sharepoint sample below starts with a MIME header instead, so
# it isn't mbox-classified.
FAILURE_SAMPLE = (
"samples/failure/DMARC Failure Report for domain.de "
"(mail-from=sharepoint@domain.de, ip=10.10.10.10).eml"
)
SMTP_TLS_SAMPLE = "samples/smtp_tls/rfc8460.json"
def setUp(self):
# SEEN_AGGREGATE_REPORT_IDS is a module-level ExpiringDict that
# dedupes report IDs across parses within one process; clear it so
# a report "seen" by an earlier test isn't silently dropped here,
# and so tests within this class don't interfere with each other.
# Precedent: TestDirectoryFilePaths.setUp above.
parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear()
self.addCleanup(parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear)
self._env_patcher = patch.dict(
os.environ, {"GITHUB_ACTIONS": "true"}, clear=False
)
self._env_patcher.start()
self.addCleanup(self._env_patcher.stop)
def _copy_sample(self, src_path, dest_dir, dest_basename=None):
dest_basename = dest_basename or os.path.basename(src_path)
dest_path = os.path.join(dest_dir, dest_basename)
with open(src_path, "rb") as src, open(dest_path, "wb") as dst:
dst.write(src.read())
return dest_path
def _expected_subdir(self, sample_path):
"""Parse *sample_path* the same way the CLI does and return the
``<year>/<month>/<type>`` subdirectory ``_archive_subdir_for_result``
computes for it. Computed dynamically rather than hardcoded because
aggregate ``begin_date`` is local-time, so month buckets are
timezone-dependent."""
result = parsedmarc.parse_report_file(sample_path, offline=True)
subdir = parsedmarc.cli._archive_subdir_for_result(result)
assert subdir is not None
return subdir
def _write_config(
self, tmp_dir, output_dirname, archive_dirname=None, n_procs=None
):
cfg_path = os.path.join(tmp_dir, "parsedmarc.ini")
output_dir = os.path.join(tmp_dir, output_dirname)
config_text = (
f"[general]\noffline = True\nsilent = True\noutput = {output_dir}\n"
)
archive_dir = None
if archive_dirname is not None:
archive_dir = os.path.join(tmp_dir, archive_dirname)
config_text += f"archive_directory = {archive_dir}\n"
if n_procs is not None:
config_text += f"n_procs = {n_procs}\n"
with open(cfg_path, "w") as f:
f.write(config_text)
return cfg_path, output_dir, archive_dir
def test_archive_moves_all_three_types(self):
"""Aggregate, failure, and SMTP TLS report files given as direct
``file_path`` arguments are moved into
``<archive>/<year>/<month>/<Aggregate|Failure|SMTP-TLS>/`` after a
successful parse, the source files are gone from the input
directory, and the aggregate JSON output is still produced."""
with tempfile.TemporaryDirectory() as tmp_dir:
input_dir = os.path.join(tmp_dir, "input")
os.makedirs(input_dir)
agg1 = self._copy_sample(self.AGGREGATE_SAMPLE_1, input_dir)
agg2 = self._copy_sample(self.AGGREGATE_SAMPLE_2, input_dir)
failure = self._copy_sample(self.FAILURE_SAMPLE, input_dir)
smtp_tls = self._copy_sample(self.SMTP_TLS_SAMPLE, input_dir)
cfg_path, output_dir, archive_dir = self._write_config(
tmp_dir, "output", archive_dirname="archive"
)
assert archive_dir is not None
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]):
parsedmarc.cli._main()
for src, sample_path in (
(agg1, self.AGGREGATE_SAMPLE_1),
(agg2, self.AGGREGATE_SAMPLE_2),
(failure, self.FAILURE_SAMPLE),
(smtp_tls, self.SMTP_TLS_SAMPLE),
):
subdir = self._expected_subdir(sample_path)
dest = os.path.join(archive_dir, subdir, os.path.basename(src))
self.assertTrue(os.path.isfile(dest), f"missing {dest}")
self.assertFalse(os.path.isfile(src))
self.assertTrue(os.path.isfile(os.path.join(output_dir, "aggregate.json")))
def test_failed_parse_moved_to_invalid(self):
"""A file that fails to parse (garbage content) is moved to
``<archive_directory>/Invalid/`` rather than left in place, and
the run completes without raising."""
with tempfile.TemporaryDirectory() as tmp_dir:
input_dir = os.path.join(tmp_dir, "input")
os.makedirs(input_dir)
garbage_path = os.path.join(input_dir, "garbage.xml")
with open(garbage_path, "wb") as f:
f.write(b"not a report")
cfg_path, output_dir, archive_dir = self._write_config(
tmp_dir, "output", archive_dirname="archive"
)
assert archive_dir is not None
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]):
parsedmarc.cli._main()
dest = os.path.join(archive_dir, "Invalid", "garbage.xml")
self.assertTrue(os.path.isfile(dest))
self.assertFalse(os.path.isfile(garbage_path))
def test_collision_appends_numeric_suffix(self):
"""A destination file that already exists at the computed archive
path is never overwritten: the newly archived file gets a
numeric suffix appended before its extension instead, and the
pre-existing file's content is untouched."""
with tempfile.TemporaryDirectory() as tmp_dir:
input_dir = os.path.join(tmp_dir, "input")
os.makedirs(input_dir)
src = self._copy_sample(self.AGGREGATE_SAMPLE_1, input_dir)
cfg_path, output_dir, archive_dir = self._write_config(
tmp_dir, "output", archive_dirname="archive"
)
assert archive_dir is not None
subdir = self._expected_subdir(self.AGGREGATE_SAMPLE_1)
dest_dir = os.path.join(archive_dir, subdir)
os.makedirs(dest_dir)
basename = os.path.basename(src)
preexisting_path = os.path.join(dest_dir, basename)
with open(preexisting_path, "wb") as f:
f.write(b"PREEXISTING DUMMY CONTENT")
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]):
parsedmarc.cli._main()
with open(preexisting_path, "rb") as f:
self.assertEqual(f.read(), b"PREEXISTING DUMMY CONTENT")
base, ext = os.path.splitext(basename)
suffixed_path = os.path.join(dest_dir, f"{base}-1{ext}")
self.assertTrue(os.path.isfile(suffixed_path))
self.assertFalse(os.path.isfile(src))
def test_second_run_skips_archived_files(self):
"""Files already inside ``archive_directory`` are excluded from
the next run's ``file_path`` expansion, so the archive may safely
live inside an input directory without its own contents being
re-parsed and re-archived on a later run. SEEN_AGGREGATE_REPORT_IDS
is cleared between runs to prove the exclusion doesn't rely on
aggregate-report dedup (the failure/SMTP-TLS samples aren't
deduped at all, so they alone would already prove this, but
clearing it removes any doubt for the aggregate sample too)."""
with tempfile.TemporaryDirectory() as tmp_dir:
input_dir = os.path.join(tmp_dir, "input")
os.makedirs(input_dir)
self._copy_sample(self.AGGREGATE_SAMPLE_1, input_dir)
self._copy_sample(self.FAILURE_SAMPLE, input_dir)
self._copy_sample(self.SMTP_TLS_SAMPLE, input_dir)
archive_dir = os.path.join(input_dir, "archive")
output_dir = os.path.join(tmp_dir, "output")
cfg_path = os.path.join(tmp_dir, "parsedmarc.ini")
config_text = (
"[general]\noffline = True\nsilent = True\n"
f"output = {output_dir}\narchive_directory = {archive_dir}\n"
)
with open(cfg_path, "w") as f:
f.write(config_text)
def _archive_tree():
tree = {}
for root, _dirs, files in os.walk(archive_dir):
for name in files:
path = os.path.join(root, name)
rel = os.path.relpath(path, archive_dir)
with open(path, "rb") as f:
tree[rel] = f.read()
return tree
with patch.object(
sys, "argv", ["parsedmarc", "-c", cfg_path, "-r", input_dir]
):
parsedmarc.cli._main()
first_run_tree = _archive_tree()
self.assertEqual(len(first_run_tree), 3)
parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear()
with patch.object(
sys, "argv", ["parsedmarc", "-c", cfg_path, "-r", input_dir]
):
parsedmarc.cli._main()
second_run_tree = _archive_tree()
self.assertEqual(first_run_tree, second_run_tree)
for rel in second_run_tree:
self.assertNotIn("-1", os.path.basename(rel))
def test_archive_with_n_procs_2(self):
"""The same archiving behavior as
``test_archive_moves_all_three_types``, but with ``n_procs = 2``
so the direct-file parsing path runs through the multiprocessing
pool."""
with tempfile.TemporaryDirectory() as tmp_dir:
input_dir = os.path.join(tmp_dir, "input")
os.makedirs(input_dir)
agg1 = self._copy_sample(self.AGGREGATE_SAMPLE_1, input_dir)
agg2 = self._copy_sample(self.AGGREGATE_SAMPLE_2, input_dir)
failure = self._copy_sample(self.FAILURE_SAMPLE, input_dir)
smtp_tls = self._copy_sample(self.SMTP_TLS_SAMPLE, input_dir)
cfg_path, output_dir, archive_dir = self._write_config(
tmp_dir, "output", archive_dirname="archive", n_procs=2
)
assert archive_dir is not None
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]):
parsedmarc.cli._main()
for src, sample_path in (
(agg1, self.AGGREGATE_SAMPLE_1),
(agg2, self.AGGREGATE_SAMPLE_2),
(failure, self.FAILURE_SAMPLE),
(smtp_tls, self.SMTP_TLS_SAMPLE),
):
subdir = self._expected_subdir(sample_path)
dest = os.path.join(archive_dir, subdir, os.path.basename(src))
self.assertTrue(os.path.isfile(dest), f"missing {dest}")
self.assertFalse(os.path.isfile(src))
self.assertTrue(os.path.isfile(os.path.join(output_dir, "aggregate.json")))
def test_no_archive_when_option_unset(self):
"""Without ``archive_directory`` configured, input files are left
in place after processing and no ``archive`` directory is
created."""
with tempfile.TemporaryDirectory() as tmp_dir:
input_dir = os.path.join(tmp_dir, "input")
os.makedirs(input_dir)
agg1 = self._copy_sample(self.AGGREGATE_SAMPLE_1, input_dir)
cfg_path, output_dir, archive_dir = self._write_config(tmp_dir, "output")
self.assertIsNone(archive_dir)
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]):
parsedmarc.cli._main()
self.assertTrue(os.path.isfile(agg1))
self.assertFalse(os.path.isdir(os.path.join(tmp_dir, "archive")))
def test_duplicate_aggregate_reports_both_archived(self):
"""The same aggregate report content under two different
filenames is deduplicated down to one report in the JSON output,
but both source files are archived normally: archiving runs for
every file that parsed successfully, independent of the
in-process report-ID dedup."""
with tempfile.TemporaryDirectory() as tmp_dir:
input_dir = os.path.join(tmp_dir, "input")
os.makedirs(input_dir)
copy1 = self._copy_sample(
self.AGGREGATE_SAMPLE_1, input_dir, dest_basename="copy-a.xml"
)
copy2 = self._copy_sample(
self.AGGREGATE_SAMPLE_1, input_dir, dest_basename="copy-b.xml"
)
cfg_path, output_dir, archive_dir = self._write_config(
tmp_dir, "output", archive_dirname="archive"
)
assert archive_dir is not None
with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path, input_dir]):
parsedmarc.cli._main()
with open(os.path.join(output_dir, "aggregate.json")) as f:
reports = json.load(f)
self.assertEqual(len(reports), 1)
subdir = self._expected_subdir(self.AGGREGATE_SAMPLE_1)
dest1 = os.path.join(archive_dir, subdir, "copy-a.xml")
dest2 = os.path.join(archive_dir, subdir, "copy-b.xml")
self.assertTrue(os.path.isfile(dest1))
self.assertTrue(os.path.isfile(dest2))
self.assertFalse(os.path.isfile(copy1))
self.assertFalse(os.path.isfile(copy2))
class TestGmailAuthModes(unittest.TestCase):
@patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
@patch("parsedmarc.cli.GmailConnection")
@@ -3734,6 +4050,398 @@ class TestExpandPath(unittest.TestCase):
self.assertEqual(_expand_path("relative/path"), "relative/path")
class TestArchiveSubdirForResult(unittest.TestCase):
"""Unit tests for _archive_subdir_for_result (issue #570): maps a
parsed report to its ``<year>/<month>/<Aggregate|Failure|SMTP-TLS>``
archive subdirectory, using minimal synthetic report dicts rather
than full sample parses."""
def test_aggregate_report_january_zero_pads_month(self):
from parsedmarc.cli import _archive_subdir_for_result
result = cast(
ParsedReport,
{
"report_type": "aggregate",
"report": {
"report_metadata": {"begin_date": "2023-01-05 00:00:00"},
},
},
)
self.assertEqual(
_archive_subdir_for_result(result),
os.path.join("2023", "01", "Aggregate"),
)
def test_failure_report_january_zero_pads_month(self):
from parsedmarc.cli import _archive_subdir_for_result
result = cast(
ParsedReport,
{
"report_type": "failure",
"report": {"arrival_date_utc": "2023-01-05 12:00:00"},
},
)
self.assertEqual(
_archive_subdir_for_result(result),
os.path.join("2023", "01", "Failure"),
)
def test_smtp_tls_report_uses_hyphenated_folder_name(self):
from parsedmarc.cli import _archive_subdir_for_result
result = cast(
ParsedReport,
{
"report_type": "smtp_tls",
"report": {"begin_date": "2016-04-01T00:00:00Z"},
},
)
self.assertEqual(
_archive_subdir_for_result(result),
os.path.join("2016", "04", "SMTP-TLS"),
)
def test_missing_date_key_returns_none(self):
from parsedmarc.cli import _archive_subdir_for_result
result = cast(
ParsedReport,
{
"report_type": "aggregate",
"report": {"report_metadata": {}},
},
)
self.assertIsNone(_archive_subdir_for_result(result))
def test_unparseable_date_returns_none(self):
from parsedmarc.cli import _archive_subdir_for_result
result = cast(
ParsedReport,
{
"report_type": "failure",
"report": {"arrival_date_utc": "not-a-real-date"},
},
)
self.assertIsNone(_archive_subdir_for_result(result))
def test_unknown_report_type_returns_none(self):
from parsedmarc.cli import _archive_subdir_for_result
result = cast(ParsedReport, {"report_type": "unknown", "report": {}})
self.assertIsNone(_archive_subdir_for_result(result))
class TestExcludeArchivedPaths(unittest.TestCase):
"""Unit tests for _exclude_archived_paths (issue #570): filters out
paths already inside the archive directory so a re-run doesn't
re-parse and re-archive its own previous output."""
def test_files_inside_archive_dir_are_excluded(self):
from parsedmarc.cli import _exclude_archived_paths
with tempfile.TemporaryDirectory() as tmp_dir:
archive_dir = os.path.join(tmp_dir, "archive")
nested_dir = os.path.join(archive_dir, "2024", "01", "Aggregate")
os.makedirs(nested_dir)
inside_path = os.path.join(nested_dir, "report.xml")
with open(inside_path, "w") as f:
f.write("x")
outside_path = os.path.join(tmp_dir, "report.xml")
with open(outside_path, "w") as f:
f.write("x")
result = _exclude_archived_paths([inside_path, outside_path], archive_dir)
self.assertEqual(result, [outside_path])
def test_relative_paths_are_resolved_before_comparison(self):
from parsedmarc.cli import _exclude_archived_paths
with tempfile.TemporaryDirectory() as tmp_dir:
archive_dir = os.path.join(tmp_dir, "archive")
os.makedirs(archive_dir)
inside_path = os.path.join(archive_dir, "report.xml")
with open(inside_path, "w") as f:
f.write("x")
cwd = os.getcwd()
try:
os.chdir(tmp_dir)
result = _exclude_archived_paths(
[os.path.join("archive", "report.xml")], "archive"
)
finally:
os.chdir(cwd)
self.assertEqual(result, [])
def test_files_outside_archive_dir_are_kept(self):
from parsedmarc.cli import _exclude_archived_paths
with tempfile.TemporaryDirectory() as tmp_dir:
archive_dir = os.path.join(tmp_dir, "archive")
os.makedirs(archive_dir)
other_path = os.path.join(tmp_dir, "elsewhere", "report.xml")
os.makedirs(os.path.dirname(other_path))
with open(other_path, "w") as f:
f.write("x")
result = _exclude_archived_paths([other_path], archive_dir)
self.assertEqual(result, [other_path])
def test_symlinked_archive_dir_still_excludes_real_path(self):
"""A symlink doesn't defeat exclusion: if archive_directory is
configured through one spelling (e.g. a ``/data`` symlink) while
an input path is discovered through the real mount-point spelling
(e.g. ``/mnt/...``), ``os.path.realpath`` resolves both to the
same canonical path so the file already inside the archive is
still recognized and excluded rather than re-archived forever."""
from parsedmarc.cli import _exclude_archived_paths
with tempfile.TemporaryDirectory() as tmp_dir:
real_archive_dir = os.path.join(tmp_dir, "real_archive")
os.makedirs(real_archive_dir)
symlinked_archive_dir = os.path.join(tmp_dir, "archive_link")
os.symlink(real_archive_dir, symlinked_archive_dir)
real_file_path = os.path.join(real_archive_dir, "report.xml")
with open(real_file_path, "w") as f:
f.write("x")
# archive_directory is configured via the symlinked spelling,
# while the file is discovered via the real spelling.
result = _exclude_archived_paths([real_file_path], symlinked_archive_dir)
self.assertEqual(result, [])
def test_non_comparable_paths_are_kept(self):
"""Per the Python docs, ``os.path.commonpath`` raises
``ValueError`` when the paths "are on the different drives"
(Windows) or mix absolute and relative pathnames
(https://docs.python.org/3/library/os.path.html#os.path.commonpath).
Both inputs here are realpath()-resolved so the mix can't occur
on POSIX, leaving the different-drives case unreachable on Linux
CI — hence the simulated raise. A path that can't be compared
with the archive root can't be inside it, so it must be kept
for parsing, and the ValueError must not propagate."""
from parsedmarc.cli import _exclude_archived_paths
with tempfile.TemporaryDirectory() as tmp_dir:
archive_dir = os.path.join(tmp_dir, "archive")
os.makedirs(archive_dir)
report_path = os.path.join(tmp_dir, "report.xml")
with open(report_path, "w") as f:
f.write("x")
with patch(
"parsedmarc.cli.os.path.commonpath",
side_effect=ValueError("Paths don't have the same drive"),
):
result = _exclude_archived_paths([report_path], archive_dir)
self.assertEqual(result, [report_path])
class TestMoveFileToArchive(unittest.TestCase):
"""Unit tests for _move_file_to_archive (issue #570): the collision
loop that appends a numeric suffix rather than overwriting an
existing destination file."""
def test_no_collision_keeps_original_basename(self):
from parsedmarc.cli import _move_file_to_archive
with tempfile.TemporaryDirectory() as tmp_dir:
src_dir = os.path.join(tmp_dir, "src")
os.makedirs(src_dir)
src_path = os.path.join(src_dir, "report.xml")
with open(src_path, "w") as f:
f.write("content")
dest_dir = os.path.join(tmp_dir, "dest")
dest_path = _move_file_to_archive(src_path, dest_dir)
self.assertEqual(dest_path, os.path.join(dest_dir, "report.xml"))
def test_collision_appends_dash_one(self):
from parsedmarc.cli import _move_file_to_archive
with tempfile.TemporaryDirectory() as tmp_dir:
src_dir = os.path.join(tmp_dir, "src")
os.makedirs(src_dir)
src_path = os.path.join(src_dir, "report.xml")
with open(src_path, "w") as f:
f.write("new content")
dest_dir = os.path.join(tmp_dir, "dest")
os.makedirs(dest_dir)
existing_path = os.path.join(dest_dir, "report.xml")
with open(existing_path, "w") as f:
f.write("original content")
dest_path = _move_file_to_archive(src_path, dest_dir)
self.assertEqual(dest_path, os.path.join(dest_dir, "report-1.xml"))
with open(existing_path) as f:
self.assertEqual(f.read(), "original content")
with open(dest_path) as f:
self.assertEqual(f.read(), "new content")
def test_two_collisions_append_dash_two(self):
from parsedmarc.cli import _move_file_to_archive
with tempfile.TemporaryDirectory() as tmp_dir:
src_dir = os.path.join(tmp_dir, "src")
os.makedirs(src_dir)
src_path = os.path.join(src_dir, "report.xml")
with open(src_path, "w") as f:
f.write("newest content")
dest_dir = os.path.join(tmp_dir, "dest")
os.makedirs(dest_dir)
for name in ("report.xml", "report-1.xml"):
with open(os.path.join(dest_dir, name), "w") as f:
f.write(f"original {name}")
dest_path = _move_file_to_archive(src_path, dest_dir)
self.assertEqual(dest_path, os.path.join(dest_dir, "report-2.xml"))
with open(os.path.join(dest_dir, "report.xml")) as f:
self.assertEqual(f.read(), "original report.xml")
with open(os.path.join(dest_dir, "report-1.xml")) as f:
self.assertEqual(f.read(), "original report-1.xml")
def test_failed_move_reraises_even_when_placeholder_cleanup_fails(self):
"""When the move fails and removing the placeholder also fails,
the move failure still propagates: the OSError from the
best-effort cleanup must be swallowed, not allowed to mask the
actionable error. The move error is deliberately a non-OSError
type so the assertion proves which of the two exceptions
escaped; the placeholder is left behind, as expected when its
cleanup fails."""
from parsedmarc.cli import _move_file_to_archive
with tempfile.TemporaryDirectory() as tmp_dir:
src_path = os.path.join(tmp_dir, "report.xml")
with open(src_path, "w") as f:
f.write("content")
dest_dir = os.path.join(tmp_dir, "dest")
with (
patch(
"parsedmarc.cli.shutil.move",
side_effect=RuntimeError("move failed"),
),
patch(
"parsedmarc.cli.os.remove",
side_effect=OSError("remove failed"),
),
):
with self.assertRaises(RuntimeError):
_move_file_to_archive(src_path, dest_dir)
# The source file is untouched and the zero-byte placeholder
# survives its failed cleanup.
self.assertTrue(os.path.isfile(src_path))
placeholder = os.path.join(dest_dir, "report.xml")
self.assertTrue(os.path.isfile(placeholder))
self.assertEqual(os.path.getsize(placeholder), 0)
# A leftover placeholder must never be executable or
# group/other-accessible: it's created with mode 0o600, not
# os.open's 0o777 default. The requested mode is masked by
# the process umask, which only clears bits, so asserting on
# the owner-exec and group/other bits is umask-independent.
placeholder_mode = stat.S_IMODE(os.stat(placeholder).st_mode)
self.assertEqual(placeholder_mode & 0o177, 0)
class TestArchiveProcessedFile(unittest.TestCase):
"""Unit tests for _archive_processed_file (issue #570): the
orchestrator that routes a processed file to Invalid/ on a parse
failure or the dated subdirectory on success, and never lets a move
failure abort the run."""
def test_exception_result_routes_to_invalid(self):
from parsedmarc.cli import _archive_processed_file
with tempfile.TemporaryDirectory() as tmp_dir:
src_path = os.path.join(tmp_dir, "garbage.xml")
with open(src_path, "w") as f:
f.write("not a report")
archive_dir = os.path.join(tmp_dir, "archive")
_archive_processed_file(
src_path, archive_dir, parsedmarc.ParserError("boom")
)
self.assertTrue(
os.path.isfile(os.path.join(archive_dir, "Invalid", "garbage.xml"))
)
def test_move_failure_is_logged_and_does_not_raise(self):
"""A filesystem error while moving a file into the archive is
logged, not raised: the file has already been definitively
classified (parsed or failed), so a move error must not abort
the run and lose that work."""
from parsedmarc.cli import _archive_processed_file
with tempfile.TemporaryDirectory() as tmp_dir:
src_path = os.path.join(tmp_dir, "garbage.xml")
with open(src_path, "w") as f:
f.write("not a report")
archive_dir = os.path.join(tmp_dir, "archive")
with patch("shutil.move", side_effect=OSError("disk full")):
with self.assertLogs("parsedmarc.log", level="ERROR") as log_ctx:
_archive_processed_file(
src_path, archive_dir, parsedmarc.ParserError("boom")
)
self.assertTrue(any("Error moving" in msg for msg in log_ctx.output))
# The move never completed, so the source file is untouched.
self.assertTrue(os.path.isfile(src_path))
def test_unresolvable_subdir_leaves_file_in_place(self):
"""When _archive_subdir_for_result can't determine a destination
(unknown report type or unparseable date, already warned about by
that helper), the file is left where it is rather than guessed
at."""
from parsedmarc.cli import _archive_processed_file
with tempfile.TemporaryDirectory() as tmp_dir:
src_path = os.path.join(tmp_dir, "report.json")
with open(src_path, "w") as f:
f.write("{}")
archive_dir = os.path.join(tmp_dir, "archive")
result = cast(ParsedReport, {"report_type": "unknown", "report": {}})
_archive_processed_file(src_path, archive_dir, result)
self.assertTrue(os.path.isfile(src_path))
def test_non_parser_error_leaves_file_in_place(self):
"""Only a ParserError (every parse-failure exception, including
InvalidSMTPTLSReport, subclasses it) routes to Invalid/. A
transient OSError — e.g. from _parse_report_file_job's broad
catch, not a report-parsing failure — leaves the file in place
so a later run can retry it, and creates no Invalid/ directory
at all, since renaming it would permanently sideline a file that
was never actually shown to be an invalid report."""
from parsedmarc.cli import _archive_processed_file
with tempfile.TemporaryDirectory() as tmp_dir:
src_path = os.path.join(tmp_dir, "report.xml")
with open(src_path, "w") as f:
f.write("<xml></xml>")
archive_dir = os.path.join(tmp_dir, "archive")
_archive_processed_file(src_path, archive_dir, OSError("transient"))
self.assertTrue(os.path.isfile(src_path))
self.assertFalse(os.path.isdir(archive_dir))
# ---------------------------------------------------------------------------
# _parse_config: per-section INI → opts mapping
#
@@ -3873,6 +4581,33 @@ class TestParseConfigGeneral(unittest.TestCase):
_parse_config(cp, opts)
self.assertEqual(opts.normalize_timespan_threshold_hours, 48.0)
def test_general_archive_directory_expands_env_var(self):
"""archive_directory goes through _expand_path, so a $VAR
reference in the INI value is expanded (issue #570)."""
from parsedmarc.cli import _parse_config
cp = _config_with(
"general", {"archive_directory": "$SOME_ARCHIVE_VAR/reports-archive"}
)
opts = _opts()
with patch.dict(os.environ, {"SOME_ARCHIVE_VAR": "/opt/dmarc"}):
_parse_config(cp, opts)
self.assertEqual(opts.archive_directory, "/opt/dmarc/reports-archive")
def test_general_archive_directory_unset_leaves_attribute_absent(self):
"""When archive_directory is absent from the INI, _parse_config
never sets opts.archive_directory at all — asserted here as the
attribute staying absent from this test's bare Namespace. (In the
real CLI the attribute pre-exists with the Namespace default of
None, so _parse_config leaving it untouched is what keeps
archiving disabled.)"""
from parsedmarc.cli import _parse_config
cp = _config_with("general", {"silent": "false"})
opts = _opts()
_parse_config(cp, opts)
self.assertFalse(hasattr(opts, "archive_directory"))
class TestParseConfigElasticsearch(unittest.TestCase):
def test_elasticsearch_basic(self):