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
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',
342 'BearerAddrPool': '2002::2',
344 'BearerV4AddrPool': '2.0.0.2',
348 'StartRate': '900.0',
350 'DisconnectRate': '900.0',
356 'transactionRate': 1000,
366 'name': 'traffic_gen',
367 'file': '/etc/yardstick/nodes/pod_landslide.yaml'
372 'file': '/etc/yardstick/nodes/pod_vepc_sut.yaml'
378 "schema": "nsb:traffic_profile:0.1",
379 "name": "LandslideProfile",
380 "description": "Spirent Landslide traffic profile",
382 "traffic_type": "LandslideProfile"
389 "description": "Basic data flow using UDP/IP",
391 "dataProtocol": "udp"
395 SUCCESS_CREATED_CODE = 201
396 SUCCESS_OK_CODE = 200
397 SUCCESS_RECORD_ID = 5
401 self._id = uuid.uuid1().int
403 self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
404 self.mock_lsapi.start()
406 self.mock_ssh_helper = mock.patch.object(sample_vnf, 'VnfSshHelper')
407 self.mock_ssh_helper.start()
408 self.vnfd = VNFD['vnfd:vnfd-catalog']['vnfd'][0]
409 self.ls_tg = tg_landslide.LandslideTrafficGen(
410 NAME, self.vnfd, self._id)
411 self.session_profile = copy.deepcopy(SESSION_PROFILE)
412 self.ls_tg.session_profile = self.session_profile
414 self.addCleanup(self._cleanup)
417 self.mock_lsapi.stop()
418 self.mock_ssh_helper.stop()
420 @mock.patch.object(net_serv_utils, 'get_nsb_option')
421 def test___init__(self, mock_get_nsb_option, *args):
422 _path_to_nsb = 'path/to/nsb'
423 mock_get_nsb_option.return_value = _path_to_nsb
424 ls_tg = tg_landslide.LandslideTrafficGen(NAME, self.vnfd, self._id)
425 self.assertIsInstance(ls_tg.resource_helper,
426 tg_landslide.LandslideResourceHelper)
427 mock_get_nsb_option.assert_called_once_with('bin_path')
428 self.assertEqual(_path_to_nsb, ls_tg.bin_path)
429 self.assertEqual(NAME, ls_tg.name)
430 self.assertTrue(ls_tg.runs_traffic)
431 self.assertFalse(ls_tg.traffic_finished)
432 self.assertIsNone(ls_tg.session_profile)
434 def test_listen_traffic(self):
435 _traffic_profile = {}
436 self.assertIsNone(self.ls_tg.listen_traffic(_traffic_profile))
438 def test_terminate(self, *args):
439 self.ls_tg.resource_helper._tcl = mock.Mock()
440 self.assertIsNone(self.ls_tg.terminate())
441 self.ls_tg.resource_helper._tcl.disconnect.assert_called_once()
443 @mock.patch.object(ctx_base.Context, 'get_context_from_server',
444 return_value='fake_context')
445 def test_instantiate(self, *args):
446 self.ls_tg._tg_process = mock.Mock()
447 self.ls_tg._tg_process.start = mock.Mock()
448 self.ls_tg.resource_helper.connect = mock.Mock()
449 self.ls_tg.resource_helper.create_test_servers = mock.Mock()
450 self.ls_tg.resource_helper.create_suts = mock.Mock()
451 self.ls_tg._load_session_profile = mock.Mock()
452 self.assertIsNone(self.ls_tg.instantiate(self.SCENARIO_CFG,
454 self.ls_tg.resource_helper.connect.assert_called_once()
455 self.ls_tg.resource_helper.create_test_servers.assert_called_once()
456 _suts_blocks_num = len([item['suts'] for item in self.vnfd['config']])
457 self.assertEqual(_suts_blocks_num,
458 self.ls_tg.resource_helper.create_suts.call_count)
459 self.ls_tg._load_session_profile.assert_called_once()
461 @mock.patch.object(tg_landslide.LandslideResourceHelper,
463 def test_run_traffic(self, mock_get_tests, *args):
464 self.ls_tg.resource_helper._url = EXAMPLE_URL
465 self.ls_tg.scenario_helper.scenario_cfg = self.SCENARIO_CFG
466 mock_traffic_profile = mock.Mock(
467 spec=landslide_profile.LandslideProfile)
468 mock_traffic_profile.dmf_config = {'keywords': 'UDP',
469 'dataProtocol': 'udp'}
470 mock_traffic_profile.params = self.TRAFFIC_PROFILE
471 self.ls_tg.resource_helper._user_id = self.TEST_USER_ID
472 mock_get_tests.return_value = [{'id': self.SUCCESS_RECORD_ID,
473 'testStateOrStep': 'COMPLETE'}]
474 mock_post = mock.Mock()
475 mock_post.status_code = self.SUCCESS_CREATED_CODE
476 mock_post.json.return_value = {'id': self.SUCCESS_RECORD_ID}
477 mock_session = mock.Mock(spec=requests.Session)
478 mock_session.post.return_value = mock_post
479 self.ls_tg.resource_helper.session = mock_session
480 self.ls_tg.resource_helper._tcl = mock.Mock()
481 _tcl = self.ls_tg.resource_helper._tcl
482 self.assertIsNone(self.ls_tg.run_traffic(mock_traffic_profile))
483 self.assertEqual(self.SUCCESS_RECORD_ID,
484 self.ls_tg.resource_helper.run_id)
485 mock_traffic_profile.update_dmf.assert_called_with(
486 self.ls_tg.scenario_helper.all_options)
487 _tcl.create_dmf.assert_called_with(mock_traffic_profile.dmf_config)
488 _tcl.create_test_session.assert_called_with(self.session_profile)
490 @mock.patch.object(tg_landslide.LandslideResourceHelper,
491 'check_running_test_state')
492 def test_collect_kpi(self, mock_check_running_test_state, *args):
493 self.ls_tg.resource_helper.run_id = self.SUCCESS_RECORD_ID
494 mock_check_running_test_state.return_value = 'COMPLETE'
495 self.assertEqual({'done': True}, self.ls_tg.collect_kpi())
496 mock_check_running_test_state.assert_called_once()
498 def test_wait_for_instantiate(self):
499 self.assertIsNone(self.ls_tg.wait_for_instantiate())
500 self.ls_tg.wait_for_instantiate()
502 def test__update_session_suts_no_tc_role(self, *args):
503 _suts = [{'role': 'epc_role'}]
504 _testcase = {'parameters': {'diff_epc_role': {'class': 'Sut'}}}
505 res = self.ls_tg._update_session_suts(_suts, _testcase)
506 self.assertEqual(_testcase, res)
508 def test__update_session_suts(self, *args):
510 def get_testnode_param(role, key, session_prof):
511 """ Get value by key from the deep nested dict to avoid calls like:
512 e.g. session_prof['tsGroups'][0]['testCases'][1]['parameters'][key]
514 for group in session_prof['tsGroups']:
515 for tc in group['testCases']:
516 tc_params = tc['parameters']
517 if tc_params.get(role):
518 return tc_params[role][key]
520 def get_sut_param(role, key, suts):
521 """ Search list of dicts for one with specific role.
522 Return the value of related dict by key. Expect key presence.
525 if sut.get('role') == role:
529 testnode_role = 'SgwControlAddr'
531 sut_role = 'SgwUserSut'
533 config_suts = [config['suts'] for config in self.vnfd['config']]
534 session_tcs = [_tc for _ts_group in self.ls_tg.session_profile['tsGroups']
535 for _tc in _ts_group['testCases']]
536 for suts, tc in zip(config_suts, session_tcs):
537 self.assertEqual(tc, self.ls_tg._update_session_suts(suts, tc))
539 # Verify TestNode class objects keys were updated
540 for _key in {'ip', 'phy', 'nextHop'}:
542 get_testnode_param(testnode_role, _key, self.ls_tg.session_profile),
543 get_sut_param(testnode_role, _key, TS1_SUTS))
544 # Verify Sut class objects name was updated
546 get_testnode_param(sut_role, 'name', self.ls_tg.session_profile),
547 get_sut_param(sut_role, 'name', TS2_SUTS))
549 def test__update_session_test_servers(self, *args):
550 for ts_index, ts in enumerate(TEST_SERVERS):
552 self.ls_tg._update_session_test_servers(ts, ts_index))
553 # Verify preResolvedArpAddress key was added
555 _item.get('preResolvedArpAddress')
556 for _item in self.ls_tg.session_profile['tsGroups']))
557 # Verify reservations key was added to session profile
558 self.assertEqual(RESERVATIONS,
559 self.ls_tg.session_profile.get('reservations'))
560 self.assertEqual('true',
561 self.ls_tg.session_profile.get('reservePorts'))
563 def test__update_session_tc_params_assoc_phys(self):
564 _tc_options = {'AssociatedPhys': 'eth1'}
566 _testcase_orig = copy.deepcopy(_testcase)
567 res = self.ls_tg._update_session_tc_params(_tc_options, _testcase)
568 self.assertNotEqual(_testcase_orig, res)
569 self.assertEqual(_tc_options, _testcase)
571 def test__update_session_tc_params(self, *args):
573 def get_session_tc_param_value(param, tc_type, session_prof):
574 """ Get param value from the deep nested dict to avoid calls like:
575 session_prof['tsGroups'][0]['testCases'][0]['parameters'][key]
577 for test_group in session_prof['tsGroups']:
578 session_tc = test_group['testCases'][0]
579 if session_tc['type'] == tc_type:
580 return session_tc['parameters'].get(param)
582 session_tcs = [_tc for _ts_group in self.ls_tg.session_profile['tsGroups']
583 for _tc in _ts_group['testCases']]
584 scenario_tcs = [_tc for _tc in
585 self.SCENARIO_CFG['options']['test_cases']]
586 for tc_options, tc in zip(scenario_tcs, session_tcs):
589 self.ls_tg._update_session_tc_params(tc_options, tc))
591 # Verify that each test case parameter was updated
592 # Params been compared are deeply nested. Using loops to ease access.
593 for _tc in self.SCENARIO_CFG['options']['test_cases']:
594 for _key, _val in _tc.items():
598 get_session_tc_param_value(_key, _tc.get('type'),
599 self.ls_tg.session_profile))
601 @mock.patch.object(common_utils, 'open_relative_file')
602 @mock.patch.object(yaml_loader, 'yaml_load')
603 @mock.patch.object(tg_landslide.LandslideTrafficGen,
604 '_update_session_test_servers')
605 @mock.patch.object(tg_landslide.LandslideTrafficGen,
606 '_update_session_suts')
607 @mock.patch.object(tg_landslide.LandslideTrafficGen,
608 '_update_session_tc_params')
609 def test__load_session_profile(self, mock_upd_ses_tc_params,
610 mock_upd_ses_suts, mock_upd_ses_ts,
611 mock_yaml_load, *args):
612 self.ls_tg.scenario_helper.scenario_cfg = \
613 copy.deepcopy(self.SCENARIO_CFG)
614 mock_yaml_load.return_value = copy.deepcopy(SESSION_PROFILE)
615 self.assertIsNone(self.ls_tg._load_session_profile())
616 self.assertIsNotNone(self.ls_tg.session_profile)
617 # Number of blocks in configuration files
618 # Number of test servers, suts and tc params blocks should be equal
619 _config_files_blocks_num = len([item['test_server']
620 for item in self.vnfd['config']])
621 self.assertEqual(_config_files_blocks_num,
622 mock_upd_ses_ts.call_count)
623 self.assertEqual(_config_files_blocks_num,
624 mock_upd_ses_suts.call_count)
625 self.assertEqual(_config_files_blocks_num,
626 mock_upd_ses_tc_params.call_count)
628 @mock.patch.object(common_utils, 'open_relative_file')
629 @mock.patch.object(yaml_loader, 'yaml_load')
630 def test__load_session_profile_unequal_num_of_cfg_blocks(
631 self, mock_yaml_load, *args):
632 vnfd = copy.deepcopy(VNFD['vnfd:vnfd-catalog']['vnfd'][0])
633 ls_traffic_gen = tg_landslide.LandslideTrafficGen(NAME, vnfd, self._id)
634 ls_traffic_gen.scenario_helper.scenario_cfg = self.SCENARIO_CFG
635 mock_yaml_load.return_value = copy.deepcopy(SESSION_PROFILE)
636 # Delete test_servers item from pod file to make it not valid
637 ls_traffic_gen.vnfd_helper['config'].pop()
638 with self.assertRaises(RuntimeError):
639 ls_traffic_gen._load_session_profile()
641 @mock.patch.object(common_utils, 'open_relative_file')
642 @mock.patch.object(yaml_loader, 'yaml_load')
643 def test__load_session_profile_test_type_mismatch(self, mock_yaml_load,
645 vnfd = copy.deepcopy(VNFD['vnfd:vnfd-catalog']['vnfd'][0])
646 # Swap test servers data in pod file
647 vnfd['config'] = list(reversed(vnfd['config']))
648 ls_tg = tg_landslide.LandslideTrafficGen(NAME, vnfd, self._id)
649 ls_tg.scenario_helper.scenario_cfg = self.SCENARIO_CFG
650 mock_yaml_load.return_value = SESSION_PROFILE
651 with self.assertRaises(RuntimeError):
652 ls_tg._load_session_profile()
655 class TestLandslideResourceHelper(unittest.TestCase):
658 EXAMPLE_URL = ''.join([TAS_INFO['proto'], '://', TAS_INFO['ip'], ':',
659 str(PROTO_PORT), '/api/'])
660 SUCCESS_CREATED_CODE = 201
661 SUCCESS_OK_CODE = 200
662 INVALID_REST_CODE = '400'
663 NOT_MODIFIED_CODE = 500810
665 SUCCESS_RECORD_ID = 11
666 EXPIRE_DATE = '2020/01/01 12:00 FLE Standard Time'
669 AUTH_DATA = {'user': TAS_INFO['user'], 'password': TAS_INFO['password']}
670 TEST_SESSION_NAME = 'default_bearer_capacity'
674 "url": ''.join([EXAMPLE_URL, 'users/', str(SUCCESS_RECORD_ID)]),
675 "id": SUCCESS_RECORD_ID,
677 "username": TEST_USER
681 CREATE_USER_DATA = {'username': TAS_INFO['user'],
682 'expiresOn': EXPIRE_DATE,
684 'contactInformation': '',
685 'fullName': 'Test User',
686 'password': TAS_INFO['password'],
692 "url": ''.join([EXAMPLE_URL, 'suts/', str(SUCCESS_RECORD_ID)]),
693 "id": SUCCESS_RECORD_ID,
697 TEST_SERVERS_DATA = {
700 "url": ''.join([EXAMPLE_URL, "testServers/1"]),
702 "name": TEST_SERVERS[0]['name'],
704 "version": "16.4.0.10"
707 "url": ''.join([EXAMPLE_URL, "testServers/2"]),
709 "name": TEST_SERVERS[1]['name'],
711 "version": "16.4.0.10"
719 RUNNING_TESTS_DATA = {
721 "url": ''.join([EXAMPLE_URL, "runningTests/{}".format(RUN_ID)]),
722 "measurementsUrl": ''.join(
724 "runningTests/{}/measurements".format(RUN_ID)]),
725 "criteriaUrl": ''.join(
727 "runningTests/{}/criteria".format(RUN_ID)]),
730 "library": SUCCESS_RECORD_ID,
731 "name": "default_bearer_capacity",
733 "criteriaStatus": "NA",
734 "testStateOrStep": "COMPLETE"
737 TEST_RESULTS_DATA = {
740 "actualTime": 1521548057296,
744 "Start Time": "Tue Mar 20 07:11:55 CDT 2018",
745 "Actual Dedicated Bearer Session Connects": "100",
746 "Actual Dedicated Bearer Session Disconnects": "100",
747 "Actual Disconnect Rate(Sessions / Second)(P - I)": "164.804",
748 "Average Session Disconnect Time(P - I)": "5.024 s",
749 "Total Data Sent + Received Packets / Sec(P - I)": "1,452.294"
753 self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
754 self.mock_lsapi.start()
756 mock_env_helper = mock.Mock()
757 self.res_helper = tg_landslide.LandslideResourceHelper(mock_env_helper)
758 self.res_helper._url = EXAMPLE_URL
760 self.addCleanup(self._cleanup)
763 self.mock_lsapi.stop()
764 self.res_helper._url = None
766 def test___init__(self, *args):
767 self.assertIsInstance(self.res_helper,
768 tg_landslide.LandslideResourceHelper)
769 self.assertEqual({}, self.res_helper._result)
770 self.assertIsNone(self.res_helper.run_id)
772 @mock.patch.object(tg_landslide.LandslideResourceHelper,
773 'stop_running_tests')
774 @mock.patch.object(tg_landslide.LandslideResourceHelper,
776 def test_abort_running_tests_no_running_tests(self, mock_get_tests,
777 mock_stop_tests, *args):
778 tests_data = [{'id': self.SUCCESS_RECORD_ID,
779 'testStateOrStep': 'COMPLETE'}]
780 mock_get_tests.return_value = tests_data
781 self.assertIsNone(self.res_helper.abort_running_tests())
782 mock_stop_tests.assert_not_called()
784 @mock.patch.object(time, 'sleep')
785 @mock.patch.object(tg_landslide.LandslideResourceHelper,
786 'stop_running_tests')
787 @mock.patch.object(tg_landslide.LandslideResourceHelper,
789 def test_abort_running_tests(self, mock_get_tests, mock_stop_tests, *args):
790 test_states_seq = iter(['RUNNING', 'COMPLETE'])
792 def configure_mock(*args):
793 return [{'id': self.SUCCESS_RECORD_ID,
794 'testStateOrStep': next(test_states_seq)}]
796 mock_get_tests.side_effect = configure_mock
797 self.assertIsNone(self.res_helper.abort_running_tests())
798 mock_stop_tests.assert_called_once_with(
799 running_test_id=self.SUCCESS_RECORD_ID,
801 self.assertEqual(2, mock_get_tests.call_count)
803 @mock.patch.object(tg_landslide.LandslideResourceHelper,
804 'stop_running_tests')
805 @mock.patch.object(tg_landslide.LandslideResourceHelper,
807 def test_abort_running_tests_error(self, mock_get_tests, mock_stop_tests,
809 tests_data = {'id': self.SUCCESS_RECORD_ID,
810 'testStateOrStep': 'RUNNING'}
811 mock_get_tests.return_value = [tests_data]
812 with self.assertRaises(RuntimeError):
813 self.res_helper.abort_running_tests(timeout=1, delay=1)
814 mock_stop_tests.assert_called_with(
815 running_test_id=self.SUCCESS_RECORD_ID,
818 def test__build_url(self, *args):
820 action = {'action': 'userCreate'}
821 expected_url = ''.join([EXAMPLE_URL, 'users?action=userCreate'])
822 self.assertEqual(expected_url,
823 self.res_helper._build_url(resource, action))
825 def test__build_url_error(self, *args):
827 action = {'action': 'userCreate'}
829 with self.assertRaises(ValueError):
830 self.res_helper._build_url(resource, action)
832 def test_get_response_params(self, *args):
835 mock_session = mock.Mock(spec=requests.Session)
836 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
837 'json.return_value': self.USERS_DATA}
838 mock_session.get.return_value.configure_mock(**get_resp_data)
839 self.res_helper.session = mock_session
840 resp = self.res_helper.get_response_params(method, resource)
841 self.assertTrue(resp)
843 @mock.patch.object(tg_landslide.LandslideResourceHelper, '_get_users')
844 @mock.patch.object(time, 'time')
845 def test__create_user(self, mock_time, mock_get_users, *args):
846 mock_time.strftime.return_value = self.EXPIRE_DATE
847 post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
848 'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
849 mock_session = mock.Mock(spec=requests.Session)
850 mock_session.post.return_value.configure_mock(**post_resp_data)
851 self.res_helper.session = mock_session
852 self.assertEqual(self.SUCCESS_RECORD_ID,
853 self.res_helper._create_user(self.AUTH_DATA))
854 mock_get_users.assert_not_called()
856 @mock.patch.object(tg_landslide.LandslideResourceHelper, '_modify_user')
857 @mock.patch.object(time, 'time')
858 def test__create_user_username_exists(self, mock_time, mock_modify_user,
860 mock_time.strftime.return_value = self.EXPIRE_DATE
861 mock_modify_user.return_value = {'id': self.SUCCESS_RECORD_ID,
862 'result': 'No changes requested'}
864 'status_code': self.ERROR_CODE,
865 'json.return_value': {'id': self.SUCCESS_OK_CODE,
866 'apiCode': self.NOT_MODIFIED_CODE}}
867 mock_session = mock.Mock(spec=requests.Session)
868 mock_session.post.return_value.configure_mock(**post_resp_data)
869 self.res_helper.session = mock_session
870 res = self.res_helper._create_user(self.AUTH_DATA)
871 mock_modify_user.assert_called_once_with(TAS_INFO['user'],
872 {'isActive': 'true'})
873 self.assertEqual(self.SUCCESS_RECORD_ID, res)
875 @mock.patch.object(time, 'time')
876 def test__create_user_error(self, mock_time, *args):
877 mock_time.strftime.return_value = self.EXPIRE_DATE
878 mock_session = mock.Mock(spec=requests.Session)
879 post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
880 'json.return_value': {'apiCode': self.ERROR_CODE}}
881 mock_session.post.return_value.configure_mock(**post_resp_data)
882 self.res_helper.session = mock_session
883 with self.assertRaises(exceptions.RestApiError):
884 self.res_helper._create_user(self.AUTH_DATA)
886 def test__modify_user(self, *args):
887 post_data = {'username': 'test_user'}
888 mock_session = mock.Mock(spec=requests.Session)
889 post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
890 'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
891 mock_session.post.return_value.configure_mock(**post_resp_data)
892 self.res_helper.session = mock_session
893 res = self.res_helper._modify_user(username=self.TEST_USER,
895 self.assertEqual(self.SUCCESS_RECORD_ID, res['id'])
897 def test__modify_user_rest_resp_fail(self, *args):
898 post_data = {'non-existing-key': ''}
899 mock_session = mock.Mock(spec=requests.Session)
900 mock_session.post.ok = False
901 self.res_helper.session = mock_session
902 self.assertRaises(exceptions.RestApiError,
903 self.res_helper._modify_user,
904 username=self.TEST_USER, fields=post_data)
905 mock_session.post.assert_called_once()
907 def test__delete_user(self, *args):
908 mock_session = mock.Mock(spec=requests.Session)
909 self.res_helper.session = mock_session
910 self.assertIsNone(self.res_helper._delete_user(
911 username=self.TEST_USER))
913 def test__get_users(self, *args):
914 mock_session = mock.Mock(spec=requests.Session)
915 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
916 'json.return_value': self.USERS_DATA}
917 mock_session.get.return_value.configure_mock(**get_resp_data)
918 self.res_helper.session = mock_session
919 self.assertEqual(self.USERS_DATA['users'],
920 self.res_helper._get_users())
922 def test_exec_rest_request(self, *args):
923 resource = 'testServers'
924 action = {'action': 'modify'}
925 expected_url = ''.join([EXAMPLE_URL, 'testServers?action=modify'])
926 post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
927 'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
928 mock_session = mock.Mock(spec=requests.Session)
929 mock_session.post.return_value.configure_mock(**post_resp_data)
930 self.res_helper.session = mock_session
931 self.res_helper.exec_rest_request('post', resource, action)
932 self.res_helper.session.post.assert_called_once_with(expected_url,
935 def test_exec_rest_request_unsupported_method_error(self, *args):
936 resource = 'testServers'
937 action = {'action': 'modify'}
938 with self.assertRaises(ValueError):
939 self.res_helper.exec_rest_request('patch', resource, action)
941 def test_exec_rest_request_missed_action_arg(self, *args):
942 resource = 'testServers'
943 with self.assertRaises(ValueError):
944 self.res_helper.exec_rest_request('post', resource)
946 def test_exec_rest_request_raise_exc(self):
948 action = {'action': 'modify'}
949 post_resp_data = {'status_code': self.ERROR_CODE,
950 'json.return_value': {
951 'status_code': self.ERROR_CODE}}
952 mock_session = mock.Mock(spec=requests.Session)
953 mock_session.post.return_value.configure_mock(**post_resp_data)
954 self.assertRaises(exceptions.RestApiError,
955 self.res_helper.exec_rest_request,
956 'post', resource, action, raise_exc=True)
958 @mock.patch.object(time, 'time')
959 def test_connect(self, mock_time, *args):
960 vnfd = VNFD['vnfd:vnfd-catalog']['vnfd'][0]
961 mock_time.strftime.return_value = self.EXPIRE_DATE
962 self.res_helper.vnfd_helper = vnfd
964 self.res_helper._tcl = mock.Mock()
965 post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
966 'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
967 mock_session = mock.Mock(spec=requests.Session, headers={})
968 mock_session.post.return_value.configure_mock(**post_resp_data)
969 self.res_helper.session = mock_session
970 self.assertIsInstance(self.res_helper.connect(), requests.Session)
971 self.res_helper._tcl.connect.assert_called_once_with(
974 TAS_INFO['password'])
976 def test_disconnect(self, *args):
977 self.res_helper._tcl = mock.Mock()
978 self.assertIsNone(self.res_helper.disconnect())
979 self.assertIsNone(self.res_helper.session)
980 self.res_helper._tcl.disconnect.assert_called_once()
982 def test_terminate(self, *args):
983 self.assertIsNone(self.res_helper.terminate())
984 self.assertEqual(self.TEST_TERMINATED,
985 self.res_helper._terminated.value)
987 def test_create_dmf(self, *args):
988 self.res_helper._tcl = mock.Mock()
989 self.assertIsNone(self.res_helper.create_dmf(DMF_CFG))
990 self.res_helper._tcl.create_dmf.assert_called_once_with(DMF_CFG)
992 def test_create_dmf_as_list(self, *args):
993 self.res_helper._tcl = mock.Mock()
994 self.assertIsNone(self.res_helper.create_dmf([DMF_CFG]))
995 self.res_helper._tcl.create_dmf.assert_called_once_with(DMF_CFG)
997 def test_delete_dmf(self, *args):
998 self.res_helper._tcl = mock.Mock()
999 self.assertIsNone(self.res_helper.delete_dmf(DMF_CFG))
1000 self.res_helper._tcl.delete_dmf.assert_called_once_with(DMF_CFG)
1002 def test_delete_dmf_as_list(self, *args):
1003 self.res_helper._tcl = mock.Mock()
1004 self.assertIsNone(self.res_helper.delete_dmf([DMF_CFG]))
1005 self.res_helper._tcl.delete_dmf.assert_called_once_with(DMF_CFG)
1007 @mock.patch.object(tg_landslide.LandslideResourceHelper, 'configure_sut')
1008 def test_create_suts(self, mock_configure_sut, *args):
1009 mock_session = mock.Mock(spec=requests.Session)
1010 post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE}
1011 mock_session.post.return_value.configure_mock(**post_resp_data)
1012 self.res_helper.session = mock_session
1013 self.assertIsNone(self.res_helper.create_suts(TS1_SUTS))
1014 mock_configure_sut.assert_not_called()
1016 @mock.patch.object(tg_landslide.LandslideResourceHelper, 'configure_sut')
1017 def test_create_suts_sut_exists(self, mock_configure_sut, *args):
1018 sut_name = 'test_sut'
1021 'role': 'SgwControlAddr',
1022 'managementIp': '12.0.1.1',
1023 'ip': '10.42.32.100'
1026 mock_session = mock.Mock(spec=requests.Session)
1027 post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
1028 mock_session.post.return_value.configure_mock(**post_resp_data)
1029 self.res_helper.session = mock_session
1030 self.assertIsNone(self.res_helper.create_suts(suts))
1031 mock_configure_sut.assert_called_once_with(
1033 json_data={k: v for k, v in suts[0].items()
1034 if k not in {'phy', 'nextHop', 'role', 'name'}})
1036 def test_get_suts(self, *args):
1037 mock_session = mock.Mock(spec=requests.Session)
1038 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1039 'json.return_value': self.SUTS_DATA}
1040 mock_session.get.return_value.configure_mock(**get_resp_data)
1041 self.res_helper.session = mock_session
1042 self.assertIsInstance(self.res_helper.get_suts(), list)
1044 def test_get_suts_single_id(self, *args):
1045 mock_session = mock.Mock(spec=requests.Session)
1046 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1047 'json.return_value': self.SUTS_DATA['suts'][0]}
1048 mock_session.get.return_value.configure_mock(**get_resp_data)
1049 self.res_helper.session = mock_session
1050 self.assertIsInstance(self.res_helper.get_suts(suts_id=2), dict)
1052 def test_configure_sut(self, *args):
1053 post_data = {'managementIp': '2.2.2.2'}
1054 mock_session = mock.Mock(spec=requests.Session)
1055 post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1056 'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
1057 mock_session.post.return_value.configure_mock(**post_resp_data)
1058 self.res_helper.session = mock_session
1059 self.assertIsNone(self.res_helper.configure_sut('test_name',
1061 mock_session.post.assert_called_once()
1063 def test_configure_sut_error(self, *args):
1064 post_data = {'managementIp': '2.2.2.2'}
1065 mock_session = mock.Mock(spec=requests.Session)
1066 post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
1067 mock_session.post.return_value.configure_mock(**post_resp_data)
1068 self.res_helper.session = mock_session
1069 with self.assertRaises(exceptions.RestApiError):
1070 self.res_helper.configure_sut('test_name', post_data)
1072 def test_delete_suts(self, *args):
1073 mock_session = mock.Mock(spec=requests.Session)
1074 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1075 'json.return_value': self.SUTS_DATA}
1076 delete_resp_data = {'status_code': self.SUCCESS_OK_CODE}
1077 mock_session.get.return_value.configure_mock(**get_resp_data)
1078 mock_session.delete.return_value.configure_mock(**delete_resp_data)
1079 self.res_helper.session = mock_session
1080 self.assertIsNone(self.res_helper.delete_suts())
1081 mock_session.delete.assert_called_once()
1083 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1085 def test__check_test_servers_state(self, mock_get_test_servers, *args):
1086 mock_get_test_servers.return_value = \
1087 self.TEST_SERVERS_DATA['testServers']
1088 self.res_helper._check_test_servers_state()
1089 mock_get_test_servers.assert_called_once()
1091 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1093 def test__check_test_servers_state_server_not_ready(
1094 self, mock_get_test_servers, *args):
1095 test_servers_not_ready = [
1097 "url": ''.join([EXAMPLE_URL, "testServers/1"]),
1099 "name": "TestServer_1",
1100 "state": "NOT_READY",
1101 "version": "16.4.0.10"
1105 mock_get_test_servers.return_value = test_servers_not_ready
1106 with self.assertRaises(RuntimeError):
1107 self.res_helper._check_test_servers_state(timeout=1, delay=0)
1109 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1110 '_check_test_servers_state')
1111 def test_create_test_servers(self, mock_check_ts_state, *args):
1112 test_servers_ids = [
1113 ts['id'] for ts in self.TEST_SERVERS_DATA['testServers']]
1115 self.res_helper.license_data['lic_id'] = TAS_INFO['license']
1116 self.res_helper._tcl.create_test_server = mock.Mock()
1117 self.res_helper._tcl.create_test_server.side_effect = test_servers_ids
1118 self.assertIsNone(self.res_helper.create_test_servers(TEST_SERVERS))
1119 mock_check_ts_state.assert_called_once_with(test_servers_ids)
1121 @mock.patch.object(tg_landslide.LandslideTclClient,
1122 'resolve_test_server_name')
1123 @mock.patch.object(tg_landslide.LsTclHandler, 'execute')
1124 def test_create_test_servers_error(self, mock_execute,
1125 mock_resolve_ts_name, *args):
1126 self.res_helper.license_data['lic_id'] = TAS_INFO['license']
1127 # Return message for case test server wasn't created
1128 mock_execute.return_value = 'TS not found'
1129 # Return message for case test server name wasn't resolved
1130 mock_resolve_ts_name.return_value = 'TS not found'
1131 with self.assertRaises(RuntimeError):
1132 self.res_helper.create_test_servers(TEST_SERVERS)
1134 def test_get_test_servers(self, *args):
1135 mock_session = mock.Mock(spec=requests.Session)
1136 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1137 'json.return_value': self.TEST_SERVERS_DATA}
1138 mock_session.get.return_value.configure_mock(**get_resp_data)
1139 self.res_helper.session = mock_session
1140 res = self.res_helper.get_test_servers()
1141 self.assertEqual(self.TEST_SERVERS_DATA['testServers'], res)
1143 def test_get_test_servers_by_id(self, *args):
1144 mock_session = mock.Mock(spec=requests.Session)
1146 _ts = self.TEST_SERVERS_DATA['testServers'][0]
1147 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1148 'json.return_value': _ts}
1149 mock_session.get.return_value.configure_mock(**get_resp_data)
1150 self.res_helper.session = mock_session
1151 res = self.res_helper.get_test_servers(test_server_ids=[_ts['id']])
1152 self.assertEqual([_ts], res)
1154 def test_configure_test_servers(self, *args):
1155 mock_session = mock.Mock(spec=requests.Session)
1156 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1157 'json.return_value': self.TEST_SERVERS_DATA}
1158 mock_session.get.return_value.configure_mock(**get_resp_data)
1159 self.res_helper.session = mock_session
1160 res = self.res_helper.configure_test_servers(
1161 action={'action': 'recycle'})
1163 [x['id'] for x in self.TEST_SERVERS_DATA['testServers']],
1165 self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']),
1166 mock_session.post.call_count)
1168 def test_delete_test_servers(self, *args):
1169 mock_session = mock.Mock(spec=requests.Session)
1170 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1171 'json.return_value': self.TEST_SERVERS_DATA}
1172 mock_session.get.return_value.configure_mock(**get_resp_data)
1173 self.res_helper.session = mock_session
1174 self.assertIsNone(self.res_helper.delete_test_servers())
1175 self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']),
1176 mock_session.delete.call_count)
1178 def test_create_test_session_res_helper(self, *args):
1179 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1180 self.res_helper._tcl = mock.Mock()
1181 test_session = {'name': 'test'}
1182 self.assertIsNone(self.res_helper.create_test_session(test_session))
1183 self.res_helper._tcl.create_test_session.assert_called_once_with(
1184 {'name': 'test', 'library': self.SUCCESS_RECORD_ID})
1186 @mock.patch.object(tg_landslide.LandslideTclClient,
1187 'resolve_test_server_name',
1188 return_value='Not Found')
1189 def test_create_test_session_ts_name_not_found(self, *args):
1190 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1193 'description': 'UE default bearer creation test case',
1194 'name': 'default_bearer_capacity',
1195 'tsGroups': [{'testCases': [{'type': 'SGW_Node',
1197 'tsId': 'TestServer_3'}]
1199 with self.assertRaises(RuntimeError):
1200 self.res_helper.create_test_session(test_session)
1202 def test_get_test_session(self, *args):
1203 test_session = {"name": self.TEST_SESSION_NAME}
1204 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1205 mock_session = mock.Mock(spec=requests.Session)
1206 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1207 'json.return_value': test_session}
1208 mock_session.get.return_value.configure_mock(**get_resp_data)
1209 self.res_helper.session = mock_session
1210 res = self.res_helper.get_test_session(self.TEST_SESSION_NAME)
1211 self.assertEqual(test_session, res)
1213 def test_configure_test_session(self, *args):
1214 test_session = {'name': self.TEST_SESSION_NAME}
1215 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1216 self.res_helper.user_lib_uri = 'libraries/{{}}/{}'.format(
1217 self.res_helper.test_session_uri)
1218 mock_session = mock.Mock(spec=requests.Session)
1219 self.res_helper.session = mock_session
1220 res = self.res_helper.configure_test_session(self.TEST_SESSION_NAME,
1222 self.assertIsNotNone(res)
1223 mock_session.post.assert_called_once()
1225 def test_delete_test_session(self, *args):
1226 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1227 self.res_helper.user_lib_uri = 'libraries/{{}}/{}'.format(
1228 self.res_helper.test_session_uri)
1229 mock_session = mock.Mock(spec=requests.Session)
1230 self.res_helper.session = mock_session
1231 res = self.res_helper.delete_test_session(self.TEST_SESSION_NAME)
1232 self.assertIsNotNone(res)
1233 mock_session.delete.assert_called_once()
1235 def test_create_running_tests(self, *args):
1236 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1237 test_session = {'id': self.SUCCESS_RECORD_ID}
1238 mock_session = mock.Mock(spec=requests.Session)
1239 post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
1240 'json.return_value': test_session}
1241 mock_session.post.return_value.configure_mock(**post_resp_data)
1242 self.res_helper.session = mock_session
1243 self.res_helper.create_running_tests(self.TEST_SESSION_NAME)
1244 self.assertEqual(self.SUCCESS_RECORD_ID, self.res_helper.run_id)
1246 def test_create_running_tests_error(self, *args):
1247 self.res_helper._user_id = self.SUCCESS_RECORD_ID
1248 mock_session = mock.Mock(spec=requests.Session)
1249 post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
1250 mock_session.post.return_value.configure_mock(**post_resp_data)
1251 self.res_helper.session = mock_session
1252 with self.assertRaises(exceptions.RestApiError):
1253 self.res_helper.create_running_tests(self.TEST_SESSION_NAME)
1255 def test_get_running_tests(self, *args):
1256 mock_session = mock.Mock(spec=requests.Session)
1257 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1258 'json.return_value': self.RUNNING_TESTS_DATA}
1259 mock_session.get.return_value.configure_mock(**get_resp_data)
1260 self.res_helper.session = mock_session
1261 res = self.res_helper.get_running_tests()
1262 self.assertEqual(self.RUNNING_TESTS_DATA['runningTests'], res)
1264 def test_delete_running_tests(self, *args):
1265 mock_session = mock.Mock(spec=requests.Session)
1266 delete_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1267 'json.return_value': self.RUNNING_TESTS_DATA}
1268 mock_session.delete.return_value.configure_mock(**delete_resp_data)
1269 self.res_helper.session = mock_session
1270 self.assertIsNone(self.res_helper.delete_running_tests())
1272 def test__running_tests_action(self, *args):
1274 mock_session = mock.Mock(spec=requests.Session)
1275 self.res_helper.session = mock_session
1276 res = self.res_helper._running_tests_action(self.SUCCESS_RECORD_ID,
1278 self.assertIsNone(res)
1280 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1281 '_running_tests_action')
1282 def test_stop_running_tests(self, mock_tests_action, *args):
1283 res = self.res_helper.stop_running_tests(self.SUCCESS_RECORD_ID)
1284 self.assertIsNone(res)
1285 mock_tests_action.assert_called_once()
1287 def test_check_running_test_state(self, *args):
1288 mock_session = mock.Mock(spec=requests.Session)
1290 'status_code': self.SUCCESS_OK_CODE,
1291 'json.return_value': self.RUNNING_TESTS_DATA["runningTests"][0]}
1292 mock_session.get.return_value.configure_mock(**get_resp_data)
1293 self.res_helper.session = mock_session
1294 res = self.res_helper.check_running_test_state(self.SUCCESS_RECORD_ID)
1296 self.RUNNING_TESTS_DATA["runningTests"][0]['testStateOrStep'],
1299 def test_get_running_tests_results(self, *args):
1300 mock_session = mock.Mock(spec=requests.Session)
1301 get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1302 'json.return_value': self.TEST_RESULTS_DATA}
1303 mock_session.get.return_value.configure_mock(**get_resp_data)
1304 self.res_helper.session = mock_session
1305 res = self.res_helper.get_running_tests_results(
1306 self.SUCCESS_RECORD_ID)
1307 self.assertEqual(self.TEST_RESULTS_DATA, res)
1309 def test__write_results(self, *args):
1310 res = self.res_helper._write_results(self.TEST_RESULTS_DATA)
1312 "Test Summary::Actual Dedicated Bearer Session Connects": 100.0,
1313 "Test Summary::Actual Dedicated Bearer Session Disconnects": 100.0,
1314 "Test Summary::Actual Disconnect Rate(Sessions / Second)(P - I)": 164.804,
1315 "Test Summary::Average Session Disconnect Time(P - I)": 5.024,
1316 "Test Summary::Total Data Sent + Received Packets / Sec(P - I)": 1452.294
1318 self.assertEqual(exp_res, res)
1320 def test__write_results_no_tabs(self, *args):
1321 _res_data = copy.deepcopy(self.TEST_RESULTS_DATA)
1322 del _res_data['tabs']
1323 # Return None if tabs not found in test results dict
1324 self.assertIsNone(self.res_helper._write_results(_res_data))
1326 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1327 'check_running_test_state')
1328 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1329 'get_running_tests_results')
1330 def test_collect_kpi_test_running(self, mock_tests_results,
1331 mock_tests_state, *args):
1332 self.res_helper.run_id = self.SUCCESS_RECORD_ID
1333 mock_tests_state.return_value = 'RUNNING'
1334 mock_tests_results.return_value = self.TEST_RESULTS_DATA
1335 res = self.res_helper.collect_kpi()
1336 self.assertNotIn('done', res)
1337 mock_tests_state.assert_called_once_with(self.res_helper.run_id)
1338 mock_tests_results.assert_called_once_with(self.res_helper.run_id)
1340 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1341 'check_running_test_state')
1342 @mock.patch.object(tg_landslide.LandslideResourceHelper,
1343 'get_running_tests_results')
1344 def test_collect_kpi_test_completed(self, mock_tests_results,
1345 mock_tests_state, *args):
1346 self.res_helper.run_id = self.SUCCESS_RECORD_ID
1347 mock_tests_state.return_value = 'COMPLETE'
1348 res = self.res_helper.collect_kpi()
1349 self.assertIsNotNone(res)
1350 mock_tests_state.assert_called_once_with(self.res_helper.run_id)
1351 mock_tests_results.assert_not_called()
1352 self.assertDictContainsSubset({'done': True}, res)
1355 class TestLandslideTclClient(unittest.TestCase):
1357 self.mock_tcl_handler = mock.Mock(spec=tg_landslide.LsTclHandler)
1358 self.ls_res_helper = mock.Mock(
1359 spec=tg_landslide.LandslideResourceHelper)
1360 self.ls_tcl_client = tg_landslide.LandslideTclClient(
1361 self.mock_tcl_handler,
1364 def test___init__(self, *args):
1365 self.ls_tcl_client = tg_landslide.LandslideTclClient(
1366 self.mock_tcl_handler,
1368 self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
1369 self.assertIsNone(self.ls_tcl_client._user)
1370 self.assertIsNone(self.ls_tcl_client._library_id)
1371 self.assertIsNone(self.ls_tcl_client._basic_library_id)
1372 self.assertEqual(set(), self.ls_tcl_client.ts_ids)
1373 self.assertIsInstance(self.ls_tcl_client._tc_types, set)
1374 self.assertIsNotNone(self.ls_tcl_client._tc_types)
1376 def test_connect_login_success(self, *args):
1378 exec_responses = ['java0x2', lib_id, lib_id]
1379 auth = ('user', 'password')
1380 self.mock_tcl_handler.execute.side_effect = exec_responses
1381 self.ls_tcl_client.connect(TAS_INFO['ip'], *auth)
1382 self.assertEqual(lib_id, self.ls_tcl_client._library_id)
1383 self.assertEqual(lib_id, self.ls_tcl_client._basic_library_id)
1384 self.assertEqual(TAS_INFO['ip'], self.ls_tcl_client.tcl_server_ip)
1385 self.assertEqual(auth[0], self.ls_tcl_client._user)
1386 self.assertEqual(len(exec_responses),
1387 self.mock_tcl_handler.execute.call_count)
1388 self.mock_tcl_handler.execute.assert_has_calls([
1389 mock.call("ls::login 1.1.1.1 user password"),
1390 mock.call("ls::get [ls::query LibraryInfo -userLibraryName user] -Id"),
1393 def test_connect_login_failed(self, *args):
1394 exec_responses = ['Login failed']
1395 auth = ('user', 'password')
1396 self.mock_tcl_handler.execute.side_effect = exec_responses
1397 self.assertRaises(exceptions.LandslideTclException,
1398 self.ls_tcl_client.connect,
1401 self.assertIsNone(self.ls_tcl_client._library_id)
1402 self.assertIsNone(self.ls_tcl_client._basic_library_id)
1403 self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
1404 self.assertIsNone(self.ls_tcl_client._user)
1405 self.assertEqual(len(exec_responses),
1406 self.mock_tcl_handler.execute.call_count)
1407 self.mock_tcl_handler.execute.assert_called_with(
1408 "ls::login 1.1.1.1 user password")
1410 def test_disconnect(self, *args):
1411 self.ls_tcl_client.disconnect()
1412 self.mock_tcl_handler.execute.assert_called_once_with("ls::logout")
1413 self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
1414 self.assertIsNone(self.ls_tcl_client._user)
1415 self.assertIsNone(self.ls_tcl_client._library_id)
1416 self.assertIsNone(self.ls_tcl_client._basic_library_id)
1418 def test_create_test_server(self, *args):
1420 self.ls_tcl_client._ts_context.vnfd_helper = \
1421 VNFD['vnfd:vnfd-catalog']['vnfd'][0]
1422 self.ls_tcl_client._ts_context.license_data = {'lic_id': return_value}
1423 self.mock_tcl_handler.execute.return_value = return_value
1424 self.ls_tcl_client._set_thread_model = mock.Mock()
1425 res = self.ls_tcl_client.create_test_server(TEST_SERVERS[1])
1426 self.assertEqual(3, self.mock_tcl_handler.execute.call_count)
1427 self.mock_tcl_handler.execute.assert_has_calls([
1428 mock.call('ls::query TsId TestServer_2'),
1429 mock.call('set ts [ls::retrieve TsInfo -Name "TestServer_2"]'),
1430 mock.call('ls::get $ts -RequestedLicense'),
1432 self.ls_tcl_client._set_thread_model.assert_called_once_with(
1433 TEST_SERVERS[1]['name'],
1434 TEST_SERVERS[1]['thread_model'])
1435 self.assertEqual(int(return_value), res)
1437 def test_create_test_server_fail_limit_reach(self, *args):
1438 self.mock_tcl_handler.execute.side_effect = ['TS not found',
1440 self.assertRaises(RuntimeError,
1441 self.ls_tcl_client.create_test_server,
1443 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1444 self.mock_tcl_handler.execute.assert_has_calls([
1445 mock.call('ls::query TsId TestServer_1'),
1446 mock.call('ls::perform AddTs -Name "TestServer_1" '
1447 '-Ip "192.168.122.101"'),
1450 def test__add_test_server(self):
1452 self.mock_tcl_handler.execute.side_effect = ['TS not found', ts_id]
1453 self.assertEqual(ts_id,
1454 self.ls_tcl_client._add_test_server('name', 'ip'))
1455 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1456 self.mock_tcl_handler.execute.assert_has_calls([
1457 mock.call('ls::query TsId name'),
1458 mock.call('ls::perform AddTs -Name "name" -Ip "ip"'),
1461 def test__add_test_server_failed(self):
1462 self.mock_tcl_handler.execute.side_effect = ['TS not found',
1464 self.assertRaises(RuntimeError, self.ls_tcl_client._add_test_server,
1466 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1467 self.mock_tcl_handler.execute.assert_has_calls([
1468 mock.call('ls::query TsId name'),
1469 mock.call('ls::perform AddTs -Name "name" -Ip "ip"'),
1472 def test__update_license(self):
1475 exec_resp = ['java0x4',
1477 TCL_SUCCESS_RESPONSE,
1478 TCL_SUCCESS_RESPONSE]
1479 self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id}
1480 self.mock_tcl_handler.execute.side_effect = exec_resp
1481 self.ls_tcl_client._update_license('name')
1482 self.assertEqual(len(exec_resp),
1483 self.mock_tcl_handler.execute.call_count)
1485 self.mock_tcl_handler.execute.assert_has_calls([
1486 mock.call('set ts [ls::retrieve TsInfo -Name "name"]'),
1487 mock.call('ls::get $ts -RequestedLicense'),
1488 mock.call('ls::config $ts -RequestedLicense 222'),
1489 mock.call('ls::perform ModifyTs $ts'),
1492 def test__update_license_same_as_current(self):
1495 exec_resp = ['java0x4', curr_lic_id]
1496 self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id}
1497 self.mock_tcl_handler.execute.side_effect = exec_resp
1498 self.ls_tcl_client._update_license('name')
1499 self.assertEqual(len(exec_resp),
1500 self.mock_tcl_handler.execute.call_count)
1501 self.mock_tcl_handler.execute.assert_has_calls([
1502 mock.call('set ts [ls::retrieve TsInfo -Name "name"]'),
1503 mock.call('ls::get $ts -RequestedLicense'),
1506 def test__set_thread_model_update_needed(self):
1507 self.ls_tcl_client._ts_context.vnfd_helper = {
1509 'cfguser_password': 'cfguser_password'
1512 exec_resp = ['java0x4', 'V0', '', '']
1513 self.mock_tcl_handler.execute.side_effect = exec_resp
1514 self.ls_tcl_client._set_thread_model('name', 'Fireball')
1515 self.assertEqual(len(exec_resp),
1516 self.mock_tcl_handler.execute.call_count)
1517 self.mock_tcl_handler.execute.assert_has_calls([
1518 mock.call('set tsc [ls::perform RetrieveTsConfiguration '
1519 '-name "name" cfguser_password]'),
1520 mock.call('ls::get $tsc -ThreadModel'),
1521 mock.call('ls::config $tsc -ThreadModel "V1_FB3"'),
1522 mock.call('ls::perform ApplyTsConfiguration $tsc cfguser_password'),
1525 def test__set_thread_model_no_update_needed(self):
1526 self.ls_tcl_client._ts_context.vnfd_helper = {
1528 'cfguser_password': 'cfguser_password'
1531 exec_resp = ['java0x4', 'V0']
1532 self.mock_tcl_handler.execute.side_effect = exec_resp
1533 self.ls_tcl_client._set_thread_model('name', 'Legacy')
1534 self.assertEqual(len(exec_resp),
1535 self.mock_tcl_handler.execute.call_count)
1536 self.mock_tcl_handler.execute.assert_has_calls([
1537 mock.call('set tsc [ls::perform RetrieveTsConfiguration '
1538 '-name "name" cfguser_password]'),
1539 mock.call('ls::get $tsc -ThreadModel'),
1542 @mock.patch.object(tg_landslide.LandslideTclClient,
1543 'resolve_test_server_name', side_effect=['4', '2'])
1544 def test_create_test_session(self, *args):
1545 _session_profile = copy.deepcopy(SESSION_PROFILE)
1546 _session_profile['reservations'] = RESERVATIONS
1547 self.ls_tcl_client._save_test_session = mock.Mock()
1548 self.ls_tcl_client._configure_ts_group = mock.Mock()
1549 self.ls_tcl_client._library_id = 42
1550 self.ls_tcl_client.create_test_session(_session_profile)
1551 self.assertEqual(17, self.mock_tcl_handler.execute.call_count)
1552 self.mock_tcl_handler.execute.assert_has_calls([
1553 mock.call('set test_ [ls::create TestSession]'),
1554 mock.call('ls::config $test_ -Library 42 '
1555 '-Name "default_bearer_capacity"'),
1556 mock.call('ls::config $test_ -Description ' \
1557 '"UE default bearer creation test case"'),
1558 mock.call('ls::config $test_ -Keywords ""'),
1559 mock.call('ls::config $test_ -Duration "60"'),
1560 mock.call('ls::config $test_ -Iterations "1"'),
1561 # _configure_reservation
1562 mock.call('set reservation_ [ls::create Reservation -under $test_]'),
1563 mock.call('ls::config $reservation_ -TsIndex 0 '
1564 '-TsId 4 -TsName "TestServer_1"'),
1565 mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
1566 mock.call('ls::config $physubnet_ -Name "eth1" -Base "10.42.32.100" '
1567 '-Mask "/24" -NumIps 20'),
1568 # _configure_reservation
1569 mock.call('set reservation_ [ls::create Reservation -under $test_]'),
1570 mock.call('ls::config $reservation_ -TsIndex 1 '
1571 '-TsId 2 -TsName "TestServer_2"'),
1572 mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
1573 mock.call('ls::config $physubnet_ -Name "eth1" -Base "10.42.32.1" '
1574 '-Mask "/24" -NumIps 100'),
1575 mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
1576 mock.call('ls::config $physubnet_ -Name "eth2" -Base "10.42.33.1" '
1577 '-Mask "/24" -NumIps 100'),
1578 # _configure_report_options
1579 mock.call('ls::config $test_.ReportOptions -Format 1 -Ts -3 -Tc -3'),
1582 def test_create_dmf(self):
1583 self.mock_tcl_handler.execute.return_value = '2'
1584 self.ls_tcl_client._save_dmf = mock.Mock()
1585 self.ls_tcl_client.create_dmf(copy.deepcopy(DMF_CFG))
1586 self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
1587 # This is needed because the dictionary is unordered and the arguments
1588 # can come in either order
1590 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"')
1592 'ls::config $dmf_ -isClientPortRange "false" -clientPort 2002')
1594 call1 in self.mock_tcl_handler.execute.mock_calls or
1595 call2 in self.mock_tcl_handler.execute.mock_calls)
1597 self.mock_tcl_handler.execute.assert_has_calls([
1598 mock.call('set dmf_ [ls::create Dmf]'),
1600 'ls::get [ls::query LibraryInfo -systemLibraryName test] -Id'),
1601 mock.call('ls::config $dmf_ -Library 2 -Name "Basic UDP"'),
1602 mock.call('ls::config $dmf_ -dataProtocol "udp"'),
1604 # 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"'),
1605 mock.call('ls::config $dmf_ -serverPort 2003'),
1608 def test_configure_dmf(self):
1609 self.mock_tcl_handler.execute.return_value = '2'
1610 self.ls_tcl_client._save_dmf = mock.Mock()
1611 self.ls_tcl_client.configure_dmf(DMF_CFG)
1612 self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
1613 # This is need because the dictionary is unordered and the arguments
1614 # can come in either order
1616 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"')
1618 'ls::config $dmf_ -isClientPortRange "false" -clientPort 2002')
1620 call1 in self.mock_tcl_handler.execute.mock_calls or
1621 call2 in self.mock_tcl_handler.execute.mock_calls)
1623 self.mock_tcl_handler.execute.assert_has_calls([
1624 mock.call('set dmf_ [ls::create Dmf]'),
1626 'ls::get [ls::query LibraryInfo -systemLibraryName test] -Id'),
1627 mock.call('ls::config $dmf_ -Library 2 -Name "Basic UDP"'),
1628 mock.call('ls::config $dmf_ -dataProtocol "udp"'),
1630 # 'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"'),
1631 mock.call('ls::config $dmf_ -serverPort 2003'),
1634 def test_delete_dmf(self):
1635 self.assertRaises(NotImplementedError,
1636 self.ls_tcl_client.delete_dmf,
1639 def test__save_dmf_valid(self):
1640 exec_resp = [TCL_SUCCESS_RESPONSE, TCL_SUCCESS_RESPONSE]
1641 self.mock_tcl_handler.execute.side_effect = exec_resp
1642 self.ls_tcl_client._save_dmf()
1643 self.assertEqual(len(exec_resp),
1644 self.mock_tcl_handler.execute.call_count)
1645 self.mock_tcl_handler.execute.assert_has_calls([
1646 mock.call('ls::perform Validate -Dmf $dmf_'),
1647 mock.call('ls::save $dmf_ -overwrite'),
1650 def test__save_dmf_invalid(self):
1651 exec_resp = ['Invalid', 'List of errors and warnings']
1652 self.mock_tcl_handler.execute.side_effect = exec_resp
1653 self.assertRaises(exceptions.LandslideTclException,
1654 self.ls_tcl_client._save_dmf)
1655 self.assertEqual(len(exec_resp),
1656 self.mock_tcl_handler.execute.call_count)
1657 self.mock_tcl_handler.execute.assert_has_calls([
1658 mock.call('ls::perform Validate -Dmf $dmf_'),
1659 mock.call('ls::get $dmf_ -ErrorsAndWarnings'),
1662 def test__configure_report_options(self):
1663 _options = {'format': 'CSV', 'PerInterval': 'false'}
1664 self.ls_tcl_client._configure_report_options(_options)
1665 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1666 self.mock_tcl_handler.execute.assert_has_calls([
1667 mock.call('ls::config $test_.ReportOptions -Format 1 -Ts -3 -Tc -3'),
1668 mock.call('ls::config $test_.ReportOptions -PerInterval false'),
1672 def test___configure_ts_group(self, *args):
1673 _ts_group = copy.deepcopy(SESSION_PROFILE['tsGroups'][0])
1674 self.ls_tcl_client._configure_tc_type = mock.Mock()
1675 self.ls_tcl_client._configure_preresolved_arp = mock.Mock()
1676 self.ls_tcl_client.resolve_test_server_name = mock.Mock(
1678 self.ls_tcl_client._configure_ts_group(_ts_group, 0)
1679 self.mock_tcl_handler.execute.assert_called_once_with(
1680 'set tss_ [ls::create TsGroup -under $test_ -tsId 2 ]')
1682 def test___configure_ts_group_resolve_ts_fail(self, *args):
1683 _ts_group = copy.deepcopy(SESSION_PROFILE['tsGroups'][0])
1684 self.ls_tcl_client._configure_tc_type = mock.Mock()
1685 self.ls_tcl_client._configure_preresolved_arp = mock.Mock()
1686 self.ls_tcl_client.resolve_test_server_name = mock.Mock(
1687 return_value='TS Not Found')
1688 self.assertRaises(RuntimeError, self.ls_tcl_client._configure_ts_group,
1690 self.mock_tcl_handler.execute.assert_not_called()
1692 def test__configure_tc_type(self):
1693 _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1694 self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE
1695 self.ls_tcl_client._configure_parameters = mock.Mock()
1696 self.ls_tcl_client._configure_tc_type(_tc, 0)
1697 self.assertEqual(7, self.mock_tcl_handler.execute.call_count)
1699 def test__configure_tc_type_optional_param_omitted(self):
1700 _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1702 self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE
1703 self.ls_tcl_client._configure_parameters = mock.Mock()
1704 self.ls_tcl_client._configure_tc_type(_tc, 0)
1705 self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
1707 def test__configure_tc_type_wrong_type(self):
1708 _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1709 _tc['type'] = 'not_supported'
1710 self.ls_tcl_client._configure_parameters = mock.Mock()
1711 self.assertRaises(RuntimeError,
1712 self.ls_tcl_client._configure_tc_type,
1714 self.mock_tcl_handler.assert_not_called()
1716 def test__configure_tc_type_not_found_basic_lib(self):
1717 _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1718 self.ls_tcl_client._configure_parameters = mock.Mock()
1719 self.mock_tcl_handler.execute.return_value = 'Invalid'
1720 self.assertRaises(RuntimeError,
1721 self.ls_tcl_client._configure_tc_type,
1724 def test__configure_parameters(self):
1725 _params = copy.deepcopy(
1726 SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters'])
1727 self.ls_tcl_client._configure_parameters(_params)
1728 self.assertEqual(16, self.mock_tcl_handler.execute.call_count)
1730 def test__configure_array_param(self):
1731 _array = {"class": "Array",
1733 self.ls_tcl_client._configure_array_param('name', _array)
1734 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1735 self.mock_tcl_handler.execute.assert_has_calls([
1736 mock.call('ls::create -Array-name -under $p_ ;'),
1737 mock.call('ls::create ArrayItem -under $p_.name -Value "0"'),
1740 def test__configure_test_node_param(self):
1741 _params = copy.deepcopy(
1742 SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters'])
1743 self.ls_tcl_client._configure_test_node_param('SgwUserAddr',
1744 _params['SgwUserAddr'])
1745 cmd = ('ls::create -TestNode-SgwUserAddr -under $p_ -Type "eth" '
1746 '-Phy "eth1" -Ip "SGW_USER_IP" -NumLinksOrNodes 1 '
1747 '-NextHop "SGW_CONTROL_NEXT_HOP" -Mac "" -MTU 1500 '
1748 '-ForcedEthInterface "" -EthStatsEnabled false -VlanId 0 '
1749 '-VlanUserPriority 0 -NumVlan 1 -UniqueVlanAddr false;')
1750 self.mock_tcl_handler.execute.assert_called_once_with(cmd)
1752 def test__configure_sut_param(self):
1753 _params = {'name': 'name'}
1754 self.ls_tcl_client._configure_sut_param('name', _params)
1755 self.mock_tcl_handler.execute.assert_called_once_with(
1756 'ls::create -Sut-name -under $p_ -Name "name";')
1758 def test__configure_dmf_param(self):
1759 _params = {"mainflows": [{"library": '111',
1760 "name": "Basic UDP"}],
1761 "instanceGroups": [{
1769 "overridePort": "false",
1773 "transport": "Any"}]
1775 self.ls_tcl_client._get_library_id = mock.Mock(return_value='111')
1776 res = self.ls_tcl_client._configure_dmf_param('name', _params)
1777 self.assertEqual(5, self.mock_tcl_handler.execute.call_count)
1778 self.assertIsNone(res)
1779 self.mock_tcl_handler.execute.assert_has_calls([
1780 mock.call('ls::create -Dmf-name -under $p_ ;'),
1781 mock.call('ls::perform AddDmfMainflow $p_.Dmf 111 "Basic UDP"'),
1782 mock.call('ls::config $p_.Dmf.InstanceGroup(0) -mixType '),
1783 mock.call('ls::config $p_.Dmf.InstanceGroup(0) -rate 0.0'),
1784 mock.call('ls::config $p_.Dmf.InstanceGroup(0).Row(0) -Node 0 '
1785 '-OverridePort false -ClientPort 0 -Context 0 -Role 0 '
1786 '-PreferredTransport Any -RatingGroup 0 '
1790 def test__configure_dmf_param_no_instance_groups(self):
1791 _params = {"mainflows": [{"library": '111',
1792 "name": "Basic UDP"}]}
1793 self.ls_tcl_client._get_library_id = mock.Mock(return_value='111')
1794 res = self.ls_tcl_client._configure_dmf_param('name', _params)
1795 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1796 self.assertIsNone(res)
1797 self.mock_tcl_handler.execute.assert_has_calls([
1798 mock.call('ls::create -Dmf-name -under $p_ ;'),
1799 mock.call('ls::perform AddDmfMainflow $p_.Dmf 111 "Basic UDP"'),
1802 def test__configure_reservation(self):
1803 _reservation = copy.deepcopy(RESERVATIONS[0])
1804 self.ls_tcl_client.resolve_test_server_name = mock.Mock(
1806 res = self.ls_tcl_client._configure_reservation(_reservation)
1807 self.assertIsNone(res)
1808 self.assertEqual(4, self.mock_tcl_handler.execute.call_count)
1809 self.mock_tcl_handler.execute.assert_has_calls([
1810 mock.call('set reservation_ [ls::create Reservation -under $test_]'),
1811 mock.call('ls::config $reservation_ -TsIndex 0 -TsId 4 ' + \
1812 '-TsName "TestServer_1"'),
1813 mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
1814 mock.call('ls::config $physubnet_ -Name "eth1" ' + \
1815 '-Base "10.42.32.100" -Mask "/24" -NumIps 20'),
1818 def test__configure_preresolved_arp(self):
1819 _arp = [{'StartingAddress': '10.81.1.10',
1821 res = self.ls_tcl_client._configure_preresolved_arp(_arp)
1822 self.mock_tcl_handler.execute.assert_called_once()
1823 self.assertIsNone(res)
1824 self.mock_tcl_handler.execute.assert_called_once_with(
1825 'ls::create PreResolvedArpAddress -under $tss_ ' + \
1826 '-StartingAddress "10.81.1.10" -NumNodes 1')
1828 def test__configure_preresolved_arp_none(self):
1829 res = self.ls_tcl_client._configure_preresolved_arp(None)
1830 self.assertIsNone(res)
1831 self.mock_tcl_handler.execute.assert_not_called()
1833 def test_delete_test_session(self):
1834 self.assertRaises(NotImplementedError,
1835 self.ls_tcl_client.delete_test_session, {})
1837 def test__save_test_session(self):
1838 self.mock_tcl_handler.execute.side_effect = [TCL_SUCCESS_RESPONSE,
1839 TCL_SUCCESS_RESPONSE]
1840 res = self.ls_tcl_client._save_test_session()
1841 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1842 self.assertIsNone(res)
1843 self.mock_tcl_handler.execute.assert_has_calls([
1844 mock.call('ls::perform Validate -TestSession $test_'),
1845 mock.call('ls::save $test_ -overwrite'),
1848 def test__save_test_session_invalid(self):
1849 self.mock_tcl_handler.execute.side_effect = ['Invalid', 'Errors']
1850 self.assertRaises(exceptions.LandslideTclException,
1851 self.ls_tcl_client._save_test_session)
1852 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1853 self.mock_tcl_handler.execute.assert_has_calls([
1854 mock.call('ls::perform Validate -TestSession $test_'),
1855 mock.call('ls::get $test_ -ErrorsAndWarnings'),
1858 def test__get_library_id_system_lib(self):
1859 self.mock_tcl_handler.execute.return_value = '111'
1860 res = self.ls_tcl_client._get_library_id('name')
1861 self.mock_tcl_handler.execute.assert_called_once()
1862 self.assertEqual('111', res)
1863 self.mock_tcl_handler.execute.assert_called_with(
1864 'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id')
1866 def test__get_library_id_user_lib(self):
1867 self.mock_tcl_handler.execute.side_effect = ['Not found', '222']
1868 res = self.ls_tcl_client._get_library_id('name')
1869 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1870 self.assertEqual('222', res)
1871 self.mock_tcl_handler.execute.assert_has_calls([
1873 'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id'),
1875 'ls::get [ls::query LibraryInfo -userLibraryName name] -Id'),
1878 def test__get_library_id_exception(self):
1879 self.mock_tcl_handler.execute.side_effect = ['Not found', 'Not found']
1880 self.assertRaises(exceptions.LandslideTclException,
1881 self.ls_tcl_client._get_library_id,
1883 self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1884 self.mock_tcl_handler.execute.assert_has_calls([
1886 'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id'),
1888 'ls::get [ls::query LibraryInfo -userLibraryName name] -Id'),
1892 class TestLsTclHandler(unittest.TestCase):
1895 self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
1896 self.mock_lsapi.start()
1898 self.addCleanup(self._cleanup)
1901 self.mock_lsapi.stop()
1903 def test___init__(self, *args):
1904 self.ls_tcl_handler = tg_landslide.LsTclHandler()
1905 self.assertEqual({}, self.ls_tcl_handler.tcl_cmds)
1906 self.ls_tcl_handler._ls.tcl.assert_called_once()
1908 def test_execute(self, *args):
1909 self.ls_tcl_handler = tg_landslide.LsTclHandler()
1910 self.ls_tcl_handler.execute('command')
1911 self.assertIn('command', self.ls_tcl_handler.tcl_cmds)