为使用 threading.Timer 的函数编排测试用例

Question

我想为一个函数创建一个测试：

调用 api 来创建订单
使用计时器等待订单被执行

# 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()

为了实现这一点，我模拟了两个 _client 函数：

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

问题是在启动 Timer 之后，该线程没有模拟的上下文，它调用了实际的 class... 有什么想法可以让 Timer 调用正确的模拟 get_order 函数吗？

Answer 1

that thread doesn't have the mocked context

那是因为test_open已经结束，补丁方法已经恢复

Any ideas how can I trick the Timer into calling the correct mocked get_order function?

这里有3个选项。我会选择选项 2。

1。睡觉

这很简单，但需要硬编码正确的睡眠时间。

order.open('BUY', 123)
sleep(1.0)  # Add this

2。跟踪并加入定时器线程

这是最清楚的方法。

t = Timer(1.0, lambda: self._check_order_status(order['orderId']))
t.start()
self._t = t  # Add this

order.open('BUY', 123)
order._t.join()  # Add this

有些人可能会觉得以上内容向模块中泄露了测试需求。
您可能希望在补丁 Timer class 中跟踪计时器线程：

class TrackedTimer(Timer):
    instance_by_caller_id = {}

    def __init__(self, interval, function, args=None, kwargs=None):
        super().__init__(interval, function, args=args, kwargs=kwargs)

        caller = inspect.currentframe().f_back.f_locals.get('self')
        if caller:
            self.instance_by_caller_id[id(caller)] = self

    @classmethod
    def join_for_caller(cls, caller):
        instance = cls.instance_by_caller_id.get(id(caller))
        if instance:
            instance.join()

mocker.patch(              # Add this
    'polls.module.Timer',  #
    TrackedTimer,          #
)                          #

# Act
order = Order()
order.open('BUY', 123)
TrackedTimer.join_for_caller(order)  # Add this

3。直接替换class或实例

上的方法

这允许在 test_open 结束后执行模拟方法。

直接替换class上的方法：

# mocker.patch(                         # Replace this
#     'polls.module.Client.get_order',  #
#     mock_get_order                    #
# )                                     #

# Act
Client.get_order = mock_get_order       # with this
order = Order()
order.open('BUY', 123)

或者直接替换实例上的方法：

order = Order()
order._client.get_order = lambda orderId: mock_get_order(order._client, orderId)  # Add this
# setattr(order._client, 'get_order', mock_get_order.__get__(order._client))      # or a bound method
order.open('BUY', 123)

为使用 threading.Timer 的函数编排测试用例

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

python

pytest

python-multithreading

1。睡觉

2。跟踪并加入定时器线程

3。直接替换class或实例