Sunday 24 October 2021

Orchestrate test case for function that's using threading.Timer

I want to create a test for a function that:

  1. Calls an api to create an order
  2. 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