openstack/deb-python-taskflow
Code Issues Proposed changes

Add tests for metadata updates

Browse Source 
This change adds couple of unit tests to persistence base test case,
to verify that metadata updates are not lost.

Change-Id: I467b21eb59e9b85a689086d09c775b714a70d02a
This commit is contained in:
Ivan A. Melnikov
2013-12-18 10:34:24 +02:00
parent 3a8faf8850
commit 08f1224c83
1 changed files with 45 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
taskflow/tests/unit/persistence/base.py
View File

@@ -69,6 +69,26 @@ class PersistenceTestMixin(object):
conn.save_logbook(lb)
conn.update_flow_details(fd)
def test_flow_detail_meta_update(self):
lb_id = uuidutils.generate_uuid()
lb_name = 'lb-%s' % (lb_id)
lb = logbook.LogBook(name=lb_name, uuid=lb_id)
fd = logbook.FlowDetail('test', uuid=uuidutils.generate_uuid())
fd.meta = {'test': 42}
lb.add(fd)
with contextlib.closing(self._get_connection()) as conn:
conn.save_logbook(lb)
conn.update_flow_details(fd)
fd.meta['test'] = 43
with contextlib.closing(self._get_connection()) as conn:
conn.update_flow_details(fd)
with contextlib.closing(self._get_connection()) as conn:
lb2 = conn.get_logbook(lb_id)
fd2 = lb2.find(fd.uuid)
self.assertEqual(fd2.meta.get('test'), 43)
def test_task_detail_save(self):
lb_id = uuidutils.generate_uuid()
lb_name = 'lb-%s' % (lb_id)
@@ -91,6 +111,31 @@ class PersistenceTestMixin(object):
conn.update_flow_details(fd)
conn.update_task_details(td)
def test_task_detail_meta_update(self):
lb_id = uuidutils.generate_uuid()
lb_name = 'lb-%s' % (lb_id)
lb = logbook.LogBook(name=lb_name, uuid=lb_id)
fd = logbook.FlowDetail('test', uuid=uuidutils.generate_uuid())
lb.add(fd)
td = logbook.TaskDetail("detail-1", uuid=uuidutils.generate_uuid())
td.meta = {'test': 42}
fd.add(td)
with contextlib.closing(self._get_connection()) as conn:
conn.save_logbook(lb)
conn.update_flow_details(fd)
conn.update_task_details(td)
td.meta['test'] = 43
with contextlib.closing(self._get_connection()) as conn:
conn.update_task_details(td)
with contextlib.closing(self._get_connection()) as conn:
lb2 = conn.get_logbook(lb_id)
fd2 = lb2.find(fd.uuid)
td2 = fd2.find(td.uuid)
self.assertEqual(td2.meta.get('test'), 43)
def test_task_detail_with_failure(self):
lb_id = uuidutils.generate_uuid()
lb_name = 'lb-%s' % (lb_id)