【发布时间】:2015-06-22 20:17:22
【问题描述】:
在这个问题被立即标记为重复之前,让我说我已经尝试过everysolution,这是与我的情况最相关的两个问题。如果有人在必要时关闭此问题之前至少可以查看我的特定问题,我将不胜感激。
我有一个名为 e 的有限状态机对象,它是一个 MCFiniteSM 对象。 e 的核心是一个名为 state_dict 的字典,它存储“进程”ID(“1”、“2”等)和一个关联的字典,它存储有关每个“进程”的更多信息。我正在运行 unittests 来添加进程,根据给定的参数更改它们的状态等。但是,在 unittest 文件中的函数调用之间,有限状态机似乎被清除了。我已经查看了上面列出的两个问题,以避免这种情况并保留更改,但无论我尝试什么,对有限状态机的更改都不会保留。这是 uniitest 文件。
from finite_state_machine import MCFiniteSM
from unittest import TestLoader, TestCase, main as unimain
from datetime import datetime
import time, calendar
class MyUnitTest(TestCase):
@classmethod
def setUpClass(cls):
cls.e = MCFiniteSM()
cls.timestamp = datetime.strftime(datetime.fromtimestamp(calendar.timegm(time.gmtime())), '%Y/%m/%d %H:%M:%S')
class TestFSM(MyUnitTest):
@classmethod
def setUpClass(cls):
super(TestFSM, cls).setUpClass()
#e = MCFiniteSM()
#timestamp = datetime.strftime(datetime.fromtimestamp(calendar.timegm(time.gmtime())), '%Y/%m/%d %H:%M:%S')
def test_add_convert_processes(self):
self.e.add_process('1', 'S', self.timestamp, 100, 'start message for process 1')
self.e.add_process('2', 'S', self.timestamp, 200, 'start message for process 2')
self.e.add_process('3', 'S', self.timestamp, 300, 'start message for process 3')
self.e.add_process('4', 'S', self.timestamp, 400, 'start message for process 4')
self.e.add_process('5', 'S', self.timestamp, 500, 'start message for process 5')
self.e.add_process('6', 'S', self.timestamp, 600, 'start message for process 6')
self.assertEqual(self.e.state_dict['1'], {'id':'1', 'message_type': 'S', 'timestamp':self.timestamp, 'state': 'i', 'threshold':100, 'message': 'start message for process 1'})
self.assertEqual(self.e.state_dict['2'], {'id':'2', 'message_type': 'S', 'timestamp':self.timestamp, 'state': 'i', 'threshold':200, 'message': 'start message for process 2'})
self.assertEqual(self.e.state_dict['3'], {'id':'3', 'message_type': 'S', 'timestamp':self.timestamp, 'state': 'i', 'threshold':300, 'message': 'start message for process 3'})
self.assertEqual(self.e.state_dict['4'], {'id':'4', 'message_type': 'S', 'timestamp':self.timestamp, 'state': 'i', 'threshold':400, 'message': 'start message for process 4'})
self.assertEqual(self.e.state_dict['5'], {'id':'5', 'message_type': 'S', 'timestamp':self.timestamp, 'state': 'i', 'threshold':500, 'message': 'start message for process 5'})
self.assertEqual(self.e.state_dict['6'], {'id':'6', 'message_type': 'S', 'timestamp':self.timestamp, 'state': 'i', 'threshold':600, 'message': 'start message for process 6'})
self.e.convert_state('1', 'S')
self.e.convert_state('2', 'S')
self.e.convert_state('3', 'S')
self.e.convert_state('4', 'S')
self.e.convert_state('5', 'S')
self.e.convert_state('6', 'S')
self.assertEqual(self.e.state_dict['2']['state'], 'a')
self.assertEqual(self.e.state_dict['3']['state'], 'a')
self.assertEqual(self.e.state_dict['4']['state'], 'a')
self.e.add_process('2', 'E', self.timestamp, None, 'end message for process 2')
self.e.add_process('3', 'E', self.timestamp, None, 'end message for process 3')
self.e.add_process('4', 'E', self.timestamp, None, 'end message for process 4')
self.assertEqual(self.e.state_dict['2']['state'], 'i')
self.assertEqual(self.e.state_dict['3']['state'], 'i')
self.assertEqual(self.e.state_dict['4']['state'], 'i')
def test_active_start_conversion(self):
print self.e
print 'trying...'
import sys
from StringIO import StringIO
orig = sys.stdout
try:
output = StringIO()
sys.stdout = output
self.e.convert_state('1', 'S')
out = output.getvalue().strip()
test = "Process ID:", '1', "\n" \
"Process Message Type:", 'S', "\n" \
"Process timestamp:", self.timestamp, "\n" \
"Process Threshold:", 100, "\n" \
"Process Message:", 'start message for process 1', "\n" \
"Log Message:", "PROCESS WITH ID 1 SENT MULTIPLE START MESSAGES", "\n"
self.assertEqual(out, test)
finally:
sys.stdout = orig
if __name__ == '__main__':
unimain()
'e' 是我想在函数调用之间保持修改的变量。当我下降到TestFSM.test_active_start_conversion 时,e 的大小打印出 0,而它应该是 6。TestFSM.test_add_convert_processes 方法运行成功。
实际错误是关键错误。这是堆栈跟踪:
Error
Traceback (most recent call last):
File "/home/Desktop/fsm_unit_tests.py", line 68, in test_active_start_conversion
self.e.convert_state('1', 'S')
File "/home/Desktop/finite_state_machine.py", line 26, in convert_state
cb_id = self.state_dict[id]['id']
KeyError: '1'
引发错误的行是self.e.convert_state('1', 'S'),我告诉 fsm 更改 ID 为“1”的进程的状态。它认为没有 ID 为 1 的进程,并且在打印 fsm 的大小时,它认为有限状态机是空的。这一定是因为 e 没有被持续维护,但我不明白为什么。
我已尝试将 self.e 转换为 self.__class__.e(与时间戳相同)以及仅将 e 和时间戳保持全局并调用 TestFSM.e 和 TestFSM.timestamp 以保持更改。这两种解决方案都列在其他问题中,但它们仍然产生了 Key Error。我尝试设置setUpClass(),但这仍然产生了一个关键错误。
我怎样才能持久化 e 和时间戳?
【问题讨论】:
-
在我看来,您依赖于执行单元测试的 order。在这种情况下,我猜
test_active_start_conversion在test_add_convert_processes之前运行。由于方法存储在类字典中并且字典是无序的,因此没有定义执行顺序(甚至可以在某些 python 版本/实现中从一次运行到下一次运行)。 -
谢谢。似乎订购是问题所在。我将测试名称更改为“test_1”和“test_2”,至少它没有抛出 KeyError。我不知道排序,因为我认为它会按顺序解释
-
您可以使用
sqlite3或shelve数据库来持久化数据,而不是将其存储到类实例中的字典中。您还可以在单独进程中运行的循环中将数据保存到字典中(例如tornado或asyncio)。 -- 我建议这样做,因为它有助于线程安全、锁定、范围,并且会在您的测试期间持续存在。考虑到您当前的设置,shelve可能是最容易使用的。
标签: python