mirror of
https://github.com/Horizontal42/netsleuth.git
synced 2026-11-10 15:42:10 +03:00
exporter: markdown rendering with sparklines and hop bars
This commit is contained in:
Andrew Pimenov
parent
71b1fc95e3
commit
879b974e57
4 changed files with 646 additions and 2 deletions
|
|
@ -20,7 +20,8 @@ src/netcheck/
|
|||
traceparse.py Pure text -> TraceHop parsers for Windows, Linux and BSD/macOS output.
|
||||
interpret.py Pure verdict engine: thresholds -> Finding[], VPN scoring, bufferbloat grade.
|
||||
speed.py Cascading speedtest, throughput math, cfL4 header parsing, bufferbloat probe.
|
||||
exporter.py JSON and Markdown rendering, atomic writes into ./logs/.
|
||||
exporter.py JSON and Markdown rendering, atomic writes into ./logs/. Deliberately over
|
||||
the file-size guideline to keep both output formats in one place.
|
||||
compare.py --compare: diff two saved JSON reports.
|
||||
watch.py --watch: periodic re-run loop with a live Rich dashboard.
|
||||
probes/
|
||||
|
|
|
|||
|
|
@ -20,7 +20,8 @@ src/netcheck/
|
|||
traceparse.py Чистые парсеры текста в TraceHop для Windows, Linux и BSD/macOS.
|
||||
interpret.py Чистый движок вердиктов: пороги -> Finding[], скоринг VPN, оценка bufferbloat.
|
||||
speed.py Каскадный замер скорости, математика пропускной способности, разбор cfL4, bufferbloat.
|
||||
exporter.py Рендер JSON и Markdown, атомарная запись в ./logs/.
|
||||
exporter.py Рендер JSON и Markdown, атомарная запись в ./logs/. Намеренно превышает
|
||||
лимит размера файла, чтобы оба формата вывода жили в одном месте.
|
||||
compare.py --compare: сравнение двух сохранённых JSON-отчётов.
|
||||
watch.py --watch: цикл периодических прогонов с живой панелью Rich.
|
||||
probes/
|
||||
|
|
|
|||
|
|
@ -95,3 +95,334 @@ def write_report(report: dict[str, Any], markdown: str, logs_dir: Path) -> tuple
|
|||
json_path = atomic_write(base / report_filename(asn, started, "json"), dump_json(report))
|
||||
md_path = atomic_write(base / report_filename(asn, started, "md"), markdown)
|
||||
return json_path, md_path
|
||||
|
||||
|
||||
from netcheck.interpret import bufferbloat_consequence
|
||||
|
||||
SPARK_CHARS = "▁▂▃▅▇"
|
||||
_BADGES = {"ok": "🟢", "info": "🟢", "warn": "🟡", "crit": "🔴"}
|
||||
_SEVERITY_RANK = {"crit": 0, "warn": 1, "info": 2, "ok": 3}
|
||||
_LOSS_JUMP_PCT = 20.0
|
||||
|
||||
|
||||
def sparkline(values: list[float | None]) -> str:
|
||||
numbers = [v for v in values if v is not None]
|
||||
if not numbers:
|
||||
return ""
|
||||
low, high = min(numbers), max(numbers)
|
||||
span = high - low
|
||||
out: list[str] = []
|
||||
for value in values:
|
||||
if value is None:
|
||||
out.append(" ")
|
||||
elif span <= 0:
|
||||
out.append(SPARK_CHARS[0])
|
||||
else:
|
||||
index = int((value - low) / span * (len(SPARK_CHARS) - 1))
|
||||
out.append(SPARK_CHARS[index])
|
||||
return "".join(out)
|
||||
|
||||
|
||||
def badge(severity: str, emoji: bool) -> str:
|
||||
return _BADGES.get(severity, "⚪") if emoji else f"[{severity}]"
|
||||
|
||||
|
||||
def first_loss_jump(hops: list[dict]) -> int | None:
|
||||
for index, hop in enumerate(hops[:-1]):
|
||||
following = hops[index + 1]
|
||||
if hop.get("loss_pct", 0.0) >= _LOSS_JUMP_PCT and following.get("loss_pct", 0.0) >= _LOSS_JUMP_PCT:
|
||||
return hop.get("ttl")
|
||||
return None
|
||||
|
||||
|
||||
def _module(report: dict, section: str) -> dict:
|
||||
return report.get(section) or {}
|
||||
|
||||
|
||||
def _data(report: dict, section: str) -> Any:
|
||||
return _module(report, section).get("data")
|
||||
|
||||
|
||||
def _num(value: Any) -> str:
|
||||
return "—" if value is None else f"{value:g}" if isinstance(value, (int, float)) else str(value)
|
||||
|
||||
|
||||
def _table(headers: list[str], rows: list[list[str]]) -> list[str]:
|
||||
if not rows:
|
||||
return ["_No rows._"]
|
||||
return (
|
||||
["| " + " | ".join(headers) + " |", "|" + "|".join(["---"] * len(headers)) + "|"]
|
||||
+ ["| " + " | ".join(row) + " |" for row in rows]
|
||||
)
|
||||
|
||||
|
||||
def _header(report: dict, emoji: bool) -> list[str]:
|
||||
meta = report.get("meta") or {}
|
||||
interp = report.get("interpretation") or {}
|
||||
target = f" · target `{meta.get('target')}`" if meta.get("target") else ""
|
||||
status = interp.get("overall_status", "ok")
|
||||
return [
|
||||
"# netcheck report",
|
||||
"",
|
||||
f"- Mode: `{meta.get('mode', 'auto')}`{target}",
|
||||
f"- Started: {meta.get('started_at', '?')} · finished: {meta.get('finished_at', '?')}",
|
||||
f"- Host OS: {meta.get('host_os', '?')} · run id `{meta.get('run_id', '?')}`",
|
||||
f"- Verdict: {badge(status, emoji)} **{status}** ({interp.get('overall_score', 0)}/100) — "
|
||||
f"{interp.get('summary_text', '')}",
|
||||
"",
|
||||
]
|
||||
|
||||
|
||||
def _findings(report: dict) -> list[dict]:
|
||||
return sorted(
|
||||
(report.get("interpretation") or {}).get("findings") or [],
|
||||
key=lambda f: _SEVERITY_RANK.get(f.get("severity"), 9),
|
||||
)
|
||||
|
||||
|
||||
def _tldr(report: dict, emoji: bool) -> list[str]:
|
||||
found = _findings(report)
|
||||
if not found:
|
||||
return ["## TL;DR", "", "No problems found on this connection.", ""]
|
||||
return ["## TL;DR", ""] + [
|
||||
f"- {badge(f.get('severity', 'info'), emoji)} **{f.get('title', '')}** — {f.get('detail', '')}"
|
||||
for f in found[:5]
|
||||
] + [""]
|
||||
|
||||
|
||||
def _connection(report: dict) -> list[str]:
|
||||
local = _data(report, "connection") or {}
|
||||
geo_bundle = _data(report, "ip_geo") or {}
|
||||
geo = geo_bundle.get("egress_v4") or {}
|
||||
v6 = geo_bundle.get("egress_v6") or {}
|
||||
cf = geo_bundle.get("cf_trace") or {}
|
||||
rows = [
|
||||
["Interface", f"{local.get('iface_name') or '?'} (MTU {local.get('iface_mtu') or '?'})"],
|
||||
["Local address", f"{local.get('local_ipv4') or '?'} / {local.get('local_ipv6') or '—'}"],
|
||||
["Default gateway", local.get("default_gateway_v4") or "?"],
|
||||
["Egress IPv4", f"{geo.get('ip') or '?'} ({geo.get('asn') or '?'} {geo.get('as_name') or geo.get('org') or ''})"],
|
||||
["Egress IPv6", f"{v6.get('ip') or '—'} ({v6.get('asn') or '—'})"],
|
||||
["Reverse DNS", geo.get("reverse_dns") or "—"],
|
||||
["Location", f"{geo.get('city') or '?'}, {geo.get('country') or geo.get('country_code') or '?'}"],
|
||||
["Address type", geo.get("ip_type") or "unknown"],
|
||||
["Cloudflare colo", cf.get("colo") or "—"],
|
||||
]
|
||||
lines = ["## Connection & identity", ""] + _table(["Field", "Value"], rows) + [""]
|
||||
if geo_bundle.get("dual_stack_note"):
|
||||
lines += [f"> {geo_bundle['dual_stack_note']}", ""]
|
||||
return lines
|
||||
|
||||
|
||||
def _vpn(report: dict) -> list[str]:
|
||||
vpn = _data(report, "vpn_assessment") or {}
|
||||
lines = [
|
||||
"## VPN / proxy assessment",
|
||||
"",
|
||||
f"Verdict: **{vpn.get('verdict', 'unknown')}** · confidence {vpn.get('confidence', 0)}"
|
||||
f" · tunnel interface {vpn.get('tunnel_iface') or '—'}",
|
||||
"",
|
||||
]
|
||||
lines += _table(
|
||||
["Signal", "Observed", "Weight", "Direction", "Note"],
|
||||
[
|
||||
[
|
||||
s.get("name", ""),
|
||||
"yes" if s.get("observed") else "no",
|
||||
_num(s.get("weight")),
|
||||
s.get("direction", ""),
|
||||
s.get("note") or "—",
|
||||
]
|
||||
for s in vpn.get("signals") or []
|
||||
],
|
||||
)
|
||||
leak = vpn.get("dns_leak") or {}
|
||||
if leak:
|
||||
lines += ["", "### DNS leak", ""]
|
||||
lines += _table(
|
||||
["Adapter", "Configured resolvers", "Echoed IP", "Echoed ASN", "Same ASN as egress"],
|
||||
[
|
||||
[
|
||||
a.get("adapter", ""),
|
||||
", ".join(a.get("configured_resolvers") or []) or "—",
|
||||
a.get("echoed_ip") or "—",
|
||||
a.get("echoed_asn") or "—",
|
||||
{True: "yes", False: "no", None: "?"}[a.get("matches_egress_asn")],
|
||||
]
|
||||
for a in leak.get("per_adapter") or []
|
||||
],
|
||||
)
|
||||
lines += ["", f"EDNS Client Subnet leaked: {'yes' if leak.get('ecs_leaked') else 'no'}", "", leak.get("note", "")]
|
||||
return lines + [""]
|
||||
|
||||
|
||||
def _bgp(report: dict) -> list[str]:
|
||||
bgp = _data(report, "bgp") or {}
|
||||
rows = [
|
||||
["ASN", f"{bgp.get('asn') or '?'} — {bgp.get('holder') or '?'}"],
|
||||
["Registry", f"{bgp.get('registry') or '?'} (allocated {bgp.get('allocated_at') or '?'})"],
|
||||
["Upstreams", ", ".join(bgp.get("upstreams") or []) or "—"],
|
||||
["Peers", ", ".join(bgp.get("peers") or []) or "—"],
|
||||
["Downstreams", ", ".join((bgp.get("downstreams") or [])[:10]) or "—"],
|
||||
["Announced prefixes", f"{bgp.get('prefix_count_v4', 0)} IPv4 / {bgp.get('prefix_count_v6', 0)} IPv6"],
|
||||
["Route stability", f"{bgp.get('stability', 'unknown')} ({len(bgp.get('flaps') or [])} updates in window)"],
|
||||
["CAIDA ASRank", f"#{_num(bgp.get('asrank'))} · cone {_num(bgp.get('cone_asns'))} ASNs / "
|
||||
f"{_num(bgp.get('cone_prefixes'))} prefixes"],
|
||||
["PeeringDB", f"{bgp.get('pdb_info_type') or '—'} · {bgp.get('pdb_traffic') or '—'}"],
|
||||
]
|
||||
lines = ["## ASN & BGP intelligence", ""] + _table(["Field", "Value"], rows)
|
||||
ixps = bgp.get("ixps") or []
|
||||
if ixps:
|
||||
lines += ["", "### IXP presence", ""]
|
||||
lines += _table(
|
||||
["Exchange", "City", "Country", "Speed (Mbps)"],
|
||||
[[i.get("name", ""), i.get("city") or "—", i.get("country") or "—", _num(i.get("speed_mbps"))] for i in ixps],
|
||||
)
|
||||
return lines + [""]
|
||||
|
||||
|
||||
def _reputation(report: dict) -> list[str]:
|
||||
rep = _data(report, "reputation") or {}
|
||||
idb = rep.get("internetdb") or {}
|
||||
rows = [
|
||||
["FireHOL hits", ", ".join(rep.get("firehol_hits") or []) or "none"],
|
||||
["Open ports (InternetDB)", ", ".join(str(p) for p in idb.get("ports") or []) or "none"],
|
||||
["Tags", ", ".join(idb.get("tags") or []) or "none"],
|
||||
["Known CVEs", ", ".join(idb.get("vulns") or []) or "none"],
|
||||
["AbuseIPDB", _num(rep.get("abuseipdb_score")) if rep.get("abuseipdb_score") is not None else "not queried"],
|
||||
]
|
||||
hits = rep.get("dnsbl_hits")
|
||||
if hits is None:
|
||||
rows.append(["DNSBL", "not run (pass --dnsbl)"])
|
||||
else:
|
||||
listed = ", ".join(h.get("zone", "") for h in hits) or "no listing"
|
||||
if rep.get("dnsbl_query_blocked"):
|
||||
listed += " · one or more zones answered with their query-error range, which is not a listing"
|
||||
rows.append(["DNSBL", listed])
|
||||
return (
|
||||
["## Reputation", "", f"Captcha/blocklist risk: **{rep.get('captcha_risk', 'unknown')}** — "
|
||||
f"{rep.get('rationale', '')}", ""]
|
||||
+ _table(["Field", "Value"], rows)
|
||||
+ [""]
|
||||
)
|
||||
|
||||
|
||||
def _latency(report: dict) -> list[str]:
|
||||
pings = _data(report, "latency") or []
|
||||
lines = ["## Latency", ""] + _table(
|
||||
["Host", "Address", "Method", "Avg ms", "Min", "Max", "Jitter", "Loss"],
|
||||
[
|
||||
[
|
||||
p.get("label", ""),
|
||||
p.get("host", ""),
|
||||
p.get("method", ""),
|
||||
_num(p.get("avg_ms")),
|
||||
_num(p.get("min_ms")),
|
||||
_num(p.get("max_ms")),
|
||||
_num(p.get("jitter_ms")),
|
||||
f"{p.get('loss_pct', 0)}%",
|
||||
]
|
||||
for p in pings
|
||||
],
|
||||
)
|
||||
if pings:
|
||||
lines += ["", "```"]
|
||||
lines += [f"{p.get('label', ''):<18}{sparkline(p.get('samples') or [])}" for p in pings]
|
||||
lines += ["```"]
|
||||
if any(p.get("method") == "tcp" for p in pings):
|
||||
lines += ["", "> Loss on a `tcp` row counts failed TCP connections, not dropped ICMP packets."]
|
||||
return lines + [""]
|
||||
|
||||
|
||||
def _bar(value: Any) -> str:
|
||||
return "#" * int(min(float(value or 0.0), 200.0) / 10)
|
||||
|
||||
|
||||
def _path(report: dict) -> list[str]:
|
||||
traces = _data(report, "path") or []
|
||||
lines = ["## Path", ""]
|
||||
for trace in traces:
|
||||
hops = trace.get("hops") or []
|
||||
jump = first_loss_jump(hops)
|
||||
lines += [
|
||||
f"### {trace.get('target') or '?'} — backend `{trace.get('backend', 'none')}`, "
|
||||
f"{'complete' if trace.get('completed') else 'incomplete'}",
|
||||
"",
|
||||
"```",
|
||||
]
|
||||
for hop in hops:
|
||||
marker = " <<" if jump is not None and hop.get("ttl") == jump else ""
|
||||
lines.append(
|
||||
f"{hop.get('ttl', 0):>3} {(hop.get('ip') or '*'):<39} "
|
||||
f"{_num(hop.get('avg_ms')):>8} ms {hop.get('loss_pct', 0):>5}% {_bar(hop.get('avg_ms'))}{marker}"
|
||||
)
|
||||
lines += ["```", ""]
|
||||
return lines
|
||||
|
||||
|
||||
def _speed(report: dict) -> list[str]:
|
||||
speed = _data(report, "speed") or {}
|
||||
grade = speed.get("bufferbloat_grade") or "?"
|
||||
rows = [
|
||||
["Method", speed.get("method", "none")],
|
||||
["Server", speed.get("server") or "—"],
|
||||
["Download", f"{_num(speed.get('download_mbps'))} Mbps"],
|
||||
["Upload", f"{_num(speed.get('upload_mbps'))} Mbps"],
|
||||
["Idle RTT", f"{_num(speed.get('idle_rtt_ms'))} ms"],
|
||||
["Loaded RTT (down / up)", f"{_num(speed.get('loaded_rtt_down_ms'))} / {_num(speed.get('loaded_rtt_up_ms'))} ms"],
|
||||
["Bufferbloat", f"grade {grade} — down +{_num(speed.get('bufferbloat_down_ms'))} ms, "
|
||||
f"up +{_num(speed.get('bufferbloat_up_ms'))} ms"],
|
||||
["Netflix OCA on-net", {True: "yes", False: "no", None: "—"}[speed.get("netflix_oca_onnet")]],
|
||||
]
|
||||
lines = ["## Speed", ""] + _table(["Field", "Value"], rows) + ["", bufferbloat_consequence(grade)]
|
||||
attempts = speed.get("tier_attempts") or []
|
||||
if attempts:
|
||||
lines += ["", "### Cascade", ""]
|
||||
lines += _table(
|
||||
["Tier", "Used", "Reason", "ms"],
|
||||
[
|
||||
[a.get("tier", ""), "yes" if a.get("ok") else "no", a.get("reason") or "—", _num(a.get("duration_ms"))]
|
||||
for a in attempts
|
||||
],
|
||||
)
|
||||
return lines + [""]
|
||||
|
||||
|
||||
def _problems(report: dict, emoji: bool) -> list[str]:
|
||||
found = _findings(report)
|
||||
if not found:
|
||||
return ["## Problems & recommendations", "", "Nothing to act on.", ""]
|
||||
lines = ["## Problems & recommendations", ""]
|
||||
for f in found:
|
||||
lines.append(f"- {badge(f.get('severity', 'info'), emoji)} **{f.get('title', '')}** — {f.get('detail', '')}")
|
||||
if f.get("advice"):
|
||||
lines.append(f" - {f['advice']}")
|
||||
return lines + [""]
|
||||
|
||||
|
||||
def _diagnostics(report: dict) -> list[str]:
|
||||
rows = []
|
||||
warnings: list[str] = []
|
||||
for section in SECTION_ORDER:
|
||||
module = _module(report, section)
|
||||
errors = "; ".join(f"{e.get('source')}: {e.get('kind')}" for e in module.get("errors") or []) or "—"
|
||||
rows.append([section, module.get("status", "skipped"), _num(module.get("duration_ms")), errors])
|
||||
warnings += list(module.get("warnings") or [])
|
||||
lines = ["## Run diagnostics", ""] + _table(["Module", "Status", "ms", "Errors"], rows)
|
||||
if warnings:
|
||||
lines += [""] + [f"- {w}" for w in warnings]
|
||||
return lines + [""]
|
||||
|
||||
|
||||
def render_markdown(report: dict[str, Any], emoji: bool = True) -> str:
|
||||
lines = _header(report, emoji)
|
||||
lines += _tldr(report, emoji)
|
||||
lines += _connection(report)
|
||||
lines += _vpn(report)
|
||||
lines += _bgp(report)
|
||||
lines += _reputation(report)
|
||||
lines += _latency(report)
|
||||
lines += _path(report)
|
||||
lines += _speed(report)
|
||||
lines += _problems(report, emoji)
|
||||
lines += _diagnostics(report)
|
||||
return "\n".join(lines).rstrip() + "\n"
|
||||
|
|
|
|||
|
|
@ -215,3 +215,314 @@ def test_write_report_emits_both_artifacts_with_matching_names(tmp_path: Path):
|
|||
assert md_path.name == "report_AS64500_20260808T191200Z.md"
|
||||
assert json.loads(json_path.read_text(encoding="utf-8"))["schema_version"] == SCHEMA_VERSION
|
||||
assert md_path.read_text(encoding="utf-8").startswith("# netcheck report")
|
||||
|
||||
|
||||
from netcheck.exporter import SPARK_CHARS, badge, first_loss_jump, render_markdown, sparkline
|
||||
from netcheck.models import (
|
||||
AdapterLeakResult,
|
||||
BgpIntel,
|
||||
CfTrace,
|
||||
DnsLeak,
|
||||
DnsblHit,
|
||||
InternetDbResult,
|
||||
IxpPresence,
|
||||
Reputation,
|
||||
Signal,
|
||||
TierAttempt,
|
||||
TraceHop,
|
||||
TraceResult,
|
||||
VpnAssessment,
|
||||
)
|
||||
|
||||
|
||||
def full_modules() -> dict[str, ModuleResult]:
|
||||
return {
|
||||
"connection": ModuleResult(
|
||||
name="connection",
|
||||
status="ok",
|
||||
data=LocalNet(
|
||||
iface_name="Wi-Fi 2",
|
||||
local_ipv4="192.168.1.34",
|
||||
iface_mtu=1500,
|
||||
default_gateway_v4="192.168.1.1",
|
||||
dns_servers_per_adapter={"Wi-Fi 2": ["192.168.1.1"]},
|
||||
),
|
||||
),
|
||||
"ip_geo": ModuleResult(
|
||||
name="ip_geo",
|
||||
status="ok",
|
||||
data={
|
||||
"egress_v4": IpGeo(
|
||||
ip="203.0.113.44",
|
||||
asn="AS64500",
|
||||
as_name="Example Telecom",
|
||||
city="Amsterdam",
|
||||
country="Netherlands",
|
||||
country_code="NL",
|
||||
ip_type="residential",
|
||||
reverse_dns="host-203-0-113-44.example.net",
|
||||
),
|
||||
"egress_v6": IpGeo(ip="2001:db8::1", asn="AS64500"),
|
||||
"cf_trace": CfTrace(ip="203.0.113.44", colo="AMS", warp="off"),
|
||||
"dual_stack_note": None,
|
||||
},
|
||||
),
|
||||
"vpn_assessment": ModuleResult(
|
||||
name="vpn_assessment",
|
||||
status="ok",
|
||||
data=VpnAssessment(
|
||||
verdict="likely",
|
||||
confidence=0.55,
|
||||
signals=[Signal(name="cf_warp", observed=True, weight=0.5, direction="vpn", note="on")],
|
||||
tunnel_iface="wg0",
|
||||
dns_leak=DnsLeak(
|
||||
per_adapter=[
|
||||
AdapterLeakResult(
|
||||
adapter="Wi-Fi 2",
|
||||
configured_resolvers=["192.168.1.1"],
|
||||
echoed_ip="203.0.113.9",
|
||||
echoed_asn="AS64501",
|
||||
matches_egress_asn=False,
|
||||
)
|
||||
],
|
||||
ecs_leaked=True,
|
||||
note="ISP resolver still active on the Wi-Fi adapter.",
|
||||
),
|
||||
),
|
||||
),
|
||||
"bgp": ModuleResult(
|
||||
name="bgp",
|
||||
status="ok",
|
||||
data=BgpIntel(
|
||||
asn="AS64500",
|
||||
holder="Example Telecom BV",
|
||||
upstreams=["AS3356"],
|
||||
prefix_count_v4=2,
|
||||
prefix_count_v6=1,
|
||||
stability="stable",
|
||||
ixps=[IxpPresence(name="AMS-IX", city="Amsterdam", country="NL", speed_mbps=100000)],
|
||||
asrank=1842,
|
||||
cone_asns=37,
|
||||
pdb_info_type="Cable/DSL/ISP",
|
||||
),
|
||||
),
|
||||
"reputation": ModuleResult(
|
||||
name="reputation",
|
||||
status="ok",
|
||||
data=Reputation(
|
||||
internetdb=InternetDbResult(ip="203.0.113.44", ports=[80, 443], tags=["cdn"]),
|
||||
firehol_hits=[],
|
||||
dnsbl_hits=[DnsblHit(zone="bl.spamcop.net", codes=["127.0.0.2"])],
|
||||
dnsbl_query_blocked=True,
|
||||
captcha_risk="high",
|
||||
rationale="listed on bl.spamcop.net",
|
||||
),
|
||||
),
|
||||
"latency": ModuleResult(
|
||||
name="latency",
|
||||
status="ok",
|
||||
data=[
|
||||
PingResult(
|
||||
label="cloudflare-dns",
|
||||
host="1.1.1.1",
|
||||
method="icmp_win",
|
||||
sent=5,
|
||||
received=5,
|
||||
avg_ms=12.4,
|
||||
min_ms=11.0,
|
||||
max_ms=15.1,
|
||||
jitter_ms=1.9,
|
||||
samples=[11.0, 12.0, None, 15.1, 12.4],
|
||||
),
|
||||
PingResult(label="github", host="github.com", method="tcp", sent=5, received=4, loss_pct=20.0, avg_ms=42.0),
|
||||
],
|
||||
duration_ms=2400,
|
||||
),
|
||||
"path": ModuleResult(
|
||||
name="path",
|
||||
status="ok",
|
||||
data=[
|
||||
TraceResult(
|
||||
target="1.1.1.1",
|
||||
backend="icmp_win",
|
||||
completed=True,
|
||||
hops=[
|
||||
TraceHop(ttl=1, ip="192.168.1.1", avg_ms=1.1, loss_pct=0.0),
|
||||
TraceHop(ttl=2, ip="10.64.0.1", avg_ms=9.0, loss_pct=60.0),
|
||||
TraceHop(ttl=3, ip="1.1.1.1", avg_ms=12.4, loss_pct=55.0),
|
||||
],
|
||||
)
|
||||
],
|
||||
),
|
||||
"speed": ModuleResult(
|
||||
name="speed",
|
||||
status="ok",
|
||||
data=SpeedResult(
|
||||
method="cloudflare",
|
||||
tier_attempts=[
|
||||
TierAttempt(tier="ookla_bin", ok=False, reason="binary not on PATH"),
|
||||
TierAttempt(tier="cloudflare", ok=True),
|
||||
],
|
||||
download_mbps=284.3,
|
||||
upload_mbps=41.7,
|
||||
server="speed.cloudflare.com",
|
||||
idle_rtt_ms=12.0,
|
||||
loaded_rtt_down_ms=48.0,
|
||||
bufferbloat_down_ms=36.0,
|
||||
bufferbloat_grade="C",
|
||||
),
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def full_report() -> dict:
|
||||
findings = [
|
||||
Finding(
|
||||
id="latency.loss.github",
|
||||
severity="warn",
|
||||
title="Loss to github.com",
|
||||
detail="20.0% connection failures over 5 probes via tcp.",
|
||||
metric="loss_pct",
|
||||
value=20.0,
|
||||
threshold=2.0,
|
||||
advice="Sustained loss on every host points at the local link.",
|
||||
)
|
||||
]
|
||||
return build_report(meta(), full_modules(), findings, {"ip-api": {"status": "success"}})
|
||||
|
||||
|
||||
def test_sparkline_maps_a_series_across_the_glyph_ramp():
|
||||
line = sparkline([1.0, 2.0, 3.0, 4.0, 5.0])
|
||||
assert len(line) == 5
|
||||
assert line[0] == SPARK_CHARS[0]
|
||||
assert line[-1] == SPARK_CHARS[-1]
|
||||
assert SPARK_CHARS == "▁▂▃▅▇"
|
||||
|
||||
|
||||
def test_sparkline_of_a_flat_series_is_all_baseline():
|
||||
assert sparkline([7.0, 7.0, 7.0]) == SPARK_CHARS[0] * 3
|
||||
|
||||
|
||||
def test_sparkline_renders_a_dropped_probe_as_a_gap():
|
||||
line = sparkline([1.0, None, 5.0])
|
||||
assert line[1] == " "
|
||||
assert len(line) == 3
|
||||
|
||||
|
||||
def test_sparkline_of_nothing_is_empty():
|
||||
assert sparkline([]) == ""
|
||||
assert sparkline([None, None]) == ""
|
||||
|
||||
|
||||
def test_badge_is_gated_behind_the_emoji_setting():
|
||||
assert badge("crit", emoji=True) == "🔴"
|
||||
assert badge("warn", emoji=True) == "🟡"
|
||||
assert badge("ok", emoji=True) == "🟢"
|
||||
assert badge("crit", emoji=False) == "[crit]"
|
||||
assert badge("warn", emoji=False) == "[warn]"
|
||||
|
||||
|
||||
def test_first_loss_jump_needs_the_loss_to_persist_downstream():
|
||||
hops = [
|
||||
{"ttl": 1, "loss_pct": 0.0},
|
||||
{"ttl": 2, "loss_pct": 60.0},
|
||||
{"ttl": 3, "loss_pct": 55.0},
|
||||
]
|
||||
assert first_loss_jump(hops) == 2
|
||||
|
||||
|
||||
def test_first_loss_jump_ignores_a_single_icmp_rate_limiting_hop():
|
||||
hops = [{"ttl": 1, "loss_pct": 0.0}, {"ttl": 2, "loss_pct": 100.0}, {"ttl": 3, "loss_pct": 0.0}]
|
||||
assert first_loss_jump(hops) is None
|
||||
assert first_loss_jump([]) is None
|
||||
|
||||
|
||||
def test_markdown_has_every_section_from_the_spec_in_order():
|
||||
text = render_markdown(full_report())
|
||||
headings = [line for line in text.splitlines() if line.startswith("## ")]
|
||||
assert headings == [
|
||||
"## TL;DR",
|
||||
"## Connection & identity",
|
||||
"## VPN / proxy assessment",
|
||||
"## ASN & BGP intelligence",
|
||||
"## Reputation",
|
||||
"## Latency",
|
||||
"## Path",
|
||||
"## Speed",
|
||||
"## Problems & recommendations",
|
||||
"## Run diagnostics",
|
||||
]
|
||||
assert text.startswith("# netcheck report")
|
||||
|
||||
|
||||
def test_markdown_header_states_mode_timestamp_and_verdict():
|
||||
text = render_markdown(full_report())
|
||||
assert "auto" in text
|
||||
assert "2026-08-08T19:12:00Z" in text
|
||||
assert "90/100" in text
|
||||
|
||||
|
||||
def test_markdown_reports_identity_and_both_stacks():
|
||||
text = render_markdown(full_report())
|
||||
assert "203.0.113.44" in text
|
||||
assert "AS64500" in text
|
||||
assert "2001:db8::1" in text
|
||||
assert "Amsterdam" in text
|
||||
assert "residential" in text
|
||||
|
||||
|
||||
def test_markdown_vpn_section_lists_signals_and_the_dns_leak():
|
||||
text = render_markdown(full_report())
|
||||
assert "likely" in text
|
||||
assert "cf_warp" in text
|
||||
assert "Wi-Fi 2" in text
|
||||
assert "AS64501" in text
|
||||
assert "ISP resolver still active" in text
|
||||
|
||||
|
||||
def test_markdown_dnsbl_error_range_is_never_shown_as_a_plain_listing():
|
||||
text = render_markdown(full_report())
|
||||
assert "bl.spamcop.net" in text
|
||||
assert "not a listing" in text
|
||||
|
||||
|
||||
def test_markdown_latency_table_carries_a_sparkline_in_a_fenced_block():
|
||||
text = render_markdown(full_report())
|
||||
assert "cloudflare-dns" in text
|
||||
assert any(char in text for char in SPARK_CHARS)
|
||||
assert text.count("```") >= 2
|
||||
|
||||
|
||||
def test_markdown_flags_that_tcp_loss_is_a_different_metric():
|
||||
text = render_markdown(full_report())
|
||||
assert "failed TCP connections" in text
|
||||
|
||||
|
||||
def test_markdown_path_section_marks_the_first_loss_jump():
|
||||
text = render_markdown(full_report())
|
||||
path_block = text.split("## Path")[1].split("## Speed")[0]
|
||||
marked = [line for line in path_block.splitlines() if line.rstrip().endswith("<<")]
|
||||
assert len(marked) == 1
|
||||
assert "10.64.0.1" in marked[0]
|
||||
|
||||
|
||||
def test_markdown_speed_section_explains_the_bufferbloat_grade_in_plain_language():
|
||||
text = render_markdown(full_report())
|
||||
assert "284.3" in text
|
||||
assert "grade C" in text
|
||||
assert "choppy" in text
|
||||
assert "ookla_bin" in text
|
||||
assert "binary not on PATH" in text
|
||||
|
||||
|
||||
def test_markdown_run_diagnostics_lists_every_module_with_its_status():
|
||||
text = render_markdown(full_report())
|
||||
block = text.split("## Run diagnostics")[1]
|
||||
for section in SECTION_ORDER:
|
||||
assert section in block
|
||||
|
||||
|
||||
def test_markdown_without_emoji_uses_text_badges():
|
||||
text = render_markdown(full_report(), emoji=False)
|
||||
assert "[warn]" in text
|
||||
assert "🟡" not in text
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue