Add some UT test cases and fix pep8 errors for create_zone

JIRA:QTIP-89

Change-Id: Ic5d88a7492f89dbffbed62ce15e3a4e04b4ef2cc
Signed-off-by: zhifeng.jiang <jiang.zhifeng@zte.com.cn>

diff --git a/func/create_zones.py b/func/create_zones.py
index 44ba756..e715dfd 100644 (file)
--- a/func/create_zones.py
+++ b/func/create_zones.py
@@ -1,127 +1,130 @@
-##############################################################################
-# Copyright (c) 2015 Dell Inc  and others.
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Apache License, Version 2.0
-# which accompanies this distribution, and is available at
-# http://www.apache.org/licenses/LICENSE-2.0
-##############################################################################
-
-
-
-from keystoneclient.auth.identity import v2
-from keystoneclient import session
-from novaclient import client
-import os
-import re
-from collections import defaultdict
-
-
-class create_zones:
-
-    def __init__(self):
-        print 'Creating Zones'
-        self._keystone_client = None
-        self._nova_client = None
-
-    def _get_keystone_client(self):
-        '''returns a keystone client instance'''
-
-        if self._keystone_client is None:
-            '''
-            self._keystone_client = keystoneclient.v2_0.client.Client(
-            auth_url=os.environ.get('OS_AUTH_URL'),
-            username=os.environ.get('OS_USERNAME'),
-            password=os.environ.get('OS_PASSWORD'),
-            tenant_name=os.environ.get('OS_TENANT_NAME'))
-            '''
-            auth = v2.Password(auth_url=os.environ.get('OS_AUTH_URL'),
-                               username=os.environ.get('OS_USERNAME'),
-                               password=os.environ.get('OS_PASSWORD'),
-                               tenant_name=os.environ.get('OS_TENANT_NAME'))
-
-            sess = session.Session(auth=auth)
-
-        return sess
-
-    def _get_nova_client(self):
-        if self._nova_client is None:
-            keystone = self._get_keystone_client()
-            self._nova_client = client.Client('2', session=keystone)
-        return self._nova_client
-
-    def check_aggregate(self, nova, agg_name):
-        list1 = nova.aggregates.list()
-
-        agg_name_exist = False
-        for x in list1:
-
-            if x.name == agg_name:
-                agg_name_exist = True
-        return agg_name_exist
-
-    def get_aggregate_id(self, nova, agg_name):
-        list1 = nova.aggregates.list()
-        agg_id = 0
-        agg_name_exist = False
-        for x in list1:
-            if x.name == agg_name:
-                agg_id = x.id
-                return agg_id
-
-    def check_host_added_to_aggregate(self, nova, agg_id, hostname):
-        host_added = False
-        list1 = nova.aggregates.get_details(agg_id)
-
-        nme = str(list1.hosts)
-        if(hostname in nme):
-            host_added = True
-        return host_added
-
-    def del_agg(self, nova, id, host):
-
-        nova.aggregates.remove_host(id, host)
-        nova.aggregates.delete(id)
-
-    def get_compute_num(self, computeName):
-
-        num = re.findall(r'\d+',computeName)
-        return (int(num[0])-1)
-
-    def create_agg(self, D):
-        nova = self._get_nova_client()
-        hyper_list = nova.hypervisors.list()
-        hostnA = []
-        zone_machine = defaultdict(list)
-
-        x = 0
-        for x in range(len(hyper_list)):
-
-            hostnA.append(hyper_list[x].service['host'])
-            hostnA[x] = str(hostnA[x])
-
-        hostnA.sort()
-        for k in D:
-
-            zone_machine[k].append(' ')
-
-        for x in range(len(zone_machine)):
-            compute_index = self.get_compute_num(D[x])
-            if compute_index > len(hyper_list):
-                print '\n The specified compute node doesnt exist. using compute 1'
-                compute_index = 1
-            if not self.check_aggregate(nova, hostnA[compute_index]):
-                agg_idA = nova.aggregates.create(hostnA[compute_index], D[x])
-                nova.aggregates.add_host(aggregate=agg_idA, host=hostnA[compute_index])
-
-            else:
-
-                id1 = self.get_aggregate_id(nova, hostnA[compute_index])
-                self.del_agg(nova, id1, hostnA[compute_index])
-                agg_idA = nova.aggregates.create(hostnA[compute_index], D[x])
-                id1 = self.get_aggregate_id(nova, hostnA[compute_index])
-
-                if not self.check_host_added_to_aggregate(
-                        nova, id1, hostnA[compute_index]):
-
-                    nova.aggregates.add_host(aggregate=id1, host=hostnA[compute_index])
+##############################################################################\r
+# Copyright (c) 2015 Dell Inc  and others.\r
+#\r
+# All rights reserved. This program and the accompanying materials\r
+# are made available under the terms of the Apache License, Version 2.0\r
+# which accompanies this distribution, and is available at\r
+# http://www.apache.org/licenses/LICENSE-2.0\r
+##############################################################################\r
+from keystoneclient.auth.identity import v2\r
+from keystoneclient import session\r
+from novaclient import client\r
+import os\r
+import re\r
+from collections import defaultdict\r
+\r
+\r
+class create_zones:\r
+\r
+    def __init__(self):\r
+        print 'Creating Zones'\r
+        self._keystone_client = None\r
+        self._nova_client = None\r
+\r
+    def _get_keystone_client(self):\r
+        """returns a keystone client instance"""\r
+\r
+        if self._keystone_client is None:\r
+            '''\r
+            self._keystone_client = keystoneclient.v2_0.client.Client(\r
+            auth_url=os.environ.get('OS_AUTH_URL'),\r
+            username=os.environ.get('OS_USERNAME'),\r
+            password=os.environ.get('OS_PASSWORD'),\r
+            tenant_name=os.environ.get('OS_TENANT_NAME'))\r
+            '''\r
+            auth = v2.Password(auth_url=os.environ.get('OS_AUTH_URL'),\r
+                               username=os.environ.get('OS_USERNAME'),\r
+                               password=os.environ.get('OS_PASSWORD'),\r
+                               tenant_name=os.environ.get('OS_TENANT_NAME'))\r
+\r
+            sess = session.Session(auth=auth)\r
+        else:\r
+            return self._keystone_client\r
+\r
+        return sess\r
+\r
+    def _get_nova_client(self):\r
+        if self._nova_client is None:\r
+            keystone = self._get_keystone_client()\r
+            self._nova_client = client.Client('2', session=keystone)\r
+        return self._nova_client\r
+\r
+    @staticmethod\r
+    def check_aggregate(nova, agg_name):\r
+        list1 = nova.aggregates.list()\r
+        agg_name_exist = False\r
+        for x in list1:\r
+            if x.name == agg_name:\r
+                agg_name_exist = True\r
+        return agg_name_exist\r
+\r
+    @staticmethod\r
+    def get_aggregate_id(nova, agg_name):\r
+        list1 = nova.aggregates.list()\r
+        for x in list1:\r
+            if x.name == agg_name:\r
+                agg_id = x.id\r
+                return agg_id\r
+\r
+    @staticmethod\r
+    def check_host_added_to_aggregate(nova, agg_id, hostname):\r
+        host_added = False\r
+        list1 = nova.aggregates.get_details(agg_id)\r
+\r
+        nme = str(list1.hosts)\r
+        if hostname in nme:\r
+            host_added = True\r
+        return host_added\r
+\r
+    @staticmethod\r
+    def del_agg(nova, id, host):\r
+\r
+        nova.aggregates.remove_host(id, host)\r
+        nova.aggregates.delete(id)\r
+\r
+    @staticmethod\r
+    def get_compute_num(compute_name):\r
+\r
+        num = re.findall(r'\d+', compute_name)\r
+        return int(num[0]) - 1\r
+\r
+    def test(self):\r
+        nova = self._get_nova_client()\r
+        hyper_list = nova.hypervisors.list()\r
+        return hyper_list\r
+\r
+    def create_agg(self, d):\r
+        nova = self._get_nova_client()\r
+        hyper_list = nova.hypervisors.list()\r
+        host_a = []\r
+        zone_machine = defaultdict(list)\r
+\r
+        for x in range(len(hyper_list)):\r
+\r
+            host_a.append(hyper_list[x].service['host'])\r
+            host_a[x] = str(host_a[x])\r
+\r
+        host_a.sort()\r
+        for k in d:\r
+\r
+            zone_machine[k].append(' ')\r
+\r
+        for x in range(len(zone_machine)):\r
+            compute_index = self.get_compute_num(d[x])\r
+            if compute_index > len(hyper_list):\r
+                print '\n The specified compute node doesnt exist. using compute 1'\r
+                compute_index = 1\r
+            if not self.check_aggregate(nova, host_a[compute_index]):\r
+                agg_id_a = nova.aggregates.create(host_a[compute_index], d[x])\r
+                nova.aggregates.add_host(aggregate=agg_id_a, host=host_a[compute_index])\r
+\r
+            else:\r
+                id1 = self.get_aggregate_id(nova, host_a[compute_index])\r
+                self.del_agg(nova, id1, host_a[compute_index])\r
+                nova.aggregates.create(host_a[compute_index], d[x])\r
+                id1 = self.get_aggregate_id(nova, host_a[compute_index])\r
+\r
+                if not self.check_host_added_to_aggregate(\r
+                        nova, id1, host_a[compute_index]):\r
+\r
+                    nova.aggregates.add_host(aggregate=id1, host=host_a[compute_index])\r

