# 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"
"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,
'description': 'UE default bearer creation test case',
'name': 'default_bearer_capacity',
'reportOptions': {'format': 'CSV'},
- "reservePorts": "true",
- "reservations": [
- {"tsId": 4,
- "tsIndex": 0,
- "tsName": TEST_SERVERS[0]['name'],
- "phySubnets": [
- {"base": "10.42.32.100", "mask": "/24", "name": "eth5",
- "numIps": 20},
- {"base": "10.42.33.100", "mask": "/24", "name": "eth6",
- "numIps": 20}
- ]},
- {"tsId": 2,
- "tsIndex": 1,
- "tsName": TEST_SERVERS[1]['name'],
- "phySubnets": [
- {"base": "10.42.32.1", "mask": "/24", "name": "eth5",
- "numIps": 100},
- {"base": "10.42.33.1", "mask": "/24", "name": "eth6",
- "numIps": 100}
- ]}
- ],
+ 'reservePorts': 'false',
'tsGroups': [
{
'testCases': [{
'type': 'SGW_Node',
'name': '',
- 'linked': False,
+ 'linked': "false",
'AssociatedPhys': '',
'parameters': {
'SgiPtpTunnelEn': 'false',
'forcedEthInterface': '',
'ip': 'SGW_USER_IP',
'class': 'TestNode',
- 'ethStatsEnabled': False,
+ 'ethStatsEnabled': "false",
'mtu': 1500
},
'SgwControlAddr': {
'forcedEthInterface': '',
'ip': 'SGW_CONTROL_IP',
'class': 'TestNode',
- 'ethStatsEnabled': False,
+ 'ethStatsEnabled': "false",
'mtu': 1500,
'nextHop': 'SGW_CONTROL_NEXT_HOP'
},
'class': 'Dmf',
'instanceGroups': [
{
- 'startPaused': False,
+ 'startPaused': "false",
'rate': 0,
'mainflowIdx': 0,
'mixType': ''
'forcedEthInterface': '',
'ip': 'MME_CONTROL_IP',
'class': 'TestNode',
- 'ethStatsEnabled': False,
+ 'ethStatsEnabled': "false",
'mtu': 1500
},
'SgwUserSut': {
'forcedEthInterface': '',
'ip': 'NET_HOST_IP',
'class': 'TestNode',
- 'ethStatsEnabled': False,
+ 'ethStatsEnabled': "false",
'mtu': 1500
},
'DedicatedsPerDefaultBearer': '0',
'forcedEthInterface': '',
'ip': 'ENB_USER_IP',
'class': 'TestNode',
- 'ethStatsEnabled': False,
+ 'ethStatsEnabled': "false",
'mtu': 1500
},
'TestType': 'SGW-NODAL'
}
+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/'])
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)
res = self.res_helper.configure_test_servers(
action={'action': 'recycle'})
self.assertEqual(
- {x['id'] for x in self.TEST_SERVERS_DATA['testServers']},
- set(res))
+ [x['id'] for x in self.TEST_SERVERS_DATA['testServers']],
+ res)
self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']),
mock_session.post.call_count)
self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']),
mock_session.delete.call_count)
- def test_create_test_session(self, *args):
+ 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()
- test_session = {'name': 'test'}
- self.assertIsNone(self.res_helper.create_test_session(test_session))
+ 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': 'test', 'library': self.SUCCESS_RECORD_ID})
+ {'name': _session['name'],
+ 'duration': 60,
+ 'library': self.SUCCESS_RECORD_ID})
@mock.patch.object(tg_landslide.LandslideTclClient,
'resolve_test_server_name',
self.assertEqual(self.TEST_RESULTS_DATA, res)
def test__write_results(self, *args):
- self.maxDiff = None
res = self.res_helper._write_results(self.TEST_RESULTS_DATA)
exp_res = {
"Test Summary::Actual Dedicated Bearer Session Connects": 100.0,
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']
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()
+ 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.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.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.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',
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'
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'
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 = {
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 = {
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', return_value='2')
+ '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(20, self.mock_tcl_handler.execute.call_count)
+ 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(DMF_CFG)
+ 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._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.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.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()
+ 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.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])
"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'])
- self.mock_tcl_handler.execute.assert_called_once()
+ 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()
+ 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',
"clientPort": 0,
"context": 0,
"node": 0,
- "overridePort": False,
+ "overridePort": "false",
"ratingGroup": 0,
"role": 0,
"serviceId": 0,
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',
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(SESSION_PROFILE['reservations'][0])
+ _reservation = copy.deepcopy(RESERVATIONS[0])
self.ls_tcl_client.resolve_test_server_name = mock.Mock(
- return_value='2')
+ return_value='4')
res = self.ls_tcl_client._configure_reservation(_reservation)
self.assertIsNone(res)
- self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
+ 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',
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,
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.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):
Code Review / yardstick.git / blobdiff