我正在尝试模拟表示设备状态的变量
在这种情况下,我正在尝试添加设备,并且有以下代码:
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(安全),当我这样做时,我将self.commandState更改为Starting。我想完成的是大约 5 秒后将其更改为自我。COMMAND_COMPLETED以成功完成操作。
有什么想法如何嘲笑这个吗?
由于您的代码是阻塞的(程序流在该循环中一直保持,直到它完成),因此在不弄乱当前代码的情况下,一个选项是跨越一个线程,该线程在给定时间后更改变量。
假设您的代码位于一个名为 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()