(self)
| 19 | Task.thread_id_pool = 0 |
| 20 | |
| 21 | def testTree(self): |
| 22 | # Build a tree. |
| 23 | spec = WorkflowSpec(name='Mock Workflow') |
| 24 | workflow = MockWorkflow(spec) |
| 25 | task1 = Simple(spec, 'Simple 1') |
| 26 | task2 = Simple(spec, 'Simple 2') |
| 27 | task3 = Simple(spec, 'Simple 3') |
| 28 | task4 = Simple(spec, 'Simple 4') |
| 29 | task5 = Simple(spec, 'Simple 5') |
| 30 | task6 = Simple(spec, 'Simple 6') |
| 31 | task7 = Simple(spec, 'Simple 7') |
| 32 | task8 = Simple(spec, 'Simple 8') |
| 33 | task9 = Simple(spec, 'Simple 9') |
| 34 | root = Task(workflow, task1) |
| 35 | c1 = root._add_child(task2) |
| 36 | c11 = c1._add_child(task3) |
| 37 | c111 = c11._add_child(task4) |
| 38 | Task(workflow, task5, c111) |
| 39 | Task(workflow, task6, c11) |
| 40 | Task(workflow, task7, c1) |
| 41 | Task(workflow, task8, root) |
| 42 | c3 = Task(workflow, task9, root) |
| 43 | c3.state = TaskState.COMPLETED |
| 44 | |
| 45 | # Check whether the tree is built properly. |
| 46 | expected = """!/0: Task of Simple 1 State: MAYBE Children: 3 |
| 47 | !/0: Task of Simple 2 State: MAYBE Children: 2 |
| 48 | !/0: Task of Simple 3 State: MAYBE Children: 2 |
| 49 | !/0: Task of Simple 4 State: MAYBE Children: 1 |
| 50 | !/0: Task of Simple 5 State: MAYBE Children: 0 |
| 51 | !/0: Task of Simple 6 State: MAYBE Children: 0 |
| 52 | !/0: Task of Simple 7 State: MAYBE Children: 0 |
| 53 | !/0: Task of Simple 8 State: MAYBE Children: 0 |
| 54 | !/0: Task of Simple 9 State: COMPLETED Children: 0""" |
| 55 | expected = re.compile(expected.replace('!', r'([0-9a-f\-]+)')) |
| 56 | self.assertTrue(expected.match(root.get_dump()), |
| 57 | 'Expected:\n' + repr(expected.pattern) + '\n' + |
| 58 | 'but got:\n' + repr(root.get_dump())) |
| 59 | |
| 60 | # Now remove one line from the expected output for testing the |
| 61 | # filtered iterator. |
| 62 | expected2 = '' |
| 63 | for line in expected.pattern.split('\n'): |
| 64 | if line.find('Simple 9') >= 0: |
| 65 | continue |
| 66 | expected2 += line.lstrip() + '\n' |
| 67 | expected2 = re.compile(expected2) |
| 68 | |
| 69 | # Run the iterator test. |
| 70 | result = '' |
| 71 | for thetask in TaskIterator(root, state=TaskState.MAYBE): |
| 72 | result += thetask.get_dump(0, False) + '\n' |
| 73 | self.assertTrue(expected2.match(result), |
| 74 | 'Expected:\n' + repr(expected2.pattern) + '\n' + |
| 75 | 'but got:\n' + repr(result)) |
nothing calls this directly
no test coverage detected