()
| 85 | |
| 86 | |
| 87 | def main(): |
| 88 | ap = argparse.ArgumentParser() |
| 89 | ap.add_argument("--port", type=int, default=14256) |
| 90 | ap.add_argument("--timeout", type=float, default=30.0) |
| 91 | args = ap.parse_args() |
| 92 | |
| 93 | config_dir = tempfile.mkdtemp(prefix="rns_interop_req_") |
| 94 | os.makedirs(os.path.join(config_dir, "storage", "resources"), exist_ok=True) |
| 95 | write_config(config_dir, args.port) |
| 96 | print(f"[python] config dir: {config_dir}", flush=True) |
| 97 | |
| 98 | try: |
| 99 | reticulum = RNS.Reticulum(config_dir) |
| 100 | except Exception as e: |
| 101 | print(f"[python] Reticulum init failed: {e}", file=sys.stderr) |
| 102 | sys.exit(3) |
| 103 | |
| 104 | identity = RNS.Identity() |
| 105 | destination = RNS.Destination(identity, RNS.Destination.IN, |
| 106 | RNS.Destination.SINGLE, APP_NAME, ASPECT) |
| 107 | destination.set_link_established_callback(on_link_established) |
| 108 | destination.set_proof_strategy(RNS.Destination.PROVE_ALL) |
| 109 | destination.register_request_handler("/echo", echo_handler, |
| 110 | allow=RNS.Destination.ALLOW_ALL) |
| 111 | |
| 112 | print(f"[python] destination hash: {destination.hash.hex()}", flush=True) |
| 113 | destination.announce() |
| 114 | print("[python] announced", flush=True) |
| 115 | |
| 116 | start = time.time() |
| 117 | last_announce = start |
| 118 | while time.time() - start < args.timeout: |
| 119 | if time.time() - last_announce >= 3.0 and not state["request_seen"]: |
| 120 | destination.announce() |
| 121 | last_announce = time.time() |
| 122 | print("[python] re-announced", flush=True) |
| 123 | |
| 124 | if state["request_seen"]: |
| 125 | if state["request_ok"]: |
| 126 | # Give the response a moment to land on the wire before |
| 127 | # exiting (the handler return value is sent asynchronously |
| 128 | # by RNS's link machinery). |
| 129 | time.sleep(0.5) |
| 130 | print("[python] SUCCESS", flush=True) |
| 131 | sys.exit(0) |
| 132 | else: |
| 133 | print("[python] FAILURE — payload mismatch on inbound", |
| 134 | flush=True) |
| 135 | sys.exit(1) |
| 136 | |
| 137 | time.sleep(0.1) |
| 138 | |
| 139 | print(f"[python] TIMEOUT request_seen={state['request_seen']}", |
| 140 | flush=True) |
| 141 | sys.exit(2) |
| 142 | |
| 143 | |
| 144 | if __name__ == "__main__": |
no test coverage detected