Remove extra runner layer and just use use machine in engine

Just directly use the built machine in the action engine and
avoid having another layer of abstraction that does not provide
that much value. This makes the code cleaner, and more easy to
understand (and so-on).

Change-Id: Iae1279098112254338258c1941c15889f1ad1a79

1 files changed, 98 insertions, 99 deletions

diff --git a/taskflow/tests/unit/action_engine/test_runner.py b/taskflow/tests/unit/action_engine/test_builder.py
index bff74cb..b406744 100644
--- a/taskflow/tests/unit/action_engine/test_runner.py
+++ b/taskflow/tests/unit/action_engine/test_builder.py
@@ -15,11 +15,12 @@
 #    under the License.
 
 from automaton import exceptions as excp
+from automaton import runners
 import six
 
+from taskflow.engines.action_engine import builder
 from taskflow.engines.action_engine import compiler
 from taskflow.engines.action_engine import executor
-from taskflow.engines.action_engine import runner
 from taskflow.engines.action_engine import runtime
 from taskflow.patterns import linear_flow as lf
 from taskflow import states as st
@@ -30,7 +31,8 @@ from taskflow.types import notifier
 from taskflow.utils import persistence_utils as pu
 
 
-class _RunnerTestMixin(object):
+class BuildersTest(test.TestCase):
+
     def _make_runtime(self, flow, initial_state=None):
         compilation = compiler.PatternCompiler(flow).compile()
         flow_detail = pu.create_flow_detail(flow)
@@ -51,17 +53,11 @@ class _RunnerTestMixin(object):
         r.compile()
         return r
 
-
-class RunnerTest(test.TestCase, _RunnerTestMixin):
-    def test_running(self):
-        flow = lf.Flow("root")
-        flow.add(*test_utils.make_many(1))
-
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        self.assertTrue(rt.runner.runnable())
-
-        rt = self._make_runtime(flow, initial_state=st.SUSPENDED)
-        self.assertFalse(rt.runner.runnable())
+    def _make_machine(self, flow, initial_state=None):
+        runtime = self._make_runtime(flow, initial_state=initial_state)
+        machine, memory = runtime.builder.build()
+        machine_runner = runners.FiniteRunner(machine)
+        return (runtime, machine, memory, machine_runner)
 
     def test_run_iterations(self):
         flow = lf.Flow("root")
@@ -69,29 +65,32 @@ class RunnerTest(test.TestCase, _RunnerTestMixin):
             1, task_cls=test_utils.TaskNoRequiresNoReturns)
         flow.add(*tasks)
 
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        self.assertTrue(rt.runner.runnable())
+        runtime, machine, memory, machine_runner = self._make_machine(
+            flow, initial_state=st.RUNNING)
 
-        it = rt.runner.run_iter()
-        state, failures = six.next(it)
-        self.assertEqual(st.RESUMING, state)
-        self.assertEqual(0, len(failures))
+        it = machine_runner.run_iter(builder.START)
+        prior_state, new_state = six.next(it)
+        self.assertEqual(st.RESUMING, new_state)
+        self.assertEqual(0, len(memory.failures))
 
-        state, failures = six.next(it)
-        self.assertEqual(st.SCHEDULING, state)
-        self.assertEqual(0, len(failures))
+        prior_state, new_state = six.next(it)
+        self.assertEqual(st.SCHEDULING, new_state)
+        self.assertEqual(0, len(memory.failures))
 
-        state, failures = six.next(it)
-        self.assertEqual(st.WAITING, state)
-        self.assertEqual(0, len(failures))
+        prior_state, new_state = six.next(it)
+        self.assertEqual(st.WAITING, new_state)
+        self.assertEqual(0, len(memory.failures))
 
-        state, failures = six.next(it)
-        self.assertEqual(st.ANALYZING, state)
-        self.assertEqual(0, len(failures))
+        prior_state, new_state = six.next(it)
+        self.assertEqual(st.ANALYZING, new_state)
+        self.assertEqual(0, len(memory.failures))
 
