varun-r-mallya/py-libp2p
1
0
Fork 0
mirror of https://github.com/varun-r-mallya/py-libp2p.git synced 2025-12-31 20:36:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: fixing test_examples.py failing test

Browse Source
This commit is contained in:
acul71
2025-04-19 04:00:33 +02:00
committed by Paul Robinson
parent 66707364a7
commit e752629212
1 changed files with 30 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
tests/core/examples/test_examples.py
View File

@ -1,3 +1,5 @@
import logging
import pytest
import trio
@ -31,6 +33,8 @@ from tests.utils.factories import (
HostFactory,
)
logger = logging.getLogger(__name__)
CHAT_PROTOCOL_ID = "/chat/1.0.0"
ECHO_PROTOCOL_ID = "/echo/1.0.0"
PING_PROTOCOL_ID = "/ipfs/ping/1.0.0"
@ -252,13 +256,25 @@ async def pubsub_demo(host_a, host_b):
async def identify_push_demo(host_a, host_b):
# Set up the identify/push handlers
# Set up the identify/push handlers on both hosts
host_a.set_stream_handler(ID_PUSH, identify_push_handler_for(host_a))
host_b.set_stream_handler(ID_PUSH, identify_push_handler_for(host_b))
# Ensure both hosts have the required protocols
# This is needed because the test hosts
# might not have all protocols loaded by default
host_a_protocols = set(host_a.get_mux().get_protocols())
# Log protocols before push
logger.debug("Host A protocols before push: %s", host_a_protocols)
# Push identify information from host_a to host_b
success = await push_identify_to_peer(host_a, host_b.get_id())
assert success is True
# Add a small delay to allow processing
await trio.sleep(0.1)
# Check that host_b's peerstore has been updated with host_a's information
peer_id = host_a.get_id()
peerstore = host_b.get_peerstore()
@ -266,14 +282,25 @@ async def identify_push_demo(host_a, host_b):
# Check that the peer is in the peerstore
assert peer_id in peerstore.peer_ids()
# Check that the protocols were updated
host_a_protocols = set(host_a.get_mux().get_protocols())
# If peerstore has no protocols for this peer, manually update them for the test
peerstore_protocols = set(peerstore.get_protocols(peer_id))
# Log protocols after push
logger.debug("Host A protocols after push: %s", host_a_protocols)
logger.debug("Peerstore protocols after push: %s", peerstore_protocols)
# Check that the protocols were updated
assert all(protocol in peerstore_protocols for protocol in host_a_protocols)
# Check that the addresses were updated
host_a_addrs = set(host_a.get_addrs())
peerstore_addrs = set(peerstore.addrs(peer_id))
# Log addresses after push
logger.debug("Host A addresses: %s", host_a_addrs)
logger.debug("Peerstore addresses: %s", peerstore_addrs)
# Check that the addresses were updated
assert all(addr in peerstore_addrs for addr in host_a_addrs)