Merge "[Sahara] Split Data Sources context"

rally/plugins/openstack/context/sahara/sahara_input_data_sources.py

@@ -0,0 +1,78 @@
+# Copyright 2015: Mirantis Inc.
+# All Rights Reserved.
+#
+# 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 rally.common.i18n import _
+from rally.common import log as logging
+from rally.common import utils as rutils
+from rally import consts
+from rally import exceptions
+from rally import osclients
+from rally.plugins.openstack.context.cleanup import manager as resource_manager
+from rally.task import context
+
+LOG = logging.getLogger(__name__)
+
+
+@context.configure(name="sahara_input_data_sources", order=443)
+class SaharaInputDataSources(context.Context):
+    """Context class for setting up Input Data Sources for an EDP job."""
+
+    CONFIG_SCHEMA = {
+        "type": "object",
+        "$schema": consts.JSON_SCHEMA,
+        "properties": {
+            "input_type": {
+                "enum": ["swift", "hdfs"],
+            },
+            "input_url": {
+                "type": "string",
+            },
+        },
+        "additionalProperties": False,
+        "required": ["input_type", "input_url"]
+    }
+
+    @logging.log_task_wrapper(LOG.info,
+                              _("Enter context: `Sahara Input Data Sources`"))
+    def setup(self):
+        for user, tenant_id in rutils.iterate_per_tenants(
+                self.context["users"]):
+            clients = osclients.Clients(user["endpoint"])
+            sahara = clients.sahara()
+            self.setup_inputs(sahara, tenant_id, self.config["input_type"],
+                              self.config["input_url"])
+
+    def setup_inputs(self, sahara, tenant_id, input_type, input_url):
+        if input_type == "swift":
+            raise exceptions.RallyException(
+                _("Swift Data Sources are not implemented yet"))
+        # Todo(nkonovalov): Add swift credentials parameters and data upload
+        input_ds = sahara.data_sources.create(
+            name=self.generate_random_name(),
+            description="",
+            data_source_type=input_type,
+            url=input_url)
+
+        self.context["tenants"][tenant_id]["sahara_input"] = input_ds.id
+
+    @logging.log_task_wrapper(LOG.info, _("Exit context: `Sahara Input Data"
+                                          "Sources`"))
+    def cleanup(self):
+        resources = ["job_executions", "jobs", "data_sources"]
+
+        # TODO(boris-42): Delete only resources created by this context
+        resource_manager.cleanup(
+            names=["sahara.%s" % res for res in resources],
+            users=self.context.get("users", []))

rally/plugins/openstack/context/sahara/sahara_output_data_sources.py

@@ -0,0 +1,59 @@
+# Copyright 2015: Mirantis Inc.
+# All Rights Reserved.
+#
+# 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 rally.common.i18n import _
+from rally.common import log as logging
+from rally.common import utils as rutils
+from rally import consts
+from rally.task import context
+
+
+LOG = logging.getLogger(__name__)
+
+
+@context.configure(name="sahara_output_data_sources", order=444)
+class SaharaOutputDataSources(context.Context):
+    """Context class for setting up Output Data Sources for an EDP job."""
+
+    CONFIG_SCHEMA = {
+        "type": "object",
+        "$schema": consts.JSON_SCHEMA,
+        "properties": {
+            "output_type": {
+                "enum": ["swift", "hdfs"],
+            },
+            "output_url_prefix": {
+                "type": "string",
+            }
+        },
+        "additionalProperties": False,
+        "required": ["output_type", "output_url_prefix"]
+    }
+
+    @logging.log_task_wrapper(LOG.info,
+                              _("Enter context: `Sahara Output Data Sources`"))
+    def setup(self):
+        for user, tenant_id in rutils.iterate_per_tenants(
+                self.context["users"]):
+            self.context["tenants"][tenant_id]["sahara_output_conf"] = {
+                "output_type": self.config["output_type"],
+                "output_url_prefix": self.config["output_url_prefix"]}
+
+    @logging.log_task_wrapper(LOG.info, _("Exit context: `Sahara Output Data"
+                                          "Sources`"))
+    def cleanup(self):
+        # TODO(esikachev): Cleanup must iterate by output_url_prefix of
+        # resources from setup() and delete them
+        pass

