Add FlowExtension

It makes possible of running data-driven flow which's the list of
the other existent extensions to be called.

Change-Id: Ib73ea4da92f291c872b7ae51e46ecc7fdd45ee16

ironic_python_agent/flow.py

@@ -0,0 +1,65 @@
+# Copyright 2014 Mirantis, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from stevedore import enabled
+
+from ironic_python_agent import base
+from ironic_python_agent import decorators
+from ironic_python_agent import errors
+from ironic_python_agent.openstack.common import log
+
+LOG = log.getLogger(__name__)
+
+
+def _load_extension(ext):
+    disabled_extension_list = ['flow']
+    return ext.name not in disabled_extension_list
+
+
+def _validate_exts(ext, flow=None):
+    for task in flow:
+        for method in task:
+            ext_name, cmd = ext.split_command(method)
+            if ext_name not in ext.ext_mgr.names():
+                raise errors.RequestedObjectNotFoundError('Extension',
+                                                          ext_name)
+            ext_obj = ext.ext_mgr[ext_name].obj
+            ext.check_cmd_presence(ext_obj, ext_name, cmd)
+
+
+class FlowExtension(base.BaseAgentExtension, base.ExecuteCommandMixin):
+    def __init__(self):
+        super(FlowExtension, self).__init__('FLOW')
+        self.command_map['start_flow'] = self.start_flow
+
+    def get_extension_manager(self):
+        return enabled.EnabledExtensionManager(
+            'ironic_python_agent.extensions',
+            _load_extension,
+            invoke_on_load=True,
+            propagate_map_exceptions=True,
+        )
+
+    @decorators.async_command(_validate_exts)
+    def start_flow(self, command_name, flow=None):
+        for task in flow:
+            for method, params in task.items():
+                LOG.info("Executing method %s for now" % method)
+                result = self.execute_command(method, **params)
+                result.join()
+                LOG.info("%s method's execution is done" % method)
+                if result.command_status == base.AgentCommandStatus.FAILED:
+                    raise errors.CommandExecutionError(
+                        "%s was failed" % method
+                    )

ironic_python_agent/tests/flow.py

@@ -0,0 +1,114 @@
+# Copyright 2014 Mirantis, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import time
+
+import mock
+from oslotest import base as test_base
+from stevedore import enabled
+from stevedore import extension
+
+from ironic_python_agent import base
+from ironic_python_agent import decorators
+from ironic_python_agent import errors
+from ironic_python_agent import flow
+
+
+FLOW_INFO = [
+    {"fake.sleep": {"sleep_info": {"time": 1}}},
+    {"fake.sleep": {"sleep_info": {"time": 2}}},
+    {"fake.sync_sleep": {"sleep_info": {"time": 3}}},
+    {"fake.sleep": {"sleep_info": {"time": 4}}},
+    {"fake.sync_sleep": {"sleep_info": {"time": 5}}},
+    {"fake.sleep": {"sleep_info": {"time": 6}}},
+    {"fake.sleep": {"sleep_info": {"time": 7}}},
+]
+
+
+class FakeExtension(base.BaseAgentExtension):
+    def __init__(self):
+        super(FakeExtension, self).__init__('FAKE')
+        self.command_map['sleep'] = self.sleep
+        self.command_map['sync_sleep'] = self.sync_sleep
+
+    @decorators.async_command()
+    def sleep(self, command_name, sleep_info=None):
+        time.sleep(sleep_info['time'])
+
+    def sync_sleep(self, command_name, sleep_info=None):
+        time.sleep(sleep_info['time'])
+
+
+class TestFlowExtension(test_base.BaseTestCase):
+    def setUp(self):
+        super(TestFlowExtension, self).setUp()
+        self.agent_extension = flow.FlowExtension()
+        self.agent_extension.ext_mgr = enabled.EnabledExtensionManager.\
+            make_test_instance([extension.Extension('fake', None,
+                                                    FakeExtension,
+                                                    FakeExtension())])
+
+    def test_flow_extension(self):
+        self.assertEqual(self.agent_extension.name, 'FLOW')
+
+    @mock.patch('time.sleep', autospec=True)
+    def test_sleep_flow_success(self, sleep_mock):
+        result = self.agent_extension.start_flow('start_flow', flow=FLOW_INFO)
+        result.join()
+        sleep_calls = [mock.call(i) for i in range(1, 8)]
+        sleep_mock.assert_has_calls(sleep_calls)
+
+    @mock.patch('time.sleep', autospec=True)
+    def test_sleep_flow_failed(self, sleep_mock):
+        sleep_mock.side_effect = errors.RESTError()
+        result = self.agent_extension.start_flow('start_flow', flow=FLOW_INFO)
+        result.join()
+        self.assertEqual(base.AgentCommandStatus.FAILED, result.command_status)
+        self.assertTrue(isinstance(result.command_error,
+                                   errors.CommandExecutionError))
+
+    @mock.patch('time.sleep', autospec=True)
+    def test_sleep_flow_failed_on_second_command(self, sleep_mock):
+        sleep_mock.side_effect = [None, Exception('foo'), None, None]
+        result = self.agent_extension.start_flow('start_flow',
+                                                 flow=FLOW_INFO[:4])
+        result.join()
+        self.assertEqual(base.AgentCommandStatus.FAILED, result.command_status)
+        self.assertTrue(isinstance(result.command_error,
+                                   errors.CommandExecutionError))
+        self.assertEqual(2, sleep_mock.call_count)
+
+    def test_validate_exts_success(self):
+        flow._validate_exts(self.agent_extension, flow=FLOW_INFO)
+
+    def test_validate_exts_failed_to_find_extension(self):
+        self.agent_extension.ext_mgr.names = mock.Mock()
+        self.agent_extension.ext_mgr.names.return_value = ['fake_fake']
+        self.assertRaises(errors.RequestedObjectNotFoundError,
+                          flow._validate_exts, self.agent_extension,
+                          flow=FLOW_INFO)
+
+    def test_validate_exts_failed_empty_command_map(self):
+        fake_ext = self.agent_extension.ext_mgr['fake'].obj
+        delattr(fake_ext, 'command_map')
+        self.assertRaises(errors.InvalidCommandParamsError,
+                          flow._validate_exts, self.agent_extension,
+                          flow=FLOW_INFO)
+
+    def test_validate_exts_failed_missing_command(self):
+        fake_ext = self.agent_extension.ext_mgr['fake'].obj
+        fake_ext.command_map = {'not_exist': 'fake'}
+        self.assertRaises(errors.InvalidCommandParamsError,
+                          flow._validate_exts, self.agent_extension,
+                          flow=FLOW_INFO)

setup.cfg

@@ -21,6 +21,7 @@ console_scripts =
 ironic_python_agent.extensions =
     standby = ironic_python_agent.standby:StandbyExtension
     decom = ironic_python_agent.decom:DecomExtension
+    flow = ironic_python_agent.flow:FlowExtension
 
 ironic_python_agent.hardware_managers =
     generic = ironic_python_agent.hardware:GenericHardwareManager