Changeset View
Changeset View
Standalone View
Standalone View
test/functional/test_framework/avatools.py
| Show All 24 Lines | from .messages import ( | ||||
| TCPAvalancheResponse, | TCPAvalancheResponse, | ||||
| ToHex, | ToHex, | ||||
| ) | ) | ||||
| from .p2p import P2PInterface, p2p_lock | from .p2p import P2PInterface, p2p_lock | ||||
| from .test_node import TestNode | from .test_node import TestNode | ||||
| from .util import ( | from .util import ( | ||||
| assert_equal, | assert_equal, | ||||
| satoshi_round, | satoshi_round, | ||||
| wait_until, | wait_until_helper, | ||||
| ) | ) | ||||
| from .wallet_util import bytes_to_wif | from .wallet_util import bytes_to_wif | ||||
| def create_coinbase_stakes( | def create_coinbase_stakes( | ||||
| node: TestNode, | node: TestNode, | ||||
| blockhashes: List[str], | blockhashes: List[str], | ||||
| priv_key: str, | priv_key: str, | ||||
| ▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Lines | def wait_for_proof(node, proofid_hex, timeout=60, expect_orphan=None): | ||||
| """ | """ | ||||
| def proof_found(): | def proof_found(): | ||||
| try: | try: | ||||
| wait_for_proof.is_orphan = node.getrawavalancheproof(proofid_hex)[ | wait_for_proof.is_orphan = node.getrawavalancheproof(proofid_hex)[ | ||||
| "orphan"] | "orphan"] | ||||
| return True | return True | ||||
| except JSONRPCException: | except JSONRPCException: | ||||
| return False | return False | ||||
| wait_until(proof_found, timeout=timeout) | wait_until_helper(proof_found, timeout=timeout) | ||||
| if expect_orphan is not None: | if expect_orphan is not None: | ||||
| assert_equal(expect_orphan, wait_for_proof.is_orphan) | assert_equal(expect_orphan, wait_for_proof.is_orphan) | ||||
| class AvaP2PInterface(P2PInterface): | class AvaP2PInterface(P2PInterface): | ||||
| """P2PInterface with avalanche capabilities""" | """P2PInterface with avalanche capabilities""" | ||||
| Show All 34 Lines | class AvaP2PInterface(P2PInterface): | ||||
| def send_avaresponse(self, round, votes, privkey): | def send_avaresponse(self, round, votes, privkey): | ||||
| response = AvalancheResponse(round, 0, votes) | response = AvalancheResponse(round, 0, votes) | ||||
| sig = privkey.sign_schnorr(response.get_hash()) | sig = privkey.sign_schnorr(response.get_hash()) | ||||
| msg = msg_tcpavaresponse() | msg = msg_tcpavaresponse() | ||||
| msg.response = TCPAvalancheResponse(response, sig) | msg.response = TCPAvalancheResponse(response, sig) | ||||
| self.send_message(msg) | self.send_message(msg) | ||||
| def wait_for_avaresponse(self, timeout=5): | def wait_for_avaresponse(self, timeout=5): | ||||
| wait_until( | self.wait_until( | ||||
| lambda: len(self.avaresponses) > 0, | lambda: len(self.avaresponses) > 0, | ||||
| timeout=timeout, | timeout=timeout) | ||||
| lock=p2p_lock) | |||||
| with p2p_lock: | with p2p_lock: | ||||
| return self.avaresponses.pop(0) | return self.avaresponses.pop(0) | ||||
| def send_poll(self, hashes): | def send_poll(self, hashes): | ||||
| msg = msg_avapoll() | msg = msg_avapoll() | ||||
| msg.poll.round = self.round | msg.poll.round = self.round | ||||
| self.round += 1 | self.round += 1 | ||||
| for h in hashes: | for h in hashes: | ||||
| msg.poll.invs.append(CInv(2, h)) | msg.poll.invs.append(CInv(2, h)) | ||||
| self.send_message(msg) | self.send_message(msg) | ||||
| def get_avapoll_if_available(self): | def get_avapoll_if_available(self): | ||||
| with p2p_lock: | with p2p_lock: | ||||
| return self.avapolls.pop(0) if len(self.avapolls) > 0 else None | return self.avapolls.pop(0) if len(self.avapolls) > 0 else None | ||||
| def wait_for_avahello(self, timeout=5): | def wait_for_avahello(self, timeout=5): | ||||
| wait_until( | self.wait_until( | ||||
| lambda: self.avahello is not None, | lambda: self.avahello is not None, | ||||
| timeout=timeout, | timeout=timeout) | ||||
| lock=p2p_lock) | |||||
| with p2p_lock: | with p2p_lock: | ||||
| return self.avahello | return self.avahello | ||||
| def send_avahello(self, delegation_hex: str, delegated_privkey: ECKey): | def send_avahello(self, delegation_hex: str, delegated_privkey: ECKey): | ||||
| delegation = FromHex(AvalancheDelegation(), delegation_hex) | delegation = FromHex(AvalancheDelegation(), delegation_hex) | ||||
| local_sighash = hash256( | local_sighash = hash256( | ||||
| delegation.getid() + | delegation.getid() + | ||||
| Show All 38 Lines | |||||