Add tests for _os_timezone function (#22)

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: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

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
 

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<n>, 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<n>, 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.",

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,

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]:

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:

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:

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(

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")

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

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

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

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"
 

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

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,

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

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]]]:

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():

tests/test_interpret_l7.py

@@ -17,7 +17,6 @@ from netsleuth.models import (
     DnsAdvanced,
     DpiCheckResult,
     PathDiversity,
-    PortProbe,
     PrefixBenchmark,
     PrefixProbe,
     ResolverProbe,

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():

tests/test_models.py

@@ -13,8 +13,8 @@ from netsleuth.models import (
     CfL4Stats,
     CfTrace,
     DnsAdvanced,
-    DnsLeak,
     DnsblHit,
+    DnsLeak,
     DpiCheckResult,
     Finding,
     InternetDbResult,

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:

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)
 

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():

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,

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

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():

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
 
 

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()

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,