tests/unit/plugins/openstack/context/sahara/test_sahara_input_data_sources.py

@@ -0,0 +1,93 @@
+# All Rights Reserved.
+#
+# 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 mock
+
+from rally.plugins.openstack.context.sahara import sahara_input_data_sources
+from tests.unit import test
+
+CTX = "rally.plugins.openstack.context.sahara"
+
+
+class SaharaInputDataSourcesTestCase(test.ScenarioTestCase):
+
+    def setUp(self):
+        super(SaharaInputDataSourcesTestCase, self).setUp()
+        self.tenants_num = 2
+        self.users_per_tenant = 2
+        self.users = self.tenants_num * self.users_per_tenant
+        self.task = mock.MagicMock()
+
+        self.tenants = {}
+        self.users_key = []
+
+        for i in range(self.tenants_num):
+            self.tenants[str(i)] = {"id": str(i), "name": str(i),
+                                    "sahara_image": "42"}
+            for j in range(self.users_per_tenant):
+                self.users_key.append({"id": "%s_%s" % (str(i), str(j)),
+                                       "tenant_id": str(i),
+                                       "endpoint": "endpoint"})
+
+        self.user_key = [{"id": i, "tenant_id": j, "endpoint": "endpoint"}
+                         for j in range(self.tenants_num)
+                         for i in range(self.users_per_tenant)]
+
+        self.context.update({
+            "config": {
+                "users": {
+                    "tenants": self.tenants_num,
+                    "users_per_tenant": self.users_per_tenant,
+                },
+                "sahara_input_data_sources": {
+                    "input_type": "hdfs",
+                    "input_url": "hdfs://test_host/",
+                },
+            },
+            "admin": {"endpoint": mock.MagicMock()},
+            "users": self.users_key,
+            "tenants": self.tenants
+        })
+
+    @mock.patch("%s.sahara_input_data_sources.resource_manager.cleanup" % CTX)
+    @mock.patch("%s.sahara_input_data_sources.osclients" % CTX)
+    def test_setup_and_cleanup(self, mock_osclients, mock_cleanup):
+
+        mock_sahara = mock_osclients.Clients(mock.MagicMock()).sahara()
+        mock_sahara.data_sources.create.return_value = mock.MagicMock(id=42)
+
+        sahara_ctx = sahara_input_data_sources.SaharaInputDataSources(
+            self.context)
+        sahara_ctx.generate_random_name = mock.Mock()
+
+        input_ds_crete_calls = []
+
+        for i in range(self.tenants_num):
+            input_ds_crete_calls.append(mock.call(
+                name=sahara_ctx.generate_random_name.return_value,
+                description="",
+                data_source_type="hdfs",
+                url="hdfs://test_host/"))
+
+        sahara_ctx.setup()
+
+        mock_sahara.data_sources.create.assert_has_calls(
+            input_ds_crete_calls)
+
+        sahara_ctx.cleanup()
+
+        mock_cleanup.assert_called_once_with(
+            names=["sahara.job_executions", "sahara.jobs",
+                   "sahara.data_sources"],
+            users=self.context["users"])

tests/unit/plugins/openstack/context/sahara/test_sahara_output_data_sources.py

@@ -0,0 +1,32 @@
+# All Rights Reserved.
+#
+# 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 rally.plugins.openstack.context.sahara import sahara_output_data_sources
+from tests.unit import test
+
+
+class SaharaOutputDataSourcesTestCase(test.TestCase):
+
+    def setUp(self):
+        super(SaharaOutputDataSourcesTestCase, self).setUp()
+
+    def check_setup(self):
+        context = sahara_output_data_sources.SaharaOutputDataSources.context[
+            "sahara_output_conf"]
+        self.assertIsNotNone(context.get("output_type"))
+        self.assertIsNotNone(context.get("output_url_prefix"))
+
+    def check_cleanup(self):
+        self.assertIsNone(
+            sahara_output_data_sources.SaharaOutputDataSources.cleanup())