-        state, failures = six.next(it)
-        self.assertEqual(st.SUCCESS, state)
-        self.assertEqual(0, len(failures))
+        prior_state, new_state = six.next(it)
+        self.assertEqual(builder.GAME_OVER, new_state)
+        self.assertEqual(0, len(memory.failures))
+        prior_state, new_state = six.next(it)
+        self.assertEqual(st.SUCCESS, new_state)
+        self.assertEqual(0, len(memory.failures))
 
         self.assertRaises(StopIteration, six.next, it)
 
@@ -101,15 +100,15 @@ class RunnerTest(test.TestCase, _RunnerTestMixin):
             1, task_cls=test_utils.TaskWithFailure)
         flow.add(*tasks)
 
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        self.assertTrue(rt.runner.runnable())
+        runtime, machine, memory, machine_runner = self._make_machine(
+            flow, initial_state=st.RUNNING)
 
-        transitions = list(rt.runner.run_iter())
-        state, failures = transitions[-1]
-        self.assertEqual(st.REVERTED, state)
-        self.assertEqual([], failures)
-
-        self.assertEqual(st.REVERTED, rt.storage.get_atom_state(tasks[0].name))
+        transitions = list(machine_runner.run_iter(builder.START))
+        prior_state, new_state = transitions[-1]
+        self.assertEqual(st.REVERTED, new_state)
+        self.assertEqual([], memory.failures)
+        self.assertEqual(st.REVERTED,
+                         runtime.storage.get_atom_state(tasks[0].name))
 
     def test_run_iterations_failure(self):
         flow = lf.Flow("root")
@@ -117,18 +116,17 @@ class RunnerTest(test.TestCase, _RunnerTestMixin):
             1, task_cls=test_utils.NastyFailingTask)
         flow.add(*tasks)
 
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        self.assertTrue(rt.runner.runnable())
+        runtime, machine, memory, machine_runner = self._make_machine(
+            flow, initial_state=st.RUNNING)
 
-        transitions = list(rt.runner.run_iter())
-        state, failures = transitions[-1]
-        self.assertEqual(st.FAILURE, state)
-        self.assertEqual(1, len(failures))
-        failure = failures[0]
+        transitions = list(machine_runner.run_iter(builder.START))
+        prior_state, new_state = transitions[-1]
+        self.assertEqual(st.FAILURE, new_state)
+        self.assertEqual(1, len(memory.failures))
+        failure = memory.failures[0]
         self.assertTrue(failure.check(RuntimeError))
-
         self.assertEqual(st.REVERT_FAILURE,
-                         rt.storage.get_atom_state(tasks[0].name))
+                         runtime.storage.get_atom_state(tasks[0].name))
 
     def test_run_iterations_suspended(self):
         flow = lf.Flow("root")
@@ -136,20 +134,22 @@ class RunnerTest(test.TestCase, _RunnerTestMixin):
             2, task_cls=test_utils.TaskNoRequiresNoReturns)
         flow.add(*tasks)
 
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        self.assertTrue(rt.runner.runnable())
+        runtime, machine, memory, machine_runner = self._make_machine(
+            flow, initial_state=st.RUNNING)
 
         transitions = []
-        for state, failures in rt.runner.run_iter():
-            transitions.append((state, failures))
-            if state == st.ANALYZING:
-                rt.storage.set_flow_state(st.SUSPENDED)
+        for prior_state, new_state in machine_runner.run_iter(builder.START):
+            transitions.append((new_state, memory.failures))
+            if new_state == st.ANALYZING:
+                runtime.storage.set_flow_state(st.SUSPENDED)
         state, failures = transitions[-1]
         self.assertEqual(st.SUSPENDED, state)
         self.assertEqual([], failures)
 
-        self.assertEqual(st.SUCCESS, rt.storage.get_atom_state(tasks[0].name))
-        self.assertEqual(st.PENDING, rt.storage.get_atom_state(tasks[1].name))
+        self.assertEqual(st.SUCCESS,
+                         runtime.storage.get_atom_state(tasks[0].name))
+        self.assertEqual(st.PENDING,
+                         runtime.storage.get_atom_state(tasks[1].name))
 
     def test_run_iterations_suspended_failure(self):
         flow = lf.Flow("root")
