Merge "validate network information during cluster create"

cue/api/controllers/v1/cluster.py

@@ -305,10 +305,10 @@ class ClusterController(rest.RestController):
             context, broker_name)
 
         job_args = {
+            'tenant_id': new_cluster.project_id,
             'flavor': cluster.flavor,
             'image': image_id,
             'volume_size': cluster.volume_size,
-            'network_id': cluster.network_id,
             'port': '5672',
             'context': context.to_dict(),
             # TODO(sputnik13: this needs to come from the create request and

cue/taskflow/flow/create_cluster.py

@@ -23,6 +23,7 @@ from cue.db.sqlalchemy import models
 from cue.taskflow.flow import create_cluster_node
 import cue.taskflow.task as cue_tasks
 import os_tasklib.common as os_common
+import os_tasklib.neutron as os_neutron
 import os_tasklib.nova as nova
 
 
@@ -103,6 +104,27 @@ def create_cluster(cluster_id, node_ids, user_network_id,
 
     flow.add(create_cluster_end_task)
 
+    show_network = os_neutron.ShowNetwork(
+        name="get tenant network information",
+        os_client=client.neutron_client(),
+        inject={'network': user_network_id},
+        provides="tenant_network_info"
+    )
+    flow.add(show_network)
+    flow.link(create_node_start_task, show_network)
+
+    validate_network_info = (lambda tenant_network_info, tenant_id:
+                             tenant_network_info['shared'] or
+                             tenant_network_info['tenant_id'] == tenant_id)
+
+    validate_tenant_network = os_common.Assert(
+        validate_network_info,
+        name="validate tenant network info",
+        requires=('tenant_network_info', 'tenant_id')
+    )
+    flow.add(validate_tenant_network)
+    flow.link(show_network, validate_tenant_network)
+
     node_check_timeout = cfg.CONF.taskflow.cluster_node_check_timeout
     node_check_max_count = cfg.CONF.taskflow.cluster_node_check_max_count
 
@@ -136,10 +158,11 @@ def create_cluster(cluster_id, node_ids, user_network_id,
                     'cluster_id': cluster_id})
         flow.add(generate_userdata)
 
-        create_cluster_node.create_cluster_node(
-            cluster_id=cluster_id, node_number=i, node_id=node_id,
-            graph_flow=flow, generate_userdata=generate_userdata,
-            start_task=create_node_start_task, post_task=check_rabbit_online,
-            user_network_id=user_network_id,
-            management_network_id=management_network_id)
+        create_cluster_node.create_cluster_node(cluster_id, i, node_id, flow,
+                                                generate_userdata,
+                                                validate_tenant_network,
+                                                check_rabbit_online,
+                                                user_network_id,
+                                                management_network_id)
+
     return flow

cue/tests/functional/fixtures/neutron.py

@@ -73,6 +73,7 @@ class NeutronClient(base.BaseFixture):
         v2_client.list_ports = self.list_ports
         v2_client.list_networks = self.list_networks
         v2_client.delete_port = self.delete_port
+        v2_client.show_network = self.show_network
 
     def create_port(self, body=None):
         """Mock'd version of neutronclient...create_port().
@@ -159,4 +160,16 @@ class NeutronClient(base.BaseFixture):
         if port_id in self._port_list:
             self._port_list.pop(port_id)
         else:
-            raise exceptions.NeutronClientException("404 Not found")
+            raise exceptions.NeutronClientException("404 Not found")
+
+    def show_network(self, network):
+        try:
+            network_id = network.id
+        except AttributeError:
+            network_id = network
+
+        if network_id in self._network_list:
+            return {'network': self._network_list[network_id]}
+        else:
+            raise exceptions.NeutronClientException(
+                "Network " + network_id + " could not be found.")

cue/tests/functional/taskflow/flow/test_check_cluster_status.py

@@ -81,6 +81,7 @@ class CheckClusterStatusTests(base.FunctionalTestCase):
 
     def test_check_cluster_status_active(self):
         flow_store_create = {
+            "tenant_id": str(self.valid_network['tenant_id']),
             "image": self.valid_image.id,
             "flavor": self.valid_flavor.id,
             "port": self.port,
@@ -147,6 +148,7 @@ class CheckClusterStatusTests(base.FunctionalTestCase):
 
     def test_check_cluster_status_down(self):
         flow_store_create = {
+            "tenant_id": str(self.valid_network['tenant_id']),
             "image": self.valid_image.id,
             "flavor": self.valid_flavor.id,
             "port": self.port,
@@ -215,6 +217,7 @@ class CheckClusterStatusTests(base.FunctionalTestCase):
 
     def test_check_cluster_status_size_one(self):
         flow_store_create = {
+            "tenant_id": str(self.valid_network['tenant_id']),
             "image": self.valid_image.id,
             "flavor": self.valid_flavor.id,
             "port": self.port,

cue/tests/functional/taskflow/flow/test_create_cluster.py

@@ -15,6 +15,8 @@
 
 import uuid
 
+from neutronclient.common import exceptions as neutron_exceptions
+
 from cue import client
 from cue.db.sqlalchemy import models
 from cue import objects
@@ -81,6 +83,7 @@ class CreateClusterTests(base.FunctionalTestCase):
 
     def test_create_cluster(self):
         flow_store = {
+            "tenant_id": str(self.valid_network['tenant_id']),
             "image": self.valid_image.id,
             "flavor": self.valid_flavor.id,
             "port": self.port,
@@ -153,6 +156,7 @@ class CreateClusterTests(base.FunctionalTestCase):
         port_list = self.neutron_client.list_ports()
 
         flow_store = {
+            "tenant_id": str(self.valid_network['tenant_id']),
             'image': self.valid_image.id,
             'flavor': self.valid_flavor.id,
             "port": self.port,
@@ -196,6 +200,7 @@ class CreateClusterTests(base.FunctionalTestCase):
         cluster_size = 3
 
         flow_store = {
+            "tenant_id": str(self.valid_network['tenant_id']),
             'image': self.valid_image.id,
             'flavor': self.valid_flavor.id,
             "port": self.port,
@@ -230,13 +235,12 @@ class CreateClusterTests(base.FunctionalTestCase):
 
         try:
             engines.run(flow, store=flow_store)
-        except taskflow_exc.WrappedFailure as err:
-            cluster_ref = objects.Cluster.get_cluster_by_id(self.context,
-                                                            new_cluster.id)
-            self.assertEqual(cluster_size, len(err._causes))
-            for failure in err._causes:
-                self.assertEqual(cluster_ref.error_detail,
-                                 failure.__str__())
+        except neutron_exceptions.NeutronClientException as err:
+            # When an incorrect user network ID is given, the neutron client
+            # returns a NeutronClientException.
+            self.assertEqual(err.message,
+                             "Network " + str(invalid_network_id) +
+                             " could not be found.")
         else:
             self.fail("Expected taskflow_exc.WrappedFailure exception.")
 
@@ -244,6 +248,7 @@ class CreateClusterTests(base.FunctionalTestCase):
         self.flags(cluster_node_anti_affinity=True, group="taskflow")
 
         flow_store = {
+            "tenant_id": str(self.valid_network['tenant_id']),
             'image': self.valid_image.id,
             'flavor': self.valid_flavor.id,
             "port": self.port,

cue/tests/functional/taskflow/flow/test_delete_cluster.py

@@ -84,6 +84,7 @@ class DeleteClusterTests(base.FunctionalTestCase):
 
     def test_delete_cluster(self):
         flow_store_create = {
+            "tenant_id": str(self.valid_network['tenant_id']),
             "image": self.valid_image.id,
             "flavor": self.valid_flavor.id,
             "port": self.port,
@@ -172,6 +173,7 @@ class DeleteClusterTests(base.FunctionalTestCase):
         self.flags(cluster_node_anti_affinity=True, group="taskflow")
 
         flow_store_create = {
+            "tenant_id": str(self.valid_network['tenant_id']),
             "image": self.valid_image.id,
             "flavor": self.valid_flavor.id,
             "port": self.port,

os_tasklib/neutron/__init__.py

@@ -14,5 +14,6 @@
 # under the License.
 
 
-from os_tasklib.neutron.create_port import CreatePort  # noqa
-from os_tasklib.neutron.delete_ports import DeletePorts  # noqa
+from create_port import CreatePort  # noqa
+from delete_ports import DeletePorts  # noqa
+from show_network import ShowNetwork  # noqa

os_tasklib/neutron/create_port.py

@@ -31,7 +31,6 @@ class CreatePort(os_tasklib.BaseTask):
     parameters provided to the Task.
 
     """
-    default_provides = 'neutron_port_id'
 
     def execute(self, network_id, port_name, security_groups=None, **kwargs):
         """Main execute method

os_tasklib/neutron/show_network.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015 Hewlett-Packard Development Company, L.P.
+#
+# 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 os_tasklib
+
+from oslo_log import log as logging
+
+
+LOG = logging.getLogger(__name__)
+
+
+class ShowNetwork(os_tasklib.BaseTask):
+    """ShowNetwork Task
+
+    This task interfaces with Neutron API and retrieves information about the
+    specified network.
+
+    """
+
+    def execute(self, network, **kwargs):
+        """Main execute method
+
+        :param network_id: Network id to connect new port to
+        :type network_id: string
+        :return: Port record provided by Neutron
+        :rtype: dict
+        """
+        network = self.os_client.show_network(network=network)
+
+        return network['network']