Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_fromraw_toraw(self):
peer = Peer.from_signed_raw(rawpeer)
rendered_peer = peer.signed_raw()
from_rendered_peer = Peer.from_signed_raw(rendered_peer)
self.assertEqual(from_rendered_peer.currency, "beta_brousouf")
self.assertEqual(
from_rendered_peer.pubkey, "HsLShAtzXTVxeUtQd7yi5Z5Zh4zNvbu8sTEZ53nfKcqY"
)
self.assertEqual(
str(from_rendered_peer.blockUID),
"8-1922C324ABC4AF7EF7656734A31F5197888DDD52",
)
self.assertEqual(len(peer.endpoints), 4)
self.assertIsInstance(peer.endpoints[0], BMAEndpoint)
self.assertIsInstance(peer.endpoints[1], BMAEndpoint)
self.assertIsInstance(peer.endpoints[2], WS2PEndpoint)
self.assertIsInstance(peer.endpoints[3], UnknownEndpoint)
self.assertEqual(from_rendered_peer.endpoints[0].server, "some.dns.name")
def test_peer_signature(self):
signed_raw = """Version: 2
Type: Peer
Currency: test_net
PublicKey: 8Fi1VSTbjkXguwThF4v2ZxC5whK7pwG2vcGTkPUPjPGU
Block: 2086-00005D6FC6E22FB308D8815A565A01C66FFB7DC761D616DE0698F6322565F1D6
Endpoints:
BASIC_MERKLED_API testnet.duniter.inso.ovh 80
4aQ/sfqFAFUeYkkLdC2OfgXqTBjCIcMptpR/GIlGqbe4aFVJcy9NEVAFx7sHiLuAb+VNnec3XHHC+xOk3MLzDA==
"""
peer = Peer.from_signed_raw(signed_raw)
pubkey = "8Fi1VSTbjkXguwThF4v2ZxC5whK7pwG2vcGTkPUPjPGU"
verifying_key = VerifyingKey(pubkey)
self.assertTrue(verifying_key.verify_document(peer))
def test_fromraw(self):
peer = Peer.from_signed_raw(rawpeer)
self.assertEqual(peer.currency, "beta_brousouf")
self.assertEqual(peer.pubkey, "HsLShAtzXTVxeUtQd7yi5Z5Zh4zNvbu8sTEZ53nfKcqY")
self.assertEqual(
str(peer.blockUID), "8-1922C324ABC4AF7EF7656734A31F5197888DDD52"
)
self.assertEqual(len(peer.endpoints), 4)
self.assertIsInstance(peer.endpoints[0], BMAEndpoint)
self.assertIsInstance(peer.endpoints[1], BMAEndpoint)
self.assertIsInstance(peer.endpoints[2], WS2PEndpoint)
self.assertIsInstance(peer.endpoints[3], UnknownEndpoint)
self.assertEqual(peer.endpoints[0].server, "some.dns.name")
self.assertEqual(peer.endpoints[0].ipv4, "88.77.66.55")
self.assertEqual(peer.endpoints[0].ipv6, "2001:42d0:52:a00::648")
self.assertEqual(peer.endpoints[0].port, 9001)
def test_incorrect(self):
peer = Peer.from_signed_raw(test_weird_ipv6_peer)
rendered_peer = peer.signed_raw_for_certified()
Peer.from_signed_raw(rendered_peer)
def test_fromraw_toraw(self):
peer = Peer.from_signed_raw(rawpeer)
rendered_peer = peer.signed_raw()
from_rendered_peer = Peer.from_signed_raw(rendered_peer)
self.assertEqual(from_rendered_peer.currency, "beta_brousouf")
self.assertEqual(
from_rendered_peer.pubkey, "HsLShAtzXTVxeUtQd7yi5Z5Zh4zNvbu8sTEZ53nfKcqY"
)
self.assertEqual(
str(from_rendered_peer.blockUID),
"8-1922C324ABC4AF7EF7656734A31F5197888DDD52",
)
self.assertEqual(len(peer.endpoints), 4)
self.assertIsInstance(peer.endpoints[0], BMAEndpoint)
self.assertIsInstance(peer.endpoints[1], BMAEndpoint)
self.assertIsInstance(peer.endpoints[2], WS2PEndpoint)
self.assertIsInstance(peer.endpoints[3], UnknownEndpoint)
def test_incorrect(self):
peer = Peer.from_signed_raw(test_weird_ipv6_peer)
rendered_peer = peer.signed_raw()
self.assertEqual(test_weird_ipv6_peer, rendered_peer)
Peer.from_signed_raw(rendered_peer)
def refresh_peer_data(self, peer_data):
if "raw" in peer_data:
try:
str_doc = "{0}{1}\n".format(peer_data['raw'],
peer_data['signature'])
peer_doc = Peer.from_signed_raw(str_doc)
self.neighbour_found.emit(peer_doc)
except MalformedDocumentError as e:
self._logger.debug("{:}:{:}".format(str(e.__class__.__name__), str(e)))
else:
self._logger.debug("Incorrect leaf reply")
def refresh_peer_data(self, peer_data):
if "raw" in peer_data:
try:
str_doc = "{0}{1}\n".format(peer_data['raw'],
peer_data['signature'])
peer_doc = Peer.from_signed_raw(str_doc)
self.neighbour_found.emit(peer_doc)
except MalformedDocumentError as e:
logging.debug(str(e))
else:
logging.debug("Incorrect leaf reply")
async def from_address(cls, currency, address, port, session):
"""
Factory method to get a node from a given address
:param str currency: The node currency. None if we don't know\
the currency it should have, for example if its the first one we add
:param str address: The node address
:param int port: The node port
:return: A new node
:rtype: sakia.core.net.Node
"""
peer_data = await bma.network.Peering(ConnectionHandler(address, port)).get(session)
peer = Peer.from_signed_raw("{0}{1}\n".format(peer_data['raw'],
peer_data['signature']))
if currency is not None:
if peer.currency != currency:
raise InvalidNodeCurrency(peer.currency, currency)
node = cls(peer,
"", peer.pubkey, None, Node.ONLINE, time.time(),
{'root': "", 'leaves': []}, "", "", 0, session)
logging.debug("Node from address : {:}".format(str(node)))
return node