#!/usr/bin/env python3
"""Gate D1/D4: a frame crosses between two swarmd daemons over a real channel.
python3 -B test/test_channel_e2e.py [--bootstrap host:port] [--seconds N]
Two daemons join one topic, a swarmd-probe client on each opens the SAME
protocol channel, and a frame sent by one arrives at the other. That is the
whole product in one assertion: everything before it is plumbing.
⛔ NO WIRE IS WRITTEN HERE. The first version hand-rolled compact-encoding in
Python and the daemon answered "undecodable message" to every frame — the test
was wrong, not the daemon. Hand-writing the wire in a client is exactly what
generating it was supposed to stop, and a test client is still a client. So this
drives tools/swarmd-probe, which uses the generated encoders, and reads its
words. The only thing Python parses here is stdout.
⛔ IT ASSERTS ITS OWN BASELINE. If the two daemons never see each other the
message result says nothing, so a swarm that never meshes exits INCONCLUSIVE
rather than FAIL — the mistake the first long-lived gate made, reporting "the
responder has gone silent" from a run in which nothing had ever connected.
"""
import os, re, subprocess, sys, tempfile, time
HERE = os.path.dirname(os.path.realpath(__file__))
ROOT = os.path.dirname(HERE)
SWARMD = os.path.join(ROOT, "swarmd")
PROBE = os.path.join(ROOT, "tools", "swarmd-probe")
EXIT_PASS, EXIT_FAIL, EXIT_INCONCLUSIVE = 0, 1, 2
TOPIC = "5a" * 32
PROTOCOL = "swarmd-gate/v1"
PAYLOAD = "hello-over-the-swarm"
# The sender goes first ON PURPOSE. It re-offers the frame until the daemon
# accepts it, so starting it before the receiver exists exercises the refusal
# path as well as the crossing — the case the fire-once-on-a-timer probe passed
# by luck on a fast testnet.
RECEIVER_DELAY = 5
def arg(name, default=None):
return sys.argv[sys.argv.index(name) + 1] if name in sys.argv else default
def read(path):
try:
with open(path, "rb") as fh:
return fh.read().decode("utf-8", "replace")
except OSError:
return ""
def wait_for(path, needle, timeout, proc):
end = time.time() + timeout
while time.time() < end:
if proc is not None and proc.poll() is not None:
return False
if needle in read(path):
return True
time.sleep(0.3)
return False
def peers_seen(text):
"""The highest peer count this daemon ever reported."""
return max([int(m) for m in re.findall(r"peers=(\d+)", text)] or [0])
def main():
bootstrap = arg("--bootstrap")
window = int(arg("--seconds", "60" if bootstrap else "180"))
for path in (SWARMD, PROBE):
if not os.path.isfile(path):
print("missing %s — run ./build.sh" % path)
return EXIT_INCONCLUSIVE
where = "bootstrap %s" % bootstrap if bootstrap else "the PUBLIC DHT"
print("two daemons on %s, %ds window" % (where, window))
tmp = tempfile.mkdtemp(prefix="swarmd-chan-")
procs, socks, logs = [], [], {}
def spawn(argv, logpath):
fh = open(logpath, "wb")
p = subprocess.Popen(argv, stdout=fh, stderr=subprocess.STDOUT)
procs.append(p)
return p
try:
daemons = {}
for tag in ("a", "b"):
sp = os.path.join(tmp, tag + ".sock")
logs[tag] = os.path.join(tmp, tag + ".log")
argv = [SWARMD, "--socket", sp]
if bootstrap:
argv += ["--bootstrap", bootstrap]
daemons[tag] = spawn(argv, logs[tag])
socks.append(sp)
# ---- baseline 1: both daemons are actually on a DHT ----
for tag in ("a", "b"):
if not wait_for(logs[tag], "dht=up", 90, daemons[tag]):
print("\nINCONCLUSIVE: daemon %s never reached dht=up." % tag)
return EXIT_INCONCLUSIVE
print(" both daemons are on the DHT")
def probe(tag, sock_tag, extra, seconds):
logs[tag] = os.path.join(tmp, tag + ".log")
argv = [PROBE, "--socket", os.path.join(tmp, sock_tag + ".sock"),
"--topic-hex", TOPIC, "--protocol", PROTOCOL,
"--seconds", str(seconds)] + extra
return spawn(argv, logs[tag])
sender = probe("send", "a", ["--send", PAYLOAD], window)
time.sleep(RECEIVER_DELAY)
receiver = probe("recv", "b", [], window - RECEIVER_DELAY)
print(" probes attached; waiting up to %ds" % window)
end = time.time() + window + 15
while time.time() < end and receiver.poll() is None:
time.sleep(1)
for p in (sender, receiver):
if p.poll() is None:
p.terminate()
try: p.wait(timeout=10)
except subprocess.TimeoutExpired: p.kill()
sender_out, receiver_out = read(logs["send"]), read(logs["recv"])
met = min(peers_seen(read(logs["a"])), peers_seen(read(logs["b"])))
accepted = "accepted after" in sender_out
print(" daemons met: %s | frame accepted by the daemon: %s"
% ("yes" if met else "no", "yes" if accepted else "no"))
# ---- baseline 2: if they never met, nothing downstream means anything ----
if not met:
print("\nINCONCLUSIVE: the two daemons never saw each other, so the "
"message result says nothing about the channel.")
print("(logs: %s)" % tmp)
return EXIT_INCONCLUSIVE
if ("RECEIVED: " + PAYLOAD) in receiver_out:
print("\nPASS: the frame crossed the swarm and arrived on the other daemon.")
return EXIT_PASS
if not accepted:
print("\nFAIL: the daemons met but neither channel ever opened — the "
"daemon refused the frame for the whole window.")
else:
print("\nFAIL: the daemon accepted the frame and it never arrived.")
print("--- sender ---\n%s--- receiver ---\n%s" % (sender_out, receiver_out))
return EXIT_FAIL
finally:
for p in procs:
if p.poll() is None:
p.terminate()
try: p.wait(timeout=5)
except subprocess.TimeoutExpired: p.kill()
for sp in socks:
try: os.unlink(sp)
except OSError: pass
print("(logs: %s)" % tmp)
if __name__ == "__main__":
sys.exit(main())