yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_landslide.py

   1 # Copyright (c) 2018 Intel Corporation
   2 #
   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
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 import copy
  16 import mock
  17 import requests
  18 import time
  19 import unittest
  20 import uuid
  21
  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
  31
  32 NAME = "tg__0"
  33
  34 EXAMPLE_URL = 'http://example.com/'
  35 TCL_SUCCESS_RESPONSE = 'ls_ok'
  36
  37 TEST_SERVERS = [
  38     {'ip': '192.168.122.101',
  39      'phySubnets': [
  40          {'mask': '/24',
  41           'base': '10.42.32.100',
  42           'numIps': 20,
  43           'name': 'eth1'}
  44      ],
  45      'role': 'SGW_Node',
  46      'name': 'TestServer_1'},
  47     {'ip': '192.168.122.102',
  48      'phySubnets': [
  49          {'mask': '/24',
  50           'base': '10.42.32.1',
  51           'numIps': 100,
  52           'name': 'eth1'
  53           },
  54          {'mask': '/24',
  55           'base': '10.42.33.1',
  56           'numIps': 100,
  57           'name': 'eth2'}
  58      ],
  59      'preResolvedArpAddress': [
  60          {'NumNodes': 1,
  61           'StartingAddress': '10.42.33.5'}
  62      ],
  63      'role': 'SGW_Nodal',
  64      'name': 'TestServer_2',
  65      'thread_model': 'Fireball'
  66      }
  67 ]
  68
  69 TS1_SUTS = [
  70     {'name': 'SGW - C TestNode',
  71      'role': 'SgwControlAddr',
  72      'managementIp': '12.0.1.1',
  73      'ip': '10.42.32.100',
  74      'phy': 'eth5',
  75      'nextHop': '10.42.32.5'
  76      },
  77     {'name': 'SGW - U TestNode',
  78      'role': 'SgwUserAddr',
  79      'managementIp': '12.0.1.2',
  80      'ip': '10.42.32.101',
  81      'phy': 'eth5',
  82      'nextHop': '10.42.32.5'
  83      }
  84 ]
  85
  86 TS2_SUTS = [
  87     {'name': 'eNodeB TestNode',
  88      'role': 'EnbUserAddr',
  89      'managementIp': '12.0.2.1',
  90      'ip': '10.42.32.2',
  91      'phy': 'eth5',
  92      'nextHop': '10.42.32.5'
  93      },
  94     {'name': 'MME TestNode',
  95      'role': 'MmeControlAddr',
  96      'managementIp': '12.0.3.1',
  97      'ip': '10.42.32.1',
  98      'phy': 'eth5',
  99      'nextHop': '10.42.32.5'
 100      },
 101     {'name': 'NetHost TestNode',
 102      'role': 'NetworkHostAddrLocal',
 103      'managementIp': '12.0.4.1',
 104      'ip': '10.42.33.1',
 105      'phy': 'eth5',
 106      'nextHop': '10.42.32.5'
 107      },
 108     {'name': 'PGW TestNode',
 109      'role': 'PgwV4Sut',
 110      'managementIp': '12.0.5.1',
 111      'ip': '10.42.32.105',
 112      'phy': 'eth5',
 113      'nextHop': '10.42.32.5'
 114      },
 115     {'name': 'SGW - C SUT',
 116      'role': 'SgwSut',
 117      'managementIp': '12.0.6.1',
 118      'ip': '10.42.32.100'
 119      },
 120     {'name': 'SGW - U SUT',
 121      'role': 'SgwUserSut',
 122      'managementIp': '12.0.6.2',
 123      'ip': '10.42.32.101'}
 124 ]
 125
 126 VNFD = {
 127     'vnfd:vnfd-catalog': {
 128         'vnfd': [{
 129             'short-name': 'landslide',
 130             'vdu': [{
 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}],
 138             'mgmt-interface': {
 139                 'vdu-id': 'landslide-tas',
 140                 'user': 'user',
 141                 'password': 'user',
 142                 'super-user': 'super-user',
 143                 'super-user-password': 'super-user-password',
 144                 'cfguser_password': 'cfguser_password',
 145                 'license': 48,
 146                 'proto': 'http',
 147                 'ip': '1.1.1.1'},
 148             'benchmark': {
 149                 'kpi': [
 150                     'tx_throughput_mbps',
 151                     'rx_throughput_mbps',
 152                     'in_packets',
 153                     'out_packets',
 154                     'activation_rate_sessps',
 155                     'deactivation_rate_sessps']},
 156             'id': 'LandslideTrafficGen',
 157             'name': 'LandslideTrafficGen'}]}}
 158
 159 TAS_INFO = VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
 160
 161 DMF_CFG = {
 162     "dmf": {
 163         "library": "test",
 164         "name": "Basic UDP"
 165     },
 166     "clientPort": {
 167         "clientPort": 2002,
 168         "isClientPortRange": "false"
 169     },
 170     "dataProtocol": "udp",
 171     "serverPort": 2003
 172 }
 173
 174 RESERVATIONS = [
 175     {'tsName': TEST_SERVERS[0]['name'],
 176      'phySubnets': TEST_SERVERS[0]['phySubnets'],
 177      'tsId': TEST_SERVERS[0]['name'],
 178      'tsIndex': 0},
 179     {'tsName': TEST_SERVERS[1]['name'],
 180      'phySubnets': TEST_SERVERS[1]['phySubnets'],
 181      'tsId': TEST_SERVERS[1]['name'],
 182      'tsIndex': 1}]
 183
 184 SESSION_PROFILE = {
 185     'keywords': '',
 186     'duration': 60,
 187     'iterations': 1,
 188     'description': 'UE default bearer creation test case',
 189     'name': 'default_bearer_capacity',
 190     'reportOptions': {'format': 'CSV'},
 191     'reservePorts': 'false',
 192     'tsGroups': [
 193         {
 194             'testCases': [{
 195                 'type': 'SGW_Node',
 196                 'name': '',
 197                 'linked': "false",
 198                 'AssociatedPhys': '',
 199                 'parameters': {
 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',
 208                     'PgwNodeEn': 'true',
 209                     'DedicatedsPerDefaultBearer': '0',
 210                     'DefaultBearers': '1',
 211                     'SgwUserAddr': {
 212                         'numLinksOrNodes': 1,
 213                         'phy': 'eth1',
 214                         'forcedEthInterface': '',
 215                         'ip': 'SGW_USER_IP',
 216                         'class': 'TestNode',
 217                         'ethStatsEnabled': "false",
 218                         'mtu': 1500
 219                     },
 220                     'SgwControlAddr': {
 221                         'numLinksOrNodes': 1,
 222                         'phy': 'eth1',
 223                         'forcedEthInterface': '',
 224                         'ip': 'SGW_CONTROL_IP',
 225                         'class': 'TestNode',
 226                         'ethStatsEnabled': "false",
 227                         'mtu': 1500,
 228                         'nextHop': 'SGW_CONTROL_NEXT_HOP'
 229                     },
 230                     'BearerAddrPool': '2001::1',
 231                     'TestType': 'SGW-NODE'
 232                 }
 233             }],
 234             'tsId': TEST_SERVERS[0]['name']},
 235         {
 236             'testCases': [{
 237                 'type': 'SGW_Nodal',
 238                 'name': '',
 239                 'parameters': {
 240                     'DataTraffic': 'Continuous',
 241                     'TrafficStartType': 'When All Sessions Established',
 242                     'NetworkHost': 'Local',
 243                     'Gtp2Imsi': '505024101215074',
 244                     'Dmf': {
 245                         'mainflows': [
 246                             {
 247                                 'name': 'Basic UDP',
 248                                 'library': 'test'
 249                             }
 250                         ],
 251                         'class': 'Dmf',
 252                         'instanceGroups': [
 253                             {
 254                                 'startPaused': "false",
 255                                 'rate': 0,
 256                                 'mainflowIdx': 0,
 257                                 'mixType': ''
 258                             }
 259                         ]
 260                     },
 261                     'S5Protocol': 'GTPv2',
 262                     'DataUserCfgFileEn': 'false',
 263                     'PgwUserSutEn': 'false',
 264                     'MmeControlAddr': {
 265                         'numLinksOrNodes': 1,
 266                         'phy': 'eth1',
 267                         'forcedEthInterface': '',
 268                         'ip': 'MME_CONTROL_IP',
 269                         'class': 'TestNode',
 270                         'ethStatsEnabled': "false",
 271                         'mtu': 1500
 272                     },
 273                     'SgwUserSut': {
 274                         'class': 'Sut',
 275                         'name': 'SGW_USER_NAME'
 276                     },
 277                     'TestActivity': 'Capacity Test',
 278                     'NetworkHostAddrLocal': {
 279                         'numLinksOrNodes': 1,
 280                         'phy': 'eth2',
 281                         'forcedEthInterface': '',
 282                         'ip': 'NET_HOST_IP',
 283                         'class': 'TestNode',
 284                         'ethStatsEnabled': "false",
 285                         'mtu': 1500
 286                     },
 287                     'DedicatedsPerDefaultBearer': '0',
 288                     'DisconnectRate': '1000.0',
 289                     'Sessions': '100000',
 290                     'SgwSut': {
 291                         'class': 'Sut',
 292                         'name': 'SGW_CONTROL_NAME'
 293                     },
 294                     'TrafficMtu': '1500',
 295                     'Gtp2Version': '13.6.0',
 296                     'Gtp2Imei': '50502410121507',
 297                     'PgwNodeEn': 'false',
 298                     'StartRate': '1000.0',
 299                     'PgwV4Sut': {
 300                         'class': 'Sut',
 301                         'name': 'PGW_SUT_NAME'
 302                     },
 303                     'DefaultBearers': '1',
 304                     'EnbUserAddr': {
 305                         'numLinksOrNodes': 1,
 306                         'phy': 'eth1',
 307                         'forcedEthInterface': '',
 308                         'ip': 'ENB_USER_IP',
 309                         'class': 'TestNode',
 310                         'ethStatsEnabled': "false",
 311                         'mtu': 1500
 312                     },
 313                     'TestType': 'SGW-NODAL'
 314                 }
 315             }],
 316             'tsId': TEST_SERVERS[1]['name']
 317         }
 318     ]
 319 }
 320
 321
 322 class TestLandslideTrafficGen(unittest.TestCase):
 323     SCENARIO_CFG = {
 324         'session_profile': '/traffic_profiles/landslide/'
 325                            'landslide_session_default_bearer.yaml',
 326         'task_path': '',
 327         'runner': {
 328             'type': 'Iteration',
 329             'iterations': 1
 330         },
 331         'nodes': {
 332             'tg__0': 'tg__0.traffic_gen',
 333             'vnf__0': 'vnf__0.vnf_epc'
 334         },
 335         'topology': 'landslide_tg_topology.yaml',
 336         'type': 'NSPerf',
 337         'traffic_profile': '../../traffic_profiles/landslide/'
 338                            'landslide_dmf_udp.yaml',
 339         'options': {
 340             'test_cases': [
 341                 {
 342                     'BearerAddrPool': '2002::2',
 343                     'type': 'SGW_Node',
 344                     'BearerV4AddrPool': '2.0.0.2',
 345                     'Sessions': '90000'
 346                 },
 347                 {
 348                     'StartRate': '900.0',
 349                     'type': 'SGW_Nodal',
 350                     'DisconnectRate': '900.0',
 351                     'Sessions': '90000'
 352                 }
 353             ],
 354             'dmf':
 355                 {
 356                     'transactionRate': 1000,
 357                     'packetSize': 512
 358                 }
 359         }
 360     }
 361
 362     CONTEXT_CFG = {
 363         'contexts': [
 364             {
 365                 'type': 'Node',
 366                 'name': 'traffic_gen',
 367                 'file': '/etc/yardstick/nodes/pod_landslide.yaml'
 368             },
 369             {
 370                 'type': 'Node',
 371                 'name': 'vnf_epc',
 372                 'file': '/etc/yardstick/nodes/pod_vepc_sut.yaml'
 373             }
 374         ]
 375     }
 376
 377     TRAFFIC_PROFILE = {
 378         "schema": "nsb:traffic_profile:0.1",
 379         "name": "LandslideProfile",
 380         "description": "Spirent Landslide traffic profile",
 381         "traffic_profile": {
 382             "traffic_type": "LandslideProfile"
 383         },
 384         "dmf_config": {
 385             "dmf": {
 386                 "library": "test",
 387                 "name": "Basic UDP"
 388             },
 389             "description": "Basic data flow using UDP/IP",
 390             "keywords": "UDP",
 391             "dataProtocol": "udp"
 392         }
 393     }
 394
 395     SUCCESS_CREATED_CODE = 201
 396     SUCCESS_OK_CODE = 200
 397     SUCCESS_RECORD_ID = 5
 398     TEST_USER_ID = 11
 399
 400     def setUp(self):
 401         self._id = uuid.uuid1().int
 402
 403         self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
 404         self.mock_lsapi.start()
 405
 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
 413
 414         self.addCleanup(self._cleanup)
 415
 416     def _cleanup(self):
 417         self.mock_lsapi.stop()
 418         self.mock_ssh_helper.stop()
 419
 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)
 433
 434     def test_listen_traffic(self):
 435         _traffic_profile = {}
 436         self.assertIsNone(self.ls_tg.listen_traffic(_traffic_profile))
 437
 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()
 442
 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,
 453                                                  self.CONTEXT_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()
 460
 461     @mock.patch.object(tg_landslide.LandslideResourceHelper,
 462                        'get_running_tests')
 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)
 489
 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()
 497
 498     def test_wait_for_instantiate(self):
 499         self.assertIsNone(self.ls_tg.wait_for_instantiate())
 500         self.ls_tg.wait_for_instantiate()
 501
 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)
 507
 508     def test__update_session_suts(self, *args):
 509
 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]
 513             """
 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]
 519
 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.
 523             """
 524             for sut in suts:
 525                 if sut.get('role') == role:
 526                     return sut[key]
 527
 528         # TestNode to verify
 529         testnode_role = 'SgwControlAddr'
 530         # SUT to verify
 531         sut_role = 'SgwUserSut'
 532
 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))
 538
 539         # Verify TestNode class objects keys were updated
 540         for _key in {'ip', 'phy', 'nextHop'}:
 541             self.assertEqual(
 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
 545         self.assertEqual(
 546             get_testnode_param(sut_role, 'name', self.ls_tg.session_profile),
 547             get_sut_param(sut_role, 'name', TS2_SUTS))
 548
 549     def test__update_session_test_servers(self, *args):
 550         for ts_index, ts in enumerate(TEST_SERVERS):
 551             self.assertIsNone(
 552                 self.ls_tg._update_session_test_servers(ts, ts_index))
 553         # Verify preResolvedArpAddress key was added
 554         self.assertTrue(any(
 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'))
 562
 563     def test__update_session_tc_params_assoc_phys(self):
 564         _tc_options = {'AssociatedPhys': 'eth1'}
 565         _testcase = {}
 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)
 570
 571     def test__update_session_tc_params(self, *args):
 572
 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]
 576             """
 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)
 581
 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):
 587             self.assertEqual(
 588                 tc,
 589                 self.ls_tg._update_session_tc_params(tc_options, tc))
 590
 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():
 595                 if _key != 'type':
 596                     self.assertEqual(
 597                         _val,
 598                         get_session_tc_param_value(_key, _tc.get('type'),
 599                                                    self.ls_tg.session_profile))
 600
 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 = self.SCENARIO_CFG
 613         mock_yaml_load.return_value = SESSION_PROFILE
 614         self.assertIsNone(self.ls_tg._load_session_profile())
 615         self.assertIsNotNone(self.ls_tg.session_profile)
 616         # Number of blocks in configuration files
 617         # Number of test servers, suts and tc params blocks should be equal
 618         _config_files_blocks_num = len([item['test_server']
 619                                         for item in self.vnfd['config']])
 620         self.assertEqual(_config_files_blocks_num,
 621                          mock_upd_ses_ts.call_count)
 622         self.assertEqual(_config_files_blocks_num,
 623                          mock_upd_ses_suts.call_count)
 624         self.assertEqual(_config_files_blocks_num,
 625                          mock_upd_ses_tc_params.call_count)
 626
 627     @mock.patch.object(common_utils, 'open_relative_file')
 628     @mock.patch.object(yaml_loader, 'yaml_load')
 629     def test__load_session_profile_unequal_num_of_cfg_blocks(
 630             self, mock_yaml_load, *args):
 631         vnfd = copy.deepcopy(VNFD['vnfd:vnfd-catalog']['vnfd'][0])
 632         ls_traffic_gen = tg_landslide.LandslideTrafficGen(NAME, vnfd, self._id)
 633         ls_traffic_gen.scenario_helper.scenario_cfg = self.SCENARIO_CFG
 634         mock_yaml_load.return_value = SESSION_PROFILE
 635         # Delete test_servers item from pod file to make it not valid
 636         ls_traffic_gen.vnfd_helper['config'].pop()
 637         with self.assertRaises(RuntimeError):
 638             ls_traffic_gen._load_session_profile()
 639
 640     @mock.patch.object(common_utils, 'open_relative_file')
 641     @mock.patch.object(yaml_loader, 'yaml_load')
 642     def test__load_session_profile_test_type_mismatch(self, mock_yaml_load,
 643                                                       *args):
 644         vnfd = copy.deepcopy(VNFD['vnfd:vnfd-catalog']['vnfd'][0])
 645         # Swap test servers data in pod file
 646         vnfd['config'] = list(reversed(vnfd['config']))
 647         ls_tg = tg_landslide.LandslideTrafficGen(NAME, vnfd, self._id)
 648         ls_tg.scenario_helper.scenario_cfg = self.SCENARIO_CFG
 649         mock_yaml_load.return_value = SESSION_PROFILE
 650         with self.assertRaises(RuntimeError):
 651             ls_tg._load_session_profile()
 652
 653
 654 class TestLandslideResourceHelper(unittest.TestCase):
 655
 656     PROTO_PORT = 8080
 657     EXAMPLE_URL = ''.join([TAS_INFO['proto'], '://', TAS_INFO['ip'], ':',
 658                            str(PROTO_PORT), '/api/'])
 659     SUCCESS_CREATED_CODE = 201
 660     SUCCESS_OK_CODE = 200
 661     INVALID_REST_CODE = '400'
 662     NOT_MODIFIED_CODE = 500810
 663     ERROR_CODE = 500800
 664     SUCCESS_RECORD_ID = 11
 665     EXPIRE_DATE = '2020/01/01 12:00 FLE Standard Time'
 666     TEST_USER = 'test'
 667     TEST_TERMINATED = 1
 668     AUTH_DATA = {'user': TAS_INFO['user'], 'password': TAS_INFO['password']}
 669     TEST_SESSION_NAME = 'default_bearer_capacity'
 670
 671     USERS_DATA = {
 672         "users": [{
 673             "url": ''.join([EXAMPLE_URL, 'users/', str(SUCCESS_RECORD_ID)]),
 674             "id": SUCCESS_RECORD_ID,
 675             "level": 1,
 676             "username": TEST_USER
 677         }]
 678     }
 679
 680     CREATE_USER_DATA = {'username': TAS_INFO['user'],
 681                         'expiresOn': EXPIRE_DATE,
 682                         'level': 1,
 683                         'contactInformation': '',
 684                         'fullName': 'Test User',
 685                         'password': TAS_INFO['password'],
 686                         'isActive': 'true'}
 687
 688     SUTS_DATA = {
 689         "suts": [
 690             {
 691                 "url": ''.join([EXAMPLE_URL, 'suts/', str(SUCCESS_RECORD_ID)]),
 692                 "id": SUCCESS_RECORD_ID,
 693                 "name": "10.41.32.1"
 694             }]}
 695
 696     TEST_SERVERS_DATA = {
 697         "testServers": [
 698             {
 699                 "url": ''.join([EXAMPLE_URL, "testServers/1"]),
 700                 "id": 1,
 701                 "name": TEST_SERVERS[0]['name'],
 702                 "state": "READY",
 703                 "version": "16.4.0.10"
 704             },
 705             {
 706                 "url": ''.join([EXAMPLE_URL, "testServers/2"]),
 707                 "id": 2,
 708                 "name": TEST_SERVERS[1]['name'],
 709                 "state": "READY",
 710                 "version": "16.4.0.10"
 711             }
 712
 713         ]
 714     }
 715
 716     RUN_ID = 3
 717
 718     RUNNING_TESTS_DATA = {
 719         "runningTests": [{
 720             "url": ''.join([EXAMPLE_URL, "runningTests/{}".format(RUN_ID)]),
 721             "measurementsUrl": ''.join(
 722                 [EXAMPLE_URL,
 723                  "runningTests/{}/measurements".format(RUN_ID)]),
 724             "criteriaUrl": ''.join(
 725                 [EXAMPLE_URL,
 726                  "runningTests/{}/criteria".format(RUN_ID)]),
 727             "noteToUser": "",
 728             "id": RUN_ID,
 729             "library": SUCCESS_RECORD_ID,
 730             "name": "default_bearer_capacity",
 731             "user": TEST_USER,
 732             "criteriaStatus": "NA",
 733             "testStateOrStep": "COMPLETE"
 734         }]}
 735
 736     TEST_RESULTS_DATA = {
 737         "interval": 0,
 738         "elapsedTime": 138,
 739         "actualTime": 1521548057296,
 740         "iteration": 1,
 741         "tabs": {
 742             "Test Summary": {
 743                 "Start Time": "Tue Mar 20 07:11:55 CDT 2018",
 744                 "Actual Dedicated Bearer Session Connects": "100",
 745                 "Actual Dedicated Bearer Session Disconnects": "100",
 746                 "Actual Disconnect Rate(Sessions / Second)(P - I)": "164.804",
 747                 "Average Session Disconnect Time(P - I)": "5.024 s",
 748                 "Total Data Sent + Received Packets / Sec(P - I)": "1,452.294"
 749             }}}
 750
 751     def setUp(self):
 752         self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
 753         self.mock_lsapi.start()
 754
 755         mock_env_helper = mock.Mock()
 756         self.res_helper = tg_landslide.LandslideResourceHelper(mock_env_helper)
 757         self.res_helper._url = EXAMPLE_URL
 758
 759         self.addCleanup(self._cleanup)
 760
 761     def _cleanup(self):
 762         self.mock_lsapi.stop()
 763         self.res_helper._url = None
 764
 765     def test___init__(self, *args):
 766         self.assertIsInstance(self.res_helper,
 767                               tg_landslide.LandslideResourceHelper)
 768         self.assertEqual({}, self.res_helper._result)
 769         self.assertIsNone(self.res_helper.run_id)
 770
 771     @mock.patch.object(tg_landslide.LandslideResourceHelper,
 772                        'stop_running_tests')
 773     @mock.patch.object(tg_landslide.LandslideResourceHelper,
 774                        'get_running_tests')
 775     def test_abort_running_tests_no_running_tests(self, mock_get_tests,
 776                                                   mock_stop_tests, *args):
 777         tests_data = [{'id': self.SUCCESS_RECORD_ID,
 778                        'testStateOrStep': 'COMPLETE'}]
 779         mock_get_tests.return_value = tests_data
 780         self.assertIsNone(self.res_helper.abort_running_tests())
 781         mock_stop_tests.assert_not_called()
 782
 783     @mock.patch.object(time, 'sleep')
 784     @mock.patch.object(tg_landslide.LandslideResourceHelper,
 785                        'stop_running_tests')
 786     @mock.patch.object(tg_landslide.LandslideResourceHelper,
 787                        'get_running_tests')
 788     def test_abort_running_tests(self, mock_get_tests, mock_stop_tests, *args):
 789         test_states_seq = iter(['RUNNING', 'COMPLETE'])
 790
 791         def configure_mock(*args):
 792             return [{'id': self.SUCCESS_RECORD_ID,
 793                      'testStateOrStep': next(test_states_seq)}]
 794
 795         mock_get_tests.side_effect = configure_mock
 796         self.assertIsNone(self.res_helper.abort_running_tests())
 797         mock_stop_tests.assert_called_once_with(
 798             running_test_id=self.SUCCESS_RECORD_ID,
 799             force=True)
 800         self.assertEqual(2, mock_get_tests.call_count)
 801
 802     @mock.patch.object(tg_landslide.LandslideResourceHelper,
 803                        'stop_running_tests')
 804     @mock.patch.object(tg_landslide.LandslideResourceHelper,
 805                        'get_running_tests')
 806     def test_abort_running_tests_error(self, mock_get_tests, mock_stop_tests,
 807                                        *args):
 808         tests_data = {'id': self.SUCCESS_RECORD_ID,
 809                       'testStateOrStep': 'RUNNING'}
 810         mock_get_tests.return_value = [tests_data]
 811         with self.assertRaises(RuntimeError):
 812             self.res_helper.abort_running_tests(timeout=1, delay=1)
 813         mock_stop_tests.assert_called_with(
 814             running_test_id=self.SUCCESS_RECORD_ID,
 815             force=True)
 816
 817     def test__build_url(self, *args):
 818         resource = 'users'
 819         action = {'action': 'userCreate'}
 820         expected_url = ''.join([EXAMPLE_URL, 'users?action=userCreate'])
 821         self.assertEqual(expected_url,
 822                          self.res_helper._build_url(resource, action))
 823
 824     def test__build_url_error(self, *args):
 825         resource = ''
 826         action = {'action': 'userCreate'}
 827
 828         with self.assertRaises(ValueError):
 829             self.res_helper._build_url(resource, action)
 830
 831     def test_get_response_params(self, *args):
 832         method = 'get'
 833         resource = 'users'
 834         mock_session = mock.Mock(spec=requests.Session)
 835         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
 836                          'json.return_value': self.USERS_DATA}
 837         mock_session.get.return_value.configure_mock(**get_resp_data)
 838         self.res_helper.session = mock_session
 839         resp = self.res_helper.get_response_params(method, resource)
 840         self.assertTrue(resp)
 841
 842     @mock.patch.object(tg_landslide.LandslideResourceHelper, '_get_users')
 843     @mock.patch.object(time, 'time')
 844     def test__create_user(self, mock_time, mock_get_users, *args):
 845         mock_time.strftime.return_value = self.EXPIRE_DATE
 846         post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
 847                           'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
 848         mock_session = mock.Mock(spec=requests.Session)
 849         mock_session.post.return_value.configure_mock(**post_resp_data)
 850         self.res_helper.session = mock_session
 851         self.assertEqual(self.SUCCESS_RECORD_ID,
 852                          self.res_helper._create_user(self.AUTH_DATA))
 853         mock_get_users.assert_not_called()
 854
 855     @mock.patch.object(tg_landslide.LandslideResourceHelper, '_modify_user')
 856     @mock.patch.object(time, 'time')
 857     def test__create_user_username_exists(self, mock_time, mock_modify_user,
 858                                           *args):
 859         mock_time.strftime.return_value = self.EXPIRE_DATE
 860         mock_modify_user.return_value = {'id': self.SUCCESS_RECORD_ID,
 861                                          'result': 'No changes requested'}
 862         post_resp_data = {
 863             'status_code': self.ERROR_CODE,
 864             'json.return_value': {'id': self.SUCCESS_OK_CODE,
 865                                   'apiCode': self.NOT_MODIFIED_CODE}}
 866         mock_session = mock.Mock(spec=requests.Session)
 867         mock_session.post.return_value.configure_mock(**post_resp_data)
 868         self.res_helper.session = mock_session
 869         res = self.res_helper._create_user(self.AUTH_DATA)
 870         mock_modify_user.assert_called_once_with(TAS_INFO['user'],
 871                                                  {'isActive': 'true'})
 872         self.assertEqual(self.SUCCESS_RECORD_ID, res)
 873
 874     @mock.patch.object(time, 'time')
 875     def test__create_user_error(self, mock_time, *args):
 876         mock_time.strftime.return_value = self.EXPIRE_DATE
 877         mock_session = mock.Mock(spec=requests.Session)
 878         post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
 879                           'json.return_value': {'apiCode': self.ERROR_CODE}}
 880         mock_session.post.return_value.configure_mock(**post_resp_data)
 881         self.res_helper.session = mock_session
 882         with self.assertRaises(exceptions.RestApiError):
 883             self.res_helper._create_user(self.AUTH_DATA)
 884
 885     def test__modify_user(self, *args):
 886         post_data = {'username': 'test_user'}
 887         mock_session = mock.Mock(spec=requests.Session)
 888         post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
 889                           'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
 890         mock_session.post.return_value.configure_mock(**post_resp_data)
 891         self.res_helper.session = mock_session
 892         res = self.res_helper._modify_user(username=self.TEST_USER,
 893                                            fields=post_data)
 894         self.assertEqual(self.SUCCESS_RECORD_ID, res['id'])
 895
 896     def test__modify_user_rest_resp_fail(self, *args):
 897         post_data = {'non-existing-key': ''}
 898         mock_session = mock.Mock(spec=requests.Session)
 899         mock_session.post.ok = False
 900         self.res_helper.session = mock_session
 901         self.assertRaises(exceptions.RestApiError,
 902                           self.res_helper._modify_user,
 903                           username=self.TEST_USER, fields=post_data)
 904         mock_session.post.assert_called_once()
 905
 906     def test__delete_user(self, *args):
 907         mock_session = mock.Mock(spec=requests.Session)
 908         self.res_helper.session = mock_session
 909         self.assertIsNone(self.res_helper._delete_user(
 910             username=self.TEST_USER))
 911
 912     def test__get_users(self, *args):
 913         mock_session = mock.Mock(spec=requests.Session)
 914         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
 915                          'json.return_value': self.USERS_DATA}
 916         mock_session.get.return_value.configure_mock(**get_resp_data)
 917         self.res_helper.session = mock_session
 918         self.assertEqual(self.USERS_DATA['users'],
 919                          self.res_helper._get_users())
 920
 921     def test_exec_rest_request(self, *args):
 922         resource = 'testServers'
 923         action = {'action': 'modify'}
 924         expected_url = ''.join([EXAMPLE_URL, 'testServers?action=modify'])
 925         post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
 926                           'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
 927         mock_session = mock.Mock(spec=requests.Session)
 928         mock_session.post.return_value.configure_mock(**post_resp_data)
 929         self.res_helper.session = mock_session
 930         self.res_helper.exec_rest_request('post', resource, action)
 931         self.res_helper.session.post.assert_called_once_with(expected_url,
 932                                                              json={})
 933
 934     def test_exec_rest_request_unsupported_method_error(self, *args):
 935         resource = 'testServers'
 936         action = {'action': 'modify'}
 937         with self.assertRaises(ValueError):
 938             self.res_helper.exec_rest_request('patch', resource, action)
 939
 940     def test_exec_rest_request_missed_action_arg(self, *args):
 941         resource = 'testServers'
 942         with self.assertRaises(ValueError):
 943             self.res_helper.exec_rest_request('post', resource)
 944
 945     def test_exec_rest_request_raise_exc(self):
 946         resource = 'users'
 947         action = {'action': 'modify'}
 948         post_resp_data = {'status_code': self.ERROR_CODE,
 949                           'json.return_value': {
 950                               'status_code': self.ERROR_CODE}}
 951         mock_session = mock.Mock(spec=requests.Session)
 952         mock_session.post.return_value.configure_mock(**post_resp_data)
 953         self.assertRaises(exceptions.RestApiError,
 954                           self.res_helper.exec_rest_request,
 955                           'post', resource, action, raise_exc=True)
 956
 957     @mock.patch.object(time, 'time')
 958     def test_connect(self, mock_time, *args):
 959         vnfd = VNFD['vnfd:vnfd-catalog']['vnfd'][0]
 960         mock_time.strftime.return_value = self.EXPIRE_DATE
 961         self.res_helper.vnfd_helper = vnfd
 962
 963         self.res_helper._tcl = mock.Mock()
 964         post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
 965                           'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
 966         mock_session = mock.Mock(spec=requests.Session, headers={})
 967         mock_session.post.return_value.configure_mock(**post_resp_data)
 968         self.res_helper.session = mock_session
 969         self.assertIsInstance(self.res_helper.connect(), requests.Session)
 970         self.res_helper._tcl.connect.assert_called_once_with(
 971             TAS_INFO['ip'],
 972             TAS_INFO['user'],
 973             TAS_INFO['password'])
 974
 975     def test_disconnect(self, *args):
 976         self.res_helper._tcl = mock.Mock()
 977         self.assertIsNone(self.res_helper.disconnect())
 978         self.assertIsNone(self.res_helper.session)
 979         self.res_helper._tcl.disconnect.assert_called_once()
 980
 981     def test_terminate(self, *args):
 982         self.assertIsNone(self.res_helper.terminate())
 983         self.assertEqual(self.TEST_TERMINATED,
 984                          self.res_helper._terminated.value)
 985
 986     def test_create_dmf(self, *args):
 987         self.res_helper._tcl = mock.Mock()
 988         self.assertIsNone(self.res_helper.create_dmf(DMF_CFG))
 989         self.res_helper._tcl.create_dmf.assert_called_once_with(DMF_CFG)
 990
 991     def test_create_dmf_as_list(self, *args):
 992         self.res_helper._tcl = mock.Mock()
 993         self.assertIsNone(self.res_helper.create_dmf([DMF_CFG]))
 994         self.res_helper._tcl.create_dmf.assert_called_once_with(DMF_CFG)
 995
 996     def test_delete_dmf(self, *args):
 997         self.res_helper._tcl = mock.Mock()
 998         self.assertIsNone(self.res_helper.delete_dmf(DMF_CFG))
 999         self.res_helper._tcl.delete_dmf.assert_called_once_with(DMF_CFG)
1000
1001     def test_delete_dmf_as_list(self, *args):
1002         self.res_helper._tcl = mock.Mock()
1003         self.assertIsNone(self.res_helper.delete_dmf([DMF_CFG]))
1004         self.res_helper._tcl.delete_dmf.assert_called_once_with(DMF_CFG)
1005
1006     @mock.patch.object(tg_landslide.LandslideResourceHelper, 'configure_sut')
1007     def test_create_suts(self, mock_configure_sut, *args):
1008         mock_session = mock.Mock(spec=requests.Session)
1009         post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE}
1010         mock_session.post.return_value.configure_mock(**post_resp_data)
1011         self.res_helper.session = mock_session
1012         self.assertIsNone(self.res_helper.create_suts(TS1_SUTS))
1013         mock_configure_sut.assert_not_called()
1014
1015     @mock.patch.object(tg_landslide.LandslideResourceHelper, 'configure_sut')
1016     def test_create_suts_sut_exists(self, mock_configure_sut, *args):
1017         sut_name = 'test_sut'
1018         suts = [
1019             {'name': sut_name,
1020              'role': 'SgwControlAddr',
1021              'managementIp': '12.0.1.1',
1022              'ip': '10.42.32.100'
1023              }
1024         ]
1025         mock_session = mock.Mock(spec=requests.Session)
1026         post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
1027         mock_session.post.return_value.configure_mock(**post_resp_data)
1028         self.res_helper.session = mock_session
1029         self.assertIsNone(self.res_helper.create_suts(suts))
1030         mock_configure_sut.assert_called_once_with(
1031             sut_name=sut_name,
1032             json_data={k: v for k, v in suts[0].items()
1033                        if k not in {'phy', 'nextHop', 'role', 'name'}})
1034
1035     def test_get_suts(self, *args):
1036         mock_session = mock.Mock(spec=requests.Session)
1037         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1038                          'json.return_value': self.SUTS_DATA}
1039         mock_session.get.return_value.configure_mock(**get_resp_data)
1040         self.res_helper.session = mock_session
1041         self.assertIsInstance(self.res_helper.get_suts(), list)
1042
1043     def test_get_suts_single_id(self, *args):
1044         mock_session = mock.Mock(spec=requests.Session)
1045         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1046                          'json.return_value': self.SUTS_DATA['suts'][0]}
1047         mock_session.get.return_value.configure_mock(**get_resp_data)
1048         self.res_helper.session = mock_session
1049         self.assertIsInstance(self.res_helper.get_suts(suts_id=2), dict)
1050
1051     def test_configure_sut(self, *args):
1052         post_data = {'managementIp': '2.2.2.2'}
1053         mock_session = mock.Mock(spec=requests.Session)
1054         post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1055                           'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
1056         mock_session.post.return_value.configure_mock(**post_resp_data)
1057         self.res_helper.session = mock_session
1058         self.assertIsNone(self.res_helper.configure_sut('test_name',
1059                                                         post_data))
1060         mock_session.post.assert_called_once()
1061
1062     def test_configure_sut_error(self, *args):
1063         post_data = {'managementIp': '2.2.2.2'}
1064         mock_session = mock.Mock(spec=requests.Session)
1065         post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
1066         mock_session.post.return_value.configure_mock(**post_resp_data)
1067         self.res_helper.session = mock_session
1068         with self.assertRaises(exceptions.RestApiError):
1069             self.res_helper.configure_sut('test_name', post_data)
1070
1071     def test_delete_suts(self, *args):
1072         mock_session = mock.Mock(spec=requests.Session)
1073         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1074                          'json.return_value': self.SUTS_DATA}
1075         delete_resp_data = {'status_code': self.SUCCESS_OK_CODE}
1076         mock_session.get.return_value.configure_mock(**get_resp_data)
1077         mock_session.delete.return_value.configure_mock(**delete_resp_data)
1078         self.res_helper.session = mock_session
1079         self.assertIsNone(self.res_helper.delete_suts())
1080         mock_session.delete.assert_called_once()
1081
1082     @mock.patch.object(tg_landslide.LandslideResourceHelper,
1083                        'get_test_servers')
1084     def test__check_test_servers_state(self, mock_get_test_servers, *args):
1085         mock_get_test_servers.return_value = \
1086             self.TEST_SERVERS_DATA['testServers']
1087         self.res_helper._check_test_servers_state()
1088         mock_get_test_servers.assert_called_once()
1089
1090     @mock.patch.object(tg_landslide.LandslideResourceHelper,
1091                        'get_test_servers')
1092     def test__check_test_servers_state_server_not_ready(
1093             self, mock_get_test_servers, *args):
1094         test_servers_not_ready = [
1095             {
1096                 "url": ''.join([EXAMPLE_URL, "testServers/1"]),
1097                 "id": 1,
1098                 "name": "TestServer_1",
1099                 "state": "NOT_READY",
1100                 "version": "16.4.0.10"
1101             }
1102         ]
1103
1104         mock_get_test_servers.return_value = test_servers_not_ready
1105         with self.assertRaises(RuntimeError):
1106             self.res_helper._check_test_servers_state(timeout=1, delay=0)
1107
1108     @mock.patch.object(tg_landslide.LandslideResourceHelper,
1109                        '_check_test_servers_state')
1110     def test_create_test_servers(self, mock_check_ts_state, *args):
1111         test_servers_ids = [
1112             ts['id'] for ts in self.TEST_SERVERS_DATA['testServers']]
1113
1114         self.res_helper.license_data['lic_id'] = TAS_INFO['license']
1115         self.res_helper._tcl.create_test_server = mock.Mock()
1116         self.res_helper._tcl.create_test_server.side_effect = test_servers_ids
1117         self.assertIsNone(self.res_helper.create_test_servers(TEST_SERVERS))
1118         mock_check_ts_state.assert_called_once_with(test_servers_ids)
1119
1120     @mock.patch.object(tg_landslide.LandslideTclClient,
1121                        'resolve_test_server_name')
1122     @mock.patch.object(tg_landslide.LsTclHandler, 'execute')
1123     def test_create_test_servers_error(self, mock_execute,
1124                                        mock_resolve_ts_name, *args):
1125         self.res_helper.license_data['lic_id'] = TAS_INFO['license']
1126         # Return message for case test server wasn't created
1127         mock_execute.return_value = 'TS not found'
1128         # Return message for case test server name wasn't resolved
1129         mock_resolve_ts_name.return_value = 'TS not found'
1130         with self.assertRaises(RuntimeError):
1131             self.res_helper.create_test_servers(TEST_SERVERS)
1132
1133     def test_get_test_servers(self, *args):
1134         mock_session = mock.Mock(spec=requests.Session)
1135         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1136                          'json.return_value': self.TEST_SERVERS_DATA}
1137         mock_session.get.return_value.configure_mock(**get_resp_data)
1138         self.res_helper.session = mock_session
1139         res = self.res_helper.get_test_servers()
1140         self.assertEqual(self.TEST_SERVERS_DATA['testServers'], res)
1141
1142     def test_get_test_servers_by_id(self, *args):
1143         mock_session = mock.Mock(spec=requests.Session)
1144
1145         _ts = self.TEST_SERVERS_DATA['testServers'][0]
1146         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1147                          'json.return_value': _ts}
1148         mock_session.get.return_value.configure_mock(**get_resp_data)
1149         self.res_helper.session = mock_session
1150         res = self.res_helper.get_test_servers(test_server_ids=[_ts['id']])
1151         self.assertEqual([_ts], res)
1152
1153     def test_configure_test_servers(self, *args):
1154         mock_session = mock.Mock(spec=requests.Session)
1155         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1156                          'json.return_value': self.TEST_SERVERS_DATA}
1157         mock_session.get.return_value.configure_mock(**get_resp_data)
1158         self.res_helper.session = mock_session
1159         res = self.res_helper.configure_test_servers(
1160             action={'action': 'recycle'})
1161         self.assertEqual(
1162             [x['id'] for x in self.TEST_SERVERS_DATA['testServers']],
1163             res)
1164         self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']),
1165                          mock_session.post.call_count)
1166
1167     def test_delete_test_servers(self, *args):
1168         mock_session = mock.Mock(spec=requests.Session)
1169         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1170                          'json.return_value': self.TEST_SERVERS_DATA}
1171         mock_session.get.return_value.configure_mock(**get_resp_data)
1172         self.res_helper.session = mock_session
1173         self.assertIsNone(self.res_helper.delete_test_servers())
1174         self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']),
1175                          mock_session.delete.call_count)
1176
1177     def test_create_test_session_res_helper(self, *args):
1178         self.res_helper._user_id = self.SUCCESS_RECORD_ID
1179         self.res_helper._tcl = mock.Mock()
1180         test_session = {'name': 'test'}
1181         self.assertIsNone(self.res_helper.create_test_session(test_session))
1182         self.res_helper._tcl.create_test_session.assert_called_once_with(
1183             {'name': 'test', 'library': self.SUCCESS_RECORD_ID})
1184
1185     @mock.patch.object(tg_landslide.LandslideTclClient,
1186                        'resolve_test_server_name',
1187                        return_value='Not Found')
1188     def test_create_test_session_ts_name_not_found(self, *args):
1189         self.res_helper._user_id = self.SUCCESS_RECORD_ID
1190         test_session = {
1191             'duration': 60,
1192             'description': 'UE default bearer creation test case',
1193             'name': 'default_bearer_capacity',
1194             'tsGroups': [{'testCases': [{'type': 'SGW_Node',
1195                                          'name': ''}],
1196                           'tsId': 'TestServer_3'}]
1197         }
1198         with self.assertRaises(RuntimeError):
1199             self.res_helper.create_test_session(test_session)
1200
1201     def test_get_test_session(self, *args):
1202         test_session = {"name": self.TEST_SESSION_NAME}
1203         self.res_helper._user_id = self.SUCCESS_RECORD_ID
1204         mock_session = mock.Mock(spec=requests.Session)
1205         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1206                          'json.return_value': test_session}
1207         mock_session.get.return_value.configure_mock(**get_resp_data)
1208         self.res_helper.session = mock_session
1209         res = self.res_helper.get_test_session(self.TEST_SESSION_NAME)
1210         self.assertEqual(test_session, res)
1211
1212     def test_configure_test_session(self, *args):
1213         test_session = {'name': self.TEST_SESSION_NAME}
1214         self.res_helper._user_id = self.SUCCESS_RECORD_ID
1215         self.res_helper.user_lib_uri = 'libraries/{{}}/{}'.format(
1216             self.res_helper.test_session_uri)
1217         mock_session = mock.Mock(spec=requests.Session)
1218         self.res_helper.session = mock_session
1219         res = self.res_helper.configure_test_session(self.TEST_SESSION_NAME,
1220                                                      test_session)
1221         self.assertIsNotNone(res)
1222         mock_session.post.assert_called_once()
1223
1224     def test_delete_test_session(self, *args):
1225         self.res_helper._user_id = self.SUCCESS_RECORD_ID
1226         self.res_helper.user_lib_uri = 'libraries/{{}}/{}'.format(
1227             self.res_helper.test_session_uri)
1228         mock_session = mock.Mock(spec=requests.Session)
1229         self.res_helper.session = mock_session
1230         res = self.res_helper.delete_test_session(self.TEST_SESSION_NAME)
1231         self.assertIsNotNone(res)
1232         mock_session.delete.assert_called_once()
1233
1234     def test_create_running_tests(self, *args):
1235         self.res_helper._user_id = self.SUCCESS_RECORD_ID
1236         test_session = {'id': self.SUCCESS_RECORD_ID}
1237         mock_session = mock.Mock(spec=requests.Session)
1238         post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
1239                           'json.return_value': test_session}
1240         mock_session.post.return_value.configure_mock(**post_resp_data)
1241         self.res_helper.session = mock_session
1242         self.res_helper.create_running_tests(self.TEST_SESSION_NAME)
1243         self.assertEqual(self.SUCCESS_RECORD_ID, self.res_helper.run_id)
1244
1245     def test_create_running_tests_error(self, *args):
1246         self.res_helper._user_id = self.SUCCESS_RECORD_ID
1247         mock_session = mock.Mock(spec=requests.Session)
1248         post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
1249         mock_session.post.return_value.configure_mock(**post_resp_data)
1250         self.res_helper.session = mock_session
1251         with self.assertRaises(exceptions.RestApiError):
1252             self.res_helper.create_running_tests(self.TEST_SESSION_NAME)
1253
1254     def test_get_running_tests(self, *args):
1255         mock_session = mock.Mock(spec=requests.Session)
1256         get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1257                          'json.return_value': self.RUNNING_TESTS_DATA}
1258         mock_session.get.return_value.configure_mock(**get_resp_data)
1259         self.res_helper.session = mock_session
1260         res = self.res_helper.get_running_tests()
1261         self.assertEqual(self.RUNNING_TESTS_DATA['runningTests'], res)
1262
1263     def test_delete_running_tests(self, *args):
1264         mock_session = mock.Mock(spec=requests.Session)
1265         delete_resp_data = {'status_code': self.SUCCESS_OK_CODE,
1266                             'json.return_value': self.RUNNING_TESTS_DATA}
1267         mock_session.delete.return_value.configure_mock(**delete_resp_data)
1268         self.res_helper.session = mock_session
1269         self.assertIsNone(self.res_helper.delete_running_tests())
1270
1271     def test__running_tests_action(self, *args):
1272         action = 'abort'
1273         mock_session = mock.Mock(spec=requests.Session)
1274         self.res_helper.session = mock_session
1275         res = self.res_helper._running_tests_action(self.SUCCESS_RECORD_ID,
1276                                                     action)
1277         self.assertIsNone(res)
1278
1279     @mock.patch.object(tg_landslide.LandslideResourceHelper,
1280                        '_running_tests_action')
1281     def test_stop_running_tests(self, mock_tests_action, *args):
1282         res = self.res_helper.stop_running_tests(self.SUCCESS_RECORD_ID)
1283         self.assertIsNone(res)
1284         mock_tests_action.assert_called_once()
1285
1286     def test_check_running_test_state(self, *args):
1287         mock_session = mock.Mock(spec=requests.Session)
1288         get_resp_data = {
1289             'status_code': self.SUCCESS_OK_CODE,
1290             'json.return_value': self.RUNNING_TESTS_DATA["runningTests"][0]}
1291         mock_session.get.return_value.configure_mock(**get_resp_data)
1292         self.res_helper.session = mock_session
1293         res = self.res_helper.check_running_test_state(self.SUCCESS_RECORD_ID)
1294         self.assertEqual(
1295             self.RUNNING_TESTS_DATA["runningTests"][0]['testStateOrStep'],
1296             res)
1297
1298     def test_get_running_tests_results(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.TEST_RESULTS_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_results(
1305             self.SUCCESS_RECORD_ID)
1306         self.assertEqual(self.TEST_RESULTS_DATA, res)
1307
1308     def test__write_results(self, *args):
1309         res = self.res_helper._write_results(self.TEST_RESULTS_DATA)
1310         exp_res = {
1311             "Test Summary::Actual Dedicated Bearer Session Connects": 100.0,
1312             "Test Summary::Actual Dedicated Bearer Session Disconnects": 100.0,
1313             "Test Summary::Actual Disconnect Rate(Sessions / Second)(P - I)": 164.804,
1314             "Test Summary::Average Session Disconnect Time(P - I)": 5.024,
1315             "Test Summary::Total Data Sent + Received Packets / Sec(P - I)": 1452.294
1316         }
1317         self.assertEqual(exp_res, res)
1318
1319     def test__write_results_no_tabs(self, *args):
1320         _res_data = copy.deepcopy(self.TEST_RESULTS_DATA)
1321         del _res_data['tabs']
1322         # Return None if tabs not found in test results dict
1323         self.assertIsNone(self.res_helper._write_results(_res_data))
1324
1325     @mock.patch.object(tg_landslide.LandslideResourceHelper,
1326                        'check_running_test_state')
1327     @mock.patch.object(tg_landslide.LandslideResourceHelper,
1328                        'get_running_tests_results')
1329     def test_collect_kpi_test_running(self, mock_tests_results,
1330                                       mock_tests_state, *args):
1331         self.res_helper.run_id = self.SUCCESS_RECORD_ID
1332         mock_tests_state.return_value = 'RUNNING'
1333         mock_tests_results.return_value = self.TEST_RESULTS_DATA
1334         res = self.res_helper.collect_kpi()
1335         self.assertNotIn('done', res)
1336         mock_tests_state.assert_called_once_with(self.res_helper.run_id)
1337         mock_tests_results.assert_called_once_with(self.res_helper.run_id)
1338
1339     @mock.patch.object(tg_landslide.LandslideResourceHelper,
1340                        'check_running_test_state')
1341     @mock.patch.object(tg_landslide.LandslideResourceHelper,
1342                        'get_running_tests_results')
1343     def test_collect_kpi_test_completed(self, mock_tests_results,
1344                                         mock_tests_state, *args):
1345         self.res_helper.run_id = self.SUCCESS_RECORD_ID
1346         mock_tests_state.return_value = 'COMPLETE'
1347         res = self.res_helper.collect_kpi()
1348         self.assertIsNotNone(res)
1349         mock_tests_state.assert_called_once_with(self.res_helper.run_id)
1350         mock_tests_results.assert_not_called()
1351         self.assertDictContainsSubset({'done': True}, res)
1352
1353
1354 class TestLandslideTclClient(unittest.TestCase):
1355     def setUp(self):
1356         self.mock_tcl_handler = mock.Mock(spec=tg_landslide.LsTclHandler)
1357         self.ls_res_helper = mock.Mock(
1358             spec=tg_landslide.LandslideResourceHelper)
1359         self.ls_tcl_client = tg_landslide.LandslideTclClient(
1360             self.mock_tcl_handler,
1361             self.ls_res_helper)
1362
1363     def test___init__(self, *args):
1364         self.ls_tcl_client = tg_landslide.LandslideTclClient(
1365             self.mock_tcl_handler,
1366             self.ls_res_helper)
1367         self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
1368         self.assertIsNone(self.ls_tcl_client._user)
1369         self.assertIsNone(self.ls_tcl_client._library_id)
1370         self.assertIsNone(self.ls_tcl_client._basic_library_id)
1371         self.assertEqual(set(), self.ls_tcl_client.ts_ids)
1372         self.assertIsInstance(self.ls_tcl_client._tc_types, set)
1373         self.assertIsNotNone(self.ls_tcl_client._tc_types)
1374
1375     def test_connect_login_success(self, *args):
1376         lib_id = '123'
1377         exec_responses = ['java0x2', lib_id, lib_id]
1378         auth = ('user', 'password')
1379         self.mock_tcl_handler.execute.side_effect = exec_responses
1380         self.ls_tcl_client.connect(TAS_INFO['ip'], *auth)
1381         self.assertEqual(lib_id, self.ls_tcl_client._library_id)
1382         self.assertEqual(lib_id, self.ls_tcl_client._basic_library_id)
1383         self.assertEqual(TAS_INFO['ip'], self.ls_tcl_client.tcl_server_ip)
1384         self.assertEqual(auth[0], self.ls_tcl_client._user)
1385         self.assertEqual(len(exec_responses),
1386                          self.mock_tcl_handler.execute.call_count)
1387
1388     def test_connect_login_failed(self, *args):
1389         exec_responses = ['Login failed']
1390         auth = ('user', 'password')
1391         self.mock_tcl_handler.execute.side_effect = exec_responses
1392         self.assertRaises(exceptions.LandslideTclException,
1393                           self.ls_tcl_client.connect,
1394                           TAS_INFO['ip'],
1395                           *auth)
1396         self.assertIsNone(self.ls_tcl_client._library_id)
1397         self.assertIsNone(self.ls_tcl_client._basic_library_id)
1398         self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
1399         self.assertIsNone(self.ls_tcl_client._user)
1400         self.assertEqual(len(exec_responses),
1401                          self.mock_tcl_handler.execute.call_count)
1402
1403     def test_disconnect(self, *args):
1404         self.ls_tcl_client.disconnect()
1405         self.mock_tcl_handler.execute.assert_called_once()
1406         self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
1407         self.assertIsNone(self.ls_tcl_client._user)
1408         self.assertIsNone(self.ls_tcl_client._library_id)
1409         self.assertIsNone(self.ls_tcl_client._basic_library_id)
1410
1411     def test_create_test_server(self, *args):
1412         return_value = '2'
1413         self.ls_tcl_client._ts_context.vnfd_helper = \
1414             VNFD['vnfd:vnfd-catalog']['vnfd'][0]
1415         self.ls_tcl_client._ts_context.license_data = {'lic_id': return_value}
1416         self.mock_tcl_handler.execute.return_value = return_value
1417         self.ls_tcl_client._set_thread_model = mock.Mock()
1418         res = self.ls_tcl_client.create_test_server(TEST_SERVERS[1])
1419         self.assertEqual(3, self.mock_tcl_handler.execute.call_count)
1420         self.ls_tcl_client._set_thread_model.assert_called_once_with(
1421             TEST_SERVERS[1]['name'],
1422             TEST_SERVERS[1]['thread_model'])
1423         self.assertEqual(int(return_value), res)
1424
1425     def test_create_test_server_fail_limit_reach(self, *args):
1426         self.mock_tcl_handler.execute.side_effect = ['TS not found',
1427                                                      'Add failed']
1428         self.assertRaises(RuntimeError,
1429                           self.ls_tcl_client.create_test_server,
1430                           TEST_SERVERS[0])
1431         self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1432
1433     def test__add_test_server(self):
1434         ts_id = '2'
1435         self.mock_tcl_handler.execute.side_effect = ['TS not found', ts_id]
1436         self.assertEqual(ts_id,
1437                          self.ls_tcl_client._add_test_server('name', 'ip'))
1438         self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1439
1440     def test__add_test_server_failed(self):
1441         self.mock_tcl_handler.execute.side_effect = ['TS not found',
1442                                                      'Add failed']
1443         self.assertRaises(RuntimeError, self.ls_tcl_client._add_test_server,
1444                           'name', 'ip')
1445         self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1446
1447     def test__update_license(self):
1448         curr_lic_id = '111'
1449         new_lic_id = '222'
1450         exec_resp = ['java0x4',
1451                      curr_lic_id,
1452                      TCL_SUCCESS_RESPONSE,
1453                      TCL_SUCCESS_RESPONSE]
1454         self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id}
1455         self.mock_tcl_handler.execute.side_effect = exec_resp
1456         self.ls_tcl_client._update_license('name')
1457         self.assertEqual(len(exec_resp),
1458                          self.mock_tcl_handler.execute.call_count)
1459
1460     def test__update_license_same_as_current(self):
1461         curr_lic_id = '111'
1462         new_lic_id = '111'
1463         exec_resp = ['java0x4', curr_lic_id]
1464         self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id}
1465         self.mock_tcl_handler.execute.side_effect = exec_resp
1466         self.ls_tcl_client._update_license('name')
1467         self.assertEqual(len(exec_resp),
1468                          self.mock_tcl_handler.execute.call_count)
1469
1470     def test__set_thread_model_update_needed(self):
1471         self.ls_tcl_client._ts_context.vnfd_helper = {
1472             'mgmt-interface': {
1473                 'cfguser_password': 'cfguser_password'
1474             }
1475         }
1476         exec_resp = ['java0x4', 'V0', '', '']
1477         self.mock_tcl_handler.execute.side_effect = exec_resp
1478         self.ls_tcl_client._set_thread_model('name', 'Fireball')
1479         self.assertEqual(len(exec_resp),
1480                          self.mock_tcl_handler.execute.call_count)
1481
1482     def test__set_thread_model_no_update_needed(self):
1483         self.ls_tcl_client._ts_context.vnfd_helper = {
1484             'mgmt-interface': {
1485                 'cfguser_password': 'cfguser_password'
1486             }
1487         }
1488         exec_resp = ['java0x4', 'V0']
1489         self.mock_tcl_handler.execute.side_effect = exec_resp
1490         self.ls_tcl_client._set_thread_model('name', 'Legacy')
1491         self.assertEqual(len(exec_resp),
1492                          self.mock_tcl_handler.execute.call_count)
1493
1494     @mock.patch.object(tg_landslide.LandslideTclClient,
1495                        'resolve_test_server_name', return_value='2')
1496     def test_create_test_session(self, *args):
1497         _session_profile = copy.deepcopy(SESSION_PROFILE)
1498         _session_profile['reservations'] = RESERVATIONS
1499         self.ls_tcl_client._save_test_session = mock.Mock()
1500         self.ls_tcl_client._configure_ts_group = mock.Mock()
1501         self.ls_tcl_client.create_test_session(_session_profile)
1502         self.assertEqual(17, self.mock_tcl_handler.execute.call_count)
1503
1504     def test_create_dmf(self):
1505         self.mock_tcl_handler.execute.return_value = '2'
1506         self.ls_tcl_client._save_dmf = mock.Mock()
1507         self.ls_tcl_client.create_dmf(DMF_CFG)
1508         self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
1509
1510     def test_configure_dmf(self):
1511         self.mock_tcl_handler.execute.return_value = '2'
1512         self.ls_tcl_client._save_dmf = mock.Mock()
1513         self.ls_tcl_client.configure_dmf(DMF_CFG)
1514         self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
1515
1516     def test_delete_dmf(self):
1517         self.assertRaises(NotImplementedError,
1518                           self.ls_tcl_client.delete_dmf,
1519                           DMF_CFG)
1520
1521     def test__save_dmf_valid(self):
1522         exec_resp = [TCL_SUCCESS_RESPONSE, TCL_SUCCESS_RESPONSE]
1523         self.mock_tcl_handler.execute.side_effect = exec_resp
1524         self.ls_tcl_client._save_dmf()
1525         self.assertEqual(len(exec_resp),
1526                          self.mock_tcl_handler.execute.call_count)
1527
1528     def test__save_dmf_invalid(self):
1529         exec_resp = ['Invalid', 'List of errors and warnings']
1530         self.mock_tcl_handler.execute.side_effect = exec_resp
1531         self.assertRaises(exceptions.LandslideTclException,
1532                           self.ls_tcl_client._save_dmf)
1533         self.assertEqual(len(exec_resp),
1534                          self.mock_tcl_handler.execute.call_count)
1535
1536     def test__configure_report_options(self):
1537         _options = {'format': 'CSV', 'PerInterval': 'false'}
1538         self.ls_tcl_client._configure_report_options(_options)
1539         self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1540
1541     def test___configure_ts_group(self, *args):
1542         _ts_group = copy.deepcopy(SESSION_PROFILE['tsGroups'][0])
1543         self.ls_tcl_client._configure_tc_type = mock.Mock()
1544         self.ls_tcl_client._configure_preresolved_arp = mock.Mock()
1545         self.ls_tcl_client.resolve_test_server_name = mock.Mock(
1546             return_value='2')
1547         self.ls_tcl_client._configure_ts_group(_ts_group, 0)
1548         self.mock_tcl_handler.execute.assert_called_once()
1549
1550     def test___configure_ts_group_resolve_ts_fail(self, *args):
1551         _ts_group = copy.deepcopy(SESSION_PROFILE['tsGroups'][0])
1552         self.ls_tcl_client._configure_tc_type = mock.Mock()
1553         self.ls_tcl_client._configure_preresolved_arp = mock.Mock()
1554         self.ls_tcl_client.resolve_test_server_name = mock.Mock(
1555             return_value='TS Not Found')
1556         self.assertRaises(RuntimeError, self.ls_tcl_client._configure_ts_group,
1557                           _ts_group, 0)
1558         self.mock_tcl_handler.execute.assert_not_called()
1559
1560     def test__configure_tc_type(self):
1561         _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1562         self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE
1563         self.ls_tcl_client._configure_parameters = mock.Mock()
1564         self.ls_tcl_client._configure_tc_type(_tc, 0)
1565         self.assertEqual(7, self.mock_tcl_handler.execute.call_count)
1566
1567     def test__configure_tc_type_optional_param_omitted(self):
1568         _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1569         del _tc['linked']
1570         self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE
1571         self.ls_tcl_client._configure_parameters = mock.Mock()
1572         self.ls_tcl_client._configure_tc_type(_tc, 0)
1573         self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
1574
1575     def test__configure_tc_type_wrong_type(self):
1576         _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1577         _tc['type'] = 'not_supported'
1578         self.ls_tcl_client._configure_parameters = mock.Mock()
1579         self.assertRaises(RuntimeError,
1580                           self.ls_tcl_client._configure_tc_type,
1581                           _tc, 0)
1582
1583     def test__configure_tc_type_not_found_basic_lib(self):
1584         _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
1585         self.ls_tcl_client._configure_parameters = mock.Mock()
1586         self.mock_tcl_handler.execute.return_value = 'Invalid'
1587         self.assertRaises(RuntimeError,
1588                           self.ls_tcl_client._configure_tc_type,
1589                           _tc, 0)
1590
1591     def test__configure_parameters(self):
1592         _params = copy.deepcopy(
1593             SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters'])
1594         self.ls_tcl_client._configure_parameters(_params)
1595         self.assertEqual(16, self.mock_tcl_handler.execute.call_count)
1596
1597     def test__configure_array_param(self):
1598         _array = {"class": "Array",
1599                   "array": ["0"]}
1600         self.ls_tcl_client._configure_array_param('name', _array)
1601         self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1602
1603     def test__configure_test_node_param(self):
1604         _params = copy.deepcopy(
1605             SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters'])
1606         self.ls_tcl_client._configure_test_node_param('SgwUserAddr',
1607                                                       _params['SgwUserAddr'])
1608         self.mock_tcl_handler.execute.assert_called_once()
1609
1610     def test__configure_sut_param(self):
1611         _params = {'name': 'name'}
1612         self.ls_tcl_client._configure_sut_param('name', _params)
1613         self.mock_tcl_handler.execute.assert_called_once()
1614
1615     def test__configure_dmf_param(self):
1616         _params = {"mainflows": [{"library": '111',
1617                                   "name": "Basic UDP"}],
1618                    "instanceGroups": [{
1619                        "mainflowIdx": 0,
1620                        "mixType": "",
1621                        "rate": 0.0,
1622                        "rows": [{
1623                            "clientPort": 0,
1624                            "context": 0,
1625                            "node": 0,
1626                            "overridePort": "false",
1627                            "ratingGroup": 0,
1628                            "role": 0,
1629                            "serviceId": 0,
1630                            "transport": "Any"}]
1631                    }]}
1632         self.ls_tcl_client._get_library_id = mock.Mock(return_value='111')
1633         res = self.ls_tcl_client._configure_dmf_param('name', _params)
1634         self.assertEqual(5, self.mock_tcl_handler.execute.call_count)
1635         self.assertIsNone(res)
1636
1637     def test__configure_dmf_param_no_instance_groups(self):
1638         _params = {"mainflows": [{"library": '111',
1639                                   "name": "Basic UDP"}]}
1640         self.ls_tcl_client._get_library_id = mock.Mock(return_value='111')
1641         res = self.ls_tcl_client._configure_dmf_param('name', _params)
1642         self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1643         self.assertIsNone(res)
1644
1645     def test__configure_reservation(self):
1646         _reservation = copy.deepcopy(RESERVATIONS[0])
1647         self.ls_tcl_client.resolve_test_server_name = mock.Mock(
1648             return_value='2')
1649         res = self.ls_tcl_client._configure_reservation(_reservation)
1650         self.assertIsNone(res)
1651         self.assertEqual(4, self.mock_tcl_handler.execute.call_count)
1652
1653     def test__configure_preresolved_arp(self):
1654         _arp = [{'StartingAddress': '10.81.1.10',
1655                  'NumNodes': 1}]
1656         res = self.ls_tcl_client._configure_preresolved_arp(_arp)
1657         self.mock_tcl_handler.execute.assert_called_once()
1658         self.assertIsNone(res)
1659
1660     def test__configure_preresolved_arp_none(self):
1661         res = self.ls_tcl_client._configure_preresolved_arp(None)
1662         self.assertIsNone(res)
1663         self.mock_tcl_handler.execute.assert_not_called()
1664
1665     def test_delete_test_session(self):
1666         self.assertRaises(NotImplementedError,
1667                           self.ls_tcl_client.delete_test_session, {})
1668
1669     def test__save_test_session(self):
1670         self.mock_tcl_handler.execute.side_effect = [TCL_SUCCESS_RESPONSE,
1671                                                      TCL_SUCCESS_RESPONSE]
1672         res = self.ls_tcl_client._save_test_session()
1673         self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1674         self.assertIsNone(res)
1675
1676     def test__save_test_session_invalid(self):
1677         self.mock_tcl_handler.execute.side_effect = ['Invalid', 'Errors']
1678         self.assertRaises(exceptions.LandslideTclException,
1679                           self.ls_tcl_client._save_test_session)
1680         self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1681
1682     def test__get_library_id_system_lib(self):
1683         self.mock_tcl_handler.execute.return_value = '111'
1684         res = self.ls_tcl_client._get_library_id('name')
1685         self.mock_tcl_handler.execute.assert_called_once()
1686         self.assertEqual('111', res)
1687
1688     def test__get_library_id_user_lib(self):
1689         self.mock_tcl_handler.execute.side_effect = ['Not found', '222']
1690         res = self.ls_tcl_client._get_library_id('name')
1691         self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1692         self.assertEqual('222', res)
1693
1694     def test__get_library_id_exception(self):
1695         self.mock_tcl_handler.execute.side_effect = ['Not found', 'Not found']
1696         self.assertRaises(exceptions.LandslideTclException,
1697                           self.ls_tcl_client._get_library_id,
1698                           'name')
1699         self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
1700
1701
1702 class TestLsTclHandler(unittest.TestCase):
1703
1704     def setUp(self):
1705         self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
1706         self.mock_lsapi.start()
1707
1708         self.addCleanup(self._cleanup)
1709
1710     def _cleanup(self):
1711         self.mock_lsapi.stop()
1712
1713     def test___init__(self, *args):
1714         self.ls_tcl_handler = tg_landslide.LsTclHandler()
1715         self.assertEqual({}, self.ls_tcl_handler.tcl_cmds)
1716         self.ls_tcl_handler._ls.tcl.assert_called_once()
1717
1718     def test_execute(self, *args):
1719         self.ls_tcl_handler = tg_landslide.LsTclHandler()
1720         self.ls_tcl_handler.execute('command')
1721         self.assertIn('command', self.ls_tcl_handler.tcl_cmds)