varun-r-mallya/py-libp2p
1
0
Fork 0
mirror of https://github.com/varun-r-mallya/py-libp2p.git synced 2026-02-09 14:40:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

small tweak

Browse Source 
Signed-off-by: GautamBytes <manchandanigautam@gmail.com>
This commit is contained in:
GautamBytes
2025-07-20 09:30:21 +00:00
parent 187418378a
commit 227a5c6441
2 changed files with 11 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
tests/interop/test_js_ws_ping.py
View File

@ -24,11 +24,11 @@ PLAINTEXT_PROTOCOL_ID = "/plaintext/1.0.0"
@pytest.mark.trio
async def test_ping_with_js_node():
# 1) Path to the JS node script
# Path to the JS node script
js_node_dir = os.path.join(os.path.dirname(__file__), "js_libp2p", "js_node", "src")
script_name = "ws_ping_node.mjs"
# 2) Launch the JS libp2p node (long-running)
# Launch the JS libp2p node (long-running)
proc = await open_process(
["node", script_name],
stdout=subprocess.PIPE,
@ -36,7 +36,7 @@ async def test_ping_with_js_node():
cwd=js_node_dir,
)
try:
# 3) Read first two lines (PeerID and multiaddr)
# Read first two lines (PeerID and multiaddr)
buffer = b""
with trio.fail_after(10):
while buffer.count(b"\n") < 2:
@ -50,7 +50,7 @@ async def test_ping_with_js_node():
peer_id = ID.from_base58(peer_id_line)
maddr = Multiaddr(addr_line)
# 4) Set up Python host
# Set up Python host
key_pair = create_new_key_pair()
py_peer_id = ID.from_pubkey(key_pair.public_key)
peer_store = PeerStore()
@ -66,19 +66,19 @@ async def test_ping_with_js_node():
swarm = Swarm(py_peer_id, peer_store, upgrader, transport)
host = BasicHost(swarm)
# 5) Connect to JS node
# Connect to JS node
peer_info = PeerInfo(peer_id, [maddr])
await host.connect(peer_info)
assert host.get_network().connections.get(peer_id) is not None
await trio.sleep(0.1)
# 6) Ping protocol
# Ping protocol
stream = await host.new_stream(peer_id, [TProtocol("/ipfs/ping/1.0.0")])
await stream.write(b"ping")
data = await stream.read(4)
assert data == b"pong"
# 7) Cleanup
# Cleanup
await host.close()
finally:
proc.send_signal(signal.SIGTERM)