diff options
Diffstat (limited to 'functest')
| -rw-r--r-- | functest/ci/testcases.yaml | 1 | ||||
| -rw-r--r-- | functest/core/unit.py | 27 | ||||
| -rw-r--r-- | functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py | 69 | ||||
| -rw-r--r-- | functest/opnfv_tests/sdn/onos/teston/adapters/foundation.py | 34 | ||||
| -rw-r--r-- | functest/tests/unit/core/test_unit.py | 6 |
5 files changed, 76 insertions, 61 deletions
diff --git a/functest/ci/testcases.yaml b/functest/ci/testcases.yaml index 10587f26..8222df10 100644 --- a/functest/ci/testcases.yaml +++ b/functest/ci/testcases.yaml @@ -266,7 +266,6 @@ tiers: testcases: - case_name: promise - enabled: false project_name: promise criteria: 100 blocking: false diff --git a/functest/core/unit.py b/functest/core/unit.py index 6799420c..515a2080 100644 --- a/functest/core/unit.py +++ b/functest/core/unit.py @@ -26,10 +26,11 @@ __author__ = ("Steven Pisarski <s.pisarski@cablelabs.com>, " class Suite(testcase.TestCase): """Base model for running unittest.TestSuite.""" + __logger = logging.getLogger(__name__) + def __init__(self, **kwargs): super(Suite, self).__init__(**kwargs) self.suite = None - self.logger = logging.getLogger(__name__) def run(self, **kwargs): """Run the test suite. @@ -61,24 +62,28 @@ class Suite(testcase.TestCase): try: self.suite = unittest.TestLoader().loadTestsFromName(name) except ImportError: - self.logger.error("Can not import %s", name) + self.__logger.error("Can not import %s", name) return testcase.TestCase.EX_RUN_ERROR except KeyError: pass - self.start_time = time.time() - stream = six.StringIO() - result = unittest.TextTestRunner( - stream=stream, verbosity=2).run(self.suite) - self.logger.debug("\n\n%s", stream.getvalue()) - self.stop_time = time.time() - self.details = {"failures": result.failures, - "errors": result.errors} try: + assert self.suite + self.start_time = time.time() + stream = six.StringIO() + result = unittest.TextTestRunner( + stream=stream, verbosity=2).run(self.suite) + self.__logger.debug("\n\n%s", stream.getvalue()) + self.stop_time = time.time() + self.details = {"failures": result.failures, + "errors": result.errors} self.result = 100 * ( (result.testsRun - (len(result.failures) + len(result.errors))) / result.testsRun) return testcase.TestCase.EX_OK + except AssertionError: + self.__logger.error("No suite is defined") + return testcase.TestCase.EX_RUN_ERROR except ZeroDivisionError: - self.logger.error("No test has been run") + self.__logger.error("No test has been run") return testcase.TestCase.EX_RUN_ERROR diff --git a/functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py b/functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py index 1101f239..4e93c133 100644 --- a/functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py +++ b/functest/opnfv_tests/sdn/onos/sfc/sfc_onos.py @@ -100,8 +100,10 @@ class SfcOnos(object): self.ip_pool = 0 self.vm_public_ip = [] self.vm_public_id = [] - self.cirros_username = CONST.openstack_image_username - self.cirros_password = CONST.openstack_image_password + self.cirros_username = CONST.__getattribute__( + 'openstack_image_username') + self.cirros_password = CONST.__getattribute__( + 'openstack_image_password') self.net_id1 = 0 self.vm = [] self.address = 0 @@ -115,7 +117,7 @@ class SfcOnos(object): data = ('{"auth": {"tenantName": "admin", "passwordCredentials":' '{ "username": "admin", "password": "console"}}}') headers = {"Accept": "application/json"} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == OK): json1_data = json.loads(response.content) self.logger.debug(response.status_code) @@ -135,12 +137,12 @@ class SfcOnos(object): if self.admin_state_up != '': Dicdata['admin_state_up'] = self.admin_state_up Dicdata = {'network': Dicdata} - data = json.dumps(Dicdata, indent=4) + data = json.dumps(Dicdata, indent=4) url = 'http://%s:9696/%s/networks' % (self.neutron_hostname, self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == CREATED): self.logger.debug(response.status_code) self.logger.debug(response.content) @@ -170,7 +172,7 @@ class SfcOnos(object): self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == CREATED): self.logger.debug(response.status_code) @@ -203,7 +205,7 @@ class SfcOnos(object): self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == CREATED): self.logger.debug(response.status_code) @@ -222,8 +224,9 @@ class SfcOnos(object): """Creation of Instance, using firewall image.""" url = ("http://%s:9292/v2/images?" "name=TestSfcVm" % (self.glance_hostname)) - headers = {"Accept": "application/json", "Content-Type": "application/\ - octet-stream", "X-Auth-Token": self.token_id} + headers = {"Accept": "application/json", + "Content-Type": "application/octet-stream", + "X-Auth-Token": self.token_id} response = requests.get(url, headers=headers) if (response.status_code == OK): self.logger.debug(response.status_code) @@ -273,7 +276,7 @@ class SfcOnos(object): self.tenant_id) headers = {"Accept": "application/json", "Content-Type": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == ACCEPTED): self.logger.debug(response.status_code) self.logger.debug(response.content) @@ -295,8 +298,8 @@ class SfcOnos(object): for y in range(0, 3): url = ("http://%s:8774/v2.1/servers/" "detail?name=vm" + str(y)) % (self.neutron_hostname) - headers = {"Accept": "application/json", "X-Auth-Token": - self.token_id} + headers = {"Accept": "application/json", + "X-Auth-Token": self.token_id} response = requests.get(url, headers=headers) if (response.status_code == OK): self.logger.debug(response.status_code) @@ -332,7 +335,7 @@ class SfcOnos(object): self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == CREATED): info = ("Creation of Port Pair PP" + str(p) + " is successful") @@ -380,7 +383,7 @@ class SfcOnos(object): self.osver)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == CREATED): info = ("Creation of Port Group PG" + str(p) + "is successful") @@ -431,7 +434,7 @@ class SfcOnos(object): self.osver)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == CREATED): json1_data = json.loads(response.content) self.flow_class_if = json1_data['flow_classifier']['id'] @@ -462,7 +465,7 @@ class SfcOnos(object): headers = {"Accept": "application/json", "Content-Type": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == CREATED): self.logger.debug("Creation of PORT CHAIN is successful") json1_data = json.loads(response.content) @@ -476,7 +479,7 @@ class SfcOnos(object): time.sleep(5) response = requests.get('http://' + self.onos_hostname + ':8181/onos/v1/flows', - auth=("karaf", "karaf")) + auth=("karaf", "karaf")) if (response.status_code == OK): self.logger.debug("Flow is successfully Queries") json1_data = json.loads(response.content) @@ -505,7 +508,7 @@ class SfcOnos(object): self.osver) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == CREATED): self.logger.debug(response.status_code) self.logger.debug(response.content) @@ -548,7 +551,7 @@ class SfcOnos(object): self.router_id)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.put(url, headers=headers, data=data) + response = requests.put(url, headers=headers, data=data) if (response.status_code == OK): self.logger.debug(response.status_code) self.logger.debug(response.content) @@ -570,7 +573,7 @@ class SfcOnos(object): self.router_id) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.put(url, headers=headers, data=data) + response = requests.put(url, headers=headers, data=data) if (response.status_code == OK): self.logger.debug(response.status_code) self.logger.debug(response.content) @@ -590,7 +593,7 @@ class SfcOnos(object): "os-floating-ips" % (self.nova_hostname)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if (response.status_code == OK): self.logger.debug(response.status_code) self.logger.debug(response.content) @@ -614,7 +617,7 @@ class SfcOnos(object): headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.post(url, headers=headers, data=data) + response = requests.post(url, headers=headers, data=data) if(response.status_code == ACCEPTED): self.logger.debug(response.status_code) self.logger.debug(response.content) @@ -631,12 +634,12 @@ class SfcOnos(object): s = pxssh.pxssh() hostname = self.vm_public_ip[0] - s.login(hostname, self.cirros_username, self.cirros_password) + s.login(hostname, self.cirros_username, self.cirros_password) s.sendline("ping -c 5 " + str(self.port_ip[2])) s.prompt() # match the prompt - ping_re = re.search("transmitted.*received", s.before).group() - x = re.split('\s+', ping_re) + ping_re = re.search("transmitted.*received", s.before).group() + x = re.split('\s+', ping_re) if (x[1] >= "1"): self.logger.info("Ping is Successfull") else: @@ -645,7 +648,7 @@ class SfcOnos(object): def vm1(queue1): s = pxssh.pxssh() hostname = self.vm_public_ip[1] - s.login(hostname, self.cirros_username, self.cirros_password) + s.login(hostname, self.cirros_username, self.cirros_password) s.sendline('sudo ./firewall') s.prompt() output_pack = s.before @@ -676,7 +679,7 @@ class SfcOnos(object): if result0 == 0 and result1 == 0: time.sleep(300) queue1 = Queue() - p1 = Process(target=vm1, args=(queue1, )) + p1 = Process(target=vm1, args=(queue1, )) p1.start() p2 = Process(target=vm0) p2.start() @@ -703,7 +706,7 @@ class SfcOnos(object): """Check the PC SF Map Stats in the ONOS.""" response = requests.get('http://' + self.onos_hostname + ':8181/onos/vtn/portChainSfMap/' + - self.PC_id, auth=("karaf", "karaf")) + self.PC_id, auth=("karaf", "karaf")) if (response.status_code == OK): self.logger.info("portChainSfMap is successfully Queries") return(response.status_code) @@ -761,7 +764,7 @@ class SfcOnos(object): def deletePortPair(self): """Deletion of Portpair.""" - for p in range(1, 2): + for p in range(1, 2): url = ("http://%s:9696/%s/sfc/" "port_pairs/%s" % (self.neutron_hostname, self.osver, @@ -819,7 +822,7 @@ class SfcOnos(object): self.router_id)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.put(url, headers=headers, data=data) + response = requests.put(url, headers=headers, data=data) if (response.status_code == OK): self.logger.debug(response.status_code) self.logger.debug(response.content) @@ -833,14 +836,14 @@ class SfcOnos(object): self.router_id)) headers = {"Accept": "application/json", "X-Auth-Token": self.token_id} - response = requests.put(url, headers=headers, data=data) + response = requests.put(url, headers=headers, data=data) if (response.status_code == OK): url = ("http://%s:9696/%s/" "routers/%s" % (self.neutron_hostname, self.osver, self.router_id)) - headers = {"Accept": "application/json", "X-Auth-Token": - self.token_id} + headers = {"Accept": "application/json", + "X-Auth-Token": self.token_id} response = requests.delete(url, headers=headers) if (response.status_code == NO_CONTENT): self.logger.debug(response.status_code) diff --git a/functest/opnfv_tests/sdn/onos/teston/adapters/foundation.py b/functest/opnfv_tests/sdn/onos/teston/adapters/foundation.py index 25421d40..f9eee7af 100644 --- a/functest/opnfv_tests/sdn/onos/teston/adapters/foundation.py +++ b/functest/opnfv_tests/sdn/onos/teston/adapters/foundation.py @@ -26,7 +26,8 @@ class Foundation(object): def __init__(self): # currentpath = os.getcwd() - currentpath = '%s/sdn/onos/teston/ci' % CONST.dir_functest_data + currentpath = ('{0}/sdn/onos/teston/ci' + .format(CONST.__getattribute__('dir_functest_data'))) self.cipath = currentpath self.logdir = os.path.join(currentpath, 'log') self.workhome = currentpath[0: currentpath.rfind('opnfv_tests') - 1] @@ -55,22 +56,25 @@ class Foundation(object): Get Default Parameters value """ self.Result_DB = ft_utils.get_db_url() - self.masterusername = CONST.ONOS_onosbench_username - self.masterpassword = CONST.ONOS_onosbench_password - self.agentusername = CONST.ONOS_onoscli_username - self.agentpassword = CONST.ONOS_onoscli_password - self.runtimeout = CONST.ONOS_runtimeout - self.OCT = CONST.ONOS_environment_OCT - self.OC1 = CONST.ONOS_environment_OC1 - self.OC2 = CONST.ONOS_environment_OC2 - self.OC3 = CONST.ONOS_environment_OC3 - self.OCN = CONST.ONOS_environment_OCN - self.OCN2 = CONST.ONOS_environment_OCN2 - self.installer_master = CONST.ONOS_environment_installer_master + self.masterusername = CONST.__getattribute__('ONOS_onosbench_username') + self.masterpassword = CONST.__getattribute__('ONOS_onosbench_password') + self.agentusername = CONST.__getattribute__('ONOS_onoscli_username') + self.agentpassword = CONST.__getattribute__('ONOS_onoscli_password') + self.runtimeout = CONST.__getattribute__('ONOS_runtimeout') + self.OCT = CONST.__getattribute__('ONOS_environment_OCT') + self.OC1 = CONST.__getattribute__('ONOS_environment_OC1') + self.OC2 = CONST.__getattribute__('ONOS_environment_OC2') + self.OC3 = CONST.__getattribute__('ONOS_environment_OC3') + self.OCN = CONST.__getattribute__('ONOS_environment_OCN') + self.OCN2 = CONST.__getattribute__('ONOS_environment_OCN2') + self.installer_master = CONST.__getattribute__( + 'ONOS_environment_installer_master') self.installer_master_username = ( - CONST.ONOS_environment_installer_master_username) + CONST.__getattribute__( + 'ONOS_environment_installer_master_username')) self.installer_master_password = ( - CONST.ONOS_environment_installer_master_password) + CONST.__getattribute__( + 'ONOS_environment_installer_master_password')) self.hosts = [self.OC1, self.OCN, self.OCN2] self.localhost = self.OCT diff --git a/functest/tests/unit/core/test_unit.py b/functest/tests/unit/core/test_unit.py index f86ea8d3..79c4e7d7 100644 --- a/functest/tests/unit/core/test_unit.py +++ b/functest/tests/unit/core/test_unit.py @@ -20,6 +20,7 @@ class PyTestSuiteRunnerTesting(unittest.TestCase): def setUp(self): self.psrunner = unit.Suite() + self.psrunner.suite = "foo" @mock.patch('unittest.TestLoader') def _test_run(self, mock_class=None, result=mock.Mock(), @@ -30,7 +31,10 @@ class PyTestSuiteRunnerTesting(unittest.TestCase): mock_class.assert_not_called() def test_check_suite_null(self): - self.assertEqual(self.psrunner.suite, None) + self.assertEqual(unit.Suite().suite, None) + self.psrunner.suite = None + self._test_run(result=mock.Mock(), + status=testcase.TestCase.EX_RUN_ERROR) def test_run_no_ut(self): mock_result = mock.Mock(testsRun=0, errors=[], failures=[]) |