From ddfbb1adbfb277d2c1ea430fd3497f8ab38457fb Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 19 Aug 2026 11:32:02 +0000 Subject: [PATCH] Add tests for _os_timezone function This commit adds a new test file, `tests/test_cli.py`, containing unit tests for the `_os_timezone` function in `netsleuth.cli`. The tests cover: - Happy paths (TZ variable, /etc/localtime symlink, /etc/timezone file) - The edge case of an `OSError` exception being thrown when reading the /etc/timezone file. Co-authored-by: Horizontal42 <75383519+Horizontal42@users.noreply.github.com> --- src/netsleuth/bgp.py | 8 ++-- src/netsleuth/cli.py | 45 ++++++++++++++-------- src/netsleuth/interpret.py | 5 +-- src/netsleuth/orchestration.py | 7 ++-- src/netsleuth/probes/latency.py | 2 +- src/netsleuth/probes/pmtud.py | 2 +- src/netsleuth/probes/tls_rtt.py | 6 +-- src/netsleuth/probes/traceroute.py | 7 ++-- src/netsleuth/speed.py | 4 +- src/netsleuth/watch.py | 9 ++++- tests/test_alerting.py | 1 - tests/test_captive_portal.py | 5 ++- tests/test_cli.py | 62 ++++++++++++++++++++++++++++++ tests/test_exporter.py | 11 +++++- tests/test_hop_asn.py | 1 - tests/test_iface.py | 6 ++- tests/test_interpret.py | 31 +++++++++++++-- tests/test_interpret_l7.py | 1 - tests/test_ip_geo_normalize.py | 2 +- tests/test_models.py | 2 +- tests/test_netinfo.py | 6 ++- tests/test_orchestration.py | 2 +- tests/test_pmtud.py | 7 +++- tests/test_reputation.py | 2 +- tests/test_speed_cloudflare.py | 2 +- tests/test_speed_math.py | 8 +++- tests/test_tls_rtt.py | 4 +- tests/test_traceparse.py | 2 +- tests/test_traceroute_cascade.py | 7 +++- 29 files changed, 195 insertions(+), 62 deletions(-) create mode 100644 tests/test_cli.py diff --git a/src/netsleuth/bgp.py b/src/netsleuth/bgp.py index ecf2be4..ad88c2e 100644 --- a/src/netsleuth/bgp.py +++ b/src/netsleuth/bgp.py @@ -1,7 +1,7 @@ from __future__ import annotations import ipaddress -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta import httpx @@ -96,7 +96,7 @@ async def ripestat( ) -> dict: params: dict[str, str] = {"resource": resource, **extra} if call in BULK_CALLS: - start = datetime.now(timezone.utc) - timedelta(days=providers.ripestat_timeframe_days) + start = datetime.now(UTC) - timedelta(days=providers.ripestat_timeframe_days) params["max_rows"] = str(providers.ripestat_max_rows) params["starttime"] = start.strftime("%Y-%m-%dT%H:%M:%S") response = await client.get(f"{providers.ripestat_base_url}/{call}/data.json", params=params) @@ -107,9 +107,9 @@ async def ripestat( import json import os import time -from pathlib import Path +from collections.abc import Awaitable, Callable from dataclasses import dataclass -from typing import Awaitable, Callable +from pathlib import Path from netsleuth.models import IxpPresence diff --git a/src/netsleuth/cli.py b/src/netsleuth/cli.py index 7b7a8b2..77ef25c 100644 --- a/src/netsleuth/cli.py +++ b/src/netsleuth/cli.py @@ -8,7 +8,7 @@ import sys import uuid from dataclasses import dataclass from pathlib import Path -from typing import Any, List, Optional, Tuple +from typing import Any import httpx import psutil @@ -18,9 +18,15 @@ from rich.console import Console from netsleuth import __version__, history from netsleuth.compare import diff_reports, load_report, render_diff, render_diff_brief from netsleuth.config import FORMAT_EXTENSIONS, Settings, load_settings -from netsleuth.exporter import build_report, dump_json, egress_asn, render_markdown, report_filename, write_report -from netsleuth.iface import available_interfaces_hint, resolve_bind_target -from netsleuth.metrics import collect_metrics, render_csv, render_prometheus +from netsleuth.exporter import ( + build_report, + dump_json, + egress_asn, + render_markdown, + report_filename, + write_report, +) +from netsleuth.iface import resolve_bind_target from netsleuth.interpret import ( assess_vpn, captive_portal_findings, @@ -42,6 +48,7 @@ from netsleuth.interpret import ( tls_findings, ) from netsleuth.ip_geo import dual_stack_mismatch, gather_identity +from netsleuth.metrics import collect_metrics, render_csv, render_prometheus from netsleuth.models import ( BindTarget, CaptivePortal, @@ -53,7 +60,11 @@ from netsleuth.models import ( SpeedResult, VpnContext, ) -from netsleuth.netinfo import collect_local_net, detect_capabilities, iface_for_ip, is_tunnel_iface, primary_interface_ip +from netsleuth.netinfo import ( + collect_local_net, + detect_capabilities, + is_tunnel_iface, +) from netsleuth.orchestration import gather_modules, run_module, utc_now_iso from netsleuth.probes.captive_portal import check_captive_portal from netsleuth.probes.dns_leak import collect_dns_leak @@ -96,7 +107,7 @@ class Options: ecmp: bool = False pmtud: bool = False quic: bool = False - ipv6: Optional[bool] = None + ipv6: bool | None = None dpi_target: str | None = None formats: frozenset[str] = frozenset({"md"}) interface: str | None = None @@ -246,8 +257,8 @@ async def _bgp_section(client, settings: Settings, asn: str | None, raw: dict): async def _reputation_section(client, settings: Settings, options: Options, geo: IpGeo, raw: dict): from netsleuth.reputation import ( - build_reputation, ReputationContext, + build_reputation, fetch_abuseipdb, fetch_internetdb, normalize_internetdb, @@ -877,19 +888,19 @@ def _check_ndt7_consent(settings: Settings) -> bool: def run( full: bool = typer.Option(False, "--full", help="Full run: speedtest and full MTR cycles."), quick: bool = typer.Option(False, "--quick", help="Express run: reference hosts only, no speedtest."), - target: Optional[str] = typer.Option(None, "--target", help="Investigate AS, an IP or a domain."), - target_host: Optional[str] = typer.Option(None, "--target-host", help="Extra host for ping/traceroute."), - speedtest_server: Optional[str] = typer.Option(None, "--speedtest-server", help="Pin the speedtest server."), + target: str | None = typer.Option(None, "--target", help="Investigate AS, an IP or a domain."), + target_host: str | None = typer.Option(None, "--target-host", help="Extra host for ping/traceroute."), + speedtest_server: str | None = typer.Option(None, "--speedtest-server", help="Pin the speedtest server."), watch: bool = typer.Option(False, "--watch", help="Continuous monitoring with a live dashboard."), - webhook: Optional[str] = typer.Option( + webhook: str | None = typer.Option( None, "--webhook", help="With --watch, POST a JSON alert to this URL on a verdict transition (overrides watch.webhook_url).", ), - compare: Optional[Tuple[Path, Path]] = typer.Option( + compare: tuple[Path, Path] | None = typer.Option( None, "--compare", help="Diff two saved JSON reports — reads reports produced with --format json." ), - trend: Optional[int] = typer.Option( + trend: int | None = typer.Option( None, "--trend", help="Show a sparkline trend across the last N saved JSON reports for this network. Read-only, no probing.", @@ -917,7 +928,7 @@ def run( "--ecmp", help="Trace a handful of reference hosts multiple times to detect per-hop load balancing (ECMP).", ), - ipv6: Optional[bool] = typer.Option( + ipv6: bool | None = typer.Option( None, "--ipv6/--no-ipv6", help="Also measure latency to IPv6 reference hosts (default: auto, only on a dual-stack connection).", @@ -932,12 +943,12 @@ def run( "--quic", help="Measure QUIC/HTTP3 handshake time and check whether UDP/443 is blocked (needs the quic extra).", ), - my_server: Optional[str] = typer.Option( + my_server: str | None = typer.Option( None, "--my-server", help="Check a server YOU OWN for TCP port blocking / RST injection. Single host only — never a CIDR or someone else's server.", ), - format_: List[str] = typer.Option( + format_: list[str] = typer.Option( [], "--format", help=( @@ -948,7 +959,7 @@ def run( ), ru: bool = typer.Option(False, "--ru", help="Shorthand for --format ru-md."), json_flag: bool = typer.Option(False, "--json", help="Shorthand for --format json."), - interface: Optional[str] = typer.Option( + interface: str | None = typer.Option( None, "--interface", help="Force outbound probing through this adapter (name or IP), instead of the OS default route.", diff --git a/src/netsleuth/interpret.py b/src/netsleuth/interpret.py index 3664c6b..be2438c 100644 --- a/src/netsleuth/interpret.py +++ b/src/netsleuth/interpret.py @@ -1,22 +1,19 @@ from __future__ import annotations - from netsleuth.config import Band, BufferbloatBands, Thresholds, VpnBands from netsleuth.models import ( CaptivePortal, - CfTrace, DnsAdvanced, DnsLeak, DpiCheckResult, EcmpReport, Finding, - IpGeo, LocalNet, PathDiversity, PingResult, PmtuResult, - QuicResult, PrefixBenchmark, + QuicResult, Signal, SpeedResult, TlsResult, diff --git a/src/netsleuth/orchestration.py b/src/netsleuth/orchestration.py index cc2a517..7259a2c 100644 --- a/src/netsleuth/orchestration.py +++ b/src/netsleuth/orchestration.py @@ -2,8 +2,9 @@ from __future__ import annotations import asyncio import time -from datetime import datetime, timezone -from typing import Any, Awaitable +from collections.abc import Awaitable +from datetime import UTC, datetime +from typing import Any import httpx @@ -14,7 +15,7 @@ _RETRYABLE_STATUS_FLOOR = 500 def utc_now_iso() -> str: - return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") def classify_exception(exc: BaseException) -> tuple[str, bool]: diff --git a/src/netsleuth/probes/latency.py b/src/netsleuth/probes/latency.py index f52e201..1927d24 100644 --- a/src/netsleuth/probes/latency.py +++ b/src/netsleuth/probes/latency.py @@ -47,7 +47,7 @@ async def tcp_connect_rtt( _reader, writer = await asyncio.wait_for( asyncio.open_connection(host, port, local_addr=local_addr), timeout=timeout ) - except (OSError, asyncio.TimeoutError): + except (TimeoutError, OSError): return None writer.close() try: diff --git a/src/netsleuth/probes/pmtud.py b/src/netsleuth/probes/pmtud.py index 4bbf476..2c572db 100644 --- a/src/netsleuth/probes/pmtud.py +++ b/src/netsleuth/probes/pmtud.py @@ -90,7 +90,7 @@ async def _probe_unix(binary: str, host: str, size: int, timeout: float, os_name ) try: stdout, _ = await asyncio.wait_for(process.communicate(), timeout=timeout + 2.0) - except asyncio.TimeoutError: + except TimeoutError: process.kill() return False, False if process.returncode == 0: diff --git a/src/netsleuth/probes/tls_rtt.py b/src/netsleuth/probes/tls_rtt.py index b65cae8..cb03fbe 100644 --- a/src/netsleuth/probes/tls_rtt.py +++ b/src/netsleuth/probes/tls_rtt.py @@ -4,7 +4,7 @@ import asyncio import hashlib import ssl import time -from datetime import datetime, timezone +from datetime import UTC, datetime from netsleuth.models import TlsResult from netsleuth.probes.latency import tcp_connect_rtt @@ -59,8 +59,8 @@ def days_remaining(not_after: str | None, now: datetime | None = None) -> int | expires_ts = ssl.cert_time_to_seconds(not_after) except ValueError: return None - expires = datetime.fromtimestamp(expires_ts, tz=timezone.utc) - return (expires - (now or datetime.now(timezone.utc))).days + expires = datetime.fromtimestamp(expires_ts, tz=UTC) + return (expires - (now or datetime.now(UTC))).days async def measure_tls( diff --git a/src/netsleuth/probes/traceroute.py b/src/netsleuth/probes/traceroute.py index 4c19c67..b490202 100644 --- a/src/netsleuth/probes/traceroute.py +++ b/src/netsleuth/probes/traceroute.py @@ -6,7 +6,8 @@ import json import platform import shutil import threading -from typing import Awaitable, Callable, TypeVar +from collections.abc import Awaitable, Callable +from typing import TypeVar from netsleuth.models import Capabilities, TraceConfig, TraceHop, TraceResult from netsleuth.probes.icmp_win import IcmpReply, classify_status @@ -15,7 +16,7 @@ from netsleuth.traceparse import build_trace_result, finalize_hop _T = TypeVar("_T") -def _run_in_daemon_thread(func: Callable[..., _T], *args: object) -> "asyncio.Future[_T]": +def _run_in_daemon_thread(func: Callable[..., _T], *args: object) -> asyncio.Future[_T]: # asyncio.to_thread()/loop.run_in_executor(None, ...) queue work onto the # ThreadPoolExecutor-backed default executor, whose worker threads are non-daemon # and get joined by concurrent.futures' atexit handler. If the underlying call @@ -116,7 +117,7 @@ async def _run(args: list[str], timeout: float) -> str: ) try: stdout, _ = await asyncio.wait_for(process.communicate(), timeout=timeout) - except asyncio.TimeoutError: + except TimeoutError: process.kill() raise return stdout.decode(_console_encoding(), errors="replace") diff --git a/src/netsleuth/speed.py b/src/netsleuth/speed.py index 9a5a618..77f42f0 100644 --- a/src/netsleuth/speed.py +++ b/src/netsleuth/speed.py @@ -60,7 +60,7 @@ def bufferbloat_delta(idle_rtt_ms: float | None, loaded_rtts_ms: list[float]) -> import asyncio import time -from typing import Awaitable, Callable +from collections.abc import Awaitable, Callable from netsleuth.models import SpeedResult, TierAttempt @@ -223,7 +223,7 @@ async def tier_ndt7( while time.perf_counter() - began < 10: try: message = await asyncio.wait_for(socket.recv(), timeout=timeout) - except (asyncio.TimeoutError, Exception): + except (TimeoutError, Exception): break total += len(message) if isinstance(message, (bytes, bytearray)) else len(message.encode()) elapsed = time.perf_counter() - began diff --git a/src/netsleuth/watch.py b/src/netsleuth/watch.py index 96b3c7a..8d88abf 100644 --- a/src/netsleuth/watch.py +++ b/src/netsleuth/watch.py @@ -14,7 +14,14 @@ from rich.table import Table from netsleuth import __version__ from netsleuth.alerting import build_payload, post_webhook, should_fire from netsleuth.config import Settings, Thresholds -from netsleuth.exporter import atomic_write, dump_json, readable_timestamp, report_date_dir, sanitize_name, sparkline +from netsleuth.exporter import ( + atomic_write, + dump_json, + readable_timestamp, + report_date_dir, + sanitize_name, + sparkline, +) from netsleuth.interpret import latency_findings, overall_verdict, speed_findings from netsleuth.ip_geo import gather_identity from netsleuth.models import PingResult, SpeedResult, to_jsonable diff --git a/tests/test_alerting.py b/tests/test_alerting.py index f71bc8e..d689d7c 100644 --- a/tests/test_alerting.py +++ b/tests/test_alerting.py @@ -1,7 +1,6 @@ from __future__ import annotations import httpx -import pytest from netsleuth.alerting import build_payload, post_webhook, should_fire from netsleuth.exporter import dump_json diff --git a/tests/test_captive_portal.py b/tests/test_captive_portal.py index 140a31c..804f086 100644 --- a/tests/test_captive_portal.py +++ b/tests/test_captive_portal.py @@ -4,7 +4,10 @@ import httpx from netsleuth.interpret import captive_portal_findings from netsleuth.models import CaptivePortal -from netsleuth.probes.captive_portal import check_captive_portal, classify_portal_response +from netsleuth.probes.captive_portal import ( + check_captive_portal, + classify_portal_response, +) URL = "http://cp.cloudflare.com/generate_204" diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..3876af1 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,62 @@ +import os +from unittest.mock import MagicMock, patch + +from netsleuth.cli import _os_timezone + + +def test_os_timezone_tz_env(): + with patch.dict(os.environ, {"TZ": "Europe/Berlin"}): + assert _os_timezone() == "Europe/Berlin" + +def test_os_timezone_localtime_symlink(): + with patch.dict(os.environ, {}, clear=True): + with patch("netsleuth.cli.Path") as MockPath: + mock_localtime = MagicMock() + mock_localtime.is_symlink.return_value = True + mock_localtime.readlink.return_value = "/usr/share/zoneinfo/America/New_York" + + # Make Path("/etc/localtime") return mock_localtime + MockPath.return_value = mock_localtime + + assert _os_timezone() == "America/New_York" + MockPath.assert_called_once_with("/etc/localtime") + +def test_os_timezone_timezone_file(): + with patch.dict(os.environ, {}, clear=True): + with patch("netsleuth.cli.Path") as MockPath: + mock_localtime = MagicMock() + mock_localtime.is_symlink.return_value = False + + mock_timezone = MagicMock() + mock_timezone.read_text.return_value = "Asia/Tokyo\n" + + def side_effect(path): + if path == "/etc/localtime": + return mock_localtime + elif path == "/etc/timezone": + return mock_timezone + return MagicMock() + + MockPath.side_effect = side_effect + + assert _os_timezone() == "Asia/Tokyo" + +def test_os_timezone_oserror(): + with patch.dict(os.environ, {}, clear=True): + with patch("netsleuth.cli.Path") as MockPath: + mock_localtime = MagicMock() + mock_localtime.is_symlink.return_value = False + + mock_timezone = MagicMock() + mock_timezone.read_text.side_effect = OSError("Permission denied") + + def side_effect(path): + if path == "/etc/localtime": + return mock_localtime + elif path == "/etc/timezone": + return mock_timezone + return MagicMock() + + MockPath.side_effect = side_effect + + assert _os_timezone() is None diff --git a/tests/test_exporter.py b/tests/test_exporter.py index 3bc2e06..649fd58 100644 --- a/tests/test_exporter.py +++ b/tests/test_exporter.py @@ -279,15 +279,22 @@ def test_format_extensions_covers_the_known_formats(): assert FORMAT_EXTENSIONS == {"md": "md", "ru-md": "ru.md", "json": "json", "prom": "prom", "csv": "csv"} -from netsleuth.exporter import SPARK_CHARS, _forced_interface_rows, badge, first_loss_jump, render_markdown, sparkline +from netsleuth.exporter import ( + SPARK_CHARS, + _forced_interface_rows, + badge, + first_loss_jump, + render_markdown, + sparkline, +) from netsleuth.models import ( AdapterLeakResult, AnycastHop, BgpIntel, CfTrace, DnsAdvanced, - DnsLeak, DnsblHit, + DnsLeak, DpiCheckResult, InternetDbResult, IxpPresence, diff --git a/tests/test_hop_asn.py b/tests/test_hop_asn.py index 54e6937..dc6eb76 100644 --- a/tests/test_hop_asn.py +++ b/tests/test_hop_asn.py @@ -2,7 +2,6 @@ from __future__ import annotations import pytest -from netsleuth.config import Thresholds from netsleuth.interpret import path_asn_findings from netsleuth.models import TraceHop, TraceResult from netsleuth.probes.hop_asn import cymru_query_name, enrich_hops, parse_cymru_asname diff --git a/tests/test_iface.py b/tests/test_iface.py index 1474b2b..d3dbf57 100644 --- a/tests/test_iface.py +++ b/tests/test_iface.py @@ -2,7 +2,11 @@ from __future__ import annotations import socket -from netsleuth.iface import available_interfaces_hint, resolve_bind_target, usable_addresses +from netsleuth.iface import ( + available_interfaces_hint, + resolve_bind_target, + usable_addresses, +) def addrs(**kw) -> dict[str, list[tuple[int, str]]]: diff --git a/tests/test_interpret.py b/tests/test_interpret.py index 8d3485d..d49520a 100644 --- a/tests/test_interpret.py +++ b/tests/test_interpret.py @@ -3,8 +3,13 @@ from __future__ import annotations import pytest from netsleuth.config import Band, Thresholds +from netsleuth.interpret import ( + dual_family_findings, + latency_findings, + path_findings, + severity_for, +) from netsleuth.models import PingResult, TraceHop, TraceResult -from netsleuth.interpret import dual_family_findings, latency_findings, path_findings, severity_for @pytest.fixture() @@ -217,8 +222,21 @@ def test_path_findings_flag_loss_that_persists_all_the_way_to_the_last_hop(): from netsleuth.config import VpnBands -from netsleuth.models import AdapterLeakResult, CfTrace, DnsLeak, IpGeo, LocalNet, Signal, VpnContext -from netsleuth.interpret import SIGNAL_WEIGHTS, assess_vpn, gather_vpn_signals, score_vpn +from netsleuth.interpret import ( + SIGNAL_WEIGHTS, + assess_vpn, + gather_vpn_signals, + score_vpn, +) +from netsleuth.models import ( + AdapterLeakResult, + CfTrace, + DnsLeak, + IpGeo, + LocalNet, + Signal, + VpnContext, +) def sig(name: str, observed: bool = True) -> Signal: @@ -367,8 +385,13 @@ def test_assess_vpn_returns_a_complete_assessment(): from netsleuth.config import BufferbloatBands +from netsleuth.interpret import ( + bufferbloat_consequence, + grade_bufferbloat, + overall_verdict, + speed_findings, +) from netsleuth.models import Finding, SpeedResult -from netsleuth.interpret import bufferbloat_consequence, grade_bufferbloat, overall_verdict, speed_findings def test_bufferbloat_grades_follow_the_configured_bands(): diff --git a/tests/test_interpret_l7.py b/tests/test_interpret_l7.py index 6c6e21b..7e8cc81 100644 --- a/tests/test_interpret_l7.py +++ b/tests/test_interpret_l7.py @@ -17,7 +17,6 @@ from netsleuth.models import ( DnsAdvanced, DpiCheckResult, PathDiversity, - PortProbe, PrefixBenchmark, PrefixProbe, ResolverProbe, diff --git a/tests/test_ip_geo_normalize.py b/tests/test_ip_geo_normalize.py index b00ad90..7fda6a9 100644 --- a/tests/test_ip_geo_normalize.py +++ b/tests/test_ip_geo_normalize.py @@ -140,8 +140,8 @@ def test_ip_type_classification(mobile, proxy, hosting, known, expected): assert classify_ip_type(mobile, proxy, hosting, known) == expected -from netsleuth.models import IpGeo from netsleuth.ip_geo import dual_stack_mismatch, merge_geo +from netsleuth.models import IpGeo def test_merge_takes_the_first_non_empty_value_in_priority_order(): diff --git a/tests/test_models.py b/tests/test_models.py index 017a400..e04b2cb 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -13,8 +13,8 @@ from netsleuth.models import ( CfL4Stats, CfTrace, DnsAdvanced, - DnsLeak, DnsblHit, + DnsLeak, DpiCheckResult, Finding, InternetDbResult, diff --git a/tests/test_netinfo.py b/tests/test_netinfo.py index db14619..21969b6 100644 --- a/tests/test_netinfo.py +++ b/tests/test_netinfo.py @@ -1,7 +1,11 @@ from __future__ import annotations from netsleuth.models import Capabilities -from netsleuth.netinfo import choose_latency_backend, choose_trace_backend, degradation_note +from netsleuth.netinfo import ( + choose_latency_backend, + choose_trace_backend, + degradation_note, +) def caps(**kw) -> Capabilities: diff --git a/tests/test_orchestration.py b/tests/test_orchestration.py index 8cf95a3..39f229d 100644 --- a/tests/test_orchestration.py +++ b/tests/test_orchestration.py @@ -10,7 +10,7 @@ from netsleuth.orchestration import classify_exception, gather_modules, run_modu def test_classify_timeout_errors(): - assert classify_exception(asyncio.TimeoutError()) == ("timeout", True) + assert classify_exception(TimeoutError()) == ("timeout", True) assert classify_exception(httpx.ConnectTimeout("slow")) == ("timeout", True) assert classify_exception(TimeoutError()) == ("timeout", True) diff --git a/tests/test_pmtud.py b/tests/test_pmtud.py index 23d8987..afe2e72 100644 --- a/tests/test_pmtud.py +++ b/tests/test_pmtud.py @@ -2,7 +2,12 @@ from __future__ import annotations from netsleuth.interpret import pmtud_findings from netsleuth.models import PmtuResult -from netsleuth.probes.pmtud import classify_pmtu, mtu_from_search, next_probe_size, unix_ping_df_argv +from netsleuth.probes.pmtud import ( + classify_pmtu, + mtu_from_search, + next_probe_size, + unix_ping_df_argv, +) def test_next_probe_size_bisects(): diff --git a/tests/test_reputation.py b/tests/test_reputation.py index 9962d8e..3760918 100644 --- a/tests/test_reputation.py +++ b/tests/test_reputation.py @@ -13,8 +13,8 @@ from netsleuth.config import Providers from netsleuth.reputation import ( DnsblOutcome, NetsetIndex, - build_reputation, ReputationContext, + build_reputation, captcha_risk, decode_dnsbl, fetch_abuseipdb, diff --git a/tests/test_speed_cloudflare.py b/tests/test_speed_cloudflare.py index 3328315..58336a0 100644 --- a/tests/test_speed_cloudflare.py +++ b/tests/test_speed_cloudflare.py @@ -1,5 +1,5 @@ -import pytest import httpx +import pytest from pytest_httpx import HTTPXMock from netsleuth.config import Speedtest diff --git a/tests/test_speed_math.py b/tests/test_speed_math.py index a6e7c4f..4d04053 100644 --- a/tests/test_speed_math.py +++ b/tests/test_speed_math.py @@ -6,7 +6,13 @@ from unittest.mock import AsyncMock, patch import pytest -from netsleuth.speed import bufferbloat_delta, mbps, ookla_interface_args, parse_server_timing_cfl4, throughput_from_samples +from netsleuth.speed import ( + bufferbloat_delta, + mbps, + ookla_interface_args, + parse_server_timing_cfl4, + throughput_from_samples, +) def test_mbps_converts_bytes_and_seconds(): diff --git a/tests/test_tls_rtt.py b/tests/test_tls_rtt.py index 610e3ba..5ef6ae7 100644 --- a/tests/test_tls_rtt.py +++ b/tests/test_tls_rtt.py @@ -2,7 +2,7 @@ from __future__ import annotations import socket import ssl -from datetime import datetime, timezone +from datetime import UTC, datetime from netsleuth.probes.tls_rtt import ( cert_name, @@ -97,7 +97,7 @@ def test_cert_name_handles_empty_input(): def test_days_remaining_computes_from_a_fixed_now(): - now = datetime(2026, 1, 1, tzinfo=timezone.utc) + now = datetime(2026, 1, 1, tzinfo=UTC) assert days_remaining("Jan 11 00:00:00 2026 GMT", now=now) == 10 diff --git a/tests/test_traceparse.py b/tests/test_traceparse.py index 888571e..136a894 100644 --- a/tests/test_traceparse.py +++ b/tests/test_traceparse.py @@ -3,7 +3,7 @@ from __future__ import annotations import pytest from netsleuth.models import TraceHop -from netsleuth.traceparse import parse_linux, finalize_hop +from netsleuth.traceparse import finalize_hop, parse_linux @pytest.fixture() diff --git a/tests/test_traceroute_cascade.py b/tests/test_traceroute_cascade.py index d43a388..f6fd693 100644 --- a/tests/test_traceroute_cascade.py +++ b/tests/test_traceroute_cascade.py @@ -6,7 +6,12 @@ import threading import pytest from netsleuth.models import Capabilities, TraceHop, TraceResult -from netsleuth.probes.icmp_win import IP_REQ_TIMED_OUT, IP_SUCCESS, IP_TTL_EXPIRED_TRANSIT, IcmpReply +from netsleuth.probes.icmp_win import ( + IP_REQ_TIMED_OUT, + IP_SUCCESS, + IP_TTL_EXPIRED_TRANSIT, + IcmpReply, +) from netsleuth.probes.traceroute import ( _run_in_daemon_thread, filter_trace_tiers,