mirror of
https://github.com/Chia-Network/chia-blockchain.git
synced 2024-09-22 00:52:32 +03:00
4b389c8fc0
* add values to time_out_assert_custom_interval() message
* hide time out assert helpers from stack traces
Get:
```
> await time_out_assert(15, dl_wallet_0.get_unconfirmed_balance, 101)
E AssertionError: Timed assertion timed
```
Instead of:
```
> await time_out_assert(15, dl_wallet_0.get_unconfirmed_balance, 101)
tests/wallet/db_wallet/test_dl_wallet.py:114:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests/time_out_assert.py:25: in time_out_assert
await time_out_assert_custom_interval(timeout, 0.05, function, value, *args, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
timeout = 15, interval = 0.05
function = <bound method DataLayerWallet.get_unconfirmed_balance of <chia.data_layer.data_layer_wallet.DataLayerWallet object at 0x7fbcbc35c7c0>>
value = 101, args = (), kwargs = {}, start = 1637077216.2218025, f_res = 0
async def time_out_assert_custom_interval(timeout: int, interval, function, value=True, *args, **kwargs):
start = time.time()
while time.time() - start < timeout:
if asyncio.iscoroutinefunction(function):
f_res = await function(*args, **kwargs)
else:
f_res = function(*args, **kwargs)
if value == f_res:
return None
await asyncio.sleep(interval)
> assert False, "Timed assertion timed out"
E AssertionError: Timed assertion timed out
tests/time_out_assert.py:21: AssertionError
```
55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
import asyncio
|
|
import logging
|
|
import time
|
|
from typing import Callable
|
|
|
|
from chia.protocols.protocol_message_types import ProtocolMessageTypes
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
async def time_out_assert_custom_interval(timeout: int, interval, function, value=True, *args, **kwargs):
|
|
__tracebackhide__ = True
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
if asyncio.iscoroutinefunction(function):
|
|
f_res = await function(*args, **kwargs)
|
|
else:
|
|
f_res = function(*args, **kwargs)
|
|
if value == f_res:
|
|
return None
|
|
await asyncio.sleep(interval)
|
|
assert False, f"Timed assertion timed out after {timeout} seconds: expected {value!r}, got {f_res!r}"
|
|
|
|
|
|
async def time_out_assert(timeout: int, function, value=True, *args, **kwargs):
|
|
__tracebackhide__ = True
|
|
await time_out_assert_custom_interval(timeout, 0.05, function, value, *args, **kwargs)
|
|
|
|
|
|
async def time_out_assert_not_none(timeout: int, function, *args, **kwargs):
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
if asyncio.iscoroutinefunction(function):
|
|
f_res = await function(*args, **kwargs)
|
|
else:
|
|
f_res = function(*args, **kwargs)
|
|
if f_res is not None:
|
|
return None
|
|
await asyncio.sleep(0.05)
|
|
assert False, "Timed assertion timed out"
|
|
|
|
|
|
def time_out_messages(incoming_queue: asyncio.Queue, msg_name: str, count: int = 1) -> Callable:
|
|
async def bool_f():
|
|
if incoming_queue.qsize() < count:
|
|
return False
|
|
for _ in range(count):
|
|
response = (await incoming_queue.get())[0].type
|
|
if ProtocolMessageTypes(response).name != msg_name:
|
|
# log.warning(f"time_out_message: found {response} instead of {msg_name}")
|
|
return False
|
|
return True
|
|
|
|
return bool_f
|