From cf5b2e121de1836ffdaff5f2f2a615968c0bfc6f Mon Sep 17 00:00:00 2001 From: "zhifeng.jiang" Date: Sun, 10 Jul 2016 23:02:27 +0800 Subject: Add some UT test cases and fix pep8 errors for create_zone JIRA:QTIP-89 Change-Id: Ic5d88a7492f89dbffbed62ce15e3a4e04b4ef2cc Signed-off-by: zhifeng.jiang --- func/create_zones.py | 257 +++++++++++++++++++++++---------------------- tests/create_zones_test.py | 89 ++++++++++++++++ 2 files changed, 219 insertions(+), 127 deletions(-) create mode 100644 tests/create_zones_test.py diff --git a/func/create_zones.py b/func/create_zones.py index 44ba7568..e715dfd4 100644 --- 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]) +############################################################################## +# 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) + else: + return self._keystone_client + + 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 + + @staticmethod + def check_aggregate(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 + + @staticmethod + def get_aggregate_id(nova, agg_name): + list1 = nova.aggregates.list() + for x in list1: + if x.name == agg_name: + agg_id = x.id + return agg_id + + @staticmethod + def check_host_added_to_aggregate(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 + + @staticmethod + def del_agg(nova, id, host): + + nova.aggregates.remove_host(id, host) + nova.aggregates.delete(id) + + @staticmethod + def get_compute_num(compute_name): + + num = re.findall(r'\d+', compute_name) + return int(num[0]) - 1 + + def test(self): + nova = self._get_nova_client() + hyper_list = nova.hypervisors.list() + return hyper_list + + def create_agg(self, d): + nova = self._get_nova_client() + hyper_list = nova.hypervisors.list() + host_a = [] + zone_machine = defaultdict(list) + + for x in range(len(hyper_list)): + + host_a.append(hyper_list[x].service['host']) + host_a[x] = str(host_a[x]) + + host_a.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, host_a[compute_index]): + agg_id_a = nova.aggregates.create(host_a[compute_index], d[x]) + nova.aggregates.add_host(aggregate=agg_id_a, host=host_a[compute_index]) + + else: + id1 = self.get_aggregate_id(nova, host_a[compute_index]) + self.del_agg(nova, id1, host_a[compute_index]) + nova.aggregates.create(host_a[compute_index], d[x]) + id1 = self.get_aggregate_id(nova, host_a[compute_index]) + + if not self.check_host_added_to_aggregate( + nova, id1, host_a[compute_index]): + + nova.aggregates.add_host(aggregate=id1, host=host_a[compute_index]) diff --git a/tests/create_zones_test.py b/tests/create_zones_test.py new file mode 100644 index 00000000..e431a907 --- /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 -- cgit 1.2.3-korg