Changeset View
Changeset View
Standalone View
Standalone View
test/functional/test_framework/avatools.py
| #!/usr/bin/env python3 | #!/usr/bin/env python3 | ||||
| # Copyright (c) 2021 The Bitcoin ABC developers | # Copyright (c) 2021 The Bitcoin ABC developers | ||||
| # Distributed under the MIT software license, see the accompanying | # Distributed under the MIT software license, see the accompanying | ||||
| # file COPYING or http://www.opensource.org/licenses/mit-license.php. | # file COPYING or http://www.opensource.org/licenses/mit-license.php. | ||||
| """Utilities for avalanche tests.""" | """Utilities for avalanche tests.""" | ||||
| from typing import Any, Optional, List, Dict | from typing import Any, Optional, List, Dict | ||||
| from .authproxy import JSONRPCException | |||||
| from .messages import ( | from .messages import ( | ||||
| AvalancheProof, | AvalancheProof, | ||||
| CTransaction, | CTransaction, | ||||
| FromHex, | FromHex, | ||||
| ToHex | ToHex | ||||
| ) | ) | ||||
| from .test_node import TestNode | from .test_node import TestNode | ||||
| from .util import satoshi_round | from .util import ( | ||||
| satoshi_round, | |||||
| wait_until, | |||||
| ) | |||||
| def create_coinbase_stakes( | def create_coinbase_stakes( | ||||
| node: TestNode, | node: TestNode, | ||||
| blockhashes: List[str], | blockhashes: List[str], | ||||
| priv_key: str, | priv_key: str, | ||||
| amount: Optional[str] = None) -> List[Dict[str, Any]]: | amount: Optional[str] = None) -> List[Dict[str, Any]]: | ||||
| """Returns a list of dictionaries representing stakes, in a format | """Returns a list of dictionaries representing stakes, in a format | ||||
| ▲ Show 20 Lines • Show All 90 Lines • ▼ Show 20 Lines | for utxo in utxos[:count]: | ||||
| }) | }) | ||||
| return stakes | return stakes | ||||
| def get_proof_ids(node): | def get_proof_ids(node): | ||||
| return [FromHex(AvalancheProof(), peer['proof'] | return [FromHex(AvalancheProof(), peer['proof'] | ||||
| ).proofid for peer in node.getavalanchepeerinfo()] | ).proofid for peer in node.getavalanchepeerinfo()] | ||||
| def wait_for_proof(node, proofid_hex, timeout=60): | |||||
| def proof_found(): | |||||
| try: | |||||
| node.getrawavalancheproof(proofid_hex) | |||||
| return True | |||||
| except JSONRPCException: | |||||
| return False | |||||
| wait_until(proof_found, timeout=timeout) | |||||