From 879b974e5724b2f22a47dece0e6fcf219c43b4b9 Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 02:44:43 +0300 Subject: [PATCH] exporter: markdown rendering with sparklines and hop bars --- ARCHITECTURE.md | 3 +- ARCHITECTURE.ru.md | 3 +- src/netcheck/exporter.py | 331 +++++++++++++++++++++++++++++++++++++++ tests/test_exporter.py | 311 ++++++++++++++++++++++++++++++++++++ 4 files changed, 646 insertions(+), 2 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index debc886..9cbb5f8 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -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/ diff --git a/ARCHITECTURE.ru.md b/ARCHITECTURE.ru.md index 817e95a..a73fe63 100644 --- a/ARCHITECTURE.ru.md +++ b/ARCHITECTURE.ru.md @@ -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/ diff --git a/src/netcheck/exporter.py b/src/netcheck/exporter.py index f830abb..4e6a54b 100644 --- a/src/netcheck/exporter.py +++ b/src/netcheck/exporter.py @@ -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" diff --git a/tests/test_exporter.py b/tests/test_exporter.py index e100a3a..36e8348 100644 --- a/tests/test_exporter.py +++ b/tests/test_exporter.py @@ -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