test: add tests for missing error paths in hop_asn _ptr

Added tests for the missing error paths in `_ptr` inside `netsleuth/probes/hop_asn.py`:
- `test_ptr_returns_none_on_dns_exception` verifying that it correctly catches `DNSException`.
- `test_ptr_returns_none_on_value_error_from_invalid_ip` verifying that it correctly catches `ValueError` caused by passing an invalid IP.

The tests ensure the function safely handles DNS issues or invalid inputs by returning `None` instead of propagating exceptions.

Co-authored-by: Horizontal42 <75383519+Horizontal42@users.noreply.github.com>
This commit is contained in:
google-labs-jules[bot] 2026-08-26 14:05:18 +00:00
parent 8c99ac78b7
commit 4622e35072

View file

@ -11,14 +11,20 @@ ORIGIN6_ZONE = "origin6.asn.cymru.com"
def test_cymru_query_name_reverses_ipv4_octets(): def test_cymru_query_name_reverses_ipv4_octets():
assert cymru_query_name("1.2.3.4", ORIGIN_ZONE, ORIGIN6_ZONE) == f"4.3.2.1.{ORIGIN_ZONE}" assert (
cymru_query_name("1.2.3.4", ORIGIN_ZONE, ORIGIN6_ZONE)
== f"4.3.2.1.{ORIGIN_ZONE}"
)
def test_cymru_query_name_reverses_ipv6_nibbles_against_a_hand_expanded_literal(): def test_cymru_query_name_reverses_ipv6_nibbles_against_a_hand_expanded_literal():
# 2001:db8::1 expands to 2001:0db8:0000:...:0001 -> 32 hex nibbles, reversed and # 2001:db8::1 expands to 2001:0db8:0000:...:0001 -> 32 hex nibbles, reversed and
# dot-joined. Verified independently via `ipaddress` before hardcoding here. # dot-joined. Verified independently via `ipaddress` before hardcoding here.
expected = "1" + ".0" * 23 + ".8.b.d.0.1.0.0.2" expected = "1" + ".0" * 23 + ".8.b.d.0.1.0.0.2"
assert cymru_query_name("2001:db8::1", ORIGIN_ZONE, ORIGIN6_ZONE) == f"{expected}.{ORIGIN6_ZONE}" assert (
cymru_query_name("2001:db8::1", ORIGIN_ZONE, ORIGIN6_ZONE)
== f"{expected}.{ORIGIN6_ZONE}"
)
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -59,7 +65,11 @@ async def test_enrich_hops_dedupes_repeated_ips_and_skips_none_ips(monkeypatch):
trace = TraceResult(target="8.8.8.8", hops=[hop_a, hop_b, hop_c]) trace = TraceResult(target="8.8.8.8", hops=[hop_a, hop_b, hop_c])
await enrich_hops( await enrich_hops(
[trace], origin_zone=ORIGIN_ZONE, origin6_zone=ORIGIN6_ZONE, asn_zone="asn.cymru.com", timeout=1.0 [trace],
origin_zone=ORIGIN_ZONE,
origin6_zone=ORIGIN6_ZONE,
asn_zone="asn.cymru.com",
timeout=1.0,
) )
assert calls == ["8.8.8.8"] assert calls == ["8.8.8.8"]
@ -79,7 +89,11 @@ async def test_enrich_hops_fills_reverse_dns_only_when_absent(monkeypatch):
trace = TraceResult(target="1.1.1.1", hops=[hop_with_rdns, hop_without_rdns]) trace = TraceResult(target="1.1.1.1", hops=[hop_with_rdns, hop_without_rdns])
await enrich_hops( await enrich_hops(
[trace], origin_zone=ORIGIN_ZONE, origin6_zone=ORIGIN6_ZONE, asn_zone="asn.cymru.com", timeout=1.0 [trace],
origin_zone=ORIGIN_ZONE,
origin6_zone=ORIGIN6_ZONE,
asn_zone="asn.cymru.com",
timeout=1.0,
) )
assert hop_with_rdns.reverse_dns == "already.set" assert hop_with_rdns.reverse_dns == "already.set"
@ -87,7 +101,9 @@ async def test_enrich_hops_fills_reverse_dns_only_when_absent(monkeypatch):
def _trace_with_countries(*countries: str | None) -> TraceResult: def _trace_with_countries(*countries: str | None) -> TraceResult:
hops = [TraceHop(ttl=i + 1, ip=f"1.1.1.{i}", country=c) for i, c in enumerate(countries)] hops = [
TraceHop(ttl=i + 1, ip=f"1.1.1.{i}", country=c) for i, c in enumerate(countries)
]
return TraceResult(target="1.1.1.1", hops=hops) return TraceResult(target="1.1.1.1", hops=hops)
@ -106,3 +122,28 @@ def test_path_asn_findings_silent_on_a_single_foreign_hop():
def test_path_asn_findings_silent_when_country_is_unknown_throughout(): def test_path_asn_findings_silent_when_country_is_unknown_throughout():
trace = _trace_with_countries(None, None, None) trace = _trace_with_countries(None, None, None)
assert path_asn_findings(trace, "RU") == [] assert path_asn_findings(trace, "RU") == []
@pytest.mark.asyncio
async def test_ptr_returns_none_on_dns_exception(monkeypatch):
from netsleuth.probes.hop_asn import _ptr
import dns.asyncresolver
import dns.exception
resolver = dns.asyncresolver.Resolver()
async def mock_resolve(*args, **kwargs):
raise dns.exception.DNSException()
monkeypatch.setattr(resolver, "resolve", mock_resolve)
assert await _ptr(resolver, "8.8.8.8") is None
@pytest.mark.asyncio
async def test_ptr_returns_none_on_value_error_from_invalid_ip():
from netsleuth.probes.hop_asn import _ptr
import dns.asyncresolver
resolver = dns.asyncresolver.Resolver()
assert await _ptr(resolver, "invalid_ip") is None