amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-07-16 13:34:56 +00:00
Code Issues Projects Releases Wiki Activity

Raise test coverage: utils.py, elastic.py, and opensearch.py to 100% (#816)

Browse Source 
* Raise test coverage: utils, elastic, and opensearch to 100%

Coverage of the shipped library rises from 88% to 90%, with
parsedmarc/utils.py 86% -> 100% and elastic.py / opensearch.py
99% -> 100%. All new tests assert on observable behaviour and mock
only at SDK boundaries (dnspython Resolver.resolve, requests.get,
subprocess.check_call, elasticsearch_dsl/opensearchpy Document.save).

New tests cover: query_dns transient-error retries, the load_ip_db
download/cache/bundled fallback chain, the IPinfo API token probe and
per-request MMDB fallbacks, _normalize_ip_record schema handling,
reverse-DNS-map invalid-CSV fallback, caller-provided reverse DNS
maps, Outlook MSG conversion (missing msgconvert and success paths),
parse_email Cc/Bcc/attachment-hash branches, aggregate-XML edge cases
(bytes input, repeated policy_published, unknown RFC 9990 override
types, missing org_name, attribute-only <email>), extract_report on
non-seekable streams, and the _AggregateReportDoc.save() override
that derives passed_dmarc.

Bugs found by the new tests, fixed in the same PR per the testing
standards:

- parse_email() crashed with KeyError: 'Headers' on messages whose
  From header is present but unparseable (e.g. a bare "From:" line):
  the fallback read parsed_email["Headers"], but the parsed headers
  are stored under lowercase "headers" (assigned a few lines up in
  the same function), so the key never exists. At the CLI surface
  this made any failure report whose embedded sample had an empty
  From: header fail to parse ("Missing value: 'Headers'").
- configure_ipinfo_api(probe=True) logged "IPinfo API configured"
  when the probe could not reach the API, contradicting its own
  docstring ("other errors are logged and the token is still
  accepted"): _ipinfo_api_lookup() returns None on network errors
  instead of raising, so the probe's exception handler was
  unreachable. The probe now checks the lookup result and warns on
  failure; 401/403 still raises InvalidIPinfoAPIKey.

Dead code deleted rather than padded with tests:

- _SMTPTLSReportDoc.add_policy() in elastic.py and opensearch.py
  (the save paths construct _SMTPTLSPolicyDoc directly).
- The no-op "for failure_index in failure_indexes: pass" loop in
  both migrate_indexes() implementations (parameter still accepted).
- The importlib.resources ImportError fallback in utils.py, which
  re-imported the same module and is unreachable on Python >= 3.10.
- The "Invalid report content" guard in extract_report(): every
  input branch assigns file_object or raises first (confirmed by
  pyright narrowing with the guard removed).

Also widens parse_aggregate_report_xml's annotation to str | bytes
to match its existing runtime behaviour (bytes are decoded with
errors ignored).

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

* Use assertGreater for the reverse-DNS-map fallback size check

Addresses the github-code-quality bot finding on PR #816: assertTrue
with a comparison inside can't show the operands on failure, while
assertGreater reports both values and the failed relation. No change
to test behavior.

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-11 18:21:04 -04:00
committed by GitHub
co-authored by Claude Fable 5
parent a1da7b3420
commit 746da77de5
9 changed files with 648 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_elastic.py
+26
View File
@@ -579,6 +579,32 @@ class TestSaveAggregateReport(unittest.TestCase):
self.assertIn("cust_dmarc_aggregate_tenant_a*", search_index)
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
no cluster is needed."""
def test_passed_dmarc_derived_from_alignment(self):
cases = [
(True, False, True),
(False, True, True),
(True, True, True),
(False, False, False),
]
for spf_aligned, dkim_aligned, expected in cases:
with self.subTest(spf=spf_aligned, dkim=dkim_aligned):
with patch.object(
elastic_module.Document, "save", return_value=None
) as mock_super_save:
doc = elastic_module._AggregateReportDoc(
spf_aligned=spf_aligned, dkim_aligned=dkim_aligned
)
doc.save()
mock_super_save.assert_called_once()
self.assertEqual(bool(doc.passed_dmarc), expected)
# ---------------------------------------------------------------------------
# save_failure_report_to_elasticsearch
# ---------------------------------------------------------------------------
tests/test_init.py
+160
View File
@@ -2954,5 +2954,165 @@ class TestAppendCsv(unittest.TestCase):
os.remove(path)
def _minimal_aggregate_xml(
policy_published: str = (
"<policy_published><domain>example.com</domain><p>none</p></policy_published>"
),
org_name: str = "TestOrg",
email: str = "test@example.com",
reason: str = "",
) -> str:
"""A minimal, valid aggregate report with substitutable sections."""
return f"""<?xml version="1.0"?>
<feedback>
<report_metadata>
<org_name>{org_name}</org_name>
<email>{email}</email>
<report_id>edge-case</report_id>
<date_range><begin>1704067200</begin><end>1704153599</end></date_range>
</report_metadata>
{policy_published}
<record>
<row>
<source_ip>192.0.2.1</source_ip>
<count>1</count>
<policy_evaluated>
<disposition>none</disposition>
<dkim>pass</dkim>
<spf>pass</spf>
{reason}
</policy_evaluated>
</row>
<identifiers><header_from>example.com</header_from></identifiers>
<auth_results>
<spf><domain>example.com</domain><result>pass</result></spf>
</auth_results>
</record>
</feedback>"""
class TestAggregateReportEdgeCases(unittest.TestCase):
"""Parsing edge cases for aggregate report XML documents."""
def testBytesInputIsDecoded(self):
"""parse_aggregate_report_xml accepts bytes input"""
xml = _minimal_aggregate_xml().encode("utf-8")
report = parsedmarc.parse_aggregate_report_xml(xml, offline=True)
self.assertEqual(report["report_metadata"]["report_id"], "edge-case")
def testPolicyPublishedListUsesFirstEntry(self):
"""When a reporter emits multiple policy_published elements, the
first one is used"""
policies = (
"<policy_published><domain>example.com</domain><p>reject</p>"
"</policy_published>"
"<policy_published><domain>other.example</domain><p>none</p>"
"</policy_published>"
)
report = parsedmarc.parse_aggregate_report_xml(
_minimal_aggregate_xml(policy_published=policies), offline=True
)
self.assertEqual(report["policy_published"]["domain"], "example.com")
self.assertEqual(report["policy_published"]["p"], "reject")
def testUnknownPolicyOverrideTypeWarnsUnderRFC9990(self):
"""An override reason type that RFC 9990 does not define (and RFC
7489 never defined) logs an 'Unknown policy override reason type'
warning; it is stored as-is. RFC 9990's PolicyOverrideType
enumeration is {local_policy, mailing_list, other,
policy_test_mode, trusted_forwarder}."""
policies = (
"<policy_published><domain>example.com</domain><p>none</p>"
"<np>none</np></policy_published>"
)
reason = "<reason><type>banana</type></reason>"
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
report = parsedmarc.parse_aggregate_report_xml(
_minimal_aggregate_xml(policy_published=policies, reason=reason),
offline=True,
)
self.assertTrue(
any(
"Unknown policy override reason type" in message
for message in cm.output
)
)
reasons = report["records"][0]["policy_evaluated"]["policy_override_reasons"]
self.assertEqual(reasons[0]["type"], "banana")
def testMissingOrgNameAndEmailIsInvalid(self):
"""A report with empty org_name and email raises
InvalidAggregateReport, since org_name has no fallback source"""
with self.assertRaises(parsedmarc.InvalidAggregateReport) as ctx:
parsedmarc.parse_aggregate_report_xml(
_minimal_aggregate_xml(org_name="", email=""), offline=True
)
self.assertIn("Organization name is missing", str(ctx.exception))
def testMalformedEmailAttributeOnlyIsDiscarded(self):
"""An <email> element that xmltodict turns into an attributes-only
dict (no text) is discarded rather than crashing"""
xml = _minimal_aggregate_xml().replace(
"<email>test@example.com</email>", '<email xml:lang="en"></email>'
)
report = parsedmarc.parse_aggregate_report_xml(xml, offline=True)
self.assertIsNone(report["report_metadata"]["org_email"])
class _NonSeekableStream:
"""A minimal non-seekable stream, like sys.stdin / a socket file."""
def __init__(self, data):
self._data = data
self._pos = 0
def seekable(self):
return False
def read(self, size=-1):
if size < 0:
result = self._data[self._pos :]
self._pos = len(self._data)
else:
result = self._data[self._pos : self._pos + size]
self._pos += size
return result
class _BrokenSeekableStream(_NonSeekableStream):
"""A stream whose seekable() itself raises, as some wrapped streams do."""
def seekable(self):
raise OSError("stream does not support seekable()")
class TestExtractReportStreams(unittest.TestCase):
"""extract_report accepts file objects that cannot seek (stdin, pipes,
sockets) and must reject text-mode streams with a clear error."""
def testNonSeekableTextStreamRaisesParserError(self):
"""A non-seekable text-mode stream raises ParserError instead of
failing later on a bytes/str mismatch"""
with open("samples/extract_report/nice-input.xml") as f:
text = f.read()
with self.assertRaises(parsedmarc.ParserError) as ctx:
parsedmarc.extract_report(cast(BinaryIO, _NonSeekableStream(text)))
self.assertIn("binary", str(ctx.exception))
def testNonSeekableBytesStreamIsExtracted(self):
"""A non-seekable binary stream is buffered and extracted"""
with open("samples/extract_report/nice-input.xml", "rb") as f:
data = f.read()
result = parsedmarc.extract_report(cast(BinaryIO, _NonSeekableStream(data)))
self.assertIn("<feedback>", result)
def testStreamWithBrokenSeekableIsExtracted(self):
"""A stream whose seekable() raises is treated as non-seekable"""
with open("samples/extract_report/nice-input.xml", "rb") as f:
data = f.read()
result = parsedmarc.extract_report(cast(BinaryIO, _BrokenSeekableStream(data)))
self.assertIn("<feedback>", result)
if __name__ == "__main__":
unittest.main(verbosity=2)
tests/test_opensearch.py
+26
View File
@@ -579,6 +579,32 @@ class TestSaveAggregateReport(unittest.TestCase):
self.assertIn("cust_dmarc_aggregate_tenant_a*", search_index)
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 (opensearchpy.Document.save) is mocked so
no cluster is needed."""
def test_passed_dmarc_derived_from_alignment(self):
cases = [
(True, False, True),
(False, True, True),
(True, True, True),
(False, False, False),
]
for spf_aligned, dkim_aligned, expected in cases:
with self.subTest(spf=spf_aligned, dkim=dkim_aligned):
with patch.object(
opensearch_module.Document, "save", return_value=None
) as mock_super_save:
doc = opensearch_module._AggregateReportDoc(
spf_aligned=spf_aligned, dkim_aligned=dkim_aligned
)
doc.save()
mock_super_save.assert_called_once()
self.assertEqual(bool(doc.passed_dmarc), expected)
# ---------------------------------------------------------------------------
# save_failure_report_to_opensearch
# ---------------------------------------------------------------------------
tests/test_utils.py
+407 -1
View File
@@ -11,6 +11,7 @@ from tempfile import NamedTemporaryFile
from unittest.mock import MagicMock, patch
import dns.exception
import dns.resolver
import requests
from expiringdict import ExpiringDict
@@ -866,13 +867,82 @@ Body"""
for att in result["attachments"]:
self.assertNotIn("payload", att)
def testEmptyFromHeaderYieldsNone(self):
"""An email whose From header is present but empty parses with
from=None instead of crashing.
Regression: mailparser omits "from" from mail_json when the From
header value is unparseable, and the headers fallback read
``parsed_email["Headers"]`` — a key that is never set (the parsed
headers are stored under lowercase "headers", see parse_email) —
so any such message raised KeyError: 'Headers'.
"""
email_str = "From:\r\nTo: a@b.com\r\nSubject: t\r\n\r\nbody\r\n"
result = parsedmarc.utils.parse_email(email_str)
self.assertIsNone(result["from"])
def testCcAndBccHeadersAreParsed(self):
"""Cc and Bcc headers are parsed into address dicts"""
email_str = (
"From: a@b.com\r\n"
"To: t@e.com\r\n"
"Cc: c@d.com, C Two <c2@d.com>\r\n"
"Bcc: e@f.com\r\n"
"Subject: Hi\r\n\r\nBody\r\n"
)
result = parsedmarc.utils.parse_email(email_str)
self.assertEqual([a["address"] for a in result["cc"]], ["c@d.com", "c2@d.com"])
self.assertEqual(result["cc"][1]["display_name"], "C Two")
self.assertEqual([a["address"] for a in result["bcc"]], ["e@f.com"])
@staticmethod
def _multipart_email(transfer_encoding: str, payload: str) -> str:
return (
"From: a@b.com\r\nTo: t@e.com\r\nSubject: att\r\n"
"MIME-Version: 1.0\r\n"
'Content-Type: multipart/mixed; boundary="B"\r\n\r\n'
"--B\r\nContent-Type: text/plain\r\n\r\nbody\r\n"
'--B\r\nContent-Type: application/octet-stream; name="a.bin"\r\n'
f"Content-Transfer-Encoding: {transfer_encoding}\r\n"
'Content-Disposition: attachment; filename="a.bin"\r\n\r\n'
f"{payload}\r\n"
"--B--\r\n"
)
def testNonBase64AttachmentIsHashed(self):
"""A non-base64 attachment's sha256 is computed over the encoded
payload text"""
import hashlib
result = parsedmarc.utils.parse_email(
self._multipart_email("7bit", "hello world")
)
attachments = result["attachments"]
self.assertEqual(len(attachments), 1)
self.assertEqual(
attachments[0]["sha256"], hashlib.sha256(b"hello world").hexdigest()
)
def testUndecodableAttachmentIsKeptWithoutHash(self):
"""An attachment whose base64 payload cannot be decoded is kept,
just without a sha256, and parsing does not crash"""
result = parsedmarc.utils.parse_email(
self._multipart_email("base64", "!!!notb64!!!")
)
attachments = result["attachments"]
self.assertEqual(len(attachments), 1)
self.assertNotIn("sha256", attachments[0])
self.assertEqual(attachments[0]["payload"], "!!!notb64!!!")
class TestUtilsOutlookMsg(unittest.TestCase):
"""Tests for Outlook MSG detection and conversion"""
MSG_MAGIC = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"
def testIsOutlookMsg(self):
"""is_outlook_msg detects MSG magic bytes"""
msg_magic = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + b"\x00" * 100
msg_magic = self.MSG_MAGIC + b"\x00" * 100
self.assertTrue(parsedmarc.utils.is_outlook_msg(msg_magic))
def testIsNotOutlookMsg(self):
@@ -885,6 +955,58 @@ class TestUtilsOutlookMsg(unittest.TestCase):
with self.assertRaises(ValueError):
parsedmarc.utils.convert_outlook_msg(b"not an msg file")
def testConvertOutlookMsgMissingUtility(self):
"""A missing msgconvert utility raises EmailParserError, and the
working directory is restored"""
old_cwd = os.getcwd()
with patch(
"parsedmarc.utils.subprocess.check_call",
side_effect=FileNotFoundError("msgconvert"),
):
with self.assertRaises(parsedmarc.utils.EmailParserError):
parsedmarc.utils.convert_outlook_msg(self.MSG_MAGIC + b"\x00" * 100)
self.assertEqual(os.getcwd(), old_cwd)
def testConvertOutlookMsgReadsConvertedFile(self):
"""convert_outlook_msg writes the .msg for msgconvert, reads back
the .eml it produces, and restores the working directory. The
subprocess boundary is mocked with a fake msgconvert that converts
the temp .msg into a fixed RFC 822 message."""
rfc822 = b"From: a@b.com\r\nSubject: converted\r\n\r\nhi\r\n"
def fake_msgconvert(args, stdout=None, stderr=None):
# msgconvert is invoked in a temp dir containing sample.msg
# and writes sample.eml next to it.
self.assertEqual(args, ["msgconvert", "sample.msg"])
with open("sample.msg", "rb") as f:
self.assertTrue(parsedmarc.utils.is_outlook_msg(f.read()))
with open("sample.eml", "wb") as f:
f.write(rfc822)
old_cwd = os.getcwd()
with patch(
"parsedmarc.utils.subprocess.check_call", side_effect=fake_msgconvert
):
result = parsedmarc.utils.convert_outlook_msg(
self.MSG_MAGIC + b"\x00" * 100
)
self.assertEqual(result, rfc822)
self.assertEqual(os.getcwd(), old_cwd)
def testParseEmailConvertsOutlookMsgBytes(self):
"""parse_email detects Outlook MSG bytes and parses the converted
RFC 822 output"""
def fake_msgconvert(args, stdout=None, stderr=None):
with open("sample.eml", "wb") as f:
f.write(b"From: a@b.com\r\nSubject: from msg\r\n\r\nhi\r\n")
with patch(
"parsedmarc.utils.subprocess.check_call", side_effect=fake_msgconvert
):
result = parsedmarc.utils.parse_email(self.MSG_MAGIC + b"\x00" * 100)
self.assertEqual(result["subject"], "from msg")
class TestUtilsReverseDnsMap(unittest.TestCase):
"""Tests for reverse DNS map loading"""
@@ -921,6 +1043,33 @@ class TestUtilsReverseDnsMap(unittest.TestCase):
parsedmarc.utils.load_reverse_dns_map(rdns_map)
self.assertTrue(len(rdns_map) > 0)
def testLoadReverseDnsMapInvalidCsvFallback(self):
"""A fetch that returns a non-map CSV body logs a warning and
falls back to the bundled map"""
response = MagicMock()
response.text = "not,the,map\nfoo,bar,baz\n"
response.raise_for_status.return_value = None
rdns_map = {}
with patch("parsedmarc.utils.requests.get", return_value=response):
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
parsedmarc.utils.load_reverse_dns_map(rdns_map)
self.assertTrue(any("Not a valid CSV file" in message for message in cm.output))
self.assertGreater(len(rdns_map), 0)
def testGetServiceUsesProvidedMap(self):
"""get_service_from_reverse_dns_base_domain consults a caller-
provided non-empty map without loading anything"""
provided: parsedmarc.utils.ReverseDNSMap = {
"custom.example": {"name": "Custom Co", "type": "SaaS"}
}
with patch("parsedmarc.utils.load_reverse_dns_map") as mock_load:
service = parsedmarc.utils.get_service_from_reverse_dns_base_domain(
"Custom.Example", reverse_dns_map=provided
)
mock_load.assert_not_called()
self.assertEqual(service["name"], "Custom Co")
self.assertEqual(service["type"], "SaaS")
class TestPslOverrides(unittest.TestCase):
"""Tests for PSL override matching"""
@@ -961,5 +1110,262 @@ class TestIsMbox(unittest.TestCase):
self.assertFalse(parsedmarc.utils.is_mbox("/nonexistent/file.mbox"))
class TestQueryDnsRetries(unittest.TestCase):
"""Tests for the query_dns transient-error retry loop, mocking at the
dnspython SDK boundary (Resolver.resolve)."""
def testTransientErrorIsRetried(self):
"""A retryable error (OSError is in _RETRYABLE_DNS_ERRORS) on the
first attempt is retried, and the second attempt's answers are
returned. A single nameserver is passed so the single-nameserver
lifetime branch is exercised too."""
answer = MagicMock()
answer.to_text.return_value = "mail.example.com."
with patch.object(
dns.resolver.Resolver,
"resolve",
side_effect=[OSError("transient network error"), [answer]],
) as mock_resolve:
records = parsedmarc.utils.query_dns(
"example.com",
"A",
nameservers=["192.0.2.53"],
timeout=0.1,
retries=1,
)
self.assertEqual(records, ["mail.example.com"])
self.assertEqual(mock_resolve.call_count, 2)
def testErrorRaisedAfterRetriesExhausted(self):
"""When every attempt fails, the last error propagates after
retries+1 total attempts."""
with patch.object(
dns.resolver.Resolver,
"resolve",
side_effect=OSError("persistent network error"),
) as mock_resolve:
with self.assertRaises(OSError):
parsedmarc.utils.query_dns(
"example.com",
"A",
nameservers=["192.0.2.53"],
timeout=0.1,
retries=2,
)
self.assertEqual(mock_resolve.call_count, 3)
class TestLoadIpDb(unittest.TestCase):
"""Tests for the load_ip_db() download/cache/bundled fallback chain,
mocking at the requests SDK boundary."""
def setUp(self):
old_ip_db_path = parsedmarc.utils._IP_DB_PATH
parsedmarc.utils._IP_DB_PATH = None
def restore():
parsedmarc.utils._IP_DB_PATH = old_ip_db_path
self.addCleanup(restore)
# Redirect the download cache into a per-test directory so the
# tests never touch (or depend on) the real tempdir cache.
self.tmp_dir = tempfile.mkdtemp()
self.addCleanup(lambda: shutil.rmtree(self.tmp_dir, ignore_errors=True))
patcher = patch(
"parsedmarc.utils.tempfile.gettempdir", return_value=self.tmp_dir
)
patcher.start()
self.addCleanup(patcher.stop)
def testExistingLocalFileIsUsedDirectly(self):
"""An existing local_file_path wins without any network request"""
local_path = os.path.join(self.tmp_dir, "local.mmdb")
with open(local_path, "wb") as f:
f.write(b"local db")
with patch("parsedmarc.utils.requests.get") as mock_get:
parsedmarc.utils.load_ip_db(local_file_path=local_path)
mock_get.assert_not_called()
self.assertEqual(parsedmarc.utils._IP_DB_PATH, local_path)
def testDownloadSuccessWritesCacheFile(self):
"""A successful download is written to the cache path and selected"""
response = MagicMock()
response.content = b"downloaded db bytes"
response.raise_for_status.return_value = None
with patch("parsedmarc.utils.requests.get", return_value=response) as mock_get:
parsedmarc.utils.load_ip_db(url="https://example.com/db.mmdb")
self.assertEqual(mock_get.call_args.args[0], "https://example.com/db.mmdb")
cached_path = os.path.join(self.tmp_dir, "parsedmarc", "ipinfo_lite.mmdb")
self.assertEqual(parsedmarc.utils._IP_DB_PATH, cached_path)
with open(cached_path, "rb") as f:
self.assertEqual(f.read(), b"downloaded db bytes")
def testDownloadFailureFallsBackToCachedCopy(self):
"""On a network error, a previously cached copy is selected"""
cache_dir = os.path.join(self.tmp_dir, "parsedmarc")
os.makedirs(cache_dir)
cached_path = os.path.join(cache_dir, "ipinfo_lite.mmdb")
with open(cached_path, "wb") as f:
f.write(b"stale cached db")
with patch(
"parsedmarc.utils.requests.get",
side_effect=requests.exceptions.ConnectionError("no network"),
):
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
parsedmarc.utils.load_ip_db()
self.assertTrue(
any("Failed to fetch IP database" in message for message in cm.output)
)
self.assertEqual(parsedmarc.utils._IP_DB_PATH, cached_path)
def testDownloadFailureFallsBackToBundledCopy(self):
"""On a network error with no cached copy, the bundled db is used"""
with patch(
"parsedmarc.utils.requests.get",
side_effect=requests.exceptions.ConnectionError("no network"),
):
parsedmarc.utils.load_ip_db()
bundled = str(files(parsedmarc.resources.ipinfo).joinpath("ipinfo_lite.mmdb"))
self.assertEqual(parsedmarc.utils._IP_DB_PATH, bundled)
def testSaveFailureFallsBackToBundledCopy(self):
"""A download that cannot be written to disk logs a warning and
falls back to the bundled db instead of crashing. The cache dir is
made uncreatable by pointing gettempdir at a regular file."""
blocker = os.path.join(self.tmp_dir, "blocker")
with open(blocker, "wb") as f:
f.write(b"not a directory")
response = MagicMock()
response.content = b"downloaded db bytes"
response.raise_for_status.return_value = None
with patch("parsedmarc.utils.tempfile.gettempdir", return_value=blocker):
with patch("parsedmarc.utils.requests.get", return_value=response):
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
parsedmarc.utils.load_ip_db()
self.assertTrue(
any("Failed to save IP database" in message for message in cm.output)
)
bundled = str(files(parsedmarc.resources.ipinfo).joinpath("ipinfo_lite.mmdb"))
self.assertEqual(parsedmarc.utils._IP_DB_PATH, bundled)
class TestConfigureIpinfoApiProbe(unittest.TestCase):
"""Tests for the configure_ipinfo_api() token probe."""
def setUp(self):
self.addCleanup(parsedmarc.utils.configure_ipinfo_api, None)
@staticmethod
def _response(status_code, json_body=None):
response = MagicMock()
response.status_code = status_code
response.ok = 200 <= status_code < 300
response.json.return_value = json_body if json_body is not None else {}
return response
def testProbeSuccessLogsConfigured(self):
"""A successful probe logs that the API is configured"""
api_json = {"ip": "1.1.1.1", "asn": "AS13335", "country_code": "US"}
with patch(
"parsedmarc.utils.requests.get",
return_value=self._response(200, api_json),
):
with self.assertLogs("parsedmarc.log", level="INFO") as cm:
parsedmarc.utils.configure_ipinfo_api("fake-token", probe=True)
self.assertTrue(
any("IPinfo API configured" in message for message in cm.output)
)
def testProbeNetworkErrorKeepsToken(self):
"""A probe network error logs a warning but keeps the token so
per-request fallback can take over later"""
with patch(
"parsedmarc.utils.requests.get",
side_effect=requests.exceptions.ConnectionError("no network"),
):
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
parsedmarc.utils.configure_ipinfo_api("fake-token", probe=True)
self.assertTrue(
any("IPinfo API probe failed" in message for message in cm.output)
)
self.assertEqual(parsedmarc.utils._IPINFO_API_TOKEN, "fake-token")
def testProbeInvalidKeyRaises(self):
"""A 401 during the probe raises InvalidIPinfoAPIKey"""
with patch("parsedmarc.utils.requests.get", return_value=self._response(401)):
with self.assertRaises(parsedmarc.utils.InvalidIPinfoAPIKey):
parsedmarc.utils.configure_ipinfo_api("bad-token", probe=True)
class TestIpinfoApiLookupFallbacks(unittest.TestCase):
"""API lookup failures other than 401/403 must fall back to the MMDB
silently: network errors, non-JSON bodies, and non-dict payloads."""
def setUp(self):
parsedmarc.utils.configure_ipinfo_api("fake-token", probe=False)
self.addCleanup(parsedmarc.utils.configure_ipinfo_api, None)
def _assert_mmdb_fallback(self, response=None, side_effect=None):
with patch(
"parsedmarc.utils.requests.get",
return_value=response,
side_effect=side_effect,
):
record = parsedmarc.utils.get_ip_address_db_record("8.8.8.8")
# The bundled MMDB attributes 8.8.8.8 to Google's ASN.
self.assertIsNotNone(record)
assert record is not None
self.assertEqual(record["asn"], 15169)
def testNetworkErrorFallsBackToMmdb(self):
self._assert_mmdb_fallback(
side_effect=requests.exceptions.ConnectionError("no network")
)
def testNonJsonBodyFallsBackToMmdb(self):
response = MagicMock()
response.status_code = 200
response.ok = True
response.json.side_effect = ValueError("not JSON")
self._assert_mmdb_fallback(response=response)
def testNonDictPayloadFallsBackToMmdb(self):
response = MagicMock()
response.status_code = 200
response.ok = True
response.json.return_value = ["not", "a", "dict"]
self._assert_mmdb_fallback(response=response)
class TestNormalizeIpRecord(unittest.TestCase):
"""_normalize_ip_record must produce the same internal shape from both
the IPinfo API schema and the MaxMind MMDB schema."""
def testMaxMindSchema(self):
"""MaxMind-style records (nested country iso_code, ASN under
autonomous_system_number/organization) normalize correctly"""
record = parsedmarc.utils._normalize_ip_record(
{
"country": {"iso_code": "US"},
"autonomous_system_number": 15169,
"autonomous_system_organization": "Google LLC",
}
)
self.assertEqual(record["country"], "US")
self.assertEqual(record["asn"], 15169)
self.assertEqual(record["as_name"], "Google LLC")
self.assertIsNone(record["as_domain"])
def testIntegerAsnPassesThrough(self):
"""An already-integer asn field is stored as-is, and as_domain is
lowercased on the way in"""
record = parsedmarc.utils._normalize_ip_record(
{"country_code": "US", "asn": 64496, "as_domain": "EXAMPLE.com"}
)
self.assertEqual(record["asn"], 64496)
self.assertEqual(record["as_domain"], "example.com")
if __name__ == "__main__":
unittest.main(verbosity=2)