diff --git a/tests/create_zones_test.py b/tests/create_zones_test.py
new file mode 100644 (file)
index 0000000..e431a90
--- /dev/null
+++ b/tests/create_zones_test.py
@@ -0,0 +1,89 @@
+import pytest
+import mock
+from mock import Mock, MagicMock
+import os
+from func.create_zones import create_zones
+
+return_list = []
+
+
+def get_agg_mock(host):
+    agg = Mock()
+    agg.name = host
+    agg.id = host
+    return agg
+
+
+class HyperMock(MagicMock):
+    def list(self):
+        mock_hypervisor = [Mock(service={'host': '10.20.0.4'}), Mock(service={'host': '10.20.0.5'})]
+        return mock_hypervisor
+
+
+class AggMock(MagicMock):
+    def get_details(self, agg_id):
+        print "get_detail:{0}".format(agg_id)
+        return Mock(hosts=[])
+
+    def create(self, host, agg):
+        print "create:{0}:{1}".format(host, agg)
+        return agg
+
+    def list(self):
+        return return_list
+
+    def delete(self, agg_id):
+        print "delete:{0}".format(agg_id)
+        pass
+
+    def add_host(self, aggregate, host):
+        print "add_host:{0}:{1}".format(aggregate, host)
+        pass
+
+    def remove_host(self, agg_id, host):
+        pass
+
+
+class NovaMock(MagicMock):
+    hypervisors = HyperMock()
+    aggregates = AggMock()
+
+
+class TestClass:
+    @pytest.mark.parametrize("test_input, expected", [
+        ([[], ['compute1', 'compute2']],
+         ['create:10.20.0.4:compute1',
+          'add_host:compute1:10.20.0.4',
+          'create:10.20.0.5:compute2',
+          'add_host:compute2:10.20.0.5']),
+        ([[get_agg_mock('10.20.0.4'), get_agg_mock('10.20.0.5')], ['compute1', 'compute2']],
+         ['delete:10.20.0.4',
+          'create:10.20.0.4:compute1',
+          'get_detail:10.20.0.4',
+          'add_host:10.20.0.4:10.20.0.4',
+          'delete:10.20.0.5',
+          'create:10.20.0.5:compute2',
+          'get_detail:10.20.0.5',
+          'add_host:10.20.0.5:10.20.0.5']),
+        ([[], ['compute1', 'compute5']],
+         ['The specified compute node doesnt exist. using compute 1'])
+    ])
+    @mock.patch('func.create_zones.client', autospec=True)
+    @mock.patch('func.create_zones.v2', autospec=True)
+    @mock.patch('func.create_zones.session')
+    def test_create_zones_success(self, mock_keystone_session, mock_keystone_v2, mock_nova_client, test_input, expected, capfd):
+        global return_list
+        return_list = test_input[0]
+        nova_obj = NovaMock()
+        mock_nova_client.Client.return_value = nova_obj()
+        k = mock.patch.dict(os.environ, {'OS_AUTH_URL': 'http://172.10.0.5:5000',
+                                         'OS_USERNAME': 'admin',
+                                         'OS_PASSWORD': 'admin',
+                                         'OS_TENANT_NAME': 'admin'})
+        k.start()
+        create = create_zones()
+        create.create_agg(test_input[1])
+        k.stop()
+        resout, reserr = capfd.readouterr()
+        for x in expected:
+            assert x in resout