0

我正在尝试模拟一个表示设备状态的变量

在这种情况下，我试图添加一个设备，并且我有以下代码：

if self.network.controller.add_node( secure ) :
    for i in range( 0, 60 ) :
        if flagStarted :
            if self.commandState == self.COMMAND_FAILED or self.commandState == self.COMMAND_FAILED :
            # Transaction Failed or Error
            self.network.controller.cancel_command( )
            self.log.warning( " *** Add Device Failed *** " )
            return False
        elif self.commandState == self.COMMAND_CANCEL :
            # Transaction Canceled
            self.log.debug( " *** Command Canceled " )
            return False
        elif self.commandState == self.COMMAND_COMPLETED :
            # Transaction Completed
            value = ZWaveProtocol.getAddedDevice( )
            if value > 0 :
                dev = DeviceCollection.getDeviceByProtocolID( value, "ZWave" )
                return dev.id
            else :
                if self.commandState == self.COMMAND_STARTING or self.commandState == self.COMMAND_WAITING :
                flagStarted = True
        sys.stdout.write( "." )
        sys.stdout.flush( )
        time.sleep( 1.0 )

    self.network.controller.cancel_command( )
    return -1
else :
    self.log.error( "Failed to add device" )

我正在做的是嘲笑 self.network.controller.add_node(secure)，当我这样做时，我将 self.commandState 更改为正在启动...。我想要完成的是在大约 5 秒后将其更改为 self。 COMMAND_COMPLETED 以成功完成操作。

任何想法如何模拟这个？

4

1 回答 1

1

由于您的代码是阻塞的（程序流一直停留在该循环上，直到它完成），一个不与当前代码混淆太多的选项是跨越一个在给定时间后更改变量的线程。

让我们假设您的代码在一个名为 的方法run_loop和一个类MyClass中。给定一个像这样的简单测试代码：

def test_1():
  obj = MyClass()
  # Mock something
  obj.run_loop()
  # Do your assertions

您可以将其更改为以下内容。我没有测试过，可以改进很多，但你明白了：

from threading import Thread
from time import sleep

def change_state(obj):
    sleep(5)
    obj.commandState = obj.COMMAND_COMPLETE

def test_1():
  obj = MyClass()

  # Launch a thread that within 5 seconds will change the state of `obj`
  Thread(target=change_state, args=[obj]).start()

  # Call the main loop, which will recognize that the state changed within 5 secs
  obj.run_loop()
于 2016-02-10T10:59:12.147 回答