rebbarb/scripts/test-analyse-pcap.py

#!/usr/bin/env python3
"""Self-test for analyse-pcap.py: synthesise a capture with known properties
and check the tool reports them."""
import os, socket, struct, subprocess, sys, tempfile

GC = socket.inet_aton("10.0.0.64")
PC = socket.inet_aton("10.0.0.1")
MAC_GC = bytes.fromhex("0009bf000001")
MAC_PC = bytes.fromhex("002b67d4156c")


def tcp_frame(src, dst, smac, dmac, seq, payload, win=4096):
    tcp = struct.pack("!HHIIBBHHH", 4000, 21, seq, 0, 5 << 4, 0x10, win, 0, 0)
    ip = struct.pack("!BBHHHBBH4s4s", 0x45, 0, 20 + len(tcp) + len(payload),
                     0, 0, 64, socket.IPPROTO_TCP, 0, src, dst)
    return dmac + smac + b"\x08\x00" + ip + tcp + payload


def write_pcap(path, pkts):
    with open(path, "wb") as f:
        f.write(struct.pack("<IHHiIII", 0xA1B2C3D4, 2, 4, 0, 0, 65535, 1))
        for ts, data in pkts:
            s, us = int(ts), int(round((ts - int(ts)) * 1e6))
            f.write(struct.pack("<IIII", s, us, len(data), len(data)))
            f.write(data)


def run(path, *args):
    r = subprocess.run([sys.executable, "-I", "scripts/analyse-pcap.py", path,
                        "--gc-ip", "10.0.0.64", *args],
                       capture_output=True, text=True)
    if r.returncode:
        print(r.stdout, r.stderr)
        raise SystemExit(f"analyser failed: {r.returncode}")
    return r.stdout


fails = []
tmp = tempfile.mkdtemp()

# Scenario A: 1000 full frames in 2 s from the GC, no retransmits.
#   1000 * 1514 B / 2 s = 757 KB/s -- deliberately the speed being measured.
pkts, seq, t = [], 1, 1000.0
for i in range(1000):
    pkts.append((t, tcp_frame(GC, PC, MAC_GC, MAC_PC, seq, b"\0" * 1460)))
    seq += 1460
    t += 0.002
pA = os.path.join(tmp, "a.pcap")
write_pcap(pA, pkts)
out = run(pA)
if "1000 frames" not in out:
    fails.append("A: frame count")
if "mean frame 1514" not in out:
    fails.append(f"A: mean frame size not 1514\n{out}")
if "retransmissions 0" not in out:
    fails.append("A: should report no retransmissions")
# 1514*1000/1.998s = 757 KB/s, ~22% of the EXI ceiling
if "% of the 27 Mbit/s EXI ceiling" not in out:
    fails.append("A: no EXI ceiling comparison")
pct = float(out.split("% of the 27 Mbit/s")[0].split(",")[-1].strip())
if not (20 <= pct <= 25):
    fails.append(f"A: EXI percentage {pct}, expected ~22")
else:
    print(f"A  1000x1514B/2s -> reported {pct}% of the EXI ceiling")

# Scenario B: small frames -- must trigger the per-frame-overhead warning.
pkts, seq, t = [], 1, 2000.0
for i in range(1000):
    pkts.append((t, tcp_frame(GC, PC, MAC_GC, MAC_PC, seq, b"\0" * 200)))
    seq += 200
    t += 0.002
pB = os.path.join(tmp, "b.pcap")
write_pcap(pB, pkts)
out = run(pB)
if "PER-FRAME overhead" not in out:
    fails.append(f"B: small frames did not trigger the overhead warning\n{out}")
else:
    print("B  200B payloads -> per-frame-overhead warning raised")

# Scenario C: retransmissions (same seq+len twice) must be counted.
pkts, t = [], 3000.0
for i in range(10):
    pkts.append((t, tcp_frame(GC, PC, MAC_GC, MAC_PC, 1, b"\0" * 1460)))
    t += 0.002
pC = os.path.join(tmp, "c.pcap")
write_pcap(pC, pkts)
out = run(pC)
if "retransmissions 9" not in out:
    fails.append(f"C: expected 9 retransmissions\n{out}")
else:
    print("C  same seq x10 -> 9 retransmissions counted")

# Scenario D: a long stall must be found and attributed.
pkts, seq, t = [], 1, 4000.0
for i in range(100):
    pkts.append((t, tcp_frame(GC, PC, MAC_GC, MAC_PC, seq, b"\0" * 1460)))
    seq += 1460
    t += 0.3 if i == 50 else 0.001      # one 300 ms hole
pD = os.path.join(tmp, "d.pcap")
write_pcap(pD, pkts)
out = run(pD)
if "300.00 ms" not in out:
    fails.append(f"D: the 300 ms gap was not reported\n{out}")
elif "NOT sending" not in out:
    fails.append(f"D: dominant stall not attributed\n{out}")
else:
    print("D  one 300ms hole in 100 frames -> found and attributed")

# Scenario E: zero-window from the GC means the GC cannot drain.
pkts, t = [], 5000.0
pkts.append((t, tcp_frame(PC, GC, MAC_PC, MAC_GC, 1, b"\0" * 100, win=0)))
pkts.append((t + 0.05, tcp_frame(PC, GC, MAC_PC, MAC_GC, 2, b"\0" * 100, win=4096)))
pE = os.path.join(tmp, "e.pcap")
write_pcap(pE, pkts)
out = run(pE)
if "zero-window" not in out or "CPU/stack bound" not in out:
    fails.append(f"E: zero-window not reported\n{out}")
else:
    print("E  zero-window span -> reported as CPU/stack bound")

# Scenario F: pcapng must be refused with a useful message, not misparsed.
pF = os.path.join(tmp, "f.pcapng")
with open(pF, "wb") as f:
    f.write(struct.pack(">I", 0x0A0D0D0A) + b"\0" * 64)
r = subprocess.run([sys.executable, "-I", "scripts/analyse-pcap.py", pF],
                   capture_output=True, text=True)
if "pcapng" not in (r.stdout + r.stderr):
    fails.append("F: pcapng not detected")
else:
    print("F  pcapng refused with a conversion hint")

print("\nFAILURES:\n  " + "\n  ".join(fails) if fails else
      "\nAll analyse-pcap self-tests passed.")
sys.exit(1 if fails else 0)