@@ -160,46 +160,44 @@ class RunnerTest(test.TestCase, _RunnerTestMixin):
             1, task_cls=test_utils.TaskNoRequiresNoReturns, offset=1)
         flow.add(*happy_tasks)
 
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        self.assertTrue(rt.runner.runnable())
+        runtime, machine, memory, machine_runner = self._make_machine(
+            flow, initial_state=st.RUNNING)
 
         transitions = []
-        for state, failures in rt.runner.run_iter():
-            transitions.append((state, failures))
-            if state == st.ANALYZING:
-                rt.storage.set_flow_state(st.SUSPENDED)
+        for prior_state, new_state in machine_runner.run_iter(builder.START):
+            transitions.append((new_state, memory.failures))
+            if new_state == st.ANALYZING:
+                runtime.storage.set_flow_state(st.SUSPENDED)
         state, failures = transitions[-1]
         self.assertEqual(st.SUSPENDED, state)
         self.assertEqual([], failures)
 
         self.assertEqual(st.PENDING,
-                         rt.storage.get_atom_state(happy_tasks[0].name))
+                         runtime.storage.get_atom_state(happy_tasks[0].name))
         self.assertEqual(st.FAILURE,
-                         rt.storage.get_atom_state(sad_tasks[0].name))
+                         runtime.storage.get_atom_state(sad_tasks[0].name))
 
-
-class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
     def test_builder_manual_process(self):
         flow = lf.Flow("root")
         tasks = test_utils.make_many(
             1, task_cls=test_utils.TaskNoRequiresNoReturns)
         flow.add(*tasks)
 
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        machine, machine_runner, memory = rt.runner.build()
-        self.assertTrue(rt.runner.runnable())
+        runtime, machine, memory, machine_runner = self._make_machine(
+            flow, initial_state=st.RUNNING)
         self.assertRaises(excp.NotInitialized, machine.process_event, 'poke')
 
         # Should now be pending...
-        self.assertEqual(st.PENDING, rt.storage.get_atom_state(tasks[0].name))
+        self.assertEqual(st.PENDING,
+                         runtime.storage.get_atom_state(tasks[0].name))
 
         machine.initialize()
-        self.assertEqual(runner._UNDEFINED, machine.current_state)
+        self.assertEqual(builder.UNDEFINED, machine.current_state)
         self.assertFalse(machine.terminated)
         self.assertRaises(excp.NotFound, machine.process_event, 'poke')
         last_state = machine.current_state
 
-        reaction, terminal = machine.process_event('start')
+        reaction, terminal = machine.process_event(builder.START)
         self.assertFalse(terminal)
         self.assertIsNotNone(reaction)
         self.assertEqual(st.RESUMING, machine.current_state)
@@ -208,7 +206,7 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
         last_state = machine.current_state
         cb, args, kwargs = reaction
         next_event = cb(last_state, machine.current_state,
-                        'start', *args, **kwargs)
+                        builder.START, *args, **kwargs)
         reaction, terminal = machine.process_event(next_event)
         self.assertFalse(terminal)
         self.assertIsNotNone(reaction)
@@ -225,7 +223,8 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
         self.assertRaises(excp.NotFound, machine.process_event, 'poke')
 
         # Should now be running...
-        self.assertEqual(st.RUNNING, rt.storage.get_atom_state(tasks[0].name))
+        self.assertEqual(st.RUNNING,
+                         runtime.storage.get_atom_state(tasks[0].name))
 
         last_state = machine.current_state
         cb, args, kwargs = reaction
@@ -243,10 +242,11 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
                         next_event, *args, **kwargs)
         reaction, terminal = machine.process_event(next_event)
         self.assertFalse(terminal)
-        self.assertEqual(runner._GAME_OVER, machine.current_state)
+        self.assertEqual(builder.GAME_OVER, machine.current_state)
 
         # Should now be done...
-        self.assertEqual(st.SUCCESS, rt.storage.get_atom_state(tasks[0].name))
+        self.assertEqual(st.SUCCESS,
+                         runtime.storage.get_atom_state(tasks[0].name))
 
     def test_builder_automatic_process(self):
         flow = lf.Flow("root")
