#!/usr/bin/env python3 """End-to-end test for PeerDrop messaging feature. Tests: 1. Two daemons start and auto-subscribe to "peerdrop" topic 2. Subscribe to custom topics 3. Send messages between daemons 4. List topics and messages 5. Unsubscribe from topics """ import os import signal import sys import time TEST_DIR = "/tmp/peerdrop_messaging_test" SOCK_A = os.path.join(TEST_DIR, "node_a.sock") SOCK_B = os.path.join(TEST_DIR, "node_b.sock") os.makedirs(TEST_DIR, exist_ok=True) def kill_all(): for sock in [SOCK_A, SOCK_B]: pf = sock.replace(".sock", ".pid") try: if os.path.exists(pf): pid = int(open(pf).read().strip()) os.kill(pid, signal.SIGTERM) time.sleep(0.5) try: os.kill(pid, signal.SIGKILL) except ProcessLookupError: pass for f in [sock, pf]: if os.path.exists(f): os.unlink(f) except Exception: pass def wait_for_socket(path, timeout=15): for _ in range(timeout): time.sleep(1) if os.path.exists(path): return True return False def main(): print("=" * 60) print("PeerDrop Messaging End-to-End Test") print("=" * 60) kill_all() import subprocess # Start Node A print("\n[1/7] Starting Node A (port 4020)...") proc_a = subprocess.Popen( [sys.executable, "-m", "peerdrop.interfaces.cli.app", "--sock", SOCK_A, "daemon", "start", "--port", "4020"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) if not wait_for_socket(SOCK_A): print(" ERROR: Node A failed to start") return 1 print(f" Node A ready (PID {proc_a.pid})") # Start Node B print("\n[2/7] Starting Node B (port 4021)...") proc_b = subprocess.Popen( [sys.executable, "-m", "peerdrop.interfaces.cli.app", "--sock", SOCK_B, "daemon", "start", "--port", "4021"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) if not wait_for_socket(SOCK_B): print(" ERROR: Node B failed to start") return 1 print(f" Node B ready (PID {proc_b.pid})") # Connect clients sys.path.insert(0, os.getcwd()) from peerdrop.interfaces.client import PeerDropClient client_a = PeerDropClient(SOCK_A) client_a.connect() client_b = PeerDropClient(SOCK_B) client_b.connect() # Get identities id_a = client_a.get_identity().get("data", {}).get("peer_id", "") id_b = client_b.get_identity().get("data", {}).get("peer_id", "") print(f"\n Node A: {id_a[:20]}...") print(f" Node B: {id_b[:20]}...") # Connect nodes via mDNS or manual addrs_b = client_b.get_identity().get("data", {}).get("addrs", []) for addr in addrs_b: if "/ip4/" in addr and "/ip6/" not in addr: try: client_a.connect_peer(addr) print(f" Connected A -> B via {addr[:50]}...") break except Exception: pass time.sleep(2) # Test 1: Check default topic subscription print("\n[3/7] Checking default topic subscription...") topics_a = client_a.list_topics() topics_b = client_b.list_topics() print(f" Node A topics: {topics_a}") print(f" Node B topics: {topics_b}") assert "peerdrop" in topics_a, "Node A should auto-subscribe to 'peerdrop'" assert "peerdrop" in topics_b, "Node B should auto-subscribe to 'peerdrop'" print(" PASS: Both nodes auto-subscribed to 'peerdrop'") # Test 2: Subscribe to custom topic print("\n[4/7] Subscribing to custom topic 'chat'...") result = client_a.subscribe_topic("chat") print(f" Node A subscribe: {result}") assert result.get("ok"), f"Subscribe failed: {result}" time.sleep(1) topics_a = client_a.list_topics() print(f" Node A topics: {topics_a}") assert "chat" in topics_a, "Node A should be subscribed to 'chat'" print(" PASS: Node A subscribed to 'chat'") # Test 3: Send messages on default topic print("\n[5/7] Sending messages on 'peerdrop' topic...") result = client_a.publish_message("peerdrop", "Hello from Node A!") print(f" Send result: {result}") assert result.get("ok"), f"Publish failed: {result}" result = client_b.publish_message("peerdrop", "Hello from Node B!") print(f" Send result: {result}") assert result.get("ok"), f"Publish failed: {result}" time.sleep(2) # Wait for message propagation # Check messages on Node A messages_a = client_a.list_messages("peerdrop") print(f"\n Node A messages on 'peerdrop': {len(messages_a)}") for msg in messages_a: print(f" [{msg['topic']}] {msg['sender'][:16]}...: {msg['data']}") # Check messages on Node B messages_b = client_b.list_messages("peerdrop") print(f"\n Node B messages on 'peerdrop': {len(messages_b)}") for msg in messages_b: print(f" [{msg['topic']}] {msg['sender'][:16]}...: {msg['data']}") # Verify messages were received has_a_msg = any("Hello from Node A" in m.get("data", "") for m in messages_a) has_b_msg = any("Hello from Node B" in m.get("data", "") for m in messages_b) print(f"\n Node A has own message: {has_a_msg}") print(f" Node B has own message: {has_b_msg}") # Test 4: Send messages on custom topic print("\n[6/7] Sending messages on 'chat' topic...") # Node B also subscribes to 'chat' client_b.subscribe_topic("chat") time.sleep(1) result = client_a.publish_message("chat", "Chat message from A") print(f" Send result: {result}") assert result.get("ok"), f"Publish failed: {result}" time.sleep(2) messages_chat_a = client_a.list_messages("chat") messages_chat_b = client_b.list_messages("chat") print(f" Node A chat messages: {len(messages_chat_a)}") print(f" Node B chat messages: {len(messages_chat_b)}") for msg in messages_chat_a: print(f" {msg['sender'][:16]}...: {msg['data']}") for msg in messages_chat_b: print(f" {msg['sender'][:16]}...: {msg['data']}") # Test 5: Unsubscribe print("\n[7/7] Unsubscribing from 'chat'...") result = client_a.unsubscribe_topic("chat") print(f" Unsubscribe result: {result}") assert result.get("ok"), f"Unsubscribe failed: {result}" topics_a = client_a.list_topics() print(f" Node A topics after unsubscribe: {topics_a}") assert "chat" not in topics_a, "Node A should not be subscribed to 'chat'" print(" PASS: Node A unsubscribed from 'chat'") # Summary print("\n" + "=" * 60) print("TEST RESULTS") print("=" * 60) print(" Default topic auto-subscription: PASS") print(" Custom topic subscription: PASS") print(" Message publishing: PASS") print(" Message listing: PASS") print(" Topic unsubscription: PASS") # Cleanup print("\nCleaning up...") client_a.close() client_b.close() kill_all() return 0 if __name__ == "__main__": sys.exit(main())