1 # Copyright (c) 2018 Intel Corporation
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
22 from yardstick.benchmark.contexts import base as ctx_base
23 from yardstick.common import exceptions
24 from yardstick.common import utils as common_utils
25 from yardstick.common import yaml_loader
26 from yardstick.network_services import utils as net_serv_utils
27 from yardstick.network_services.traffic_profile import landslide_profile
28 from yardstick.network_services.vnf_generic.vnf import sample_vnf
29 from yardstick.network_services.vnf_generic.vnf import tg_landslide
30 from yardstick.network_services.vnf_generic.vnf import base as vnf_base
34 EXAMPLE_URL = 'http://example.com/'
35 TCL_SUCCESS_RESPONSE = 'ls_ok'
38 {'ip': '192.168.122.101',
41 'base': '10.42.32.100',
46 'name': 'TestServer_1'},
47 {'ip': '192.168.122.102',
59 'preResolvedArpAddress': [
61 'StartingAddress': '10.42.33.5'}
64 'name': 'TestServer_2',
65 'thread_model': 'Fireball'
70 {'name': 'SGW - C TestNode',
71 'role': 'SgwControlAddr',
72 'managementIp': '12.0.1.1',
75 'nextHop': '10.42.32.5'
77 {'name': 'SGW - U TestNode',
78 'role': 'SgwUserAddr',
79 'managementIp': '12.0.1.2',
82 'nextHop': '10.42.32.5'
87 {'name': 'eNodeB TestNode',
88 'role': 'EnbUserAddr',
89 'managementIp': '12.0.2.1',
92 'nextHop': '10.42.32.5'
94 {'name': 'MME TestNode',
95 'role': 'MmeControlAddr',
96 'managementIp': '12.0.3.1',
99 'nextHop': '10.42.32.5'
101 {'name': 'NetHost TestNode',
102 'role': 'NetworkHostAddrLocal',
103 'managementIp': '12.0.4.1',
106 'nextHop': '10.42.32.5'
108 {'name': 'PGW TestNode',
110 'managementIp': '12.0.5.1',
111 'ip': '10.42.32.105',
113 'nextHop': '10.42.32.5'
115 {'name': 'SGW - C SUT',
117 'managementIp': '12.0.6.1',
120 {'name': 'SGW - U SUT',
121 'role': 'SgwUserSut',
122 'managementIp': '12.0.6.2',
123 'ip': '10.42.32.101'}
127 'vnfd:vnfd-catalog': {
129 'short-name': 'landslide',
131 'description': 'AB client interface details',
132 'name': 'abclient-baremetal',
133 'id': 'abclient-baremetal',
134 'external-interface': []}],
135 'description': 'Spirent Landslide traffic generator',
136 'config': [{'test_server': TEST_SERVERS[0], 'suts': TS1_SUTS},
137 {'test_server': TEST_SERVERS[1], 'suts': TS2_SUTS}],
139 'vdu-id': 'landslide-tas',
142 'super-user': 'super-user',
143 'super-user-password': 'super-user-password',
144 'cfguser_password': 'cfguser_password',
150 'tx_throughput_mbps',
151 'rx_throughput_mbps',
154 'activation_rate_sessps',
155 'deactivation_rate_sessps']},
156 'id': 'LandslideTrafficGen',
157 'name': 'LandslideTrafficGen'}]}}
159 TAS_INFO = VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
168 "isClientPortRange": "false"
170 "dataProtocol": "udp",
175 {'tsName': TEST_SERVERS[0]['name'],
176 'phySubnets': TEST_SERVERS[0]['phySubnets'],
177 'tsId': TEST_SERVERS[0]['name'],
179 {'tsName': TEST_SERVERS[1]['name'],
180 'phySubnets': TEST_SERVERS[1]['phySubnets'],
181 'tsId': TEST_SERVERS[1]['name'],
188 'description': 'UE default bearer creation test case',
189 'name': 'default_bearer_capacity',
190 'reportOptions': {'format': 'CSV'},
191 'reservePorts': 'false',
198 'AssociatedPhys': '',
200 'SgiPtpTunnelEn': 'false',
201 'Gtp2Imsi': '505024101215074',
202 'Sessions': '100000',
203 'S5Protocol': 'GTPv2',
204 'TrafficMtu': '1500',
205 'Gtp2Version': '13.6.0',
206 'BearerV4AddrPool': '1.0.0.1',
207 'Gtp2Imei': '50502410121507',
209 'DedicatedsPerDefaultBearer': '0',
210 'DefaultBearers': '1',
212 'numLinksOrNodes': 1,
214 'forcedEthInterface': '',
217 'ethStatsEnabled': "false",
221 'numLinksOrNodes': 1,
223 'forcedEthInterface': '',
224 'ip': 'SGW_CONTROL_IP',
226 'ethStatsEnabled': "false",
228 'nextHop': 'SGW_CONTROL_NEXT_HOP'
230 'BearerAddrPool': '2001::1',
231 'TestType': 'SGW-NODE'
234 'tsId': TEST_SERVERS[0]['name']},
240 'DataTraffic': 'Continuous',
241 'TrafficStartType': 'When All Sessions Established',
242 'NetworkHost': 'Local',
243 'Gtp2Imsi': '505024101215074',
254 'startPaused': "false",
261 'S5Protocol': 'GTPv2',
262 'DataUserCfgFileEn': 'false',
263 'PgwUserSutEn': 'false',
265 'numLinksOrNodes': 1,
267 'forcedEthInterface': '',
268 'ip': 'MME_CONTROL_IP',
270 'ethStatsEnabled': "false",
275 'name': 'SGW_USER_NAME'
277 'TestActivity': 'Capacity Test',
278 'NetworkHostAddrLocal': {
279 'numLinksOrNodes': 1,
281 'forcedEthInterface': '',
284 'ethStatsEnabled': "false",
287 'DedicatedsPerDefaultBearer': '0',
288 'DisconnectRate': '1000.0',
289 'Sessions': '100000',
292 'name': 'SGW_CONTROL_NAME'
294 'TrafficMtu': '1500',
295 'Gtp2Version': '13.6.0',
296 'Gtp2Imei': '50502410121507',
297 'PgwNodeEn': 'false',
298 'StartRate': '1000.0',
301 'name': 'PGW_SUT_NAME'
303 'DefaultBearers': '1',
305 'numLinksOrNodes': 1,
307 'forcedEthInterface': '',
310 'ethStatsEnabled': "false",
313 'TestType': 'SGW-NODAL'
316 'tsId': TEST_SERVERS[1]['name']
322 class TestLandslideTrafficGen(unittest.TestCase):
324 'session_profile': '/traffic_profiles/landslide/'
325 'landslide_session_default_bearer.yaml',
332 'tg__0': 'tg__0.traffic_gen',
333 'vnf__0': 'vnf__0.vnf_epc'
335 'topology': 'landslide_tg_topology.yaml',
337 'traffic_profile': '../../traffic_profiles/landslide/'
338 'landslide_dmf_udp.yaml',
340 'traffic_duration': 71,
343 'BearerAddrPool': '2002::2',
345 'BearerV4AddrPool': '2.0.0.2',
349 'StartRate': '900.0',
351 'DisconnectRate': '900.0',
357 'transactionRate': 1000,
367 'name': 'traffic_gen',
368 'file': '/etc/yardstick/nodes/pod_landslide.yaml'
373 'file': '/etc/yardstick/nodes/pod_vepc_sut.yaml'
379 "schema": "nsb:traffic_profile:0.1",
380 "name": "LandslideProfile",
381 "description": "Spirent Landslide traffic profile",
383 "traffic_type": "LandslideProfile"
390 "description": "Basic data flow using UDP/IP",
392 "dataProtocol": "udp"
396 SUCCESS_CREATED_CODE = 201
397 SUCCESS_OK_CODE = 200
398 SUCCESS_RECORD_ID = 5
402 self._id = uuid.uuid1().int
404 self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
405 self.mock_lsapi.start()
407 self.mock_ssh_helper = mock.patch.object(sample_vnf, 'VnfSshHelper')
408 self.mock_ssh_helper.start()
409 self.vnfd = VNFD['vnfd:vnfd-catalog']['vnfd'][0]
410 self.ls_tg = tg_landslide.LandslideTrafficGen(
411 NAME, self.vnfd, self._id)
412 self.session_profile = copy.deepcopy(SESSION_PROFILE)
413 self.ls_tg.session_profile = self.session_profile
415 self.addCleanup(self._cleanup)
418 self.mock_lsapi.stop()
419 self.mock_ssh_helper.stop()
421 @mock.patch.object(net_serv_utils, 'get_nsb_option')
422 def test___init__(self, mock_get_nsb_option, *args):
423 _path_to_nsb = 'path/to/nsb'
424 mock_get_nsb_option.return_value = _path_to_nsb
425 ls_tg = tg_landslide.LandslideTrafficGen(NAME, self.vnfd, self._id)
426 self.assertIsInstance(ls_tg.resource_helper,
427 tg_landslide.LandslideResourceHelper)
428 mock_get_nsb_option.assert_called_once_with('bin_path')
429 self.assertEqual(_path_to_nsb, ls_tg.bin_path)
430 self.assertEqual(NAME, ls_tg.name)
431 self.assertTrue(ls_tg.runs_traffic)
432 self.assertFalse(ls_tg.traffic_finished)
433 self.assertIsNone(ls_tg.session_profile)
435 def test_listen_traffic(self):
436 _traffic_profile = {}
437 self.assertIsNone(self.ls_tg.listen_traffic(_traffic_profile))
439 def test_terminate(self, *args):
440 self.ls_tg.resource_helper._tcl = mock.Mock()
441 self.assertIsNone(self.ls_tg.terminate())
442 self.ls_tg.resource_helper._tcl.disconnect.assert_called_once()
444 @mock.patch.object(ctx_base.Context, 'get_context_from_server',
445 return_value='fake_context')
446 def test_instantiate(self, *args):
447 self.ls_tg._tg_process = mock.Mock()
448 self.ls_tg._tg_process.start = mock.Mock()
449 self.ls_tg.resource_helper.connect = mock.Mock()
450 self.ls_tg.resource_helper.create_test_servers = mock.Mock()
451 self.ls_tg.resource_helper.create_suts = mock.Mock()
452 self.ls_tg._load_session_profile = mock.Mock()
453 self.assertIsNone(self.ls_tg.instantiate(self.SCENARIO_CFG,
455 self.ls_tg.resource_helper.connect.assert_called_once()
456 self.ls_tg.resource_helper.create_test_servers.assert_called_once()
457 _suts_blocks_num = len([item['suts'] for item in self.vnfd['config']])
458 self.assertEqual(_suts_blocks_num,
459 self.ls_tg.resource_helper.create_suts.call_count)
460 self.ls_tg._load_session_profile.assert_called_once()
462 @mock.patch.object(tg_landslide.LandslideResourceHelper,
464 def test_run_traffic(self, mock_get_tests, *args):
465 self.ls_tg.resource_helper._url = EXAMPLE_URL
466 self.ls_tg.scenario_helper.scenario_cfg = self.SCENARIO_CFG
467 mock_traffic_profile = mock.Mock(
468 spec=landslide_profile.LandslideProfile)
469 mock_traffic_profile.dmf_config = {
471 'dataProtocol': 'udp',
472 'dmf': {'library': 'test', 'name': 'name'}}
473 mock_traffic_profile.params = self.TRAFFIC_PROFILE
474 self.ls_tg.resource_helper._user_id = self.TEST_USER_ID
475 mock_get_tests.return_value = [{'id': self.SUCCESS_RECORD_ID,
476 'testStateOrStep': 'COMPLETE'}]
477 mock_post = mock.Mock()
478 mock_post.status_code = self.SUCCESS_CREATED_CODE
479 mock_post.json.return_value = {'id': self.SUCCESS_RECORD_ID}
480 mock_session = mock.Mock(spec=requests.Session)
481 mock_session.post.return_value = mock_post
482 self.ls_tg.resource_helper.session = mock_session
483 self.ls_tg.resource_helper._tcl = mock.Mock()
484 _tcl = self.ls_tg.resource_helper._tcl
485 self.assertIsNone(self.ls_tg.run_traffic(mock_traffic_profile))
486 self.assertEqual(self.SUCCESS_RECORD_ID,
487 self.ls_tg.resource_helper.run_id)
488 mock_traffic_profile.update_dmf.assert_called_with(
489 self.ls_tg.scenario_helper.all_options)
490 _tcl.create_dmf.assert_called_with(mock_traffic_profile.dmf_config)
491 _tcl.create_test_session.assert_called_with(self.session_profile)
493 @mock.patch.object(tg_landslide.LandslideResourceHelper,
494 'check_running_test_state')
495 def test_collect_kpi(self, mock_check_running_test_state, *args):
496 self.ls_tg.resource_helper.run_id = self.SUCCESS_RECORD_ID
497 mock_check_running_test_state.return_value = 'COMPLETE'
498 self.assertEqual({'done': True}, self.ls_tg.collect_kpi())
499 mock_check_running_test_state.assert_called_once()
501 def test_wait_for_instantiate(self):
502 self.assertIsNone(self.ls_tg.wait_for_instantiate())
503 self.ls_tg.wait_for_instantiate()
505 def test__update_session_suts_no_tc_role(self, *args):
506 _suts = [{'role': 'epc_role'}]
507 _testcase = {'parameters': {'diff_epc_role': {'class': 'Sut'}}}
508 res = self.ls_tg._update_session_suts(_suts, _testcase)
509 self.assertEqual(_testcase, res)
511 def test__update_session_suts(self, *args):
513 def get_testnode_param(role, key, session_prof):
514 """ Get value by key from the deep nested dict to avoid calls like:
515 e.g. session_prof['tsGroups'][0]['testCases'][1]['parameters'][key]
517 for group in session_prof['tsGroups']:
518 for tc in group['testCases']:
519 tc_params = tc['parameters']
520 if tc_params.get(role):
521 return tc_params[role][key]
523 def get_sut_param(role, key, suts):
524 """ Search list of dicts for one with specific role.
525 Return the value of related dict by key. Expect key presence.
528 if sut.get('role') == role:
532 testnode_role = 'SgwControlAddr'
534 sut_role = 'SgwUserSut'
536 config_suts = [config['suts'] for config in self.vnfd['config']]
537 session_tcs = [_tc for _ts_group in self.ls_tg.session_profile['tsGroups']
538 for _tc in _ts_group['testCases']]
539 for suts, tc in zip(config_suts, session_tcs):
540 self.assertEqual(tc, self.ls_tg._update_session_suts(suts, tc))
542 # Verify TestNode class objects keys were updated
543 for _key in {'ip', 'phy', 'nextHop'}:
545 get_testnode_param(testnode_role, _key, self.ls_tg.session_profile),
546 get_sut_param(testnode_role, _key, TS1_SUTS))
547 # Verify Sut class objects name was updated
549 get_testnode_param(sut_role, 'name', self.ls_tg.session_profile),
550 get_sut_param(sut_role, 'name', TS2_SUTS))
552 def test__update_session_test_servers(self, *args):
553 for ts_index, ts in enumerate(TEST_SERVERS):
555 self.ls_tg._update_session_test_servers(ts, ts_index))
556 # Verify preResolvedArpAddress key was added
558 _item.get('preResolvedArpAddress')
559 for _item in self.ls_tg.session_profile['tsGroups']))
560 # Verify reservations key was added to session profile
561 self.assertEqual(RESERVATIONS,
562 self.ls_tg.session_profile.get('reservations'))
563 self.assertEqual('true',
564 self.ls_tg.session_profile.get('reservePorts'))
566 def test__update_session_tc_params_assoc_phys(self):
567 _tc_options = {'AssociatedPhys': 'eth1'}
569 _testcase_orig = copy.deepcopy(_testcase)
570 res = self.ls_tg._update_session_tc_params(_tc_options, _testcase)
571 self.assertNotEqual(_testcase_orig, res)
572 self.assertEqual(_tc_options, _testcase)
574 def test__update_session_tc_params(self, *args):
576 def get_session_tc_param_value(param, tc_type, session_prof):
577 """ Get param value from the deep nested dict to avoid calls like:
578 session_prof['tsGroups'][0]['testCases'][0]['parameters'][key]
580 for test_group in session_prof['tsGroups']:
581 session_tc = test_group['testCases'][0]
582 if session_tc['type'] == tc_type:
583 return session_tc['parameters'].get(param)
585 session_tcs = [_tc for _ts_group in self.ls_tg.session_profile['tsGroups']
586 for _tc in _ts_group['testCases']]
587 scenario_tcs = [_tc for _tc in
588 self.SCENARIO_CFG['options']['test_cases']]
589 for tc_options, tc in zip(scenario_tcs, session_tcs):
592 self.ls_tg._update_session_tc_params(tc_options, tc))
594 # Verify that each test case parameter was updated
595 # Params been compared are deeply nested. Using loops to ease access.
596 for _tc in self.SCENARIO_CFG['options']['test_cases']:
597 for _key, _val in _tc.items():
601 get_session_tc_param_value(_key, _tc.get('type'),
602 self.ls_tg.session_profile))
604 def test__update_session_library_name(self, *args):
605 _session = copy.deepcopy(SESSION_PROFILE)
606 _session['tsGroups'].pop(0)
607 self.ls_tg.vnfd_helper = mock.MagicMock()
608 self.ls_tg.vnfd_helper.mgmt_interface.__getitem__.side_effect = {
609 'user': TAS_INFO['user']}
610 self.ls_tg._update_session_library_name(_session)
611 _dmf = _session['tsGroups'][0]['testCases'][0]['parameters']['Dmf']
612 # Expect DMF library name updated in Nodal test types
613 self.assertEqual(TAS_INFO['user'], _dmf['mainflows'][0]['library'])
615 def test__update_session_library_name_wrong_tc_type(self, *args):
616 _session = copy.deepcopy(SESSION_PROFILE)
617 _session['tsGroups'].pop(1)
618 self.ls_tg.vnfd_helper = mock.MagicMock()
619 self.ls_tg.vnfd_helper.mgmt_interface.__getitem__.side_effect = {
620 'user': TAS_INFO['user']}
621 # Expect DMF library name not updated in Node test types
622 self.assertNotIn('Dmf',
623 _session['tsGroups'][0]['testCases'][0]['parameters'])
624 self.ls_tg._update_session_library_name(_session)
626 @mock.patch.object(common_utils, 'open_relative_file')
627 @mock.patch.object(yaml_loader, 'yaml_load')
628 @mock.patch.object(tg_landslide.LandslideTrafficGen,
629 '_update_session_test_servers')
630 @mock.patch.object(tg_landslide.LandslideTrafficGen,
631 '_update_session_suts')
632 @mock.patch.object(tg_landslide.LandslideTrafficGen,
633 '_update_session_tc_params')
634 def test__load_session_profile(self, mock_upd_ses_tc_params,
635 mock_upd_ses_suts, mock_upd_ses_ts,
636 mock_yaml_load, *args):
637 self.ls_tg.scenario_helper.scenario_cfg = \
638 copy.deepcopy(self.SCENARIO_CFG)
639 mock_yaml_load.return_value = copy.deepcopy(SESSION_PROFILE)
640 self.assertIsNone(self.ls_tg._load_session_profile())
641 self.assertIsNotNone(self.ls_tg.session_profile)
642 # Number of blocks in configuration files
643 # Number of test servers, suts and tc params blocks should be equal
644 _config_files_blocks_num = len([item['test_server']
645 for item in self.vnfd['config']])
646 self.assertEqual(_config_files_blocks_num,
647 mock_upd_ses_ts.call_count)
648 self.assertEqual(_config_files_blocks_num,
649 mock_upd_ses_suts.call_count)
650 self.assertEqual(_config_files_blocks_num,
651 mock_upd_ses_tc_params.call_count)
653 @mock.patch.object(common_utils, 'open_relative_file')
654 @mock.patch.object(yaml_loader, 'yaml_load')
655 def test__load_session_profile_unequal_num_of_cfg_blocks(
656 self, mock_yaml_load, *args):
657 vnfd = copy.deepcopy(VNFD['vnfd:vnfd-catalog']['vnfd'][0])
658 ls_traffic_gen = tg_landslide.LandslideTrafficGen(NAME, vnfd, self._id)
659 ls_traffic_gen.scenario_helper.scenario_cfg = self.SCENARIO_CFG
660 mock_yaml_load.return_value = copy.deepcopy(SESSION_PROFILE)
661 # Delete test_servers item from pod file to make it not valid
662 ls_traffic_gen.vnfd_helper['config'].pop()
663 with self.assertRaises(RuntimeError):
664 ls_traffic_gen._load_session_profile()
666 @mock.patch.object(common_utils, 'open_relative_file')
667 @mock.patch.object(yaml_loader, 'yaml_load')
668 def test__load_session_profile_test_type_mismatch(self, mock_yaml_load,
670 vnfd = copy.deepcopy(VNFD['vnfd:vnfd-catalog']['vnfd'][0])
671 # Swap test servers data in pod file
672 vnfd['config'] = list(reversed(vnfd['config']))
673 ls_tg = tg_landslide.LandslideTrafficGen(NAME, vnfd, self._id)
674 ls_tg.scenario_helper.scenario_cfg = self.SCENARIO_CFG
675 mock_yaml_load.return_value = SESSION_PROFILE
676 with self.assertRaises(RuntimeError):
677 ls_tg._load_session_profile()
680 class TestLandslideResourceHelper(unittest.TestCase):
683 EXAMPLE_URL = ''.join([TAS_INFO['proto'], '://', TAS_INFO['ip'], ':',
684 str(PROTO_PORT), '/api/'])
685 SUCCESS_CREATED_CODE = 201
686 SUCCESS_OK_CODE = 200
687 INVALID_REST_CODE = '400'
688 NOT_MODIFIED_CODE = 500810
690 SUCCESS_RECORD_ID = 11
691 EXPIRE_DATE = '2020/01/01 12:00 FLE Standard Time'
694 AUTH_DATA = {'user': TAS_INFO['user'], 'password': TAS_INFO['password']}
695 TEST_SESSION_NAME = 'default_bearer_capacity'
699 "url": ''.join([EXAMPLE_URL, 'users/', str(SUCCESS_RECORD_ID)]),
700 "id": SUCCESS_RECORD_ID,
702 "username": TEST_USER
706 CREATE_USER_DATA = {'username': TAS_INFO['user'],
707 'expiresOn': EXPIRE_DATE,
709 'contactInformation': '',
710 'fullName': 'Test User',
711 'password': TAS_INFO['password'],
717 "url": ''.join([EXAMPLE_URL, 'suts/', str(SUCCESS_RECORD_ID)]),
718 "id": SUCCESS_RECORD_ID,
722 TEST_SERVERS_DATA = {
725 "url": ''.join([EXAMPLE_URL, "testServers/1"]),
727 "name": TEST_SERVERS[0]['name'],
729 "version": "16.4.0.10"
732 "url": ''.join([EXAMPLE_URL, "testServers/2"]),
734 "name": TEST_SERVERS[1]['name'],
736 "version": "16.4.0.10"
744 RUNNING_TESTS_DATA = {
746 "url": ''.join([EXAMPLE_URL, "runningTests/{}".format(RUN_ID)]),
747 "measurementsUrl": ''.join(
749 "runningTests/{}/measurements".format(RUN_ID)]),
750 "criteriaUrl": ''.join(
752 "runningTests/{}/criteria".format(RUN_ID)]),
755 "library": SUCCESS_RECORD_ID,
756 "name": "default_bearer_capacity",
758 "criteriaStatus": "NA",
759 "testStateOrStep": "COMPLETE"
762 TEST_RESULTS_DATA = {
765 "actualTime": 1521548057296,
769 "Start Time": "Tue Mar 20 07:11:55 CDT 2018",
770 "Actual Dedicated Bearer Session Connects": "100",
771 "Actual Dedicated Bearer Session Disconnects": "100",
772 "Actual Disconnect Rate(Sessions / Second)(P - I)": "164.804",
773 "Average Session Disconnect Time(P - I)": "5.024 s",
774 "Total Data Sent + Received Packets / Sec(P - I)": "1,452.294"
778 self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
779 self.mock_lsapi.start()
781 mock_env_helper = mock.Mock()
782 self.res_helper = tg_landslide.LandslideResourceHelper(mock_env_helper)
783 self.res_helper._url = EXAMPLE_URL
785 self.addCleanup(self._cleanup)
788 self.mock_lsapi.stop()
789 self.res_helper._url = None
791 def test___init__(self, *args):
792 self.assertIsInstance(self.res_helper,
793 tg_landslide.LandslideResourceHelper)
794 self.assertEqual({}, self.res_helper._result)
795 self.assertIsNone(self.res_helper.run_id)
797 @mock.patch.object(tg_landslide.LandslideResourceHelper,
798 'stop_running_tests')
799 @mock.patch.object(tg_landslide.LandslideResourceHelper,
801 def test_abort_running_tests_no_running_tests(self, mock_get_tests,
802 mock_stop_tests, *args):
803 tests_data = [{'id': self.SUCCESS_RECORD_ID,
804 'testStateOrStep': 'COMPLETE'}]
805 mock_get_tests.return_value = tests_data
806 self.assertIsNone(self.res_helper.abort_running_tests())
807 mock_stop_tests.assert_not_called()
809 @mock.patch.object(time, 'sleep')
810 @mock.patch.object(tg_landslide.LandslideResourceHelper,
811 'stop_running_tests')
812 @mock.patch.object(tg_landslide.LandslideResourceHelper,
814 def test_abort_running_tests(self, mock_get_tests, mock_stop_tests, *args):
815 test_states_seq = iter(['RUNNING', 'COMPLETE'])
817 def configure_mock(*args):
818 return [{'id': self.SUCCESS_RECORD_ID,
819 'testStateOrStep': next(test_states_seq)}]
821 mock_get_tests.side_effect = configure_mock
822 self.assertIsNone(self.res_helper.abort_running_tests())
823 mock_stop_tests.assert_called_once_with(
824 running_test_id=self.SUCCESS_RECORD_ID,
826 self.assertEqual(2, mock_get_tests.call_count)
828 @mock.patch.object(tg_landslide.LandslideResourceHelper,
829 'stop_running_tests')
830 @mock.patch.object(tg_landslide.LandslideResourceHelper,
832 def test_abort_running_tests_error(self, mock_get_tests, mock_stop_tests,
834 tests_data = {'id': self.SUCCESS_RECORD_ID,
835 'testStateOrStep': 'RUNNING'}
836 mock_get_tests.return_value = [tests_data]
837 with self.assertRaises(RuntimeError):
838 self.res_helper.abort_running_tests(timeout=1, delay=1)
839 mock_stop_tests.assert_called_with(
840 running_test_id=self.SUCCESS_RECORD_ID,
843 def test__build_url(self, *args):
845 action = {'action': 'userCreate'}
846 expected_url = ''.join([EXAMPLE_URL, 'users?action=userCreate'])
847 self.assertEqual(expected_url,
848 self.res_helper._build_url(resource, action))
850 def test__build_url_error(self, *args):
852 action = {'action': 'userCreate'}
854 with self.assertRaises(ValueError):
855 self.res_helper._build_url(resource, action)
857 def test_get_response_params(self, *args):
860 mock_session = mock.Mock(spec=requests.Session)
861 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
862 'json.return_value': self.USERS_DATA}
863 mock_session.get.return_value.configure_mock(**get_resp_data)
864 self.res_helper.session = mock_session
865 resp = self.res_helper.get_response_params(method, resource)
866 self.assertTrue(resp)
868 @mock.patch.object(tg_landslide.LandslideResourceHelper, '_get_users')
869 @mock.patch.object(time, 'time')
870 def test__create_user(self, mock_time, mock_get_users, *args):
871 mock_time.strftime.return_value = self.EXPIRE_DATE
872 post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
873 'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
874 mock_session = mock.Mock(spec=requests.Session)
875 mock_session.post.return_value.configure_mock(**post_resp_data)
876 self.res_helper.session = mock_session
877 self.assertEqual(self.SUCCESS_RECORD_ID,
878 self.res_helper._create_user(self.AUTH_DATA))
879 mock_get_users.assert_not_called()
881 @mock.patch.object(tg_landslide.LandslideResourceHelper, '_modify_user')
882 @mock.patch.object(time, 'time')
883 def test__create_user_username_exists(self, mock_time, mock_modify_user,
885 mock_time.strftime.return_value = self.EXPIRE_DATE
886 mock_modify_user.return_value = {'id': self.SUCCESS_RECORD_ID,
887 'result': 'No changes requested'}
889 'status_code': self.ERROR_CODE,
890 'json.return_value': {'id': self.SUCCESS_OK_CODE,
891 'apiCode': self.NOT_MODIFIED_CODE}}
892 mock_session = mock.Mock(spec=requests.Session)
893 mock_session.post.return_value.configure_mock(**post_resp_data)
894 self.res_helper.session = mock_session
895 res = self.res_helper._create_user(self.AUTH_DATA)
896 mock_modify_user.assert_called_once_with(TAS_INFO['user'],
897 {'isActive': 'true'})
898 self.assertEqual(self.SUCCESS_RECORD_ID, res)
900 @mock.patch.object(time, 'time')
901 def test__create_user_error(self, mock_time, *args):
902 mock_time.strftime.return_value = self.EXPIRE_DATE
903 mock_session = mock.Mock(spec=requests.Session)
904 post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
905 'json.return_value': {'apiCode': self.ERROR_CODE}}
906 mock_session.post.return_value.configure_mock(**post_resp_data)
907 self.res_helper.session = mock_session
908 with self.assertRaises(exceptions.RestApiError):
909 self.res_helper._create_user(self.AUTH_DATA)
911 def test__modify_user(self, *args):
912 post_data = {'username': 'test_user'}
913 mock_session = mock.Mock(spec=requests.Session)
914 post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
915 'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
916 mock_session.post.return_value.configure_mock(**post_resp_data)
917 self.res_helper.session = mock_session
918 res = self.res_helper._modify_user(username=self.TEST_USER,
920 self.assertEqual(self.SUCCESS_RECORD_ID, res['id'])
922 def test__modify_user_rest_resp_fail(self, *args):
923 post_data = {'non-existing-key': ''}
924 mock_session = mock.Mock(spec=requests.Session)
925 mock_session.post.ok = False
926 self.res_helper.session = mock_session
927 self.assertRaises(exceptions.RestApiError,
928 self.res_helper._modify_user,
929 username=self.TEST_USER, fields=post_data)
930 mock_session.post.assert_called_once()
932 def test__delete_user(self, *args):
933 mock_session = mock.Mock(spec=requests.Session)
934 self.res_helper.session = mock_session
935 self.assertIsNone(self.res_helper._delete_user(
936 username=self.TEST_USER))
938 def test__get_users(self, *args):
939 mock_session = mock.Mock(spec=requests.Session)
940 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
941 'json.return_value': self.USERS_DATA}
942 mock_session.get.return_value.configure_mock(**get_resp_data)
943 self.res_helper.session = mock_session
944 self.assertEqual(self.USERS_DATA['users'],
945 self.res_helper._get_users())
947 def test_exec_rest_request(self, *args):
948 resource = 'testServers'
949 action = {'action': 'modify'}
950 expected_url = ''.join([EXAMPLE_URL, 'testServers?action=modify'])
951 post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
952 'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
953 mock_session = mock.Mock(spec=requests.Session)
954 mock_session.post.return_value.configure_mock(**post_resp_data)
955 self.res_helper.session = mock_session
956 self.res_helper.exec_rest_request('post', resource, action)
957 self.res_helper.session.post.assert_called_once_with(expected_url,
960 def test_exec_rest_request_unsupported_method_error(self, *args):
961 resource = 'testServers'
962 action = {'action': 'modify'}
963 with self.assertRaises(ValueError):
964 self.res_helper.exec_rest_request('patch', resource, action)
966 def test_exec_rest_request_missed_action_arg(self, *args):
967 resource = 'testServers'
968 with self.assertRaises(ValueError):
969 self.res_helper.exec_rest_request('post', resource)
971 def test_exec_rest_request_raise_exc(self):
973 action = {'action': 'modify'}
974 post_resp_data = {'status_code': self.ERROR_CODE,
975 'json.return_value': {
976 'status_code': self.ERROR_CODE}}
977 mock_session = mock.Mock(spec=requests.Session)
978 mock_session.post.return_value.configure_mock(**post_resp_data)
979 self.assertRaises(exceptions.RestApiError,
980 self.res_helper.exec_rest_request,
981 'post', resource, action, raise_exc=True)
983 @mock.patch.object(time, 'time')
984 def test_connect(self, mock_time, *args):
985 vnfd = VNFD['vnfd:vnfd-catalog']['vnfd'][0]
986 mock_time.strftime.return_value = self.EXPIRE_DATE
987 self.res_helper.vnfd_helper = vnfd
989 self.res_helper._tcl = mock.Mock()
990 post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
991 'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
992 mock_session = mock.Mock(spec=requests.Session, headers={})
993 mock_session.post.return_value.configure_mock(**post_resp_data)
994 self.res_helper.session = mock_session
995 self.assertIsInstance(self.res_helper.connect(), requests.Session)
996 self.res_helper._tcl.connect.assert_called_once_with(
999 TAS_INFO['password'])
1001 def test_disconnect(self, *args):
1002 self.res_helper._tcl = mock.Mock()
1003 self.assertIsNone(self.res_helper.disconnect())
1004 self.assertIsNone(self.res_helper.session)
1005 self.res_helper._tcl.disconnect.assert_called_once()
1007 def test_terminate(self, *args):
1008 self.assertIsNone(self.res_helper.terminate())
1009 self.assertEqual(self.TEST_TERMINATED,
1010 self.res_helper._terminated.value)
1012 def test_create_dmf(self, *args):
1013 self.res_helper._tcl = mock.Mock()
1014 self.res_helper.vnfd_helper = mock.Mock(spec=vnf_base.VnfdHelper)
1015 self.res_helper.vnfd_helper.mgmt_interface = {'user': TAS_INFO['user']}
1016 self.assertIsNone(self.res_helper.create_dmf(DMF_CFG))
1017 self.res_helper._tcl.create_dmf.assert_called_once_with(DMF_CFG)
1019 def test_create_dmf_as_list(self, *args):
1020 self.res_helper._tcl = mock.Mock()
1021 self.res_helper.vnfd_helper = mock.Mock(spec=vnf_base.VnfdHelper)
1022 self.res_helper.vnfd_helper.mgmt_interface = {'user': TAS_INFO['user']}
1023 self.assertIsNone(self.res_helper.create_dmf([DMF_CFG]))
1024 self.res_helper._tcl.create_dmf.assert_called_once_with(DMF_CFG)
1026 def test_delete_dmf(self, *args):
1027 self.res_helper._tcl = mock.Mock()
1028 self.assertIsNone(self.res_helper.delete_dmf(DMF_CFG))
1029 self.res_helper._tcl.delete_dmf.assert_called_once_with(DMF_CFG)
1031 def test_delete_dmf_as_list(self, *args):
1032 self.res_helper._tcl = mock.Mock()
1033 self.assertIsNone(self.res_helper.delete_dmf([DMF_CFG]))
1034 self.res_helper._tcl.delete_dmf.assert_called_once_with(DMF_CFG)
1036 @mock.patch.object(tg_landslide.LandslideResourceHelper, 'configure_sut')
1037 def test_create_suts(self, mock_configure_sut, *args):
1038 mock_session = mock.Mock(spec=requests.Session)
1039 post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE}
1040 mock_session.post.return_value.configure_mock(**post_resp_data)
1041 self.res_helper.session = mock_session
1042 self.assertIsNone(self.res_helper.create_suts(TS1_SUTS))
1043 mock_configure_sut.assert_not_called()
1045 @mock.patch.object(tg_landslide.LandslideResourceHelper, 'configure_sut')
1046 def test_create_suts_sut_exists(self, mock_configure_sut, *args):
1047 sut_name = 'test_sut'
1050 'role': 'SgwControlAddr',
1051 'managementIp': '12.0.1.1',
1052 'ip': '10.42.32.100'
1055 mock_session = mock.Mock(spec=requests.Session)
1056 post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
1057 mock_session.post.return_value.configure_mock(**post_resp_data)
1058 self.res_helper.session = mock_session
1059 self.assertIsNone(self.res_helper.create_suts(suts))
1060 mock_configure_sut.assert_called_once_with(
1062 json_data={k: v for k, v in suts[0].items()
1063 if k not in {'phy', 'nextHop', 'role', 'name'}})
1065 def test_get_suts(self, *args):
1066 mock_session = mock.Mock(spec=requests.Session)
1067 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1068 'json.return_value': self.SUTS_DATA}
1069 mock_session.get.return_value.configure_mock(**get_resp_data)
1070 self.res_helper.session = mock_session
1071 self.assertIsInstance(self.res_helper.get_suts(), list)
1073 def test_get_suts_single_id(self, *args):
1074 mock_session = mock.Mock(spec=requests.Session)
1075 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1076 'json.return_value': self.SUTS_DATA['suts'][0]}
1077 mock_session.get.return_value.configure_mock(**get_resp_data)
1078 self.res_helper.session = mock_session
1079 self.assertIsInstance(self.res_helper.get_suts(suts_id=2), dict)
1081 def test_configure_sut(self, *args):
1082 post_data = {'managementIp': '2.2.2.2'}
1083 mock_session = mock.Mock(spec=requests.Session)
1084 post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1085 'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
1086 mock_session.post.return_value.configure_mock(**post_resp_data)
1087 self.res_helper.session = mock_session
1088 self.assertIsNone(self.res_helper.configure_sut('test_name',
1090 mock_session.post.assert_called_once()
1092 def test_configure_sut_error(self, *args):
1093 post_data = {'managementIp': '2.2.2.2'}
1094 mock_session = mock.Mock(spec=requests.Session)
1095 post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
1096 mock_session.post.return_value.configure_mock(**post_resp_data)
1097 self.res_helper.session = mock_session
1098 with self.assertRaises(exceptions.RestApiError):
1099 self.res_helper.configure_sut('test_name', post_data)
1101 def test_delete_suts(self, *args):
1102 mock_session = mock.Mock(spec=requests.Session)
1103 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1104 'json.return_value': self.SUTS_DATA}
1105 delete_resp_data = {'status_code': self.SUCCESS_OK_CODE}
1106 mock_session.get.return_value.configure_mock(**get_resp_data)
1107 mock_session.delete.return_value.configure_mock(**delete_resp_data)
1108 self.res_helper.session = mock_session
1109 self.assertIsNone(self.res_helper.delete_suts())
1110 mock_session.delete.assert_called_once()
1112 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1114 def test__check_test_servers_state(self, mock_get_test_servers, *args):
1115 mock_get_test_servers.return_value = \
1116 self.TEST_SERVERS_DATA['testServers']
1117 self.res_helper._check_test_servers_state()
1118 mock_get_test_servers.assert_called_once()
1120 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1122 def test__check_test_servers_state_server_not_ready(
1123 self, mock_get_test_servers, *args):
1124 test_servers_not_ready = [
1126 "url": ''.join([EXAMPLE_URL, "testServers/1"]),
1128 "name": "TestServer_1",
1129 "state": "NOT_READY",
1130 "version": "16.4.0.10"
1134 mock_get_test_servers.return_value = test_servers_not_ready
1135 with self.assertRaises(RuntimeError):
1136 self.res_helper._check_test_servers_state(timeout=1, delay=0)
1138 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1139 '_check_test_servers_state')
1140 def test_create_test_servers(self, mock_check_ts_state, *args):
1141 test_servers_ids = [
1142 ts['id'] for ts in self.TEST_SERVERS_DATA['testServers']]
1144 self.res_helper.license_data['lic_id'] = TAS_INFO['license']
1145 self.res_helper._tcl.create_test_server = mock.Mock()
1146 self.res_helper._tcl.create_test_server.side_effect = test_servers_ids
1147 self.assertIsNone(self.res_helper.create_test_servers(TEST_SERVERS))
1148 mock_check_ts_state.assert_called_once_with(test_servers_ids)
1150 @mock.patch.object(tg_landslide.LandslideTclClient,
1151 'resolve_test_server_name')
1152 @mock.patch.object(tg_landslide.LsTclHandler, 'execute')
1153 def test_create_test_servers_error(self, mock_execute,
1154 mock_resolve_ts_name, *args):
1155 self.res_helper.license_data['lic_id'] = TAS_INFO['license']
1156 # Return message for case test server wasn't created
1157 mock_execute.return_value = 'TS not found'
1158 # Return message for case test server name wasn't resolved
1159 mock_resolve_ts_name.return_value = 'TS not found'
1160 with self.assertRaises(RuntimeError):
1161 self.res_helper.create_test_servers(TEST_SERVERS)
1163 def test_get_test_servers(self, *args):
1164 mock_session = mock.Mock(spec=requests.Session)
1165 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1166 'json.return_value': self.TEST_SERVERS_DATA}
1167 mock_session.get.return_value.configure_mock(**get_resp_data)
1168 self.res_helper.session = mock_session
1169 res = self.res_helper.get_test_servers()
1170 self.assertEqual(self.TEST_SERVERS_DATA['testServers'], res)
1172 def test_get_test_servers_by_id(self, *args):
1173 mock_session = mock.Mock(spec=requests.Session)
1175 _ts = self.TEST_SERVERS_DATA['testServers'][0]
1176 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1177 'json.return_value': _ts}
1178 mock_session.get.return_value.configure_mock(**get_resp_data)
1179 self.res_helper.session = mock_session
1180 res = self.res_helper.get_test_servers(test_server_ids=[_ts['id']])
1181 self.assertEqual([_ts], res)
1183 def test_configure_test_servers(self, *args):
1184 mock_session = mock.Mock(spec=requests.Session)
1185 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1186 'json.return_value': self.TEST_SERVERS_DATA}
1187 mock_session.get.return_value.configure_mock(**get_resp_data)
1188 self.res_helper.session = mock_session
1189 res = self.res_helper.configure_test_servers(
1190 action={'action': 'recycle'})
1192 [x['id'] for x in self.TEST_SERVERS_DATA['testServers']],
1194 self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']),
1195 mock_session.post.call_count)
1197 def test_delete_test_servers(self, *args):
1198 mock_session = mock.Mock(spec=requests.Session)
1199 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1200 'json.return_value': self.TEST_SERVERS_DATA}
1201 mock_session.get.return_value.configure_mock(**get_resp_data)
1202 self.res_helper.session = mock_session
1203 self.assertIsNone(self.res_helper.delete_test_servers())
1204 self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']),
1205 mock_session.delete.call_count)
1207 def test_create_test_session_res_helper(self, *args):
1208 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1209 self.res_helper._tcl = mock.Mock()
1210 self.res_helper.scenario_helper.all_options = {'traffic_duration': 71}
1211 _session = {'name': 'test', 'duration': 60}
1212 self.assertIsNone(self.res_helper.create_test_session(_session))
1213 self.res_helper._tcl.create_test_session.assert_called_once_with(
1214 {'name': _session['name'],
1216 'library': self.SUCCESS_RECORD_ID})
1218 def test_create_test_session_res_helper_no_traffic_duration(self, *args):
1219 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1220 self.res_helper._tcl = mock.Mock()
1221 self.res_helper.scenario_helper.all_options = {}
1222 _session = {'name': 'test', 'duration': 60}
1223 self.assertIsNone(self.res_helper.create_test_session(_session))
1224 self.res_helper._tcl.create_test_session.assert_called_once_with(
1225 {'name': _session['name'],
1227 'library': self.SUCCESS_RECORD_ID})
1229 @mock.patch.object(tg_landslide.LandslideTclClient,
1230 'resolve_test_server_name',
1231 return_value='Not Found')
1232 def test_create_test_session_ts_name_not_found(self, *args):
1233 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1236 'description': 'UE default bearer creation test case',
1237 'name': 'default_bearer_capacity',
1238 'tsGroups': [{'testCases': [{'type': 'SGW_Node',
1240 'tsId': 'TestServer_3'}]
1242 with self.assertRaises(RuntimeError):
1243 self.res_helper.create_test_session(test_session)
1245 def test_get_test_session(self, *args):
1246 test_session = {"name": self.TEST_SESSION_NAME}
1247 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1248 mock_session = mock.Mock(spec=requests.Session)
1249 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1250 'json.return_value': test_session}
1251 mock_session.get.return_value.configure_mock(**get_resp_data)
1252 self.res_helper.session = mock_session
1253 res = self.res_helper.get_test_session(self.TEST_SESSION_NAME)
1254 self.assertEqual(test_session, res)
1256 def test_configure_test_session(self, *args):
1257 test_session = {'name': self.TEST_SESSION_NAME}
1258 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1259 self.res_helper.user_lib_uri = 'libraries/{{}}/{}'.format(
1260 self.res_helper.test_session_uri)
1261 mock_session = mock.Mock(spec=requests.Session)
1262 self.res_helper.session = mock_session
1263 res = self.res_helper.configure_test_session(self.TEST_SESSION_NAME,
1265 self.assertIsNotNone(res)
1266 mock_session.post.assert_called_once()
1268 def test_delete_test_session(self, *args):
1269 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1270 self.res_helper.user_lib_uri = 'libraries/{{}}/{}'.format(
1271 self.res_helper.test_session_uri)
1272 mock_session = mock.Mock(spec=requests.Session)
1273 self.res_helper.session = mock_session
1274 res = self.res_helper.delete_test_session(self.TEST_SESSION_NAME)
1275 self.assertIsNotNone(res)
1276 mock_session.delete.assert_called_once()
1278 def test_create_running_tests(self, *args):
1279 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1280 test_session = {'id': self.SUCCESS_RECORD_ID}
1281 mock_session = mock.Mock(spec=requests.Session)
1282 post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
1283 'json.return_value': test_session}
1284 mock_session.post.return_value.configure_mock(**post_resp_data)
1285 self.res_helper.session = mock_session
1286 self.res_helper.create_running_tests(self.TEST_SESSION_NAME)
1287 self.assertEqual(self.SUCCESS_RECORD_ID, self.res_helper.run_id)
1289 def test_create_running_tests_error(self, *args):
1290 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1291 mock_session = mock.Mock(spec=requests.Session)
1292 post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
1293 mock_session.post.return_value.configure_mock(**post_resp_data)
1294 self.res_helper.session = mock_session
1295 with self.assertRaises(exceptions.RestApiError):
1296 self.res_helper.create_running_tests(self.TEST_SESSION_NAME)
1298 def test_get_running_tests(self, *args):
1299 mock_session = mock.Mock(spec=requests.Session)
1300 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1301 'json.return_value': self.RUNNING_TESTS_DATA}
1302 mock_session.get.return_value.configure_mock(**get_resp_data)
1303 self.res_helper.session = mock_session
1304 res = self.res_helper.get_running_tests()
1305 self.assertEqual(self.RUNNING_TESTS_DATA['runningTests'], res)
1307 def test_delete_running_tests(self, *args):
1308 mock_session = mock.Mock(spec=requests.Session)
1309 delete_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1310 'json.return_value': self.RUNNING_TESTS_DATA}
1311 mock_session.delete.return_value.configure_mock(**delete_resp_data)
1312 self.res_helper.session = mock_session
1313 self.assertIsNone(self.res_helper.delete_running_tests())
1315 def test__running_tests_action(self, *args):
1317 mock_session = mock.Mock(spec=requests.Session)
1318 self.res_helper.session = mock_session
1319 res = self.res_helper._running_tests_action(self.SUCCESS_RECORD_ID,
1321 self.assertIsNone(res)
1323 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1324 '_running_tests_action')
1325 def test_stop_running_tests(self, mock_tests_action, *args):
1326 res = self.res_helper.stop_running_tests(self.SUCCESS_RECORD_ID)
1327 self.assertIsNone(res)
1328 mock_tests_action.assert_called_once()
1330 def test_check_running_test_state(self, *args):
1331 mock_session = mock.Mock(spec=requests.Session)
1333 'status_code': self.SUCCESS_OK_CODE,
1334 'json.return_value': self.RUNNING_TESTS_DATA["runningTests"][0]}
1335 mock_session.get.return_value.configure_mock(**get_resp_data)
1336 self.res_helper.session = mock_session
1337 res = self.res_helper.check_running_test_state(self.SUCCESS_RECORD_ID)
1339 self.RUNNING_TESTS_DATA["runningTests"][0]['testStateOrStep'],
1342 def test_get_running_tests_results(self, *args):
1343 mock_session = mock.Mock(spec=requests.Session)
1344 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1345 'json.return_value': self.TEST_RESULTS_DATA}
1346 mock_session.get.return_value.configure_mock(**get_resp_data)
1347 self.res_helper.session = mock_session
1348 res = self.res_helper.get_running_tests_results(
1349 self.SUCCESS_RECORD_ID)
1350 self.assertEqual(self.TEST_RESULTS_DATA, res)
1352 def test__write_results(self, *args):
1353 res = self.res_helper._write_results(self.TEST_RESULTS_DATA)
1355 "Test Summary::Actual Dedicated Bearer Session Connects": 100.0,
1356 "Test Summary::Actual Dedicated Bearer Session Disconnects": 100.0,
1357 "Test Summary::Actual Disconnect Rate(Sessions / Second)(P - I)": 164.804,
1358 "Test Summary::Average Session Disconnect Time(P - I)": 5.024,
1359 "Test Summary::Total Data Sent + Received Packets / Sec(P - I)": 1452.294
1361 self.assertEqual(exp_res, res)
1363 def test__write_results_no_tabs(self, *args):
1364 _res_data = copy.deepcopy(self.TEST_RESULTS_DATA)
1365 del _res_data['tabs']
1366 # Return None if tabs not found in test results dict
1367 self.assertIsNone(self.res_helper._write_results(_res_data))
1369 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1370 'check_running_test_state')
1371 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1372 'get_running_tests_results')
1373 def test_collect_kpi_test_running(self, mock_tests_results,
1374 mock_tests_state, *args):
1375 self.res_helper.run_id = self.SUCCESS_RECORD_ID
1376 mock_tests_state.return_value = 'RUNNING'
1377 mock_tests_results.return_value = self.TEST_RESULTS_DATA
1378 res = self.res_helper.collect_kpi()
1379 self.assertNotIn('done', res)
1380 mock_tests_state.assert_called_once_with(self.res_helper.run_id)
1381 mock_tests_results.assert_called_once_with(self.res_helper.run_id)
1383 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1384 'check_running_test_state')
1385 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1386 'get_running_tests_results')
1387 def test_collect_kpi_test_completed(self, mock_tests_results,
1388 mock_tests_state, *args):
1389 self.res_helper.run_id = self.SUCCESS_RECORD_ID
1390 mock_tests_state.return_value = 'COMPLETE'
1391 res = self.res_helper.collect_kpi()
1392 self.assertIsNotNone(res)
1393 mock_tests_state.assert_called_once_with(self.res_helper.run_id)
1394 mock_tests_results.assert_not_called()
1395 self.assertDictContainsSubset({'done': True}, res)
1398 class TestLandslideTclClient(unittest.TestCase):
1400 self.mock_tcl_handler = mock.Mock(spec=tg_landslide.LsTclHandler)
1401 self.ls_res_helper = mock.Mock(
1402 spec=tg_landslide.LandslideResourceHelper)
1403 self.ls_tcl_client = tg_landslide.LandslideTclClient(
1404 self.mock_tcl_handler,
1407 def test___init__(self, *args):
1408 self.ls_tcl_client = tg_landslide.LandslideTclClient(
1409 self.mock_tcl_handler,
1411 self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
1412 self.assertIsNone(self.ls_tcl_client._user)
1413 self.assertIsNone(self.ls_tcl_client._library_id)
1414 self.assertIsNone(self.ls_tcl_client._basic_library_id)
1415 self.assertEqual(set(), self.ls_tcl_client.ts_ids)
1416 self.assertIsInstance(self.ls_tcl_client._tc_types, set)
1417 self.assertIsNotNone(self.ls_tcl_client._tc_types)
1419 def test_connect_login_success(self, *args):
1421 exec_responses = ['java0x2', lib_id, lib_id]
1422 auth = ('user', 'password')
1423 self.mock_tcl_handler.execute.side_effect = exec_responses
1424 self.ls_tcl_client.connect(TAS_INFO['ip'], *auth)
1425 self.assertEqual(lib_id, self.ls_tcl_client._library_id)
1426 self.assertEqual(lib_id, self.ls_tcl_client._basic_library_id)
1427 self.assertEqual(TAS_INFO['ip'], self.ls_tcl_client.tcl_server_ip)
1428 self.assertEqual(auth[0], self.ls_tcl_client._user)
1429 self.assertEqual(len(exec_responses),
1430 self.mock_tcl_handler.execute.call_count)
1431 self.mock_tcl_handler.execute.assert_has_calls([
1432 mock.call("ls::login 1.1.1.1 user password"),
1433 mock.call("ls::get [ls::query LibraryInfo -userLibraryName user] -Id"),
1436 def test_connect_login_failed(self, *args):
1437 exec_responses = ['Login failed']
1438 auth = ('user', 'password')
1439 self.mock_tcl_handler.execute.side_effect = exec_responses
1440 self.assertRaises(exceptions.LandslideTclException,
1441 self.ls_tcl_client.connect,
1444 self.assertIsNone(self.ls_tcl_client._library_id)
1445 self.assertIsNone(self.ls_tcl_client._basic_library_id)
1446 self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
1447 self.assertIsNone(self.ls_tcl_client._user)
1448 self.assertEqual(len(exec_responses),
1449 self.mock_tcl_handler.execute.call_count)
1450 self.mock_tcl_handler.execute.assert_called_with(
1451 "ls::login 1.1.1.1 user password")
1453 def test_disconnect(self, *args):
1454 self.ls_tcl_client.disconnect()
1455 self.mock_tcl_handler.execute.assert_called_once_with("ls::logout")
1456 self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
1457 self.assertIsNone(self.ls_tcl_client._user)
1458 self.assertIsNone(self.ls_tcl_client._library_id)
1459 self.assertIsNone(self.ls_tcl_client._basic_library_id)
1461 def test_create_test_server(self, *args):
1463 self.ls_tcl_client._ts_context.vnfd_helper = \
1464 VNFD['vnfd:vnfd-catalog']['vnfd'][0]
1465 self.ls_tcl_client._ts_context.license_data = {'lic_id': return_value}
1466 self.mock_tcl_handler.execute.return_value = return_value
1467 self.ls_tcl_client._set_thread_model = mock.Mock()
1468 res = self.ls_tcl_client.create_test_server(TEST_SERVERS[1])
1469 self.assertEqual(3, self.mock_tcl_handler.execute.call_count)
1470 self.mock_tcl_handler.execute.assert_has_calls([
1471 mock.call('ls::query TsId TestServer_2'),
1472 mock.call('set ts [ls::retrieve TsInfo -Name "TestServer_2"]'),
1473 mock.call('ls::get $ts -RequestedLicense'),
1475 self.ls_tcl_client._set_thread_model.assert_called_once_with(
1476 TEST_SERVERS[1]['name'],
1477 TEST_SERVERS[1]['thread_model'])
1478 self.assertEqual(int(return_value), res)
1480 def test_create_test_server_fail_limit_reach(self, *args):
1481 self.mock_tcl_handler.execute.side_effect = ['TS not found',
1483 self.assertRaises(RuntimeError,
1484 self.ls_tcl_client.create_test_server,
1486 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1487 self.mock_tcl_handler.execute.assert_has_calls([
1488 mock.call('ls::query TsId TestServer_1'),
1489 mock.call('ls::perform AddTs -Name "TestServer_1" '
1490 '-Ip "192.168.122.101"'),
1493 def test__add_test_server(self):
1495 self.mock_tcl_handler.execute.side_effect = ['TS not found', ts_id]
1496 self.assertEqual(ts_id,
1497 self.ls_tcl_client._add_test_server('name', 'ip'))
1498 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1499 self.mock_tcl_handler.execute.assert_has_calls([
1500 mock.call('ls::query TsId name'),
1501 mock.call('ls::perform AddTs -Name "name" -Ip "ip"'),
1504 def test__add_test_server_failed(self):
1505 self.mock_tcl_handler.execute.side_effect = ['TS not found',
1507 self.assertRaises(RuntimeError, self.ls_tcl_client._add_test_server,
1509 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1510 self.mock_tcl_handler.execute.assert_has_calls([
1511 mock.call('ls::query TsId name'),
1512 mock.call('ls::perform AddTs -Name "name" -Ip "ip"'),
1515 def test__update_license(self):
1518 exec_resp = ['java0x4',
1520 TCL_SUCCESS_RESPONSE,
1521 TCL_SUCCESS_RESPONSE]
1522 self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id}
1523 self.mock_tcl_handler.execute.side_effect = exec_resp
1524 self.ls_tcl_client._update_license('name')
1525 self.assertEqual(len(exec_resp),
1526 self.mock_tcl_handler.execute.call_count)
1528 self.mock_tcl_handler.execute.assert_has_calls([
1529 mock.call('set ts [ls::retrieve TsInfo -Name "name"]'),
1530 mock.call('ls::get $ts -RequestedLicense'),
1531 mock.call('ls::config $ts -RequestedLicense 222'),
1532 mock.call('ls::perform ModifyTs $ts'),
1535 def test__update_license_same_as_current(self):
1538 exec_resp = ['java0x4', curr_lic_id]
1539 self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id}
1540 self.mock_tcl_handler.execute.side_effect = exec_resp
1541 self.ls_tcl_client._update_license('name')
1542 self.assertEqual(len(exec_resp),
1543 self.mock_tcl_handler.execute.call_count)
1544 self.mock_tcl_handler.execute.assert_has_calls([
1545 mock.call('set ts [ls::retrieve TsInfo -Name "name"]'),
1546 mock.call('ls::get $ts -RequestedLicense'),
1549 def test__set_thread_model_update_needed(self):
1550 self.ls_tcl_client._ts_context.vnfd_helper = {
1552 'cfguser_password': 'cfguser_password'
1555 exec_resp = ['java0x4', 'V0', '', '']
1556 self.mock_tcl_handler.execute.side_effect = exec_resp
1557 self.ls_tcl_client._set_thread_model('name', 'Fireball')
1558 self.assertEqual(len(exec_resp),
1559 self.mock_tcl_handler.execute.call_count)
1560 self.mock_tcl_handler.execute.assert_has_calls([
1561 mock.call('set tsc [ls::perform RetrieveTsConfiguration '
1562 '-name "name" cfguser_password]'),
1563 mock.call('ls::get $tsc -ThreadModel'),
1564 mock.call('ls::config $tsc -ThreadModel "V1_FB3"'),
1565 mock.call('ls::perform ApplyTsConfiguration $tsc cfguser_password'),
1568 def test__set_thread_model_no_update_needed(self):
1569 self.ls_tcl_client._ts_context.vnfd_helper = {
1571 'cfguser_password': 'cfguser_password'
1574 exec_resp = ['java0x4', 'V0']
1575 self.mock_tcl_handler.execute.side_effect = exec_resp
1576 self.ls_tcl_client._set_thread_model('name', 'Legacy')
1577 self.assertEqual(len(exec_resp),
1578 self.mock_tcl_handler.execute.call_count)
1579 self.mock_tcl_handler.execute.assert_has_calls([
1580 mock.call('set tsc [ls::perform RetrieveTsConfiguration '
1581 '-name "name" cfguser_password]'),
1582 mock.call('ls::get $tsc -ThreadModel'),
1585 @mock.patch.object(tg_landslide.LandslideTclClient,
1586 'resolve_test_server_name', side_effect=['4', '2'])
1587 def test_create_test_session(self, *args):
1588 _session_profile = copy.deepcopy(SESSION_PROFILE)
1589 _session_profile['reservations'] = RESERVATIONS
1590 self.ls_tcl_client._save_test_session = mock.Mock()
1591 self.ls_tcl_client._configure_ts_group = mock.Mock()
1592 self.ls_tcl_client._library_id = 42
1593 self.ls_tcl_client.create_test_session(_session_profile)
1594 self.assertEqual(17, self.mock_tcl_handler.execute.call_count)
1595 self.mock_tcl_handler.execute.assert_has_calls([
1596 mock.call('set test_ [ls::create TestSession]'),
1597 mock.call('ls::config $test_ -Library 42 '
1598 '-Name "default_bearer_capacity"'),
1599 mock.call('ls::config $test_ -Description ' \
1600 '"UE default bearer creation test case"'),
1601 mock.call('ls::config $test_ -Keywords ""'),
1602 mock.call('ls::config $test_ -Duration "60"'),
1603 mock.call('ls::config $test_ -Iterations "1"'),
1604 # _configure_reservation
1605 mock.call('set reservation_ [ls::create Reservation -under $test_]'),
1606 mock.call('ls::config $reservation_ -TsIndex 0 '
1607 '-TsId 4 -TsName "TestServer_1"'),
1608 mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
1609 mock.call('ls::config $physubnet_ -Name "eth1" -Base "10.42.32.100" '
1610 '-Mask "/24" -NumIps 20'),
1611 # _configure_reservation
1612 mock.call('set reservation_ [ls::create Reservation -under $test_]'),
1613 mock.call('ls::config $reservation_ -TsIndex 1 '
1614 '-TsId 2 -TsName "TestServer_2"'),
1615 mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
1616 mock.call('ls::config $physubnet_ -Name "eth1" -Base "10.42.32.1" '
1617 '-Mask "/24" -NumIps 100'),
1618 mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
1619 mock.call('ls::config $physubnet_ -Name "eth2" -Base "10.42.33.1" '
1620 '-Mask "/24" -NumIps 100'),
1621 # _configure_report_options
1622 mock.call('ls::config $test_.ReportOptions -Format 1 -Ts -3 -Tc -3'),
1625 def test_create_dmf(self):
1626 self.mock_tcl_handler.execute.return_value = '2'
1627 self.ls_tcl_client._save_dmf = mock.Mock()
1628 self.ls_tcl_client.create_dmf(copy.deepcopy(DMF_CFG))
1629 self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
1630 # This is needed because the dictionary is unordered and the arguments
1631 # can come in either order
1633 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"')
1635 'ls::config $dmf_ -isClientPortRange "false" -clientPort 2002')
1637 call1 in self.mock_tcl_handler.execute.mock_calls or
1638 call2 in self.mock_tcl_handler.execute.mock_calls)
1640 self.mock_tcl_handler.execute.assert_has_calls([
1641 mock.call('set dmf_ [ls::create Dmf]'),
1643 'ls::get [ls::query LibraryInfo -systemLibraryName user] -Id'),
1644 mock.call('ls::config $dmf_ -Library 2 -Name "Basic UDP"'),
1645 mock.call('ls::config $dmf_ -dataProtocol "udp"'),
1647 # 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"'),
1648 mock.call('ls::config $dmf_ -serverPort 2003'),
1651 def test_configure_dmf(self):
1652 self.mock_tcl_handler.execute.return_value = '2'
1653 self.ls_tcl_client._save_dmf = mock.Mock()
1654 self.ls_tcl_client.configure_dmf(DMF_CFG)
1655 self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
1656 # This is need because the dictionary is unordered and the arguments
1657 # can come in either order
1659 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"')
1661 'ls::config $dmf_ -isClientPortRange "false" -clientPort 2002')
1663 call1 in self.mock_tcl_handler.execute.mock_calls or
1664 call2 in self.mock_tcl_handler.execute.mock_calls)
1666 self.mock_tcl_handler.execute.assert_has_calls([
1667 mock.call('set dmf_ [ls::create Dmf]'),
1669 'ls::get [ls::query LibraryInfo -systemLibraryName user] -Id'),
1670 mock.call('ls::config $dmf_ -Library 2 -Name "Basic UDP"'),
1671 mock.call('ls::config $dmf_ -dataProtocol "udp"'),
1673 # 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"'),
1674 mock.call('ls::config $dmf_ -serverPort 2003'),
1677 def test_delete_dmf(self):
1678 self.assertRaises(NotImplementedError,
1679 self.ls_tcl_client.delete_dmf,
1682 def test__save_dmf_valid(self):
1683 exec_resp = [TCL_SUCCESS_RESPONSE, TCL_SUCCESS_RESPONSE]
1684 self.mock_tcl_handler.execute.side_effect = exec_resp
1685 self.ls_tcl_client._save_dmf()
1686 self.assertEqual(len(exec_resp),
1687 self.mock_tcl_handler.execute.call_count)
1688 self.mock_tcl_handler.execute.assert_has_calls([
1689 mock.call('ls::perform Validate -Dmf $dmf_'),
1690 mock.call('ls::save $dmf_ -overwrite'),
1693 def test__save_dmf_invalid(self):
1694 exec_resp = ['Invalid', 'List of errors and warnings']
1695 self.mock_tcl_handler.execute.side_effect = exec_resp
1696 self.assertRaises(exceptions.LandslideTclException,
1697 self.ls_tcl_client._save_dmf)
1698 self.assertEqual(len(exec_resp),
1699 self.mock_tcl_handler.execute.call_count)
1700 self.mock_tcl_handler.execute.assert_has_calls([
1701 mock.call('ls::perform Validate -Dmf $dmf_'),
1702 mock.call('ls::get $dmf_ -ErrorsAndWarnings'),
1705 def test__configure_report_options(self):
1706 _options = {'format': 'CSV', 'PerInterval': 'false'}
1707 self.ls_tcl_client._configure_report_options(_options)
1708 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1709 self.mock_tcl_handler.execute.assert_has_calls([
1710 mock.call('ls::config $test_.ReportOptions -Format 1 -Ts -3 -Tc -3'),
1711 mock.call('ls::config $test_.ReportOptions -PerInterval false'),
1715 def test___configure_ts_group(self, *args):
1716 _ts_group = copy.deepcopy(SESSION_PROFILE['tsGroups'][0])
1717 self.ls_tcl_client._configure_tc_type = mock.Mock()
1718 self.ls_tcl_client._configure_preresolved_arp = mock.Mock()
1719 self.ls_tcl_client.resolve_test_server_name = mock.Mock(
1721 self.ls_tcl_client._configure_ts_group(_ts_group, 0)
1722 self.mock_tcl_handler.execute.assert_called_once_with(
1723 'set tss_ [ls::create TsGroup -under $test_ -tsId 2 ]')
1725 def test___configure_ts_group_resolve_ts_fail(self, *args):
1726 _ts_group = copy.deepcopy(SESSION_PROFILE['tsGroups'][0])
1727 self.ls_tcl_client._configure_tc_type = mock.Mock()
1728 self.ls_tcl_client._configure_preresolved_arp = mock.Mock()
1729 self.ls_tcl_client.resolve_test_server_name = mock.Mock(
1730 return_value='TS Not Found')
1731 self.assertRaises(RuntimeError, self.ls_tcl_client._configure_ts_group,
1733 self.mock_tcl_handler.execute.assert_not_called()
1735 def test__configure_tc_type(self):
1736 _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1737 self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE
1738 self.ls_tcl_client._configure_parameters = mock.Mock()
1739 self.ls_tcl_client._configure_tc_type(_tc, 0)
1740 self.assertEqual(7, self.mock_tcl_handler.execute.call_count)
1742 def test__configure_tc_type_optional_param_omitted(self):
1743 _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1745 self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE
1746 self.ls_tcl_client._configure_parameters = mock.Mock()
1747 self.ls_tcl_client._configure_tc_type(_tc, 0)
1748 self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
1750 def test__configure_tc_type_wrong_type(self):
1751 _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1752 _tc['type'] = 'not_supported'
1753 self.ls_tcl_client._configure_parameters = mock.Mock()
1754 self.assertRaises(RuntimeError,
1755 self.ls_tcl_client._configure_tc_type,
1757 self.mock_tcl_handler.assert_not_called()
1759 def test__configure_tc_type_not_found_basic_lib(self):
1760 _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1761 self.ls_tcl_client._configure_parameters = mock.Mock()
1762 self.mock_tcl_handler.execute.return_value = 'Invalid'
1763 self.assertRaises(RuntimeError,
1764 self.ls_tcl_client._configure_tc_type,
1767 def test__configure_parameters(self):
1768 _params = copy.deepcopy(
1769 SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters'])
1770 self.ls_tcl_client._configure_parameters(_params)
1771 self.assertEqual(16, self.mock_tcl_handler.execute.call_count)
1773 def test__configure_array_param(self):
1774 _array = {"class": "Array",
1776 self.ls_tcl_client._configure_array_param('name', _array)
1777 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1778 self.mock_tcl_handler.execute.assert_has_calls([
1779 mock.call('ls::create -Array-name -under $p_ ;'),
1780 mock.call('ls::create ArrayItem -under $p_.name -Value "0"'),
1783 def test__configure_test_node_param(self):
1784 _params = copy.deepcopy(
1785 SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters'])
1786 self.ls_tcl_client._configure_test_node_param('SgwUserAddr',
1787 _params['SgwUserAddr'])
1788 cmd = ('ls::create -TestNode-SgwUserAddr -under $p_ -Type "eth" '
1789 '-Phy "eth1" -Ip "SGW_USER_IP" -NumLinksOrNodes 1 '
1790 '-NextHop "SGW_CONTROL_NEXT_HOP" -Mac "" -MTU 1500 '
1791 '-ForcedEthInterface "" -EthStatsEnabled false -VlanId 0 '
1792 '-VlanUserPriority 0 -NumVlan 1 -UniqueVlanAddr false;')
1793 self.mock_tcl_handler.execute.assert_called_once_with(cmd)
1795 def test__configure_sut_param(self):
1796 _params = {'name': 'name'}
1797 self.ls_tcl_client._configure_sut_param('name', _params)
1798 self.mock_tcl_handler.execute.assert_called_once_with(
1799 'ls::create -Sut-name -under $p_ -Name "name";')
1801 def test__configure_dmf_param(self):
1802 _params = {"mainflows": [{"library": '111',
1803 "name": "Basic UDP"}],
1804 "instanceGroups": [{
1812 "overridePort": "false",
1816 "transport": "Any"}]
1818 self.ls_tcl_client._get_library_id = mock.Mock(return_value='111')
1819 res = self.ls_tcl_client._configure_dmf_param('name', _params)
1820 self.assertEqual(5, self.mock_tcl_handler.execute.call_count)
1821 self.assertIsNone(res)
1822 self.mock_tcl_handler.execute.assert_has_calls([
1823 mock.call('ls::create -Dmf-name -under $p_ ;'),
1824 mock.call('ls::perform AddDmfMainflow $p_.Dmf 111 "Basic UDP"'),
1825 mock.call('ls::config $p_.Dmf.InstanceGroup(0) -mixType '),
1826 mock.call('ls::config $p_.Dmf.InstanceGroup(0) -rate 0.0'),
1827 mock.call('ls::config $p_.Dmf.InstanceGroup(0).Row(0) -Node 0 '
1828 '-OverridePort false -ClientPort 0 -Context 0 -Role 0 '
1829 '-PreferredTransport Any -RatingGroup 0 '
1833 def test__configure_dmf_param_no_instance_groups(self):
1834 _params = {"mainflows": [{"library": '111',
1835 "name": "Basic UDP"}]}
1836 self.ls_tcl_client._get_library_id = mock.Mock(return_value='111')
1837 res = self.ls_tcl_client._configure_dmf_param('name', _params)
1838 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1839 self.assertIsNone(res)
1840 self.mock_tcl_handler.execute.assert_has_calls([
1841 mock.call('ls::create -Dmf-name -under $p_ ;'),
1842 mock.call('ls::perform AddDmfMainflow $p_.Dmf 111 "Basic UDP"'),
1845 def test__configure_reservation(self):
1846 _reservation = copy.deepcopy(RESERVATIONS[0])
1847 self.ls_tcl_client.resolve_test_server_name = mock.Mock(
1849 res = self.ls_tcl_client._configure_reservation(_reservation)
1850 self.assertIsNone(res)
1851 self.assertEqual(4, self.mock_tcl_handler.execute.call_count)
1852 self.mock_tcl_handler.execute.assert_has_calls([
1853 mock.call('set reservation_ [ls::create Reservation -under $test_]'),
1854 mock.call('ls::config $reservation_ -TsIndex 0 -TsId 4 ' + \
1855 '-TsName "TestServer_1"'),
1856 mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
1857 mock.call('ls::config $physubnet_ -Name "eth1" ' + \
1858 '-Base "10.42.32.100" -Mask "/24" -NumIps 20'),
1861 def test__configure_preresolved_arp(self):
1862 _arp = [{'StartingAddress': '10.81.1.10',
1864 res = self.ls_tcl_client._configure_preresolved_arp(_arp)
1865 self.mock_tcl_handler.execute.assert_called_once()
1866 self.assertIsNone(res)
1867 self.mock_tcl_handler.execute.assert_called_once_with(
1868 'ls::create PreResolvedArpAddress -under $tss_ ' + \
1869 '-StartingAddress "10.81.1.10" -NumNodes 1')
1871 def test__configure_preresolved_arp_none(self):
1872 res = self.ls_tcl_client._configure_preresolved_arp(None)
1873 self.assertIsNone(res)
1874 self.mock_tcl_handler.execute.assert_not_called()
1876 def test_delete_test_session(self):
1877 self.assertRaises(NotImplementedError,
1878 self.ls_tcl_client.delete_test_session, {})
1880 def test__save_test_session(self):
1881 self.mock_tcl_handler.execute.side_effect = [TCL_SUCCESS_RESPONSE,
1882 TCL_SUCCESS_RESPONSE]
1883 res = self.ls_tcl_client._save_test_session()
1884 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1885 self.assertIsNone(res)
1886 self.mock_tcl_handler.execute.assert_has_calls([
1887 mock.call('ls::perform Validate -TestSession $test_'),
1888 mock.call('ls::save $test_ -overwrite'),
1891 def test__save_test_session_invalid(self):
1892 self.mock_tcl_handler.execute.side_effect = ['Invalid', 'Errors']
1893 self.assertRaises(exceptions.LandslideTclException,
1894 self.ls_tcl_client._save_test_session)
1895 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1896 self.mock_tcl_handler.execute.assert_has_calls([
1897 mock.call('ls::perform Validate -TestSession $test_'),
1898 mock.call('ls::get $test_ -ErrorsAndWarnings'),
1901 def test__get_library_id_system_lib(self):
1902 self.mock_tcl_handler.execute.return_value = '111'
1903 res = self.ls_tcl_client._get_library_id('name')
1904 self.mock_tcl_handler.execute.assert_called_once()
1905 self.assertEqual('111', res)
1906 self.mock_tcl_handler.execute.assert_called_with(
1907 'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id')
1909 def test__get_library_id_user_lib(self):
1910 self.mock_tcl_handler.execute.side_effect = ['Not found', '222']
1911 res = self.ls_tcl_client._get_library_id('name')
1912 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1913 self.assertEqual('222', res)
1914 self.mock_tcl_handler.execute.assert_has_calls([
1916 'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id'),
1918 'ls::get [ls::query LibraryInfo -userLibraryName name] -Id'),
1921 def test__get_library_id_exception(self):
1922 self.mock_tcl_handler.execute.side_effect = ['Not found', 'Not found']
1923 self.assertRaises(exceptions.LandslideTclException,
1924 self.ls_tcl_client._get_library_id,
1926 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1927 self.mock_tcl_handler.execute.assert_has_calls([
1929 'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id'),
1931 'ls::get [ls::query LibraryInfo -userLibraryName name] -Id'),
1935 class TestLsTclHandler(unittest.TestCase):
1938 self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
1939 self.mock_lsapi.start()
1941 self.addCleanup(self._cleanup)
1944 self.mock_lsapi.stop()
1946 def test___init__(self, *args):
1947 self.ls_tcl_handler = tg_landslide.LsTclHandler()
1948 self.assertEqual({}, self.ls_tcl_handler.tcl_cmds)
1949 self.ls_tcl_handler._ls.tcl.assert_called_once()
1951 def test_execute(self, *args):
1952 self.ls_tcl_handler = tg_landslide.LsTclHandler()
1953 self.ls_tcl_handler.execute('command')
1954 self.assertIn('command', self.ls_tcl_handler.tcl_cmds)