@@ -254,26 +254,25 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
             1, task_cls=test_utils.TaskNoRequiresNoReturns)
         flow.add(*tasks)
 
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        machine, machine_runner, memory = rt.runner.build()
-        self.assertTrue(rt.runner.runnable())
+        runtime, machine, memory, machine_runner = self._make_machine(
+            flow, initial_state=st.RUNNING)
 
-        transitions = list(machine_runner.run_iter('start'))
-        self.assertEqual((runner._UNDEFINED, st.RESUMING), transitions[0])
-        self.assertEqual((runner._GAME_OVER, st.SUCCESS), transitions[-1])
-        self.assertEqual(st.SUCCESS, rt.storage.get_atom_state(tasks[0].name))
+        transitions = list(machine_runner.run_iter(builder.START))
+        self.assertEqual((builder.UNDEFINED, st.RESUMING), transitions[0])
+        self.assertEqual((builder.GAME_OVER, st.SUCCESS), transitions[-1])
+        self.assertEqual(st.SUCCESS,
+                         runtime.storage.get_atom_state(tasks[0].name))
 
     def test_builder_automatic_process_failure(self):
         flow = lf.Flow("root")
         tasks = test_utils.make_many(1, task_cls=test_utils.NastyFailingTask)
         flow.add(*tasks)
 
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        machine, machine_runner, memory = rt.runner.build()
-        self.assertTrue(rt.runner.runnable())
+        runtime, machine, memory, machine_runner = self._make_machine(
+            flow, initial_state=st.RUNNING)
 
-        transitions = list(machine_runner.run_iter('start'))
-        self.assertEqual((runner._GAME_OVER, st.FAILURE), transitions[-1])
+        transitions = list(machine_runner.run_iter(builder.START))
+        self.assertEqual((builder.GAME_OVER, st.FAILURE), transitions[-1])
         self.assertEqual(1, len(memory.failures))
 
     def test_builder_automatic_process_reverted(self):
@@ -281,13 +280,13 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
         tasks = test_utils.make_many(1, task_cls=test_utils.TaskWithFailure)
         flow.add(*tasks)
 
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        machine, machine_runner, memory = rt.runner.build()
-        self.assertTrue(rt.runner.runnable())
+        runtime, machine, memory, machine_runner = self._make_machine(
+            flow, initial_state=st.RUNNING)
 
-        transitions = list(machine_runner.run_iter('start'))
-        self.assertEqual((runner._GAME_OVER, st.REVERTED), transitions[-1])
-        self.assertEqual(st.REVERTED, rt.storage.get_atom_state(tasks[0].name))
+        transitions = list(machine_runner.run_iter(builder.START))
+        self.assertEqual((builder.GAME_OVER, st.REVERTED), transitions[-1])
+        self.assertEqual(st.REVERTED,
+                         runtime.storage.get_atom_state(tasks[0].name))
 
     def test_builder_expected_transition_occurrences(self):
         flow = lf.Flow("root")
@@ -295,16 +294,16 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
             10, task_cls=test_utils.TaskNoRequiresNoReturns)
         flow.add(*tasks)
 
-        rt = self._make_runtime(flow, initial_state=st.RUNNING)
-        machine, machine_runner, memory = rt.runner.build()
-        transitions = list(machine_runner.run_iter('start'))
+        runtime, machine, memory, machine_runner = self._make_machine(
+            flow, initial_state=st.RUNNING)
+        transitions = list(machine_runner.run_iter(builder.START))
 
         occurrences = dict((t, transitions.count(t)) for t in transitions)
         self.assertEqual(10, occurrences.get((st.SCHEDULING, st.WAITING)))
         self.assertEqual(10, occurrences.get((st.WAITING, st.ANALYZING)))
         self.assertEqual(9, occurrences.get((st.ANALYZING, st.SCHEDULING)))
-        self.assertEqual(1, occurrences.get((runner._GAME_OVER, st.SUCCESS)))
-        self.assertEqual(1, occurrences.get((runner._UNDEFINED, st.RESUMING)))
+        self.assertEqual(1, occurrences.get((builder.GAME_OVER, st.SUCCESS)))
+        self.assertEqual(1, occurrences.get((builder.UNDEFINED, st.RESUMING)))
 
         self.assertEqual(0, len(memory.next_nodes))
         self.assertEqual(0, len(memory.not_done))