I want to create a test for a function that:
- Calls an api to create an order
- Uses a timer to wait for the order to be filled
# This is module
class Order:
def __init__(self):
self._client = Client()
def open(self, volume: float, type: str):
try:
order = self._client.create_order(
type=type,
quantity=volume
)
except Exception as e:
logger.error(
f'Exception on open_trade in {symbol} {TradeSide.BUY} \n {e} \n')
return None
t = Timer(1.0, lambda: self._check_order_status(order['orderId']))
t.start()
return order
def _check_order_status(self, order_id: str) -> Dict:
try:
order = self._client.get_order(orderId=order_id)
except Exception as e:
logger.error(
f'Exception on getting order status {order_id} {e} ')
order = None
if order and order['status'] == FILLED:
self.state.update_order(
self, order)
else:
t = Timer(2.0, lambda: self._check_order_status(order_id))
t.start()
To achieve this I mocked the two _client
functions:
def test_open(mocker: MockFixture,
new_open: Dict, get_open_order: Dict):
# Arange
def mock_create_order(self, type, quantity):
return new_open
mocker.patch(
'module.Client.create_order',
mock_create_order
)
def mock_get_order(self, orderId):
return get_open_order
mocker.patch(
'module.Client.get_order',
mock_get_order
)
# Act
order = Order()
order.open('BUY', 123)
# Assert
assert len(mock_state.open) == 1
The problem is that after starting the Timer, that thread doesn't have the mocked context and it calls the actual class... Any ideas how can I trick the Timer into calling the correct mocked get_order
function?
from Orchestrate test case for function that's using threading.Timer
No comments:
Post a Comment