# Copyright (c) 2018 Intel Corporation # # 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 copy import mock import requests import time import unittest import uuid from yardstick.benchmark.contexts import base as ctx_base from yardstick.common import exceptions from yardstick.common import utils as common_utils from yardstick.common import yaml_loader from yardstick.network_services import utils as net_serv_utils from yardstick.network_services.traffic_profile import landslide_profile from yardstick.network_services.vnf_generic.vnf import sample_vnf from yardstick.network_services.vnf_generic.vnf import tg_landslide from yardstick.network_services.vnf_generic.vnf import base as vnf_base NAME = "tg__0" EXAMPLE_URL = 'http://example.com/' TCL_SUCCESS_RESPONSE = 'ls_ok' TEST_SERVERS = [ {'ip': '192.168.122.101', 'phySubnets': [ {'mask': '/24', 'base': '10.42.32.100', 'numIps': 20, 'name': 'eth1'} ], 'role': 'SGW_Node', 'name': 'TestServer_1'}, {'ip': '192.168.122.102', 'phySubnets': [ {'mask': '/24', 'base': '10.42.32.1', 'numIps': 100, 'name': 'eth1' }, {'mask': '/24', 'base': '10.42.33.1', 'numIps': 100, 'name': 'eth2'} ], 'preResolvedArpAddress': [ {'NumNodes': 1, 'StartingAddress': '10.42.33.5'} ], 'role': 'SGW_Nodal', 'name': 'TestServer_2', 'thread_model': 'Fireball' } ] TS1_SUTS = [ {'name': 'SGW - C TestNode', 'role': 'SgwControlAddr', 'managementIp': '12.0.1.1', 'ip': '10.42.32.100', 'phy': 'eth5', 'nextHop': '10.42.32.5' }, {'name': 'SGW - U TestNode', 'role': 'SgwUserAddr', 'managementIp': '12.0.1.2', 'ip': '10.42.32.101', 'phy': 'eth5', 'nextHop': '10.42.32.5' } ] TS2_SUTS = [ {'name': 'eNodeB TestNode', 'role': 'EnbUserAddr', 'managementIp': '12.0.2.1', 'ip': '10.42.32.2', 'phy': 'eth5', 'nextHop': '10.42.32.5' }, {'name': 'MME TestNode', 'role': 'MmeControlAddr', 'managementIp': '12.0.3.1', 'ip': '10.42.32.1', 'phy': 'eth5', 'nextHop': '10.42.32.5' }, {'name': 'NetHost TestNode', 'role': 'NetworkHostAddrLocal', 'managementIp': '12.0.4.1', 'ip': '10.42.33.1', 'phy': 'eth5', 'nextHop': '10.42.32.5' }, {'name': 'PGW TestNode', 'role': 'PgwV4Sut', 'managementIp': '12.0.5.1', 'ip': '10.42.32.105', 'phy': 'eth5', 'nextHop': '10.42.32.5' }, {'name': 'SGW - C SUT', 'role': 'SgwSut', 'managementIp': '12.0.6.1', 'ip': '10.42.32.100' }, {'name': 'SGW - U SUT', 'role': 'SgwUserSut', 'managementIp': '12.0.6.2', 'ip': '10.42.32.101'} ] VNFD = { 'vnfd:vnfd-catalog': { 'vnfd': [{ 'short-name': 'landslide', 'vdu': [{ 'description': 'AB client interface details', 'name': 'abclient-baremetal', 'id': 'abclient-baremetal', 'external-interface': []}], 'description': 'Spirent Landslide traffic generator', 'config': [{'test_server': TEST_SERVERS[0], 'suts': TS1_SUTS}, {'test_server': TEST_SERVERS[1], 'suts': TS2_SUTS}], 'mgmt-interface': { 'vdu-id': 'landslide-tas', 'user': 'user', 'password': 'user', 'super-user': 'super-user', 'super-user-password': 'super-user-password', 'cfguser_password': 'cfguser_password', 'license': 48, 'proto': 'http', 'ip': '1.1.1.1'}, 'benchmark': { 'kpi': [ 'tx_throughput_mbps', 'rx_throughput_mbps', 'in_packets', 'out_packets', 'activation_rate_sessps', 'deactivation_rate_sessps']}, 'id': 'LandslideTrafficGen', 'name': 'LandslideTrafficGen'}]}} TAS_INFO = VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface'] DMF_CFG = { "dmf": { "library": "test", "name": "Basic UDP" }, "clientPort": { "clientPort": 2002, "isClientPortRange": "false" }, "dataProtocol": "udp", "serverPort": 2003 } RESERVATIONS = [ {'tsName': TEST_SERVERS[0]['name'], 'phySubnets': TEST_SERVERS[0]['phySubnets'], 'tsId': TEST_SERVERS[0]['name'], 'tsIndex': 0}, {'tsName': TEST_SERVERS[1]['name'], 'phySubnets': TEST_SERVERS[1]['phySubnets'], 'tsId': TEST_SERVERS[1]['name'], 'tsIndex': 1}] SESSION_PROFILE = { 'keywords': '', 'duration': 60, 'iterations': 1, 'description': 'UE default bearer creation test case', 'name': 'default_bearer_capacity', 'reportOptions': {'format': 'CSV'}, 'reservePorts': 'false', 'tsGroups': [ { 'testCases': [{ 'type': 'SGW_Node', 'name': '', 'linked': "false", 'AssociatedPhys': '', 'parameters': { 'SgiPtpTunnelEn': 'false', 'Gtp2Imsi': '505024101215074', 'Sessions': '100000', 'S5Protocol': 'GTPv2', 'TrafficMtu': '1500', 'Gtp2Version': '13.6.0', 'BearerV4AddrPool': '1.0.0.1', 'Gtp2Imei': '50502410121507', 'PgwNodeEn': 'true', 'DedicatedsPerDefaultBearer': '0', 'DefaultBearers': '1', 'SgwUserAddr': { 'numLinksOrNodes': 1, 'phy': 'eth1', 'forcedEthInterface': '', 'ip': 'SGW_USER_IP', 'class': 'TestNode', 'ethStatsEnabled': "false", 'mtu': 1500 }, 'SgwControlAddr': { 'numLinksOrNodes': 1, 'phy': 'eth1', 'forcedEthInterface': '', 'ip': 'SGW_CONTROL_IP', 'class': 'TestNode', 'ethStatsEnabled': "false", 'mtu': 1500, 'nextHop': 'SGW_CONTROL_NEXT_HOP' }, 'BearerAddrPool': '2001::1', 'TestType': 'SGW-NODE' } }], 'tsId': TEST_SERVERS[0]['name']}, { 'testCases': [{ 'type': 'SGW_Nodal', 'name': '', 'parameters': { 'DataTraffic': 'Continuous', 'TrafficStartType': 'When All Sessions Established', 'NetworkHost': 'Local', 'Gtp2Imsi': '505024101215074', 'Dmf': { 'mainflows': [ { 'name': 'Basic UDP', 'library': 'test' } ], 'class': 'Dmf', 'instanceGroups': [ { 'startPaused': "false", 'rate': 0, 'mainflowIdx': 0, 'mixType': '' } ] }, 'S5Protocol': 'GTPv2', 'DataUserCfgFileEn': 'false', 'PgwUserSutEn': 'false', 'MmeControlAddr': { 'numLinksOrNodes': 1, 'phy': 'eth1', 'forcedEthInterface': '', 'ip': 'MME_CONTROL_IP', 'class': 'TestNode', 'ethStatsEnabled': "false", 'mtu': 1500 }, 'SgwUserSut': { 'class': 'Sut', 'name': 'SGW_USER_NAME' }, 'TestActivity': 'Capacity Test', 'NetworkHostAddrLocal': { 'numLinksOrNodes': 1, 'phy': 'eth2', 'forcedEthInterface': '', 'ip': 'NET_HOST_IP', 'class': 'TestNode', 'ethStatsEnabled': "false", 'mtu': 1500 }, 'DedicatedsPerDefaultBearer': '0', 'DisconnectRate': '1000.0', 'Sessions': '100000', 'SgwSut': { 'class': 'Sut', 'name': 'SGW_CONTROL_NAME' }, 'TrafficMtu': '1500', 'Gtp2Version': '13.6.0', 'Gtp2Imei': '50502410121507', 'PgwNodeEn': 'false', 'StartRate': '1000.0', 'PgwV4Sut': { 'class': 'Sut', 'name': 'PGW_SUT_NAME' }, 'DefaultBearers': '1', 'EnbUserAddr': { 'numLinksOrNodes': 1, 'phy': 'eth1', 'forcedEthInterface': '', 'ip': 'ENB_USER_IP', 'class': 'TestNode', 'ethStatsEnabled': "false", 'mtu': 1500 }, 'TestType': 'SGW-NODAL' } }], 'tsId': TEST_SERVERS[1]['name'] } ] } class TestLandslideTrafficGen(unittest.TestCase): SCENARIO_CFG = { 'session_profile': '/traffic_profiles/landslide/' 'landslide_session_default_bearer.yaml', 'task_path': '', 'runner': { 'type': 'Iteration', 'iterations': 1 }, 'nodes': { 'tg__0': 'tg__0.traffic_gen', 'vnf__0': 'vnf__0.vnf_epc' }, 'topology': 'landslide_tg_topology.yaml', 'type': 'NSPerf', 'traffic_profile': '../../traffic_profiles/landslide/' 'landslide_dmf_udp.yaml', 'options': { 'traffic_duration': 71, 'test_cases': [ { 'BearerAddrPool': '2002::2', 'type': 'SGW_Node', 'BearerV4AddrPool': '2.0.0.2', 'Sessions': '90000' }, { 'StartRate': '900.0', 'type': 'SGW_Nodal', 'DisconnectRate': '900.0', 'Sessions': '90000' } ], 'dmf': { 'transactionRate': 1000, 'packetSize': 512 } } } CONTEXT_CFG = { 'contexts': [ { 'type': 'Node', 'name': 'traffic_gen', 'file': '/etc/yardstick/nodes/pod_landslide.yaml' }, { 'type': 'Node', 'name': 'vnf_epc', 'file': '/etc/yardstick/nodes/pod_vepc_sut.yaml' } ] } TRAFFIC_PROFILE = { "schema": "nsb:traffic_profile:0.1", "name": "LandslideProfile", "description": "Spirent Landslide traffic profile", "traffic_profile": { "traffic_type": "LandslideProfile" }, "dmf_config": { "dmf": { "library": "test", "name": "Basic UDP" }, "description": "Basic data flow using UDP/IP", "keywords": "UDP", "dataProtocol": "udp" } } SUCCESS_CREATED_CODE = 201 SUCCESS_OK_CODE = 200 SUCCESS_RECORD_ID = 5 TEST_USER_ID = 11 def setUp(self): self._id = uuid.uuid1().int self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi') self.mock_lsapi.start() self.mock_ssh_helper = mock.patch.object(sample_vnf, 'VnfSshHelper') self.mock_ssh_helper.start() self.vnfd = VNFD['vnfd:vnfd-catalog']['vnfd'][0] self.ls_tg = tg_landslide.LandslideTrafficGen( NAME, self.vnfd, self._id) self.session_profile = copy.deepcopy(SESSION_PROFILE) self.ls_tg.session_profile = self.session_profile self.addCleanup(self._cleanup) def _cleanup(self): self.mock_lsapi.stop() self.mock_ssh_helper.stop() @mock.patch.object(net_serv_utils, 'get_nsb_option') def test___init__(self, mock_get_nsb_option, *args): _path_to_nsb = 'path/to/nsb' mock_get_nsb_option.return_value = _path_to_nsb ls_tg = tg_landslide.LandslideTrafficGen(NAME, self.vnfd, self._id) self.assertIsInstance(ls_tg.resource_helper, tg_landslide.LandslideResourceHelper) mock_get_nsb_option.assert_called_once_with('bin_path') self.assertEqual(_path_to_nsb, ls_tg.bin_path) self.assertEqual(NAME, ls_tg.name) self.assertTrue(ls_tg.runs_traffic) self.assertFalse(ls_tg.traffic_finished) self.assertIsNone(ls_tg.session_profile) def test_listen_traffic(self): _traffic_profile = {} self.assertIsNone(self.ls_tg.listen_traffic(_traffic_profile)) def test_terminate(self, *args): self.ls_tg.resource_helper._tcl = mock.Mock() self.assertIsNone(self.ls_tg.terminate()) self.ls_tg.resource_helper._tcl.disconnect.assert_called_once() @mock.patch.object(ctx_base.Context, 'get_context_from_server', return_value='fake_context') def test_instantiate(self, *args): self.ls_tg._tg_process = mock.Mock() self.ls_tg._tg_process.start = mock.Mock() self.ls_tg.resource_helper.connect = mock.Mock() self.ls_tg.resource_helper.create_test_servers = mock.Mock() self.ls_tg.resource_helper.create_suts = mock.Mock() self.ls_tg._load_session_profile = mock.Mock() self.assertIsNone(self.ls_tg.instantiate(self.SCENARIO_CFG, self.CONTEXT_CFG)) self.ls_tg.resource_helper.connect.assert_called_once() self.ls_tg.resource_helper.create_test_servers.assert_called_once() _suts_blocks_num = len([item['suts'] for item in self.vnfd['config']]) self.assertEqual(_suts_blocks_num, self.ls_tg.resource_helper.create_suts.call_count) self.ls_tg._load_session_profile.assert_called_once() @mock.patch.object(tg_landslide.LandslideResourceHelper, 'get_running_tests') def test_run_traffic(self, mock_get_tests, *args): self.ls_tg.resource_helper._url = EXAMPLE_URL self.ls_tg.scenario_helper.scenario_cfg = self.SCENARIO_CFG mock_traffic_profile = mock.Mock( spec=landslide_profile.LandslideProfile) mock_traffic_profile.dmf_config = { 'keywords': 'UDP', 'dataProtocol': 'udp', 'dmf': {'library': 'test', 'name': 'name'}} mock_traffic_profile.params = self.TRAFFIC_PROFILE self.ls_tg.resource_helper._user_id = self.TEST_USER_ID mock_get_tests.return_value = [{'id': self.SUCCESS_RECORD_ID, 'testStateOrStep': 'COMPLETE'}] mock_post = mock.Mock() mock_post.status_code = self.SUCCESS_CREATED_CODE mock_post.json.return_value = {'id': self.SUCCESS_RECORD_ID} mock_session = mock.Mock(spec=requests.Session) mock_session.post.return_value = mock_post self.ls_tg.resource_helper.session = mock_session self.ls_tg.resource_helper._tcl = mock.Mock() _tcl = self.ls_tg.resource_helper._tcl self.assertIsNone(self.ls_tg.run_traffic(mock_traffic_profile)) self.assertEqual(self.SUCCESS_RECORD_ID, self.ls_tg.resource_helper.run_id) mock_traffic_profile.update_dmf.assert_called_with( self.ls_tg.scenario_helper.all_options) _tcl.create_dmf.assert_called_with(mock_traffic_profile.dmf_config) _tcl.create_test_session.assert_called_with(self.session_profile) @mock.patch.object(tg_landslide.LandslideResourceHelper, 'check_running_test_state') def test_collect_kpi(self, mock_check_running_test_state, *args): self.ls_tg.resource_helper.run_id = self.SUCCESS_RECORD_ID mock_check_running_test_state.return_value = 'COMPLETE' self.assertEqual({'done': True}, self.ls_tg.collect_kpi()) mock_check_running_test_state.assert_called_once() def test_wait_for_instantiate(self): self.assertIsNone(self.ls_tg.wait_for_instantiate()) self.ls_tg.wait_for_instantiate() def test__update_session_suts_no_tc_role(self, *args): _suts = [{'role': 'epc_role'}] _testcase = {'parameters': {'diff_epc_role': {'class': 'Sut'}}} res = self.ls_tg._update_session_suts(_suts, _testcase) self.assertEqual(_testcase, res) def test__update_session_suts(self, *args): def get_testnode_param(role, key, session_prof): """ Get value by key from the deep nested dict to avoid calls like: e.g. session_prof['tsGroups'][0]['testCases'][1]['parameters'][key] """ for group in session_prof['tsGroups']: for tc in group['testCases']: tc_params = tc['parameters'] if tc_params.get(role): return tc_params[role][key] def get_sut_param(role, key, suts): """ Search list of dicts for one with specific role. Return the value of related dict by key. Expect key presence. """ for sut in suts: if sut.get('role') == role: return sut[key] # TestNode to verify testnode_role = 'SgwControlAddr' # SUT to verify sut_role = 'SgwUserSut' config_suts = [config['suts'] for config in self.vnfd['config']] session_tcs = [_tc for _ts_group in self.ls_tg.session_profile['tsGroups'] for _tc in _ts_group['testCases']] for suts, tc in zip(config_suts, session_tcs): self.assertEqual(tc, self.ls_tg._update_session_suts(suts, tc)) # Verify TestNode class objects keys were updated for _key in {'ip', 'phy', 'nextHop'}: self.assertEqual( get_testnode_param(testnode_role, _key, self.ls_tg.session_profile), get_sut_param(testnode_role, _key, TS1_SUTS)) # Verify Sut class objects name was updated self.assertEqual( get_testnode_param(sut_role, 'name', self.ls_tg.session_profile), get_sut_param(sut_role, 'name', TS2_SUTS)) def test__update_session_test_servers(self, *args): for ts_index, ts in enumerate(TEST_SERVERS): self.assertIsNone( self.ls_tg._update_session_test_servers(ts, ts_index)) # Verify preResolvedArpAddress key was added self.assertTrue(any( _item.get('preResolvedArpAddress') for _item in self.ls_tg.session_profile['tsGroups'])) # Verify reservations key was added to session profile self.assertEqual(RESERVATIONS, self.ls_tg.session_profile.get('reservations')) self.assertEqual('true', self.ls_tg.session_profile.get('reservePorts')) def test__update_session_tc_params_assoc_phys(self): _tc_options = {'AssociatedPhys': 'eth1'} _testcase = {} _testcase_orig = copy.deepcopy(_testcase) res = self.ls_tg._update_session_tc_params(_tc_options, _testcase) self.assertNotEqual(_testcase_orig, res) self.assertEqual(_tc_options, _testcase) def test__update_session_tc_params(self, *args): def get_session_tc_param_value(param, tc_type, session_prof): """ Get param value from the deep nested dict to avoid calls like: session_prof['tsGroups'][0]['testCases'][0]['parameters'][key] """ for test_group in session_prof['tsGroups']: session_tc = test_group['testCases'][0] if session_tc['type'] == tc_type: return session_tc['parameters'].get(param) session_tcs = [_tc for _ts_group in self.ls_tg.session_profile['tsGroups'] for _tc in _ts_group['testCases']] scenario_tcs = [_tc for _tc in self.SCENARIO_CFG['options']['test_cases']] for tc_options, tc in zip(scenario_tcs, session_tcs): self.assertEqual( tc, self.ls_tg._update_session_tc_params(tc_options, tc)) # Verify that each test case parameter was updated # Params been compared are deeply nested. Using loops to ease access. for _tc in self.SCENARIO_CFG['options']['test_cases']: for _key, _val in _tc.items(): if _key != 'type': self.assertEqual( _val, get_session_tc_param_value(_key, _tc.get('type'), self.ls_tg.session_profile)) def test__update_session_library_name(self, *args): _session = copy.deepcopy(SESSION_PROFILE) _session['tsGroups'].pop(0) self.ls_tg.vnfd_helper = mock.MagicMock() self.ls_tg.vnfd_helper.mgmt_interface.__getitem__.side_effect = { 'user': TAS_INFO['user']} self.ls_tg._update_session_library_name(_session) _dmf = _session['tsGroups'][0]['testCases'][0]['parameters']['Dmf'] # Expect DMF library name updated in Nodal test types self.assertEqual(TAS_INFO['user'], _dmf['mainflows'][0]['library']) def test__update_session_library_name_wrong_tc_type(self, *args): _session = copy.deepcopy(SESSION_PROFILE) _session['tsGroups'].pop(1) self.ls_tg.vnfd_helper = mock.MagicMock() self.ls_tg.vnfd_helper.mgmt_interface.__getitem__.side_effect = { 'user': TAS_INFO['user']} # Expect DMF library name not updated in Node test types self.assertNotIn('Dmf', _session['tsGroups'][0]['testCases'][0]['parameters']) self.ls_tg._update_session_library_name(_session) @mock.patch.object(common_utils, 'open_relative_file') @mock.patch.object(yaml_loader, 'yaml_load') @mock.patch.object(tg_landslide.LandslideTrafficGen, '_update_session_test_servers') @mock.patch.object(tg_landslide.LandslideTrafficGen, '_update_session_suts') @mock.patch.object(tg_landslide.LandslideTrafficGen, '_update_session_tc_params') def test__load_session_profile(self, mock_upd_ses_tc_params, mock_upd_ses_suts, mock_upd_ses_ts, mock_yaml_load, *args): self.ls_tg.scenario_helper.scenario_cfg = \ copy.deepcopy(self.SCENARIO_CFG) mock_yaml_load.return_value = copy.deepcopy(SESSION_PROFILE) self.assertIsNone(self.ls_tg._load_session_profile()) self.assertIsNotNone(self.ls_tg.session_profile) # Number of blocks in configuration files # Number of test servers, suts and tc params blocks should be equal _config_files_blocks_num = len([item['test_server'] for item in self.vnfd['config']]) self.assertEqual(_config_files_blocks_num, mock_upd_ses_ts.call_count) self.assertEqual(_config_files_blocks_num, mock_upd_ses_suts.call_count) self.assertEqual(_config_files_blocks_num, mock_upd_ses_tc_params.call_count) @mock.patch.object(common_utils, 'open_relative_file') @mock.patch.object(yaml_loader, 'yaml_load') def test__load_session_profile_unequal_num_of_cfg_blocks( self, mock_yaml_load, *args): vnfd = copy.deepcopy(VNFD['vnfd:vnfd-catalog']['vnfd'][0]) ls_traffic_gen = tg_landslide.LandslideTrafficGen(NAME, vnfd, self._id) ls_traffic_gen.scenario_helper.scenario_cfg = self.SCENARIO_CFG mock_yaml_load.return_value = copy.deepcopy(SESSION_PROFILE) # Delete test_servers item from pod file to make it not valid ls_traffic_gen.vnfd_helper['config'].pop() with self.assertRaises(RuntimeError): ls_traffic_gen._load_session_profile() @mock.patch.object(common_utils, 'open_relative_file') @mock.patch.object(yaml_loader, 'yaml_load') def test__load_session_profile_test_type_mismatch(self, mock_yaml_load, *args): vnfd = copy.deepcopy(VNFD['vnfd:vnfd-catalog']['vnfd'][0]) # Swap test servers data in pod file vnfd['config'] = list(reversed(vnfd['config'])) ls_tg = tg_landslide.LandslideTrafficGen(NAME, vnfd, self._id) ls_tg.scenario_helper.scenario_cfg = self.SCENARIO_CFG mock_yaml_load.return_value = SESSION_PROFILE with self.assertRaises(RuntimeError): ls_tg._load_session_profile() class TestLandslideResourceHelper(unittest.TestCase): PROTO_PORT = 8080 EXAMPLE_URL = ''.join([TAS_INFO['proto'], '://', TAS_INFO['ip'], ':', str(PROTO_PORT), '/api/']) SUCCESS_CREATED_CODE = 201 SUCCESS_OK_CODE = 200 INVALID_REST_CODE = '400' NOT_MODIFIED_CODE = 500810 ERROR_CODE = 500800 SUCCESS_RECORD_ID = 11 EXPIRE_DATE = '2020/01/01 12:00 FLE Standard Time' TEST_USER = 'test' TEST_TERMINATED = 1 AUTH_DATA = {'user': TAS_INFO['user'], 'password': TAS_INFO['password']} TEST_SESSION_NAME = 'default_bearer_capacity' USERS_DATA = { "users": [{ "url": ''.join([EXAMPLE_URL, 'users/', str(SUCCESS_RECORD_ID)]), "id": SUCCESS_RECORD_ID, "level": 1, "username": TEST_USER }] } CREATE_USER_DATA = {'username': TAS_INFO['user'], 'expiresOn': EXPIRE_DATE, 'level': 1, 'contactInformation': '', 'fullName': 'Test User', 'password': TAS_INFO['password'], 'isActive': 'true'} SUTS_DATA = { "suts": [ { "url": ''.join([EXAMPLE_URL, 'suts/', str(SUCCESS_RECORD_ID)]), "id": SUCCESS_RECORD_ID, "name": "10.41.32.1" }]} TEST_SERVERS_DATA = { "testServers": [ { "url": ''.join([EXAMPLE_URL, "testServers/1"]), "id": 1, "name": TEST_SERVERS[0]['name'], "state": "READY", "version": "16.4.0.10" }, { "url": ''.join([EXAMPLE_URL, "testServers/2"]), "id": 2, "name": TEST_SERVERS[1]['name'], "state": "READY", "version": "16.4.0.10" } ] } RUN_ID = 3 RUNNING_TESTS_DATA = { "runningTests": [{ "url": ''.join([EXAMPLE_URL, "runningTests/{}".format(RUN_ID)]), "measurementsUrl": ''.join( [EXAMPLE_URL, "runningTests/{}/measurements".format(RUN_ID)]), "criteriaUrl": ''.join( [EXAMPLE_URL, "runningTests/{}/criteria".format(RUN_ID)]), "noteToUser": "", "id": RUN_ID, "library": SUCCESS_RECORD_ID, "name": "default_bearer_capacity", "user": TEST_USER, "criteriaStatus": "NA", "testStateOrStep": "COMPLETE" }]} TEST_RESULTS_DATA = { "interval": 0, "elapsedTime": 138, "actualTime": 1521548057296, "iteration": 1, "tabs": { "Test Summary": { "Start Time": "Tue Mar 20 07:11:55 CDT 2018", "Actual Dedicated Bearer Session Connects": "100", "Actual Dedicated Bearer Session Disconnects": "100", "Actual Disconnect Rate(Sessions / Second)(P - I)": "164.804", "Average Session Disconnect Time(P - I)": "5.024 s", "Total Data Sent + Received Packets / Sec(P - I)": "1,452.294" }}} def setUp(self): self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi') self.mock_lsapi.start() mock_env_helper = mock.Mock() self.res_helper = tg_landslide.LandslideResourceHelper(mock_env_helper) self.res_helper._url = EXAMPLE_URL self.addCleanup(self._cleanup) def _cleanup(self): self.mock_lsapi.stop() self.res_helper._url = None def test___init__(self, *args): self.assertIsInstance(self.res_helper, tg_landslide.LandslideResourceHelper) self.assertEqual({}, self.res_helper._result) self.assertIsNone(self.res_helper.run_id) @mock.patch.object(tg_landslide.LandslideResourceHelper, 'stop_running_tests') @mock.patch.object(tg_landslide.LandslideResourceHelper, 'get_running_tests') def test_abort_running_tests_no_running_tests(self, mock_get_tests, mock_stop_tests, *args): tests_data = [{'id': self.SUCCESS_RECORD_ID, 'testStateOrStep': 'COMPLETE'}] mock_get_tests.return_value = tests_data self.assertIsNone(self.res_helper.abort_running_tests()) mock_stop_tests.assert_not_called() @mock.patch.object(time, 'sleep') @mock.patch.object(tg_landslide.LandslideResourceHelper, 'stop_running_tests') @mock.patch.object(tg_landslide.LandslideResourceHelper, 'get_running_tests') def test_abort_running_tests(self, mock_get_tests, mock_stop_tests, *args): test_states_seq = iter(['RUNNING', 'COMPLETE']) def configure_mock(*args): return [{'id': self.SUCCESS_RECORD_ID, 'testStateOrStep': next(test_states_seq)}] mock_get_tests.side_effect = configure_mock self.assertIsNone(self.res_helper.abort_running_tests()) mock_stop_tests.assert_called_once_with( running_test_id=self.SUCCESS_RECORD_ID, force=True) self.assertEqual(2, mock_get_tests.call_count) @mock.patch.object(tg_landslide.LandslideResourceHelper, 'stop_running_tests') @mock.patch.object(tg_landslide.LandslideResourceHelper, 'get_running_tests') def test_abort_running_tests_error(self, mock_get_tests, mock_stop_tests, *args): tests_data = {'id': self.SUCCESS_RECORD_ID, 'testStateOrStep': 'RUNNING'} mock_get_tests.return_value = [tests_data] with self.assertRaises(RuntimeError): self.res_helper.abort_running_tests(timeout=1, delay=1) mock_stop_tests.assert_called_with( running_test_id=self.SUCCESS_RECORD_ID, force=True) def test__build_url(self, *args): resource = 'users' action = {'action': 'userCreate'} expected_url = ''.join([EXAMPLE_URL, 'users?action=userCreate']) self.assertEqual(expected_url, self.res_helper._build_url(resource, action)) def test__build_url_error(self, *args): resource = '' action = {'action': 'userCreate'} with self.assertRaises(ValueError): self.res_helper._build_url(resource, action) def test_get_response_params(self, *args): method = 'get' resource = 'users' mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.USERS_DATA} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session resp = self.res_helper.get_response_params(method, resource) self.assertTrue(resp) @mock.patch.object(tg_landslide.LandslideResourceHelper, '_get_users') @mock.patch.object(time, 'time') def test__create_user(self, mock_time, mock_get_users, *args): mock_time.strftime.return_value = self.EXPIRE_DATE post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE, 'json.return_value': {'id': self.SUCCESS_RECORD_ID}} mock_session = mock.Mock(spec=requests.Session) mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session self.assertEqual(self.SUCCESS_RECORD_ID, self.res_helper._create_user(self.AUTH_DATA)) mock_get_users.assert_not_called() @mock.patch.object(tg_landslide.LandslideResourceHelper, '_modify_user') @mock.patch.object(time, 'time') def test__create_user_username_exists(self, mock_time, mock_modify_user, *args): mock_time.strftime.return_value = self.EXPIRE_DATE mock_modify_user.return_value = {'id': self.SUCCESS_RECORD_ID, 'result': 'No changes requested'} post_resp_data = { 'status_code': self.ERROR_CODE, 'json.return_value': {'id': self.SUCCESS_OK_CODE, 'apiCode': self.NOT_MODIFIED_CODE}} mock_session = mock.Mock(spec=requests.Session) mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session res = self.res_helper._create_user(self.AUTH_DATA) mock_modify_user.assert_called_once_with(TAS_INFO['user'], {'isActive': 'true'}) self.assertEqual(self.SUCCESS_RECORD_ID, res) @mock.patch.object(time, 'time') def test__create_user_error(self, mock_time, *args): mock_time.strftime.return_value = self.EXPIRE_DATE mock_session = mock.Mock(spec=requests.Session) post_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': {'apiCode': self.ERROR_CODE}} mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session with self.assertRaises(exceptions.RestApiError): self.res_helper._create_user(self.AUTH_DATA) def test__modify_user(self, *args): post_data = {'username': 'test_user'} mock_session = mock.Mock(spec=requests.Session) post_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': {'id': self.SUCCESS_RECORD_ID}} mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session res = self.res_helper._modify_user(username=self.TEST_USER, fields=post_data) self.assertEqual(self.SUCCESS_RECORD_ID, res['id']) def test__modify_user_rest_resp_fail(self, *args): post_data = {'non-existing-key': ''} mock_session = mock.Mock(spec=requests.Session) mock_session.post.ok = False self.res_helper.session = mock_session self.assertRaises(exceptions.RestApiError, self.res_helper._modify_user, username=self.TEST_USER, fields=post_data) mock_session.post.assert_called_once() def test__delete_user(self, *args): mock_session = mock.Mock(spec=requests.Session) self.res_helper.session = mock_session self.assertIsNone(self.res_helper._delete_user( username=self.TEST_USER)) def test__get_users(self, *args): mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.USERS_DATA} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session self.assertEqual(self.USERS_DATA['users'], self.res_helper._get_users()) def test_exec_rest_request(self, *args): resource = 'testServers' action = {'action': 'modify'} expected_url = ''.join([EXAMPLE_URL, 'testServers?action=modify']) post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE, 'json.return_value': {'id': self.SUCCESS_RECORD_ID}} mock_session = mock.Mock(spec=requests.Session) mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session self.res_helper.exec_rest_request('post', resource, action) self.res_helper.session.post.assert_called_once_with(expected_url, json={}) def test_exec_rest_request_unsupported_method_error(self, *args): resource = 'testServers' action = {'action': 'modify'} with self.assertRaises(ValueError): self.res_helper.exec_rest_request('patch', resource, action) def test_exec_rest_request_missed_action_arg(self, *args): resource = 'testServers' with self.assertRaises(ValueError): self.res_helper.exec_rest_request('post', resource) def test_exec_rest_request_raise_exc(self): resource = 'users' action = {'action': 'modify'} post_resp_data = {'status_code': self.ERROR_CODE, 'json.return_value': { 'status_code': self.ERROR_CODE}} mock_session = mock.Mock(spec=requests.Session) mock_session.post.return_value.configure_mock(**post_resp_data) self.assertRaises(exceptions.RestApiError, self.res_helper.exec_rest_request, 'post', resource, action, raise_exc=True) @mock.patch.object(time, 'time') def test_connect(self, mock_time, *args): vnfd = VNFD['vnfd:vnfd-catalog']['vnfd'][0] mock_time.strftime.return_value = self.EXPIRE_DATE self.res_helper.vnfd_helper = vnfd self.res_helper._tcl = mock.Mock() post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE, 'json.return_value': {'id': self.SUCCESS_RECORD_ID}} mock_session = mock.Mock(spec=requests.Session, headers={}) mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session self.assertIsInstance(self.res_helper.connect(), requests.Session) self.res_helper._tcl.connect.assert_called_once_with( TAS_INFO['ip'], TAS_INFO['user'], TAS_INFO['password']) def test_disconnect(self, *args): self.res_helper._tcl = mock.Mock() self.assertIsNone(self.res_helper.disconnect()) self.assertIsNone(self.res_helper.session) self.res_helper._tcl.disconnect.assert_called_once() def test_terminate(self, *args): self.assertIsNone(self.res_helper.terminate()) self.assertEqual(self.TEST_TERMINATED, self.res_helper._terminated.value) def test_create_dmf(self, *args): self.res_helper._tcl = mock.Mock() self.res_helper.vnfd_helper = mock.Mock(spec=vnf_base.VnfdHelper) self.res_helper.vnfd_helper.mgmt_interface = {'user': TAS_INFO['user']} self.assertIsNone(self.res_helper.create_dmf(DMF_CFG)) self.res_helper._tcl.create_dmf.assert_called_once_with(DMF_CFG) def test_create_dmf_as_list(self, *args): self.res_helper._tcl = mock.Mock() self.res_helper.vnfd_helper = mock.Mock(spec=vnf_base.VnfdHelper) self.res_helper.vnfd_helper.mgmt_interface = {'user': TAS_INFO['user']} self.assertIsNone(self.res_helper.create_dmf([DMF_CFG])) self.res_helper._tcl.create_dmf.assert_called_once_with(DMF_CFG) def test_delete_dmf(self, *args): self.res_helper._tcl = mock.Mock() self.assertIsNone(self.res_helper.delete_dmf(DMF_CFG)) self.res_helper._tcl.delete_dmf.assert_called_once_with(DMF_CFG) def test_delete_dmf_as_list(self, *args): self.res_helper._tcl = mock.Mock() self.assertIsNone(self.res_helper.delete_dmf([DMF_CFG])) self.res_helper._tcl.delete_dmf.assert_called_once_with(DMF_CFG) @mock.patch.object(tg_landslide.LandslideResourceHelper, 'configure_sut') def test_create_suts(self, mock_configure_sut, *args): mock_session = mock.Mock(spec=requests.Session) post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE} mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session self.assertIsNone(self.res_helper.create_suts(TS1_SUTS)) mock_configure_sut.assert_not_called() @mock.patch.object(tg_landslide.LandslideResourceHelper, 'configure_sut') def test_create_suts_sut_exists(self, mock_configure_sut, *args): sut_name = 'test_sut' suts = [ {'name': sut_name, 'role': 'SgwControlAddr', 'managementIp': '12.0.1.1', 'ip': '10.42.32.100' } ] mock_session = mock.Mock(spec=requests.Session) post_resp_data = {'status_code': self.NOT_MODIFIED_CODE} mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session self.assertIsNone(self.res_helper.create_suts(suts)) mock_configure_sut.assert_called_once_with( sut_name=sut_name, json_data={k: v for k, v in suts[0].items() if k not in {'phy', 'nextHop', 'role', 'name'}}) def test_get_suts(self, *args): mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.SUTS_DATA} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session self.assertIsInstance(self.res_helper.get_suts(), list) def test_get_suts_single_id(self, *args): mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.SUTS_DATA['suts'][0]} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session self.assertIsInstance(self.res_helper.get_suts(suts_id=2), dict) def test_configure_sut(self, *args): post_data = {'managementIp': '2.2.2.2'} mock_session = mock.Mock(spec=requests.Session) post_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': {'id': self.SUCCESS_RECORD_ID}} mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session self.assertIsNone(self.res_helper.configure_sut('test_name', post_data)) mock_session.post.assert_called_once() def test_configure_sut_error(self, *args): post_data = {'managementIp': '2.2.2.2'} mock_session = mock.Mock(spec=requests.Session) post_resp_data = {'status_code': self.NOT_MODIFIED_CODE} mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session with self.assertRaises(exceptions.RestApiError): self.res_helper.configure_sut('test_name', post_data) def test_delete_suts(self, *args): mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.SUTS_DATA} delete_resp_data = {'status_code': self.SUCCESS_OK_CODE} mock_session.get.return_value.configure_mock(**get_resp_data) mock_session.delete.return_value.configure_mock(**delete_resp_data) self.res_helper.session = mock_session self.assertIsNone(self.res_helper.delete_suts()) mock_session.delete.assert_called_once() @mock.patch.object(tg_landslide.LandslideResourceHelper, 'get_test_servers') def test__check_test_servers_state(self, mock_get_test_servers, *args): mock_get_test_servers.return_value = \ self.TEST_SERVERS_DATA['testServers'] self.res_helper._check_test_servers_state() mock_get_test_servers.assert_called_once() @mock.patch.object(tg_landslide.LandslideResourceHelper, 'get_test_servers') def test__check_test_servers_state_server_not_ready( self, mock_get_test_servers, *args): test_servers_not_ready = [ { "url": ''.join([EXAMPLE_URL, "testServers/1"]), "id": 1, "name": "TestServer_1", "state": "NOT_READY", "version": "16.4.0.10" } ] mock_get_test_servers.return_value = test_servers_not_ready with self.assertRaises(RuntimeError): self.res_helper._check_test_servers_state(timeout=1, delay=0) @mock.patch.object(tg_landslide.LandslideResourceHelper, '_check_test_servers_state') def test_create_test_servers(self, mock_check_ts_state, *args): test_servers_ids = [ ts['id'] for ts in self.TEST_SERVERS_DATA['testServers']] self.res_helper.license_data['lic_id'] = TAS_INFO['license'] self.res_helper._tcl.create_test_server = mock.Mock() self.res_helper._tcl.create_test_server.side_effect = test_servers_ids self.assertIsNone(self.res_helper.create_test_servers(TEST_SERVERS)) mock_check_ts_state.assert_called_once_with(test_servers_ids) @mock.patch.object(tg_landslide.LandslideTclClient, 'resolve_test_server_name') @mock.patch.object(tg_landslide.LsTclHandler, 'execute') def test_create_test_servers_error(self, mock_execute, mock_resolve_ts_name, *args): self.res_helper.license_data['lic_id'] = TAS_INFO['license'] # Return message for case test server wasn't created mock_execute.return_value = 'TS not found' # Return message for case test server name wasn't resolved mock_resolve_ts_name.return_value = 'TS not found' with self.assertRaises(RuntimeError): self.res_helper.create_test_servers(TEST_SERVERS) def test_get_test_servers(self, *args): mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.TEST_SERVERS_DATA} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session res = self.res_helper.get_test_servers() self.assertEqual(self.TEST_SERVERS_DATA['testServers'], res) def test_get_test_servers_by_id(self, *args): mock_session = mock.Mock(spec=requests.Session) _ts = self.TEST_SERVERS_DATA['testServers'][0] get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': _ts} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session res = self.res_helper.get_test_servers(test_server_ids=[_ts['id']]) self.assertEqual([_ts], res) def test_configure_test_servers(self, *args): mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.TEST_SERVERS_DATA} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session res = self.res_helper.configure_test_servers( action={'action': 'recycle'}) self.assertEqual( [x['id'] for x in self.TEST_SERVERS_DATA['testServers']], res) self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']), mock_session.post.call_count) def test_delete_test_servers(self, *args): mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.TEST_SERVERS_DATA} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session self.assertIsNone(self.res_helper.delete_test_servers()) self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']), mock_session.delete.call_count) def test_create_test_session_res_helper(self, *args): self.res_helper._user_id = self.SUCCESS_RECORD_ID self.res_helper._tcl = mock.Mock() self.res_helper.scenario_helper.all_options = {'traffic_duration': 71} _session = {'name': 'test', 'duration': 60} self.assertIsNone(self.res_helper.create_test_session(_session)) self.res_helper._tcl.create_test_session.assert_called_once_with( {'name': _session['name'], 'duration': 71, 'library': self.SUCCESS_RECORD_ID}) def test_create_test_session_res_helper_no_traffic_duration(self, *args): self.res_helper._user_id = self.SUCCESS_RECORD_ID self.res_helper._tcl = mock.Mock() self.res_helper.scenario_helper.all_options = {} _session = {'name': 'test', 'duration': 60} self.assertIsNone(self.res_helper.create_test_session(_session)) self.res_helper._tcl.create_test_session.assert_called_once_with( {'name': _session['name'], 'duration': 60, 'library': self.SUCCESS_RECORD_ID}) @mock.patch.object(tg_landslide.LandslideTclClient, 'resolve_test_server_name', return_value='Not Found') def test_create_test_session_ts_name_not_found(self, *args): self.res_helper._user_id = self.SUCCESS_RECORD_ID test_session = { 'duration': 60, 'description': 'UE default bearer creation test case', 'name': 'default_bearer_capacity', 'tsGroups': [{'testCases': [{'type': 'SGW_Node', 'name': ''}], 'tsId': 'TestServer_3'}] } with self.assertRaises(RuntimeError): self.res_helper.create_test_session(test_session) def test_get_test_session(self, *args): test_session = {"name": self.TEST_SESSION_NAME} self.res_helper._user_id = self.SUCCESS_RECORD_ID mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': test_session} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session res = self.res_helper.get_test_session(self.TEST_SESSION_NAME) self.assertEqual(test_session, res) def test_configure_test_session(self, *args): test_session = {'name': self.TEST_SESSION_NAME} self.res_helper._user_id = self.SUCCESS_RECORD_ID self.res_helper.user_lib_uri = 'libraries/{{}}/{}'.format( self.res_helper.test_session_uri) mock_session = mock.Mock(spec=requests.Session) self.res_helper.session = mock_session res = self.res_helper.configure_test_session(self.TEST_SESSION_NAME, test_session) self.assertIsNotNone(res) mock_session.post.assert_called_once() def test_delete_test_session(self, *args): self.res_helper._user_id = self.SUCCESS_RECORD_ID self.res_helper.user_lib_uri = 'libraries/{{}}/{}'.format( self.res_helper.test_session_uri) mock_session = mock.Mock(spec=requests.Session) self.res_helper.session = mock_session res = self.res_helper.delete_test_session(self.TEST_SESSION_NAME) self.assertIsNotNone(res) mock_session.delete.assert_called_once() def test_create_running_tests(self, *args): self.res_helper._user_id = self.SUCCESS_RECORD_ID test_session = {'id': self.SUCCESS_RECORD_ID} mock_session = mock.Mock(spec=requests.Session) post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE, 'json.return_value': test_session} mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session self.res_helper.create_running_tests(self.TEST_SESSION_NAME) self.assertEqual(self.SUCCESS_RECORD_ID, self.res_helper.run_id) def test_create_running_tests_error(self, *args): self.res_helper._user_id = self.SUCCESS_RECORD_ID mock_session = mock.Mock(spec=requests.Session) post_resp_data = {'status_code': self.NOT_MODIFIED_CODE} mock_session.post.return_value.configure_mock(**post_resp_data) self.res_helper.session = mock_session with self.assertRaises(exceptions.RestApiError): self.res_helper.create_running_tests(self.TEST_SESSION_NAME) def test_get_running_tests(self, *args): mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.RUNNING_TESTS_DATA} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session res = self.res_helper.get_running_tests() self.assertEqual(self.RUNNING_TESTS_DATA['runningTests'], res) def test_delete_running_tests(self, *args): mock_session = mock.Mock(spec=requests.Session) delete_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.RUNNING_TESTS_DATA} mock_session.delete.return_value.configure_mock(**delete_resp_data) self.res_helper.session = mock_session self.assertIsNone(self.res_helper.delete_running_tests()) def test__running_tests_action(self, *args): action = 'abort' mock_session = mock.Mock(spec=requests.Session) self.res_helper.session = mock_session res = self.res_helper._running_tests_action(self.SUCCESS_RECORD_ID, action) self.assertIsNone(res) @mock.patch.object(tg_landslide.LandslideResourceHelper, '_running_tests_action') def test_stop_running_tests(self, mock_tests_action, *args): res = self.res_helper.stop_running_tests(self.SUCCESS_RECORD_ID) self.assertIsNone(res) mock_tests_action.assert_called_once() def test_check_running_test_state(self, *args): mock_session = mock.Mock(spec=requests.Session) get_resp_data = { 'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.RUNNING_TESTS_DATA["runningTests"][0]} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session res = self.res_helper.check_running_test_state(self.SUCCESS_RECORD_ID) self.assertEqual( self.RUNNING_TESTS_DATA["runningTests"][0]['testStateOrStep'], res) def test_get_running_tests_results(self, *args): mock_session = mock.Mock(spec=requests.Session) get_resp_data = {'status_code': self.SUCCESS_OK_CODE, 'json.return_value': self.TEST_RESULTS_DATA} mock_session.get.return_value.configure_mock(**get_resp_data) self.res_helper.session = mock_session res = self.res_helper.get_running_tests_results( self.SUCCESS_RECORD_ID) self.assertEqual(self.TEST_RESULTS_DATA, res) def test__write_results(self, *args): res = self.res_helper._write_results(self.TEST_RESULTS_DATA) exp_res = { "Test Summary::Actual Dedicated Bearer Session Connects": 100.0, "Test Summary::Actual Dedicated Bearer Session Disconnects": 100.0, "Test Summary::Actual Disconnect Rate(Sessions / Second)(P - I)": 164.804, "Test Summary::Average Session Disconnect Time(P - I)": 5.024, "Test Summary::Total Data Sent + Received Packets / Sec(P - I)": 1452.294 } self.assertEqual(exp_res, res) def test__write_results_no_tabs(self, *args): _res_data = copy.deepcopy(self.TEST_RESULTS_DATA) del _res_data['tabs'] # Return None if tabs not found in test results dict self.assertIsNone(self.res_helper._write_results(_res_data)) @mock.patch.object(tg_landslide.LandslideResourceHelper, 'check_running_test_state') @mock.patch.object(tg_landslide.LandslideResourceHelper, 'get_running_tests_results') def test_collect_kpi_test_running(self, mock_tests_results, mock_tests_state, *args): self.res_helper.run_id = self.SUCCESS_RECORD_ID mock_tests_state.return_value = 'RUNNING' mock_tests_results.return_value = self.TEST_RESULTS_DATA res = self.res_helper.collect_kpi() self.assertNotIn('done', res) mock_tests_state.assert_called_once_with(self.res_helper.run_id) mock_tests_results.assert_called_once_with(self.res_helper.run_id) @mock.patch.object(tg_landslide.LandslideResourceHelper, 'check_running_test_state') @mock.patch.object(tg_landslide.LandslideResourceHelper, 'get_running_tests_results') def test_collect_kpi_test_completed(self, mock_tests_results, mock_tests_state, *args): self.res_helper.run_id = self.SUCCESS_RECORD_ID mock_tests_state.return_value = 'COMPLETE' res = self.res_helper.collect_kpi() self.assertIsNotNone(res) mock_tests_state.assert_called_once_with(self.res_helper.run_id) mock_tests_results.assert_not_called() self.assertDictContainsSubset({'done': True}, res) class TestLandslideTclClient(unittest.TestCase): def setUp(self): self.mock_tcl_handler = mock.Mock(spec=tg_landslide.LsTclHandler) self.ls_res_helper = mock.Mock( spec=tg_landslide.LandslideResourceHelper) self.ls_tcl_client = tg_landslide.LandslideTclClient( self.mock_tcl_handler, self.ls_res_helper) def test___init__(self, *args): self.ls_tcl_client = tg_landslide.LandslideTclClient( self.mock_tcl_handler, self.ls_res_helper) self.assertIsNone(self.ls_tcl_client.tcl_server_ip) self.assertIsNone(self.ls_tcl_client._user) self.assertIsNone(self.ls_tcl_client._library_id) self.assertIsNone(self.ls_tcl_client._basic_library_id) self.assertEqual(set(), self.ls_tcl_client.ts_ids) self.assertIsInstance(self.ls_tcl_client._tc_types, set) self.assertIsNotNone(self.ls_tcl_client._tc_types) def test_connect_login_success(self, *args): lib_id = '123' exec_responses = ['java0x2', lib_id, lib_id] auth = ('user', 'password') self.mock_tcl_handler.execute.side_effect = exec_responses self.ls_tcl_client.connect(TAS_INFO['ip'], *auth) self.assertEqual(lib_id, self.ls_tcl_client._library_id) self.assertEqual(lib_id, self.ls_tcl_client._basic_library_id) self.assertEqual(TAS_INFO['ip'], self.ls_tcl_client.tcl_server_ip) self.assertEqual(auth[0], self.ls_tcl_client._user) self.assertEqual(len(exec_responses), self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call("ls::login 1.1.1.1 user password"), mock.call("ls::get [ls::query LibraryInfo -userLibraryName user] -Id"), ]) def test_connect_login_failed(self, *args): exec_responses = ['Login failed'] auth = ('user', 'password') self.mock_tcl_handler.execute.side_effect = exec_responses self.assertRaises(exceptions.LandslideTclException, self.ls_tcl_client.connect, TAS_INFO['ip'], *auth) self.assertIsNone(self.ls_tcl_client._library_id) self.assertIsNone(self.ls_tcl_client._basic_library_id) self.assertIsNone(self.ls_tcl_client.tcl_server_ip) self.assertIsNone(self.ls_tcl_client._user) self.assertEqual(len(exec_responses), self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_called_with( "ls::login 1.1.1.1 user password") def test_disconnect(self, *args): self.ls_tcl_client.disconnect() self.mock_tcl_handler.execute.assert_called_once_with("ls::logout") self.assertIsNone(self.ls_tcl_client.tcl_server_ip) self.assertIsNone(self.ls_tcl_client._user) self.assertIsNone(self.ls_tcl_client._library_id) self.assertIsNone(self.ls_tcl_client._basic_library_id) def test_create_test_server(self, *args): return_value = '2' self.ls_tcl_client._ts_context.vnfd_helper = \ VNFD['vnfd:vnfd-catalog']['vnfd'][0] self.ls_tcl_client._ts_context.license_data = {'lic_id': return_value} self.mock_tcl_handler.execute.return_value = return_value self.ls_tcl_client._set_thread_model = mock.Mock() res = self.ls_tcl_client.create_test_server(TEST_SERVERS[1]) self.assertEqual(3, self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::query TsId TestServer_2'), mock.call('set ts [ls::retrieve TsInfo -Name "TestServer_2"]'), mock.call('ls::get $ts -RequestedLicense'), ]) self.ls_tcl_client._set_thread_model.assert_called_once_with( TEST_SERVERS[1]['name'], TEST_SERVERS[1]['thread_model']) self.assertEqual(int(return_value), res) def test_create_test_server_fail_limit_reach(self, *args): self.mock_tcl_handler.execute.side_effect = ['TS not found', 'Add failed'] self.assertRaises(RuntimeError, self.ls_tcl_client.create_test_server, TEST_SERVERS[0]) self.assertEqual(2, self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::query TsId TestServer_1'), mock.call('ls::perform AddTs -Name "TestServer_1" ' '-Ip "192.168.122.101"'), ]) def test__add_test_server(self): ts_id = '2' self.mock_tcl_handler.execute.side_effect = ['TS not found', ts_id] self.assertEqual(ts_id, self.ls_tcl_client._add_test_server('name', 'ip')) self.assertEqual(2, self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::query TsId name'), mock.call('ls::perform AddTs -Name "name" -Ip "ip"'), ]) def test__add_test_server_failed(self): self.mock_tcl_handler.execute.side_effect = ['TS not found', 'Add failed'] self.assertRaises(RuntimeError, self.ls_tcl_client._add_test_server, 'name', 'ip') self.assertEqual(2, self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::query TsId name'), mock.call('ls::perform AddTs -Name "name" -Ip "ip"'), ]) def test__update_license(self): curr_lic_id = '111' new_lic_id = '222' exec_resp = ['java0x4', curr_lic_id, TCL_SUCCESS_RESPONSE, TCL_SUCCESS_RESPONSE] self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id} self.mock_tcl_handler.execute.side_effect = exec_resp self.ls_tcl_client._update_license('name') self.assertEqual(len(exec_resp), self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('set ts [ls::retrieve TsInfo -Name "name"]'), mock.call('ls::get $ts -RequestedLicense'), mock.call('ls::config $ts -RequestedLicense 222'), mock.call('ls::perform ModifyTs $ts'), ]) def test__update_license_same_as_current(self): curr_lic_id = '111' new_lic_id = '111' exec_resp = ['java0x4', curr_lic_id] self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id} self.mock_tcl_handler.execute.side_effect = exec_resp self.ls_tcl_client._update_license('name') self.assertEqual(len(exec_resp), self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('set ts [ls::retrieve TsInfo -Name "name"]'), mock.call('ls::get $ts -RequestedLicense'), ]) def test__set_thread_model_update_needed(self): self.ls_tcl_client._ts_context.vnfd_helper = { 'mgmt-interface': { 'cfguser_password': 'cfguser_password' } } exec_resp = ['java0x4', 'V0', '', ''] self.mock_tcl_handler.execute.side_effect = exec_resp self.ls_tcl_client._set_thread_model('name', 'Fireball') self.assertEqual(len(exec_resp), self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('set tsc [ls::perform RetrieveTsConfiguration ' '-name "name" cfguser_password]'), mock.call('ls::get $tsc -ThreadModel'), mock.call('ls::config $tsc -ThreadModel "V1_FB3"'), mock.call('ls::perform ApplyTsConfiguration $tsc cfguser_password'), ]) def test__set_thread_model_no_update_needed(self): self.ls_tcl_client._ts_context.vnfd_helper = { 'mgmt-interface': { 'cfguser_password': 'cfguser_password' } } exec_resp = ['java0x4', 'V0'] self.mock_tcl_handler.execute.side_effect = exec_resp self.ls_tcl_client._set_thread_model('name', 'Legacy') self.assertEqual(len(exec_resp), self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('set tsc [ls::perform RetrieveTsConfiguration ' '-name "name" cfguser_password]'), mock.call('ls::get $tsc -ThreadModel'), ]) @mock.patch.object(tg_landslide.LandslideTclClient, 'resolve_test_server_name', side_effect=['4', '2']) def test_create_test_session(self, *args): _session_profile = copy.deepcopy(SESSION_PROFILE) _session_profile['reservations'] = RESERVATIONS self.ls_tcl_client._save_test_session = mock.Mock() self.ls_tcl_client._configure_ts_group = mock.Mock() self.ls_tcl_client._library_id = 42 self.ls_tcl_client.create_test_session(_session_profile) self.assertEqual(17, self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('set test_ [ls::create TestSession]'), mock.call('ls::config $test_ -Library 42 ' '-Name "default_bearer_capacity"'), mock.call('ls::config $test_ -Description ' \ '"UE default bearer creation test case"'), mock.call('ls::config $test_ -Keywords ""'), mock.call('ls::config $test_ -Duration "60"'), mock.call('ls::config $test_ -Iterations "1"'), # _configure_reservation mock.call('set reservation_ [ls::create Reservation -under $test_]'), mock.call('ls::config $reservation_ -TsIndex 0 ' '-TsId 4 -TsName "TestServer_1"'), mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'), mock.call('ls::config $physubnet_ -Name "eth1" -Base "10.42.32.100" ' '-Mask "/24" -NumIps 20'), # _configure_reservation mock.call('set reservation_ [ls::create Reservation -under $test_]'), mock.call('ls::config $reservation_ -TsIndex 1 ' '-TsId 2 -TsName "TestServer_2"'), mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'), mock.call('ls::config $physubnet_ -Name "eth1" -Base "10.42.32.1" ' '-Mask "/24" -NumIps 100'), mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'), mock.call('ls::config $physubnet_ -Name "eth2" -Base "10.42.33.1" ' '-Mask "/24" -NumIps 100'), # _configure_report_options mock.call('ls::config $test_.ReportOptions -Format 1 -Ts -3 -Tc -3'), ]) def test_create_dmf(self): self.mock_tcl_handler.execute.return_value = '2' self.ls_tcl_client._save_dmf = mock.Mock() self.ls_tcl_client.create_dmf(copy.deepcopy(DMF_CFG)) self.assertEqual(6, self.mock_tcl_handler.execute.call_count) # This is needed because the dictionary is unordered and the arguments # can come in either order call1 = mock.call( 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"') call2 = mock.call( 'ls::config $dmf_ -isClientPortRange "false" -clientPort 2002') self.assertTrue( call1 in self.mock_tcl_handler.execute.mock_calls or call2 in self.mock_tcl_handler.execute.mock_calls) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('set dmf_ [ls::create Dmf]'), mock.call( 'ls::get [ls::query LibraryInfo -systemLibraryName user] -Id'), mock.call('ls::config $dmf_ -Library 2 -Name "Basic UDP"'), mock.call('ls::config $dmf_ -dataProtocol "udp"'), # mock.call( # 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"'), mock.call('ls::config $dmf_ -serverPort 2003'), ], any_order=True) def test_configure_dmf(self): self.mock_tcl_handler.execute.return_value = '2' self.ls_tcl_client._save_dmf = mock.Mock() self.ls_tcl_client.configure_dmf(DMF_CFG) self.assertEqual(6, self.mock_tcl_handler.execute.call_count) # This is need because the dictionary is unordered and the arguments # can come in either order call1 = mock.call( 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"') call2 = mock.call( 'ls::config $dmf_ -isClientPortRange "false" -clientPort 2002') self.assertTrue( call1 in self.mock_tcl_handler.execute.mock_calls or call2 in self.mock_tcl_handler.execute.mock_calls) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('set dmf_ [ls::create Dmf]'), mock.call( 'ls::get [ls::query LibraryInfo -systemLibraryName user] -Id'), mock.call('ls::config $dmf_ -Library 2 -Name "Basic UDP"'), mock.call('ls::config $dmf_ -dataProtocol "udp"'), # mock.call( # 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"'), mock.call('ls::config $dmf_ -serverPort 2003'), ], any_order=True) def test_delete_dmf(self): self.assertRaises(NotImplementedError, self.ls_tcl_client.delete_dmf, DMF_CFG) def test__save_dmf_valid(self): exec_resp = [TCL_SUCCESS_RESPONSE, TCL_SUCCESS_RESPONSE] self.mock_tcl_handler.execute.side_effect = exec_resp self.ls_tcl_client._save_dmf() self.assertEqual(len(exec_resp), self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::perform Validate -Dmf $dmf_'), mock.call('ls::save $dmf_ -overwrite'), ]) def test__save_dmf_invalid(self): exec_resp = ['Invalid', 'List of errors and warnings'] self.mock_tcl_handler.execute.side_effect = exec_resp self.assertRaises(exceptions.LandslideTclException, self.ls_tcl_client._save_dmf) self.assertEqual(len(exec_resp), self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::perform Validate -Dmf $dmf_'), mock.call('ls::get $dmf_ -ErrorsAndWarnings'), ]) def test__configure_report_options(self): _options = {'format': 'CSV', 'PerInterval': 'false'} self.ls_tcl_client._configure_report_options(_options) self.assertEqual(2, self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::config $test_.ReportOptions -Format 1 -Ts -3 -Tc -3'), mock.call('ls::config $test_.ReportOptions -PerInterval false'), ], any_order=True) def test___configure_ts_group(self, *args): _ts_group = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]) self.ls_tcl_client._configure_tc_type = mock.Mock() self.ls_tcl_client._configure_preresolved_arp = mock.Mock() self.ls_tcl_client.resolve_test_server_name = mock.Mock( return_value='2') self.ls_tcl_client._configure_ts_group(_ts_group, 0) self.mock_tcl_handler.execute.assert_called_once_with( 'set tss_ [ls::create TsGroup -under $test_ -tsId 2 ]') def test___configure_ts_group_resolve_ts_fail(self, *args): _ts_group = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]) self.ls_tcl_client._configure_tc_type = mock.Mock() self.ls_tcl_client._configure_preresolved_arp = mock.Mock() self.ls_tcl_client.resolve_test_server_name = mock.Mock( return_value='TS Not Found') self.assertRaises(RuntimeError, self.ls_tcl_client._configure_ts_group, _ts_group, 0) self.mock_tcl_handler.execute.assert_not_called() def test__configure_tc_type(self): _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0]) self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE self.ls_tcl_client._configure_parameters = mock.Mock() self.ls_tcl_client._configure_tc_type(_tc, 0) self.assertEqual(7, self.mock_tcl_handler.execute.call_count) def test__configure_tc_type_optional_param_omitted(self): _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0]) del _tc['linked'] self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE self.ls_tcl_client._configure_parameters = mock.Mock() self.ls_tcl_client._configure_tc_type(_tc, 0) self.assertEqual(6, self.mock_tcl_handler.execute.call_count) def test__configure_tc_type_wrong_type(self): _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0]) _tc['type'] = 'not_supported' self.ls_tcl_client._configure_parameters = mock.Mock() self.assertRaises(RuntimeError, self.ls_tcl_client._configure_tc_type, _tc, 0) self.mock_tcl_handler.assert_not_called() def test__configure_tc_type_not_found_basic_lib(self): _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0]) self.ls_tcl_client._configure_parameters = mock.Mock() self.mock_tcl_handler.execute.return_value = 'Invalid' self.assertRaises(RuntimeError, self.ls_tcl_client._configure_tc_type, _tc, 0) def test__configure_parameters(self): _params = copy.deepcopy( SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters']) self.ls_tcl_client._configure_parameters(_params) self.assertEqual(16, self.mock_tcl_handler.execute.call_count) def test__configure_array_param(self): _array = {"class": "Array", "array": ["0"]} self.ls_tcl_client._configure_array_param('name', _array) self.assertEqual(2, self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::create -Array-name -under $p_ ;'), mock.call('ls::create ArrayItem -under $p_.name -Value "0"'), ]) def test__configure_test_node_param(self): _params = copy.deepcopy( SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters']) self.ls_tcl_client._configure_test_node_param('SgwUserAddr', _params['SgwUserAddr']) cmd = ('ls::create -TestNode-SgwUserAddr -under $p_ -Type "eth" ' '-Phy "eth1" -Ip "SGW_USER_IP" -NumLinksOrNodes 1 ' '-NextHop "SGW_CONTROL_NEXT_HOP" -Mac "" -MTU 1500 ' '-ForcedEthInterface "" -EthStatsEnabled false -VlanId 0 ' '-VlanUserPriority 0 -NumVlan 1 -UniqueVlanAddr false;') self.mock_tcl_handler.execute.assert_called_once_with(cmd) def test__configure_sut_param(self): _params = {'name': 'name'} self.ls_tcl_client._configure_sut_param('name', _params) self.mock_tcl_handler.execute.assert_called_once_with( 'ls::create -Sut-name -under $p_ -Name "name";') def test__configure_dmf_param(self): _params = {"mainflows": [{"library": '111', "name": "Basic UDP"}], "instanceGroups": [{ "mainflowIdx": 0, "mixType": "", "rate": 0.0, "rows": [{ "clientPort": 0, "context": 0, "node": 0, "overridePort": "false", "ratingGroup": 0, "role": 0, "serviceId": 0, "transport": "Any"}] }]} self.ls_tcl_client._get_library_id = mock.Mock(return_value='111') res = self.ls_tcl_client._configure_dmf_param('name', _params) self.assertEqual(5, self.mock_tcl_handler.execute.call_count) self.assertIsNone(res) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::create -Dmf-name -under $p_ ;'), mock.call('ls::perform AddDmfMainflow $p_.Dmf 111 "Basic UDP"'), mock.call('ls::config $p_.Dmf.InstanceGroup(0) -mixType '), mock.call('ls::config $p_.Dmf.InstanceGroup(0) -rate 0.0'), mock.call('ls::config $p_.Dmf.InstanceGroup(0).Row(0) -Node 0 ' '-OverridePort false -ClientPort 0 -Context 0 -Role 0 ' '-PreferredTransport Any -RatingGroup 0 ' '-ServiceID 0'), ]) def test__configure_dmf_param_no_instance_groups(self): _params = {"mainflows": [{"library": '111', "name": "Basic UDP"}]} self.ls_tcl_client._get_library_id = mock.Mock(return_value='111') res = self.ls_tcl_client._configure_dmf_param('name', _params) self.assertEqual(2, self.mock_tcl_handler.execute.call_count) self.assertIsNone(res) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::create -Dmf-name -under $p_ ;'), mock.call('ls::perform AddDmfMainflow $p_.Dmf 111 "Basic UDP"'), ]) def test__configure_reservation(self): _reservation = copy.deepcopy(RESERVATIONS[0]) self.ls_tcl_client.resolve_test_server_name = mock.Mock( return_value='4') res = self.ls_tcl_client._configure_reservation(_reservation) self.assertIsNone(res) self.assertEqual(4, self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('set reservation_ [ls::create Reservation -under $test_]'), mock.call('ls::config $reservation_ -TsIndex 0 -TsId 4 ' + \ '-TsName "TestServer_1"'), mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'), mock.call('ls::config $physubnet_ -Name "eth1" ' + \ '-Base "10.42.32.100" -Mask "/24" -NumIps 20'), ]) def test__configure_preresolved_arp(self): _arp = [{'StartingAddress': '10.81.1.10', 'NumNodes': 1}] res = self.ls_tcl_client._configure_preresolved_arp(_arp) self.mock_tcl_handler.execute.assert_called_once() self.assertIsNone(res) self.mock_tcl_handler.execute.assert_called_once_with( 'ls::create PreResolvedArpAddress -under $tss_ ' + \ '-StartingAddress "10.81.1.10" -NumNodes 1') def test__configure_preresolved_arp_none(self): res = self.ls_tcl_client._configure_preresolved_arp(None) self.assertIsNone(res) self.mock_tcl_handler.execute.assert_not_called() def test_delete_test_session(self): self.assertRaises(NotImplementedError, self.ls_tcl_client.delete_test_session, {}) def test__save_test_session(self): self.mock_tcl_handler.execute.side_effect = [TCL_SUCCESS_RESPONSE, TCL_SUCCESS_RESPONSE] res = self.ls_tcl_client._save_test_session() self.assertEqual(2, self.mock_tcl_handler.execute.call_count) self.assertIsNone(res) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::perform Validate -TestSession $test_'), mock.call('ls::save $test_ -overwrite'), ]) def test__save_test_session_invalid(self): self.mock_tcl_handler.execute.side_effect = ['Invalid', 'Errors'] self.assertRaises(exceptions.LandslideTclException, self.ls_tcl_client._save_test_session) self.assertEqual(2, self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call('ls::perform Validate -TestSession $test_'), mock.call('ls::get $test_ -ErrorsAndWarnings'), ]) def test__get_library_id_system_lib(self): self.mock_tcl_handler.execute.return_value = '111' res = self.ls_tcl_client._get_library_id('name') self.mock_tcl_handler.execute.assert_called_once() self.assertEqual('111', res) self.mock_tcl_handler.execute.assert_called_with( 'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id') def test__get_library_id_user_lib(self): self.mock_tcl_handler.execute.side_effect = ['Not found', '222'] res = self.ls_tcl_client._get_library_id('name') self.assertEqual(2, self.mock_tcl_handler.execute.call_count) self.assertEqual('222', res) self.mock_tcl_handler.execute.assert_has_calls([ mock.call( 'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id'), mock.call( 'ls::get [ls::query LibraryInfo -userLibraryName name] -Id'), ]) def test__get_library_id_exception(self): self.mock_tcl_handler.execute.side_effect = ['Not found', 'Not found'] self.assertRaises(exceptions.LandslideTclException, self.ls_tcl_client._get_library_id, 'name') self.assertEqual(2, self.mock_tcl_handler.execute.call_count) self.mock_tcl_handler.execute.assert_has_calls([ mock.call( 'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id'), mock.call( 'ls::get [ls::query LibraryInfo -userLibraryName name] -Id'), ]) class TestLsTclHandler(unittest.TestCase): def setUp(self): self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi') self.mock_lsapi.start() self.addCleanup(self._cleanup) def _cleanup(self): self.mock_lsapi.stop() def test___init__(self, *args): self.ls_tcl_handler = tg_landslide.LsTclHandler() self.assertEqual({}, self.ls_tcl_handler.tcl_cmds) self.ls_tcl_handler._ls.tcl.assert_called_once() def test_execute(self, *args): self.ls_tcl_handler = tg_landslide.LsTclHandler() self.ls_tcl_handler.execute('command') self.assertIn('command', self.ls_tcl_handler.tcl_cmds)