path: root/yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_landslide.py

# Copyright (c) 2018 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import copy
import mock
import requests
import time
import unittest
import uuid

from yardstick.benchmark.contexts import base as ctx_base
from yardstick.common import exceptions
from yardstick.common import utils as common_utils
from yardstick.common import yaml_loader
from yardstick.network_services import utils as net_serv_utils
from yardstick.network_services.traffic_profile import landslide_profile
from yardstick.network_services.vnf_generic.vnf import sample_vnf
from yardstick.network_services.vnf_generic.vnf import tg_landslide


NAME = "tg__0"

EXAMPLE_URL = 'http://example.com/'
TCL_SUCCESS_RESPONSE = 'ls_ok'

TEST_SERVERS = [
    {'ip': '192.168.122.101',
     'phySubnets': [
         {'mask': '/24',
          'base': '10.42.32.100',
          'numIps': 20,
          'name': 'eth1'}
     ],
     'role': 'SGW_Node',
     'name': 'TestServer_1'},
    {'ip': '192.168.122.102',
     'phySubnets': [
         {'mask': '/24',
          'base': '10.42.32.1',
          'numIps': 100,
          'name': 'eth1'
          },
         {'mask': '/24',
          'base': '10.42.33.1',
          'numIps': 100,
          'name': 'eth2'}
     ],
     'preResolvedArpAddress': [
         {'NumNodes': 1,
          'StartingAddress': '10.42.33.5'}
     ],
     'role': 'SGW_Nodal',
     'name': 'TestServer_2',
     'thread_model': 'Fireball'
     }
]

TS1_SUTS = [
    {'name': 'SGW - C TestNode',
     'role': 'SgwControlAddr',
     'managementIp': '12.0.1.1',
     'ip': '10.42.32.100',
     'phy': 'eth5',
     'nextHop': '10.42.32.5'
     },
    {'name': 'SGW - U TestNode',
     'role': 'SgwUserAddr',
     'managementIp': '12.0.1.2',
     'ip': '10.42.32.101',
     'phy': 'eth5',
     'nextHop': '10.42.32.5'
     }
]

TS2_SUTS = [
    {'name': 'eNodeB TestNode',
     'role': 'EnbUserAddr',
     'managementIp': '12.0.2.1',
     'ip': '10.42.32.2',
     'phy': 'eth5',
     'nextHop': '10.42.32.5'
     },
    {'name': 'MME TestNode',
     'role': 'MmeControlAddr',
     'managementIp': '12.0.3.1',
     'ip': '10.42.32.1',
     'phy': 'eth5',
     'nextHop': '10.42.32.5'
     },
    {'name': 'NetHost TestNode',
     'role': 'NetworkHostAddrLocal',
     'managementIp': '12.0.4.1',
     'ip': '10.42.33.1',
     'phy': 'eth5',
     'nextHop': '10.42.32.5'
     },
    {'name': 'PGW TestNode',
     'role': 'PgwV4Sut',
     'managementIp': '12.0.5.1',
     'ip': '10.42.32.105',
     'phy': 'eth5',
     'nextHop': '10.42.32.5'
     },
    {'name': 'SGW - C SUT',
     'role': 'SgwSut',
     'managementIp': '12.0.6.1',
     'ip': '10.42.32.100'
     },
    {'name': 'SGW - U SUT',
     'role': 'SgwUserSut',
     'managementIp': '12.0.6.2',
     'ip': '10.42.32.101'}
]

VNFD = {
    'vnfd:vnfd-catalog': {
        'vnfd': [{
            'short-name': 'landslide',
            'vdu': [{
                'description': 'AB client interface details',
                'name': 'abclient-baremetal',
                'id': 'abclient-baremetal',
                'external-interface': []}],
            'description': 'Spirent Landslide traffic generator',
            'config': [{'test_server': TEST_SERVERS[0], 'suts': TS1_SUTS},
                       {'test_server': TEST_SERVERS[1], 'suts': TS2_SUTS}],
            'mgmt-interface': {
                'vdu-id': 'landslide-tas',
                'user': 'user',
                'password': 'user',
                'super-user': 'super-user',
                'super-user-password': 'super-user-password',
                'cfguser_password': 'cfguser_password',
                'license': 48,
                'proto': 'http',
                'ip': '1.1.1.1'},
            'benchmark': {
                'kpi': [
                    'tx_throughput_mbps',
                    'rx_throughput_mbps',
                    'in_packets',
                    'out_packets',
                    'activation_rate_sessps',
                    'deactivation_rate_sessps']},
            'id': 'LandslideTrafficGen',
            'name': 'LandslideTrafficGen'}]}}

TAS_INFO = VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']

DMF_CFG = {
    "dmf": {
        "library": "test",
        "name": "Basic UDP"
    },
    "clientPort": {
        "clientPort": 2002,
        "isClientPortRange": "false"
    },
    "dataProtocol": "udp",
    "serverPort": 2003
}

RESERVATIONS = [
    {'tsName': TEST_SERVERS[0]['name'],
     'phySubnets': TEST_SERVERS[0]['phySubnets'],
     'tsId': TEST_SERVERS[0]['name'],
     'tsIndex': 0},
    {'tsName': TEST_SERVERS[1]['name'],
     'phySubnets': TEST_SERVERS[1]['phySubnets'],
     'tsId': TEST_SERVERS[1]['name'],
     'tsIndex': 1}]

SESSION_PROFILE = {
    'keywords': '',
    'duration': 60,
    'iterations': 1,
    'description': 'UE default bearer creation test case',
    'name': 'default_bearer_capacity',
    'reportOptions': {'format': 'CSV'},
    'reservePorts': 'false',
    'tsGroups': [
        {
            'testCases': [{
                'type': 'SGW_Node',
                'name': '',
                'linked': "false",
                'AssociatedPhys': '',
                'parameters': {
                    'SgiPtpTunnelEn': 'false',
                    'Gtp2Imsi': '505024101215074',
                    'Sessions': '100000',
                    'S5Protocol': 'GTPv2',
                    'TrafficMtu': '1500',
                    'Gtp2Version': '13.6.0',
                    'BearerV4AddrPool': '1.0.0.1',
                    'Gtp2Imei': '50502410121507',
                    'PgwNodeEn': 'true',
                    'DedicatedsPerDefaultBearer': '0',
                    'DefaultBearers': '1',
                    'SgwUserAddr': {
                        'numLinksOrNodes': 1,
                        'phy': 'eth1',
                        'forcedEthInterface': '',
                        'ip': 'SGW_USER_IP',
                        'class': 'TestNode',
                        'ethStatsEnabled': "false",
                        'mtu': 1500
                    },
                    'SgwControlAddr': {
                        'numLinksOrNodes': 1,
                        'phy': 'eth1',
                        'forcedEthInterface': '',
                        'ip': 'SGW_CONTROL_IP',
                        'class': 'TestNode',
                        'ethStatsEnabled': "false",
                        'mtu': 1500,
                        'nextHop': 'SGW_CONTROL_NEXT_HOP'
                    },
                    'BearerAddrPool': '2001::1',
                    'TestType': 'SGW-NODE'
                }
            }],
            'tsId': TEST_SERVERS[0]['name']},
        {
            'testCases': [{
                'type': 'SGW_Nodal',
                'name': '',
                'parameters': {
                    'DataTraffic': 'Continuous',
                    'TrafficStartType': 'When All Sessions Established',
                    'NetworkHost': 'Local',
                    'Gtp2Imsi': '505024101215074',
                    'Dmf': {
                        'mainflows': [
                            {
                                'name': 'Basic UDP',
                                'library': 'test'
                            }
                        ],
                        'class': 'Dmf',
                        'instanceGroups': [
                            {
                                'startPaused': "false",
                                'rate': 0,
                                'mainflowIdx': 0,
                                'mixType': ''
                            }
                        ]
                    },
                    'S5Protocol': 'GTPv2',
                    'DataUserCfgFileEn': 'false',
                    'PgwUserSutEn': 'false',
                    'MmeControlAddr': {
                        'numLinksOrNodes': 1,
                        'phy': 'eth1',
                        'forcedEthInterface': '',
                        'ip': 'MME_CONTROL_IP',
                        'class': 'TestNode',
                        'ethStatsEnabled': "false",
                        'mtu': 1500
                    },
                    'SgwUserSut': {
                        'class': 'Sut',
                        'name': 'SGW_USER_NAME'
                    },
                    'TestActivity': 'Capacity Test',
                    'NetworkHostAddrLocal': {
                        'numLinksOrNodes': 1,
                        'phy': 'eth2',
                        'forcedEthInterface': '',
                        'ip': 'NET_HOST_IP',
                        'class': 'TestNode',
                        'ethStatsEnabled': "false",
                        'mtu': 1500
                    },
                    'DedicatedsPerDefaultBearer': '0',
                    'DisconnectRate': '1000.0',
                    'Sessions': '100000',
                    'SgwSut': {
                        'class': 'Sut',
                        'name': 'SGW_CONTROL_NAME'
                    },
                    'TrafficMtu': '1500',
                    'Gtp2Version': '13.6.0',
                    'Gtp2Imei': '50502410121507',
                    'PgwNodeEn': 'false',
                    'StartRate': '1000.0',
                    'PgwV4Sut': {
                        'class': 'Sut',
                        'name': 'PGW_SUT_NAME'
                    },
                    'DefaultBearers': '1',
                    'EnbUserAddr': {
                        'numLinksOrNodes': 1,
                        'phy': 'eth1',
                        'forcedEthInterface': '',
                        'ip': 'ENB_USER_IP',
                        'class': 'TestNode',
                        'ethStatsEnabled': "false",
                        'mtu': 1500
                    },
                    'TestType': 'SGW-NODAL'
                }
            }],
            'tsId': TEST_SERVERS[1]['name']
        }
    ]
}


class TestLandslideTrafficGen(unittest.TestCase):
    SCENARIO_CFG = {
        'session_profile': '/traffic_profiles/landslide/'
                           'landslide_session_default_bearer.yaml',
        'task_path': '',
        'runner': {
            'type': 'Iteration',
            'iterations': 1
        },
        'nodes': {
            'tg__0': 'tg__0.traffic_gen',
            'vnf__0': 'vnf__0.vnf_epc'
        },
        'topology': 'landslide_tg_topology.yaml',
        'type': 'NSPerf',
        'traffic_profile': '../../traffic_profiles/landslide/'
                           'landslide_dmf_udp.yaml',
        'options': {
            'test_cases': [
                {
                    'BearerAddrPool': '2002::2',
                    'type': 'SGW_Node',
                    'BearerV4AddrPool': '2.0.0.2',
                    'Sessions': '90000'
                },
                {
                    'StartRate': '900.0',
                    'type': 'SGW_Nodal',
                    'DisconnectRate': '900.0',
                    'Sessions': '90000'
                }
            ],
            'dmf':
                {
                    'transactionRate': 1000,
                    'packetSize': 512
                }
        }
    }

    CONTEXT_CFG = {
        'contexts': [
            {
                'type': 'Node',
                'name': 'traffic_gen',
                'file': '/etc/yardstick/nodes/pod_landslide.yaml'
            },
            {
                'type': 'Node',
                'name': 'vnf_epc',
                'file': '/etc/yardstick/nodes/pod_vepc_sut.yaml'
            }
        ]
    }

    TRAFFIC_PROFILE = {
        "schema": "nsb:traffic_profile:0.1",
        "name": "LandslideProfile",
        "description": "Spirent Landslide traffic profile",
        "traffic_profile": {
            "traffic_type": "LandslideProfile"
        },
        "dmf_config": {
            "dmf": {
                "library": "test",
                "name": "Basic UDP"
            },
            "description": "Basic data flow using UDP/IP",
            "keywords": "UDP",
            "dataProtocol": "udp"
        }
    }

    SUCCESS_CREATED_CODE = 201
    SUCCESS_OK_CODE = 200
    SUCCESS_RECORD_ID = 5
    TEST_USER_ID = 11

    def setUp(self):
        self._id = uuid.uuid1().int

        self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
        self.mock_lsapi.start()

        self.mock_ssh_helper = mock.patch.object(sample_vnf, 'VnfSshHelper')
        self.mock_ssh_helper.start()
        self.vnfd = VNFD['vnfd:vnfd-catalog']['vnfd'][0]
        self.ls_tg = tg_landslide.LandslideTrafficGen(
            NAME, self.vnfd, self._id)
        self.session_profile = copy.deepcopy(SESSION_PROFILE)
        self.ls_tg.session_profile = self.session_profile

        self.addCleanup(self._cleanup)

    def _cleanup(self):
        self.mock_lsapi.stop()
        self.mock_ssh_helper.stop()

    @mock.patch.object(net_serv_utils, 'get_nsb_option')
    def test___init__(self, mock_get_nsb_option, *args):
        _path_to_nsb = 'path/to/nsb'
        mock_get_nsb_option.return_value = _path_to_nsb
        ls_tg = tg_landslide.LandslideTrafficGen(NAME, self.vnfd, self._id)
        self.assertIsInstance(ls_tg.resource_helper,
                              tg_landslide.LandslideResourceHelper)
        mock_get_nsb_option.assert_called_once_with('bin_path')
        self.assertEqual(_path_to_nsb, ls_tg.bin_path)
        self.assertEqual(NAME, ls_tg.name)
        self.assertTrue(ls_tg.runs_traffic)
        self.assertFalse(ls_tg.traffic_finished)
        self.assertIsNone(ls_tg.session_profile)

    def test_listen_traffic(self):
        _traffic_profile = {}
        self.assertIsNone(self.ls_tg.listen_traffic(_traffic_profile))

    def test_terminate(self, *args):
        self.ls_tg.resource_helper._tcl = mock.Mock()
        self.assertIsNone(self.ls_tg.terminate())
        self.ls_tg.resource_helper._tcl.disconnect.assert_called_once()

    @mock.patch.object(ctx_base.Context, 'get_context_from_server',
                       return_value='fake_context')
    def test_instantiate(self, *args):
        self.ls_tg._tg_process = mock.Mock()
        self.ls_tg._tg_process.start = mock.Mock()
        self.ls_tg.resource_helper.connect = mock.Mock()
        self.ls_tg.resource_helper.create_test_servers = mock.Mock()
        self.ls_tg.resource_helper.create_suts = mock.Mock()
        self.ls_tg._load_session_profile = mock.Mock()
        self.assertIsNone(self.ls_tg.instantiate(self.SCENARIO_CFG,
                                                 self.CONTEXT_CFG))
        self.ls_tg.resource_helper.connect.assert_called_once()
        self.ls_tg.resource_helper.create_test_servers.assert_called_once()
        _suts_blocks_num = len([item['suts'] for item in self.vnfd['config']])
        self.assertEqual(_suts_blocks_num,
                         self.ls_tg.resource_helper.create_suts.call_count)
        self.ls_tg._load_session_profile.assert_called_once()

    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'get_running_tests')
    def test_run_traffic(self, mock_get_tests, *args):
        self.ls_tg.resource_helper._url = EXAMPLE_URL
        self.ls_tg.scenario_helper.scenario_cfg = self.SCENARIO_CFG
        mock_traffic_profile = mock.Mock(
            spec=landslide_profile.LandslideProfile)
        mock_traffic_profile.dmf_config = {'keywords': 'UDP',
                                           'dataProtocol': 'udp'}
        mock_traffic_profile.params = self.TRAFFIC_PROFILE
        self.ls_tg.resource_helper._user_id = self.TEST_USER_ID
        mock_get_tests.return_value = [{'id': self.SUCCESS_RECORD_ID,
                                        'testStateOrStep': 'COMPLETE'}]
        mock_post = mock.Mock()
        mock_post.status_code = self.SUCCESS_CREATED_CODE
        mock_post.json.return_value = {'id': self.SUCCESS_RECORD_ID}
        mock_session = mock.Mock(spec=requests.Session)
        mock_session.post.return_value = mock_post
        self.ls_tg.resource_helper.session = mock_session
        self.ls_tg.resource_helper._tcl = mock.Mock()
        _tcl = self.ls_tg.resource_helper._tcl
        self.assertIsNone(self.ls_tg.run_traffic(mock_traffic_profile))
        self.assertEqual(self.SUCCESS_RECORD_ID,
                         self.ls_tg.resource_helper.run_id)
        mock_traffic_profile.update_dmf.assert_called_with(
            self.ls_tg.scenario_helper.all_options)
        _tcl.create_dmf.assert_called_with(mock_traffic_profile.dmf_config)
        _tcl.create_test_session.assert_called_with(self.session_profile)

    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'check_running_test_state')
    def test_collect_kpi(self, mock_check_running_test_state, *args):
        self.ls_tg.resource_helper.run_id = self.SUCCESS_RECORD_ID
        mock_check_running_test_state.return_value = 'COMPLETE'
        self.assertEqual({'done': True}, self.ls_tg.collect_kpi())
        mock_check_running_test_state.assert_called_once()

    def test_wait_for_instantiate(self):
        self.assertIsNone(self.ls_tg.wait_for_instantiate())
        self.ls_tg.wait_for_instantiate()

    def test__update_session_suts_no_tc_role(self, *args):
        _suts = [{'role': 'epc_role'}]
        _testcase = {'parameters': {'diff_epc_role': {'class': 'Sut'}}}
        res = self.ls_tg._update_session_suts(_suts, _testcase)
        self.assertEqual(_testcase, res)

    def test__update_session_suts(self, *args):

        def get_testnode_param(role, key, session_prof):
            """ Get value by key from the deep nested dict to avoid calls like:
            e.g. session_prof['tsGroups'][0]['testCases'][1]['parameters'][key]
            """
            for group in session_prof['tsGroups']:
                for tc in group['testCases']:
                    tc_params = tc['parameters']
                    if tc_params.get(role):
                        return tc_params[role][key]

        def get_sut_param(role, key, suts):
            """ Search list of dicts for one with specific role.
            Return the value of related dict by key. Expect key presence.
            """
            for sut in suts:
                if sut.get('role') == role:
                    return sut[key]

        # TestNode to verify
        testnode_role = 'SgwControlAddr'
        # SUT to verify
        sut_role = 'SgwUserSut'

        config_suts = [config['suts'] for config in self.vnfd['config']]
        session_tcs = [_tc for _ts_group in self.ls_tg.session_profile['tsGroups']
                       for _tc in _ts_group['testCases']]
        for suts, tc in zip(config_suts, session_tcs):
            self.assertEqual(tc, self.ls_tg._update_session_suts(suts, tc))

        # Verify TestNode class objects keys were updated
        for _key in {'ip', 'phy', 'nextHop'}:
            self.assertEqual(
                get_testnode_param(testnode_role, _key, self.ls_tg.session_profile),
                get_sut_param(testnode_role, _key, TS1_SUTS))
        # Verify Sut class objects name was updated
        self.assertEqual(
            get_testnode_param(sut_role, 'name', self.ls_tg.session_profile),
            get_sut_param(sut_role, 'name', TS2_SUTS))

    def test__update_session_test_servers(self, *args):
        for ts_index, ts in enumerate(TEST_SERVERS):
            self.assertIsNone(
                self.ls_tg._update_session_test_servers(ts, ts_index))
        # Verify preResolvedArpAddress key was added
        self.assertTrue(any(
            _item.get('preResolvedArpAddress')
            for _item in self.ls_tg.session_profile['tsGroups']))
        # Verify reservations key was added to session profile
        self.assertEqual(RESERVATIONS,
                         self.ls_tg.session_profile.get('reservations'))
        self.assertEqual('true',
                         self.ls_tg.session_profile.get('reservePorts'))

    def test__update_session_tc_params_assoc_phys(self):
        _tc_options = {'AssociatedPhys': 'eth1'}
        _testcase = {}
        _testcase_orig = copy.deepcopy(_testcase)
        res = self.ls_tg._update_session_tc_params(_tc_options, _testcase)
        self.assertNotEqual(_testcase_orig, res)
        self.assertEqual(_tc_options, _testcase)

    def test__update_session_tc_params(self, *args):

        def get_session_tc_param_value(param, tc_type, session_prof):
            """ Get param value from the deep nested dict to avoid calls like:
            session_prof['tsGroups'][0]['testCases'][0]['parameters'][key]
            """
            for test_group in session_prof['tsGroups']:
                session_tc = test_group['testCases'][0]
                if session_tc['type'] == tc_type:
                    return session_tc['parameters'].get(param)

        session_tcs = [_tc for _ts_group in self.ls_tg.session_profile['tsGroups']
                       for _tc in _ts_group['testCases']]
        scenario_tcs = [_tc for _tc in
                        self.SCENARIO_CFG['options']['test_cases']]
        for tc_options, tc in zip(scenario_tcs, session_tcs):
            self.assertEqual(
                tc,
                self.ls_tg._update_session_tc_params(tc_options, tc))

        # Verify that each test case parameter was updated
        # Params been compared are deeply nested. Using loops to ease access.
        for _tc in self.SCENARIO_CFG['options']['test_cases']:
            for _key, _val in _tc.items():
                if _key != 'type':
                    self.assertEqual(
                        _val,
                        get_session_tc_param_value(_key, _tc.get('type'),
                                                   self.ls_tg.session_profile))

    @mock.patch.object(common_utils, 'open_relative_file')
    @mock.patch.object(yaml_loader, 'yaml_load')
    @mock.patch.object(tg_landslide.LandslideTrafficGen,
                       '_update_session_test_servers')
    @mock.patch.object(tg_landslide.LandslideTrafficGen,
                       '_update_session_suts')
    @mock.patch.object(tg_landslide.LandslideTrafficGen,
                       '_update_session_tc_params')
    def test__load_session_profile(self, mock_upd_ses_tc_params,
                                   mock_upd_ses_suts, mock_upd_ses_ts,
                                   mock_yaml_load, *args):
        self.ls_tg.scenario_helper.scenario_cfg = \
            copy.deepcopy(self.SCENARIO_CFG)
        mock_yaml_load.return_value = copy.deepcopy(SESSION_PROFILE)
        self.assertIsNone(self.ls_tg._load_session_profile())
        self.assertIsNotNone(self.ls_tg.session_profile)
        # Number of blocks in configuration files
        # Number of test servers, suts and tc params blocks should be equal
        _config_files_blocks_num = len([item['test_server']
                                        for item in self.vnfd['config']])
        self.assertEqual(_config_files_blocks_num,
                         mock_upd_ses_ts.call_count)
        self.assertEqual(_config_files_blocks_num,
                         mock_upd_ses_suts.call_count)
        self.assertEqual(_config_files_blocks_num,
                         mock_upd_ses_tc_params.call_count)

    @mock.patch.object(common_utils, 'open_relative_file')
    @mock.patch.object(yaml_loader, 'yaml_load')
    def test__load_session_profile_unequal_num_of_cfg_blocks(
            self, mock_yaml_load, *args):
        vnfd = copy.deepcopy(VNFD['vnfd:vnfd-catalog']['vnfd'][0])
        ls_traffic_gen = tg_landslide.LandslideTrafficGen(NAME, vnfd, self._id)
        ls_traffic_gen.scenario_helper.scenario_cfg = self.SCENARIO_CFG
        mock_yaml_load.return_value = copy.deepcopy(SESSION_PROFILE)
        # Delete test_servers item from pod file to make it not valid
        ls_traffic_gen.vnfd_helper['config'].pop()
        with self.assertRaises(RuntimeError):
            ls_traffic_gen._load_session_profile()

    @mock.patch.object(common_utils, 'open_relative_file')
    @mock.patch.object(yaml_loader, 'yaml_load')
    def test__load_session_profile_test_type_mismatch(self, mock_yaml_load,
                                                      *args):
        vnfd = copy.deepcopy(VNFD['vnfd:vnfd-catalog']['vnfd'][0])
        # Swap test servers data in pod file
        vnfd['config'] = list(reversed(vnfd['config']))
        ls_tg = tg_landslide.LandslideTrafficGen(NAME, vnfd, self._id)
        ls_tg.scenario_helper.scenario_cfg = self.SCENARIO_CFG
        mock_yaml_load.return_value = SESSION_PROFILE
        with self.assertRaises(RuntimeError):
            ls_tg._load_session_profile()


class TestLandslideResourceHelper(unittest.TestCase):

    PROTO_PORT = 8080
    EXAMPLE_URL = ''.join([TAS_INFO['proto'], '://', TAS_INFO['ip'], ':',
                           str(PROTO_PORT), '/api/'])
    SUCCESS_CREATED_CODE = 201
    SUCCESS_OK_CODE = 200
    INVALID_REST_CODE = '400'
    NOT_MODIFIED_CODE = 500810
    ERROR_CODE = 500800
    SUCCESS_RECORD_ID = 11
    EXPIRE_DATE = '2020/01/01 12:00 FLE Standard Time'
    TEST_USER = 'test'
    TEST_TERMINATED = 1
    AUTH_DATA = {'user': TAS_INFO['user'], 'password': TAS_INFO['password']}
    TEST_SESSION_NAME = 'default_bearer_capacity'

    USERS_DATA = {
        "users": [{
            "url": ''.join([EXAMPLE_URL, 'users/', str(SUCCESS_RECORD_ID)]),
            "id": SUCCESS_RECORD_ID,
            "level": 1,
            "username": TEST_USER
        }]
    }

    CREATE_USER_DATA = {'username': TAS_INFO['user'],
                        'expiresOn': EXPIRE_DATE,
                        'level': 1,
                        'contactInformation': '',
                        'fullName': 'Test User',
                        'password': TAS_INFO['password'],
                        'isActive': 'true'}

    SUTS_DATA = {
        "suts": [
            {
                "url": ''.join([EXAMPLE_URL, 'suts/', str(SUCCESS_RECORD_ID)]),
                "id": SUCCESS_RECORD_ID,
                "name": "10.41.32.1"
            }]}

    TEST_SERVERS_DATA = {
        "testServers": [
            {
                "url": ''.join([EXAMPLE_URL, "testServers/1"]),
                "id": 1,
                "name": TEST_SERVERS[0]['name'],
                "state": "READY",
                "version": "16.4.0.10"
            },
            {
                "url": ''.join([EXAMPLE_URL, "testServers/2"]),
                "id": 2,
                "name": TEST_SERVERS[1]['name'],
                "state": "READY",
                "version": "16.4.0.10"
            }

        ]
    }

    RUN_ID = 3

    RUNNING_TESTS_DATA = {
        "runningTests": [{
            "url": ''.join([EXAMPLE_URL, "runningTests/{}".format(RUN_ID)]),
            "measurementsUrl": ''.join(
                [EXAMPLE_URL,
                 "runningTests/{}/measurements".format(RUN_ID)]),
            "criteriaUrl": ''.join(
                [EXAMPLE_URL,
                 "runningTests/{}/criteria".format(RUN_ID)]),
            "noteToUser": "",
            "id": RUN_ID,
            "library": SUCCESS_RECORD_ID,
            "name": "default_bearer_capacity",
            "user": TEST_USER,
            "criteriaStatus": "NA",
            "testStateOrStep": "COMPLETE"
        }]}

    TEST_RESULTS_DATA = {
        "interval": 0,
        "elapsedTime": 138,
        "actualTime": 1521548057296,
        "iteration": 1,
        "tabs": {
            "Test Summary": {
                "Start Time": "Tue Mar 20 07:11:55 CDT 2018",
                "Actual Dedicated Bearer Session Connects": "100",
                "Actual Dedicated Bearer Session Disconnects": "100",
                "Actual Disconnect Rate(Sessions / Second)(P - I)": "164.804",
                "Average Session Disconnect Time(P - I)": "5.024 s",
                "Total Data Sent + Received Packets / Sec(P - I)": "1,452.294"
            }}}

    def setUp(self):
        self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
        self.mock_lsapi.start()

        mock_env_helper = mock.Mock()
        self.res_helper = tg_landslide.LandslideResourceHelper(mock_env_helper)
        self.res_helper._url = EXAMPLE_URL

        self.addCleanup(self._cleanup)

    def _cleanup(self):
        self.mock_lsapi.stop()
        self.res_helper._url = None

    def test___init__(self, *args):
        self.assertIsInstance(self.res_helper,
                              tg_landslide.LandslideResourceHelper)
        self.assertEqual({}, self.res_helper._result)
        self.assertIsNone(self.res_helper.run_id)

    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'stop_running_tests')
    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'get_running_tests')
    def test_abort_running_tests_no_running_tests(self, mock_get_tests,
                                                  mock_stop_tests, *args):
        tests_data = [{'id': self.SUCCESS_RECORD_ID,
                       'testStateOrStep': 'COMPLETE'}]
        mock_get_tests.return_value = tests_data
        self.assertIsNone(self.res_helper.abort_running_tests())
        mock_stop_tests.assert_not_called()

    @mock.patch.object(time, 'sleep')
    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'stop_running_tests')
    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'get_running_tests')
    def test_abort_running_tests(self, mock_get_tests, mock_stop_tests, *args):
        test_states_seq = iter(['RUNNING', 'COMPLETE'])

        def configure_mock(*args):
            return [{'id': self.SUCCESS_RECORD_ID,
                     'testStateOrStep': next(test_states_seq)}]

        mock_get_tests.side_effect = configure_mock
        self.assertIsNone(self.res_helper.abort_running_tests())
        mock_stop_tests.assert_called_once_with(
            running_test_id=self.SUCCESS_RECORD_ID,
            force=True)
        self.assertEqual(2, mock_get_tests.call_count)

    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'stop_running_tests')
    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'get_running_tests')
    def test_abort_running_tests_error(self, mock_get_tests, mock_stop_tests,
                                       *args):
        tests_data = {'id': self.SUCCESS_RECORD_ID,
                      'testStateOrStep': 'RUNNING'}
        mock_get_tests.return_value = [tests_data]
        with self.assertRaises(RuntimeError):
            self.res_helper.abort_running_tests(timeout=1, delay=1)
        mock_stop_tests.assert_called_with(
            running_test_id=self.SUCCESS_RECORD_ID,
            force=True)

    def test__build_url(self, *args):
        resource = 'users'
        action = {'action': 'userCreate'}
        expected_url = ''.join([EXAMPLE_URL, 'users?action=userCreate'])
        self.assertEqual(expected_url,
                         self.res_helper._build_url(resource, action))

    def test__build_url_error(self, *args):
        resource = ''
        action = {'action': 'userCreate'}

        with self.assertRaises(ValueError):
            self.res_helper._build_url(resource, action)

    def test_get_response_params(self, *args):
        method = 'get'
        resource = 'users'
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': self.USERS_DATA}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        resp = self.res_helper.get_response_params(method, resource)
        self.assertTrue(resp)

    @mock.patch.object(tg_landslide.LandslideResourceHelper, '_get_users')
    @mock.patch.object(time, 'time')
    def test__create_user(self, mock_time, mock_get_users, *args):
        mock_time.strftime.return_value = self.EXPIRE_DATE
        post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
                          'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
        mock_session = mock.Mock(spec=requests.Session)
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        self.assertEqual(self.SUCCESS_RECORD_ID,
                         self.res_helper._create_user(self.AUTH_DATA))
        mock_get_users.assert_not_called()

    @mock.patch.object(tg_landslide.LandslideResourceHelper, '_modify_user')
    @mock.patch.object(time, 'time')
    def test__create_user_username_exists(self, mock_time, mock_modify_user,
                                          *args):
        mock_time.strftime.return_value = self.EXPIRE_DATE
        mock_modify_user.return_value = {'id': self.SUCCESS_RECORD_ID,
                                         'result': 'No changes requested'}
        post_resp_data = {
            'status_code': self.ERROR_CODE,
            'json.return_value': {'id': self.SUCCESS_OK_CODE,
                                  'apiCode': self.NOT_MODIFIED_CODE}}
        mock_session = mock.Mock(spec=requests.Session)
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        res = self.res_helper._create_user(self.AUTH_DATA)
        mock_modify_user.assert_called_once_with(TAS_INFO['user'],
                                                 {'isActive': 'true'})
        self.assertEqual(self.SUCCESS_RECORD_ID, res)

    @mock.patch.object(time, 'time')
    def test__create_user_error(self, mock_time, *args):
        mock_time.strftime.return_value = self.EXPIRE_DATE
        mock_session = mock.Mock(spec=requests.Session)
        post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                          'json.return_value': {'apiCode': self.ERROR_CODE}}
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        with self.assertRaises(exceptions.RestApiError):
            self.res_helper._create_user(self.AUTH_DATA)

    def test__modify_user(self, *args):
        post_data = {'username': 'test_user'}
        mock_session = mock.Mock(spec=requests.Session)
        post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                          'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        res = self.res_helper._modify_user(username=self.TEST_USER,
                                           fields=post_data)
        self.assertEqual(self.SUCCESS_RECORD_ID, res['id'])

    def test__modify_user_rest_resp_fail(self, *args):
        post_data = {'non-existing-key': ''}
        mock_session = mock.Mock(spec=requests.Session)
        mock_session.post.ok = False
        self.res_helper.session = mock_session
        self.assertRaises(exceptions.RestApiError,
                          self.res_helper._modify_user,
                          username=self.TEST_USER, fields=post_data)
        mock_session.post.assert_called_once()

    def test__delete_user(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        self.res_helper.session = mock_session
        self.assertIsNone(self.res_helper._delete_user(
            username=self.TEST_USER))

    def test__get_users(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': self.USERS_DATA}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        self.assertEqual(self.USERS_DATA['users'],
                         self.res_helper._get_users())

    def test_exec_rest_request(self, *args):
        resource = 'testServers'
        action = {'action': 'modify'}
        expected_url = ''.join([EXAMPLE_URL, 'testServers?action=modify'])
        post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
                          'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
        mock_session = mock.Mock(spec=requests.Session)
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        self.res_helper.exec_rest_request('post', resource, action)
        self.res_helper.session.post.assert_called_once_with(expected_url,
                                                             json={})

    def test_exec_rest_request_unsupported_method_error(self, *args):
        resource = 'testServers'
        action = {'action': 'modify'}
        with self.assertRaises(ValueError):
            self.res_helper.exec_rest_request('patch', resource, action)

    def test_exec_rest_request_missed_action_arg(self, *args):
        resource = 'testServers'
        with self.assertRaises(ValueError):
            self.res_helper.exec_rest_request('post', resource)

    def test_exec_rest_request_raise_exc(self):
        resource = 'users'
        action = {'action': 'modify'}
        post_resp_data = {'status_code': self.ERROR_CODE,
                          'json.return_value': {
                              'status_code': self.ERROR_CODE}}
        mock_session = mock.Mock(spec=requests.Session)
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.assertRaises(exceptions.RestApiError,
                          self.res_helper.exec_rest_request,
                          'post', resource, action, raise_exc=True)

    @mock.patch.object(time, 'time')
    def test_connect(self, mock_time, *args):
        vnfd = VNFD['vnfd:vnfd-catalog']['vnfd'][0]
        mock_time.strftime.return_value = self.EXPIRE_DATE
        self.res_helper.vnfd_helper = vnfd

        self.res_helper._tcl = mock.Mock()
        post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
                          'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
        mock_session = mock.Mock(spec=requests.Session, headers={})
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        self.assertIsInstance(self.res_helper.connect(), requests.Session)
        self.res_helper._tcl.connect.assert_called_once_with(
            TAS_INFO['ip'],
            TAS_INFO['user'],
            TAS_INFO['password'])

    def test_disconnect(self, *args):
        self.res_helper._tcl = mock.Mock()
        self.assertIsNone(self.res_helper.disconnect())
        self.assertIsNone(self.res_helper.session)
        self.res_helper._tcl.disconnect.assert_called_once()

    def test_terminate(self, *args):
        self.assertIsNone(self.res_helper.terminate())
        self.assertEqual(self.TEST_TERMINATED,
                         self.res_helper._terminated.value)

    def test_create_dmf(self, *args):
        self.res_helper._tcl = mock.Mock()
        self.assertIsNone(self.res_helper.create_dmf(DMF_CFG))
        self.res_helper._tcl.create_dmf.assert_called_once_with(DMF_CFG)

    def test_create_dmf_as_list(self, *args):
        self.res_helper._tcl = mock.Mock()
        self.assertIsNone(self.res_helper.create_dmf([DMF_CFG]))
        self.res_helper._tcl.create_dmf.assert_called_once_with(DMF_CFG)

    def test_delete_dmf(self, *args):
        self.res_helper._tcl = mock.Mock()
        self.assertIsNone(self.res_helper.delete_dmf(DMF_CFG))
        self.res_helper._tcl.delete_dmf.assert_called_once_with(DMF_CFG)

    def test_delete_dmf_as_list(self, *args):
        self.res_helper._tcl = mock.Mock()
        self.assertIsNone(self.res_helper.delete_dmf([DMF_CFG]))
        self.res_helper._tcl.delete_dmf.assert_called_once_with(DMF_CFG)

    @mock.patch.object(tg_landslide.LandslideResourceHelper, 'configure_sut')
    def test_create_suts(self, mock_configure_sut, *args):
        mock_session = mock.Mock(spec=requests.Session)
        post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE}
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        self.assertIsNone(self.res_helper.create_suts(TS1_SUTS))
        mock_configure_sut.assert_not_called()

    @mock.patch.object(tg_landslide.LandslideResourceHelper, 'configure_sut')
    def test_create_suts_sut_exists(self, mock_configure_sut, *args):
        sut_name = 'test_sut'
        suts = [
            {'name': sut_name,
             'role': 'SgwControlAddr',
             'managementIp': '12.0.1.1',
             'ip': '10.42.32.100'
             }
        ]
        mock_session = mock.Mock(spec=requests.Session)
        post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        self.assertIsNone(self.res_helper.create_suts(suts))
        mock_configure_sut.assert_called_once_with(
            sut_name=sut_name,
            json_data={k: v for k, v in suts[0].items()
                       if k not in {'phy', 'nextHop', 'role', 'name'}})

    def test_get_suts(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': self.SUTS_DATA}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        self.assertIsInstance(self.res_helper.get_suts(), list)

    def test_get_suts_single_id(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': self.SUTS_DATA['suts'][0]}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        self.assertIsInstance(self.res_helper.get_suts(suts_id=2), dict)

    def test_configure_sut(self, *args):
        post_data = {'managementIp': '2.2.2.2'}
        mock_session = mock.Mock(spec=requests.Session)
        post_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                          'json.return_value': {'id': self.SUCCESS_RECORD_ID}}
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        self.assertIsNone(self.res_helper.configure_sut('test_name',
                                                        post_data))
        mock_session.post.assert_called_once()

    def test_configure_sut_error(self, *args):
        post_data = {'managementIp': '2.2.2.2'}
        mock_session = mock.Mock(spec=requests.Session)
        post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        with self.assertRaises(exceptions.RestApiError):
            self.res_helper.configure_sut('test_name', post_data)

    def test_delete_suts(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': self.SUTS_DATA}
        delete_resp_data = {'status_code': self.SUCCESS_OK_CODE}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        mock_session.delete.return_value.configure_mock(**delete_resp_data)
        self.res_helper.session = mock_session
        self.assertIsNone(self.res_helper.delete_suts())
        mock_session.delete.assert_called_once()

    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'get_test_servers')
    def test__check_test_servers_state(self, mock_get_test_servers, *args):
        mock_get_test_servers.return_value = \
            self.TEST_SERVERS_DATA['testServers']
        self.res_helper._check_test_servers_state()
        mock_get_test_servers.assert_called_once()

    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'get_test_servers')
    def test__check_test_servers_state_server_not_ready(
            self, mock_get_test_servers, *args):
        test_servers_not_ready = [
            {
                "url": ''.join([EXAMPLE_URL, "testServers/1"]),
                "id": 1,
                "name": "TestServer_1",
                "state": "NOT_READY",
                "version": "16.4.0.10"
            }
        ]

        mock_get_test_servers.return_value = test_servers_not_ready
        with self.assertRaises(RuntimeError):
            self.res_helper._check_test_servers_state(timeout=1, delay=0)

    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       '_check_test_servers_state')
    def test_create_test_servers(self, mock_check_ts_state, *args):
        test_servers_ids = [
            ts['id'] for ts in self.TEST_SERVERS_DATA['testServers']]

        self.res_helper.license_data['lic_id'] = TAS_INFO['license']
        self.res_helper._tcl.create_test_server = mock.Mock()
        self.res_helper._tcl.create_test_server.side_effect = test_servers_ids
        self.assertIsNone(self.res_helper.create_test_servers(TEST_SERVERS))
        mock_check_ts_state.assert_called_once_with(test_servers_ids)

    @mock.patch.object(tg_landslide.LandslideTclClient,
                       'resolve_test_server_name')
    @mock.patch.object(tg_landslide.LsTclHandler, 'execute')
    def test_create_test_servers_error(self, mock_execute,
                                       mock_resolve_ts_name, *args):
        self.res_helper.license_data['lic_id'] = TAS_INFO['license']
        # Return message for case test server wasn't created
        mock_execute.return_value = 'TS not found'
        # Return message for case test server name wasn't resolved
        mock_resolve_ts_name.return_value = 'TS not found'
        with self.assertRaises(RuntimeError):
            self.res_helper.create_test_servers(TEST_SERVERS)

    def test_get_test_servers(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': self.TEST_SERVERS_DATA}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        res = self.res_helper.get_test_servers()
        self.assertEqual(self.TEST_SERVERS_DATA['testServers'], res)

    def test_get_test_servers_by_id(self, *args):
        mock_session = mock.Mock(spec=requests.Session)

        _ts = self.TEST_SERVERS_DATA['testServers'][0]
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': _ts}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        res = self.res_helper.get_test_servers(test_server_ids=[_ts['id']])
        self.assertEqual([_ts], res)

    def test_configure_test_servers(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': self.TEST_SERVERS_DATA}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        res = self.res_helper.configure_test_servers(
            action={'action': 'recycle'})
        self.assertEqual(
            [x['id'] for x in self.TEST_SERVERS_DATA['testServers']],
            res)
        self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']),
                         mock_session.post.call_count)

    def test_delete_test_servers(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': self.TEST_SERVERS_DATA}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        self.assertIsNone(self.res_helper.delete_test_servers())
        self.assertEqual(len(self.TEST_SERVERS_DATA['testServers']),
                         mock_session.delete.call_count)

    def test_create_test_session_res_helper(self, *args):
        self.res_helper._user_id = self.SUCCESS_RECORD_ID
        self.res_helper._tcl = mock.Mock()
        test_session = {'name': 'test'}
        self.assertIsNone(self.res_helper.create_test_session(test_session))
        self.res_helper._tcl.create_test_session.assert_called_once_with(
            {'name': 'test', 'library': self.SUCCESS_RECORD_ID})

    @mock.patch.object(tg_landslide.LandslideTclClient,
                       'resolve_test_server_name',
                       return_value='Not Found')
    def test_create_test_session_ts_name_not_found(self, *args):
        self.res_helper._user_id = self.SUCCESS_RECORD_ID
        test_session = {
            'duration': 60,
            'description': 'UE default bearer creation test case',
            'name': 'default_bearer_capacity',
            'tsGroups': [{'testCases': [{'type': 'SGW_Node',
                                         'name': ''}],
                          'tsId': 'TestServer_3'}]
        }
        with self.assertRaises(RuntimeError):
            self.res_helper.create_test_session(test_session)

    def test_get_test_session(self, *args):
        test_session = {"name": self.TEST_SESSION_NAME}
        self.res_helper._user_id = self.SUCCESS_RECORD_ID
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': test_session}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        res = self.res_helper.get_test_session(self.TEST_SESSION_NAME)
        self.assertEqual(test_session, res)

    def test_configure_test_session(self, *args):
        test_session = {'name': self.TEST_SESSION_NAME}
        self.res_helper._user_id = self.SUCCESS_RECORD_ID
        self.res_helper.user_lib_uri = 'libraries/{{}}/{}'.format(
            self.res_helper.test_session_uri)
        mock_session = mock.Mock(spec=requests.Session)
        self.res_helper.session = mock_session
        res = self.res_helper.configure_test_session(self.TEST_SESSION_NAME,
                                                     test_session)
        self.assertIsNotNone(res)
        mock_session.post.assert_called_once()

    def test_delete_test_session(self, *args):
        self.res_helper._user_id = self.SUCCESS_RECORD_ID
        self.res_helper.user_lib_uri = 'libraries/{{}}/{}'.format(
            self.res_helper.test_session_uri)
        mock_session = mock.Mock(spec=requests.Session)
        self.res_helper.session = mock_session
        res = self.res_helper.delete_test_session(self.TEST_SESSION_NAME)
        self.assertIsNotNone(res)
        mock_session.delete.assert_called_once()

    def test_create_running_tests(self, *args):
        self.res_helper._user_id = self.SUCCESS_RECORD_ID
        test_session = {'id': self.SUCCESS_RECORD_ID}
        mock_session = mock.Mock(spec=requests.Session)
        post_resp_data = {'status_code': self.SUCCESS_CREATED_CODE,
                          'json.return_value': test_session}
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        self.res_helper.create_running_tests(self.TEST_SESSION_NAME)
        self.assertEqual(self.SUCCESS_RECORD_ID, self.res_helper.run_id)

    def test_create_running_tests_error(self, *args):
        self.res_helper._user_id = self.SUCCESS_RECORD_ID
        mock_session = mock.Mock(spec=requests.Session)
        post_resp_data = {'status_code': self.NOT_MODIFIED_CODE}
        mock_session.post.return_value.configure_mock(**post_resp_data)
        self.res_helper.session = mock_session
        with self.assertRaises(exceptions.RestApiError):
            self.res_helper.create_running_tests(self.TEST_SESSION_NAME)

    def test_get_running_tests(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': self.RUNNING_TESTS_DATA}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        res = self.res_helper.get_running_tests()
        self.assertEqual(self.RUNNING_TESTS_DATA['runningTests'], res)

    def test_delete_running_tests(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        delete_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                            'json.return_value': self.RUNNING_TESTS_DATA}
        mock_session.delete.return_value.configure_mock(**delete_resp_data)
        self.res_helper.session = mock_session
        self.assertIsNone(self.res_helper.delete_running_tests())

    def test__running_tests_action(self, *args):
        action = 'abort'
        mock_session = mock.Mock(spec=requests.Session)
        self.res_helper.session = mock_session
        res = self.res_helper._running_tests_action(self.SUCCESS_RECORD_ID,
                                                    action)
        self.assertIsNone(res)

    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       '_running_tests_action')
    def test_stop_running_tests(self, mock_tests_action, *args):
        res = self.res_helper.stop_running_tests(self.SUCCESS_RECORD_ID)
        self.assertIsNone(res)
        mock_tests_action.assert_called_once()

    def test_check_running_test_state(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {
            'status_code': self.SUCCESS_OK_CODE,
            'json.return_value': self.RUNNING_TESTS_DATA["runningTests"][0]}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        res = self.res_helper.check_running_test_state(self.SUCCESS_RECORD_ID)
        self.assertEqual(
            self.RUNNING_TESTS_DATA["runningTests"][0]['testStateOrStep'],
            res)

    def test_get_running_tests_results(self, *args):
        mock_session = mock.Mock(spec=requests.Session)
        get_resp_data = {'status_code': self.SUCCESS_OK_CODE,
                         'json.return_value': self.TEST_RESULTS_DATA}
        mock_session.get.return_value.configure_mock(**get_resp_data)
        self.res_helper.session = mock_session
        res = self.res_helper.get_running_tests_results(
            self.SUCCESS_RECORD_ID)
        self.assertEqual(self.TEST_RESULTS_DATA, res)

    def test__write_results(self, *args):
        res = self.res_helper._write_results(self.TEST_RESULTS_DATA)
        exp_res = {
            "Test Summary::Actual Dedicated Bearer Session Connects": 100.0,
            "Test Summary::Actual Dedicated Bearer Session Disconnects": 100.0,
            "Test Summary::Actual Disconnect Rate(Sessions / Second)(P - I)": 164.804,
            "Test Summary::Average Session Disconnect Time(P - I)": 5.024,
            "Test Summary::Total Data Sent + Received Packets / Sec(P - I)": 1452.294
        }
        self.assertEqual(exp_res, res)

    def test__write_results_no_tabs(self, *args):
        _res_data = copy.deepcopy(self.TEST_RESULTS_DATA)
        del _res_data['tabs']
        # Return None if tabs not found in test results dict
        self.assertIsNone(self.res_helper._write_results(_res_data))

    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'check_running_test_state')
    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'get_running_tests_results')
    def test_collect_kpi_test_running(self, mock_tests_results,
                                      mock_tests_state, *args):
        self.res_helper.run_id = self.SUCCESS_RECORD_ID
        mock_tests_state.return_value = 'RUNNING'
        mock_tests_results.return_value = self.TEST_RESULTS_DATA
        res = self.res_helper.collect_kpi()
        self.assertNotIn('done', res)
        mock_tests_state.assert_called_once_with(self.res_helper.run_id)
        mock_tests_results.assert_called_once_with(self.res_helper.run_id)

    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'check_running_test_state')
    @mock.patch.object(tg_landslide.LandslideResourceHelper,
                       'get_running_tests_results')
    def test_collect_kpi_test_completed(self, mock_tests_results,
                                        mock_tests_state, *args):
        self.res_helper.run_id = self.SUCCESS_RECORD_ID
        mock_tests_state.return_value = 'COMPLETE'
        res = self.res_helper.collect_kpi()
        self.assertIsNotNone(res)
        mock_tests_state.assert_called_once_with(self.res_helper.run_id)
        mock_tests_results.assert_not_called()
        self.assertDictContainsSubset({'done': True}, res)


class TestLandslideTclClient(unittest.TestCase):
    def setUp(self):
        self.mock_tcl_handler = mock.Mock(spec=tg_landslide.LsTclHandler)
        self.ls_res_helper = mock.Mock(
            spec=tg_landslide.LandslideResourceHelper)
        self.ls_tcl_client = tg_landslide.LandslideTclClient(
            self.mock_tcl_handler,
            self.ls_res_helper)

    def test___init__(self, *args):
        self.ls_tcl_client = tg_landslide.LandslideTclClient(
            self.mock_tcl_handler,
            self.ls_res_helper)
        self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
        self.assertIsNone(self.ls_tcl_client._user)
        self.assertIsNone(self.ls_tcl_client._library_id)
        self.assertIsNone(self.ls_tcl_client._basic_library_id)
        self.assertEqual(set(), self.ls_tcl_client.ts_ids)
        self.assertIsInstance(self.ls_tcl_client._tc_types, set)
        self.assertIsNotNone(self.ls_tcl_client._tc_types)

    def test_connect_login_success(self, *args):
        lib_id = '123'
        exec_responses = ['java0x2', lib_id, lib_id]
        auth = ('user', 'password')
        self.mock_tcl_handler.execute.side_effect = exec_responses
        self.ls_tcl_client.connect(TAS_INFO['ip'], *auth)
        self.assertEqual(lib_id, self.ls_tcl_client._library_id)
        self.assertEqual(lib_id, self.ls_tcl_client._basic_library_id)
        self.assertEqual(TAS_INFO['ip'], self.ls_tcl_client.tcl_server_ip)
        self.assertEqual(auth[0], self.ls_tcl_client._user)
        self.assertEqual(len(exec_responses),
                         self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call("ls::login 1.1.1.1 user password"),
            mock.call("ls::get [ls::query LibraryInfo -userLibraryName user] -Id"),
        ])

    def test_connect_login_failed(self, *args):
        exec_responses = ['Login failed']
        auth = ('user', 'password')
        self.mock_tcl_handler.execute.side_effect = exec_responses
        self.assertRaises(exceptions.LandslideTclException,
                          self.ls_tcl_client.connect,
                          TAS_INFO['ip'],
                          *auth)
        self.assertIsNone(self.ls_tcl_client._library_id)
        self.assertIsNone(self.ls_tcl_client._basic_library_id)
        self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
        self.assertIsNone(self.ls_tcl_client._user)
        self.assertEqual(len(exec_responses),
                         self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_called_with(
            "ls::login 1.1.1.1 user password")

    def test_disconnect(self, *args):
        self.ls_tcl_client.disconnect()
        self.mock_tcl_handler.execute.assert_called_once_with("ls::logout")
        self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
        self.assertIsNone(self.ls_tcl_client._user)
        self.assertIsNone(self.ls_tcl_client._library_id)
        self.assertIsNone(self.ls_tcl_client._basic_library_id)

    def test_create_test_server(self, *args):
        return_value = '2'
        self.ls_tcl_client._ts_context.vnfd_helper = \
            VNFD['vnfd:vnfd-catalog']['vnfd'][0]
        self.ls_tcl_client._ts_context.license_data = {'lic_id': return_value}
        self.mock_tcl_handler.execute.return_value = return_value
        self.ls_tcl_client._set_thread_model = mock.Mock()
        res = self.ls_tcl_client.create_test_server(TEST_SERVERS[1])
        self.assertEqual(3, self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('ls::query TsId TestServer_2'),
            mock.call('set ts [ls::retrieve TsInfo -Name "TestServer_2"]'),
            mock.call('ls::get $ts -RequestedLicense'),
        ])
        self.ls_tcl_client._set_thread_model.assert_called_once_with(
            TEST_SERVERS[1]['name'],
            TEST_SERVERS[1]['thread_model'])
        self.assertEqual(int(return_value), res)

    def test_create_test_server_fail_limit_reach(self, *args):
        self.mock_tcl_handler.execute.side_effect = ['TS not found',
                                                     'Add failed']
        self.assertRaises(RuntimeError,
                          self.ls_tcl_client.create_test_server,
                          TEST_SERVERS[0])
        self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('ls::query TsId TestServer_1'),
            mock.call('ls::perform AddTs -Name "TestServer_1" '
                      '-Ip "192.168.122.101"'),
        ])

    def test__add_test_server(self):
        ts_id = '2'
        self.mock_tcl_handler.execute.side_effect = ['TS not found', ts_id]
        self.assertEqual(ts_id,
                         self.ls_tcl_client._add_test_server('name', 'ip'))
        self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('ls::query TsId name'),
            mock.call('ls::perform AddTs -Name "name" -Ip "ip"'),
        ])

    def test__add_test_server_failed(self):
        self.mock_tcl_handler.execute.side_effect = ['TS not found',
                                                     'Add failed']
        self.assertRaises(RuntimeError, self.ls_tcl_client._add_test_server,
                          'name', 'ip')
        self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('ls::query TsId name'),
            mock.call('ls::perform AddTs -Name "name" -Ip "ip"'),
        ])

    def test__update_license(self):
        curr_lic_id = '111'
        new_lic_id = '222'
        exec_resp = ['java0x4',
                     curr_lic_id,
                     TCL_SUCCESS_RESPONSE,
                     TCL_SUCCESS_RESPONSE]
        self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id}
        self.mock_tcl_handler.execute.side_effect = exec_resp
        self.ls_tcl_client._update_license('name')
        self.assertEqual(len(exec_resp),
                         self.mock_tcl_handler.execute.call_count)

        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('set ts [ls::retrieve TsInfo -Name "name"]'),
            mock.call('ls::get $ts -RequestedLicense'),
            mock.call('ls::config $ts -RequestedLicense 222'),
            mock.call('ls::perform ModifyTs $ts'),
        ])

    def test__update_license_same_as_current(self):
        curr_lic_id = '111'
        new_lic_id = '111'
        exec_resp = ['java0x4', curr_lic_id]
        self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id}
        self.mock_tcl_handler.execute.side_effect = exec_resp
        self.ls_tcl_client._update_license('name')
        self.assertEqual(len(exec_resp),
                         self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('set ts [ls::retrieve TsInfo -Name "name"]'),
            mock.call('ls::get $ts -RequestedLicense'),
        ])

    def test__set_thread_model_update_needed(self):
        self.ls_tcl_client._ts_context.vnfd_helper = {
            'mgmt-interface': {
                'cfguser_password': 'cfguser_password'
            }
        }
        exec_resp = ['java0x4', 'V0', '', '']
        self.mock_tcl_handler.execute.side_effect = exec_resp
        self.ls_tcl_client._set_thread_model('name', 'Fireball')
        self.assertEqual(len(exec_resp),
                         self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('set tsc [ls::perform RetrieveTsConfiguration '
                      '-name "name" cfguser_password]'),
            mock.call('ls::get $tsc -ThreadModel'),
            mock.call('ls::config $tsc -ThreadModel "V1_FB3"'),
            mock.call('ls::perform ApplyTsConfiguration $tsc cfguser_password'),
        ])

    def test__set_thread_model_no_update_needed(self):
        self.ls_tcl_client._ts_context.vnfd_helper = {
            'mgmt-interface': {
                'cfguser_password': 'cfguser_password'
            }
        }
        exec_resp = ['java0x4', 'V0']
        self.mock_tcl_handler.execute.side_effect = exec_resp
        self.ls_tcl_client._set_thread_model('name', 'Legacy')
        self.assertEqual(len(exec_resp),
                         self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('set tsc [ls::perform RetrieveTsConfiguration '
                      '-name "name" cfguser_password]'),
            mock.call('ls::get $tsc -ThreadModel'),
        ])

    @mock.patch.object(tg_landslide.LandslideTclClient,
                       'resolve_test_server_name', side_effect=['4', '2'])
    def test_create_test_session(self, *args):
        _session_profile = copy.deepcopy(SESSION_PROFILE)
        _session_profile['reservations'] = RESERVATIONS
        self.ls_tcl_client._save_test_session = mock.Mock()
        self.ls_tcl_client._configure_ts_group = mock.Mock()
        self.ls_tcl_client._library_id = 42
        self.ls_tcl_client.create_test_session(_session_profile)
        self.assertEqual(17, self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('set test_ [ls::create TestSession]'),
            mock.call('ls::config $test_ -Library 42 '
                      '-Name "default_bearer_capacity"'),
            mock.call('ls::config $test_ -Description ' \
                      '"UE default bearer creation test case"'),
            mock.call('ls::config $test_ -Keywords ""'),
            mock.call('ls::config $test_ -Duration "60"'),
            mock.call('ls::config $test_ -Iterations "1"'),
            # _configure_reservation
            mock.call('set reservation_ [ls::create Reservation -under $test_]'),
            mock.call('ls::config $reservation_ -TsIndex 0 '
                      '-TsId 4 -TsName "TestServer_1"'),
            mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
            mock.call('ls::config $physubnet_ -Name "eth1" -Base "10.42.32.100" '
                      '-Mask "/24" -NumIps 20'),
            # _configure_reservation
            mock.call('set reservation_ [ls::create Reservation -under $test_]'),
            mock.call('ls::config $reservation_ -TsIndex 1 '
                      '-TsId 2 -TsName "TestServer_2"'),
            mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
            mock.call('ls::config $physubnet_ -Name "eth1" -Base "10.42.32.1" '
                      '-Mask "/24" -NumIps 100'),
            mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
            mock.call('ls::config $physubnet_ -Name "eth2" -Base "10.42.33.1" '
                      '-Mask "/24" -NumIps 100'),
            # _configure_report_options
            mock.call('ls::config $test_.ReportOptions -Format 1 -Ts -3 -Tc -3'),
        ])

    def test_create_dmf(self):
        self.mock_tcl_handler.execute.return_value = '2'
        self.ls_tcl_client._save_dmf = mock.Mock()
        self.ls_tcl_client.create_dmf(copy.deepcopy(DMF_CFG))
        self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
        # This is needed because the dictionary is unordered and the arguments
        # can come in either order
        call1 = mock.call(
            'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"')
        call2 = mock.call(
            'ls::config $dmf_ -isClientPortRange "false" -clientPort 2002')
        self.assertTrue(
            call1 in self.mock_tcl_handler.execute.mock_calls or
            call2 in self.mock_tcl_handler.execute.mock_calls)

        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('set dmf_ [ls::create Dmf]'),
            mock.call(
                'ls::get [ls::query LibraryInfo -systemLibraryName test] -Id'),
            mock.call('ls::config $dmf_ -Library 2 -Name "Basic UDP"'),
            mock.call('ls::config $dmf_ -dataProtocol "udp"'),
            # mock.call(
            #    'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"'),
            mock.call('ls::config $dmf_ -serverPort 2003'),
        ], any_order=True)

    def test_configure_dmf(self):
        self.mock_tcl_handler.execute.return_value = '2'
        self.ls_tcl_client._save_dmf = mock.Mock()
        self.ls_tcl_client.configure_dmf(DMF_CFG)
        self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
        # This is need because the dictionary is unordered and the arguments
        # can come in either order
        call1 = mock.call(
            'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"')
        call2 = mock.call(
            'ls::config $dmf_ -isClientPortRange "false" -clientPort 2002')
        self.assertTrue(
            call1 in self.mock_tcl_handler.execute.mock_calls or
            call2 in self.mock_tcl_handler.execute.mock_calls)

        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('set dmf_ [ls::create Dmf]'),
            mock.call(
                'ls::get [ls::query LibraryInfo -systemLibraryName test] -Id'),
            mock.call('ls::config $dmf_ -Library 2 -Name "Basic UDP"'),
            mock.call('ls::config $dmf_ -dataProtocol "udp"'),
            # mock.call(
            #    'ls::config $dmf_ -clientPort 2002 -isClientPortRange "false"'),
            mock.call('ls::config $dmf_ -serverPort 2003'),
        ], any_order=True)

    def test_delete_dmf(self):
        self.assertRaises(NotImplementedError,
                          self.ls_tcl_client.delete_dmf,
                          DMF_CFG)

    def test__save_dmf_valid(self):
        exec_resp = [TCL_SUCCESS_RESPONSE, TCL_SUCCESS_RESPONSE]
        self.mock_tcl_handler.execute.side_effect = exec_resp
        self.ls_tcl_client._save_dmf()
        self.assertEqual(len(exec_resp),
                         self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
           mock.call('ls::perform Validate -Dmf $dmf_'),
           mock.call('ls::save $dmf_ -overwrite'),
        ])

    def test__save_dmf_invalid(self):
        exec_resp = ['Invalid', 'List of errors and warnings']
        self.mock_tcl_handler.execute.side_effect = exec_resp
        self.assertRaises(exceptions.LandslideTclException,
                          self.ls_tcl_client._save_dmf)
        self.assertEqual(len(exec_resp),
                         self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
           mock.call('ls::perform Validate -Dmf $dmf_'),
           mock.call('ls::get $dmf_ -ErrorsAndWarnings'),
        ])

    def test__configure_report_options(self):
        _options = {'format': 'CSV', 'PerInterval': 'false'}
        self.ls_tcl_client._configure_report_options(_options)
        self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
           mock.call('ls::config $test_.ReportOptions -Format 1 -Ts -3 -Tc -3'),
           mock.call('ls::config $test_.ReportOptions -PerInterval false'),
           ],
           any_order=True)

    def test___configure_ts_group(self, *args):
        _ts_group = copy.deepcopy(SESSION_PROFILE['tsGroups'][0])
        self.ls_tcl_client._configure_tc_type = mock.Mock()
        self.ls_tcl_client._configure_preresolved_arp = mock.Mock()
        self.ls_tcl_client.resolve_test_server_name = mock.Mock(
            return_value='2')
        self.ls_tcl_client._configure_ts_group(_ts_group, 0)
        self.mock_tcl_handler.execute.assert_called_once_with(
            'set tss_ [ls::create TsGroup -under $test_ -tsId 2 ]')

    def test___configure_ts_group_resolve_ts_fail(self, *args):
        _ts_group = copy.deepcopy(SESSION_PROFILE['tsGroups'][0])
        self.ls_tcl_client._configure_tc_type = mock.Mock()
        self.ls_tcl_client._configure_preresolved_arp = mock.Mock()
        self.ls_tcl_client.resolve_test_server_name = mock.Mock(
            return_value='TS Not Found')
        self.assertRaises(RuntimeError, self.ls_tcl_client._configure_ts_group,
                          _ts_group, 0)
        self.mock_tcl_handler.execute.assert_not_called()

    def test__configure_tc_type(self):
        _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
        self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE
        self.ls_tcl_client._configure_parameters = mock.Mock()
        self.ls_tcl_client._configure_tc_type(_tc, 0)
        self.assertEqual(7, self.mock_tcl_handler.execute.call_count)

    def test__configure_tc_type_optional_param_omitted(self):
        _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
        del _tc['linked']
        self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE
        self.ls_tcl_client._configure_parameters = mock.Mock()
        self.ls_tcl_client._configure_tc_type(_tc, 0)
        self.assertEqual(6, self.mock_tcl_handler.execute.call_count)

    def test__configure_tc_type_wrong_type(self):
        _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
        _tc['type'] = 'not_supported'
        self.ls_tcl_client._configure_parameters = mock.Mock()
        self.assertRaises(RuntimeError,
                          self.ls_tcl_client._configure_tc_type,
                          _tc, 0)
        self.mock_tcl_handler.assert_not_called()

    def test__configure_tc_type_not_found_basic_lib(self):
        _tc = copy.deepcopy(SESSION_PROFILE['tsGroups'][0]['testCases'][0])
        self.ls_tcl_client._configure_parameters = mock.Mock()
        self.mock_tcl_handler.execute.return_value = 'Invalid'
        self.assertRaises(RuntimeError,
                          self.ls_tcl_client._configure_tc_type,
                          _tc, 0)

    def test__configure_parameters(self):
        _params = copy.deepcopy(
            SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters'])
        self.ls_tcl_client._configure_parameters(_params)
        self.assertEqual(16, self.mock_tcl_handler.execute.call_count)

    def test__configure_array_param(self):
        _array = {"class": "Array",
                  "array": ["0"]}
        self.ls_tcl_client._configure_array_param('name', _array)
        self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('ls::create -Array-name -under $p_ ;'),
            mock.call('ls::create ArrayItem -under $p_.name -Value "0"'),
        ])

    def test__configure_test_node_param(self):
        _params = copy.deepcopy(
            SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters'])
        self.ls_tcl_client._configure_test_node_param('SgwUserAddr',
                                                      _params['SgwUserAddr'])
        cmd = ('ls::create -TestNode-SgwUserAddr -under $p_ -Type "eth" '
        '-Phy "eth1" -Ip "SGW_USER_IP" -NumLinksOrNodes 1 '
        '-NextHop "SGW_CONTROL_NEXT_HOP" -Mac "" -MTU 1500 '
        '-ForcedEthInterface "" -EthStatsEnabled false -VlanId 0 '
        '-VlanUserPriority 0 -NumVlan 1 -UniqueVlanAddr false;')
        self.mock_tcl_handler.execute.assert_called_once_with(cmd)

    def test__configure_sut_param(self):
        _params = {'name': 'name'}
        self.ls_tcl_client._configure_sut_param('name', _params)
        self.mock_tcl_handler.execute.assert_called_once_with(
            'ls::create -Sut-name -under $p_ -Name "name";')

    def test__configure_dmf_param(self):
        _params = {"mainflows": [{"library": '111',
                                  "name": "Basic UDP"}],
                   "instanceGroups": [{
                       "mainflowIdx": 0,
                       "mixType": "",
                       "rate": 0.0,
                       "rows": [{
                           "clientPort": 0,
                           "context": 0,
                           "node": 0,
                           "overridePort": "false",
                           "ratingGroup": 0,
                           "role": 0,
                           "serviceId": 0,
                           "transport": "Any"}]
                   }]}
        self.ls_tcl_client._get_library_id = mock.Mock(return_value='111')
        res = self.ls_tcl_client._configure_dmf_param('name', _params)
        self.assertEqual(5, self.mock_tcl_handler.execute.call_count)
        self.assertIsNone(res)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('ls::create -Dmf-name -under $p_ ;'),
            mock.call('ls::perform AddDmfMainflow $p_.Dmf 111 "Basic UDP"'),
            mock.call('ls::config $p_.Dmf.InstanceGroup(0) -mixType '),
            mock.call('ls::config $p_.Dmf.InstanceGroup(0) -rate 0.0'),
            mock.call('ls::config $p_.Dmf.InstanceGroup(0).Row(0) -Node 0 '
                      '-OverridePort false -ClientPort 0 -Context 0 -Role 0 '
                      '-PreferredTransport Any -RatingGroup 0 '
                      '-ServiceID 0'),
        ])

    def test__configure_dmf_param_no_instance_groups(self):
        _params = {"mainflows": [{"library": '111',
                                  "name": "Basic UDP"}]}
        self.ls_tcl_client._get_library_id = mock.Mock(return_value='111')
        res = self.ls_tcl_client._configure_dmf_param('name', _params)
        self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
        self.assertIsNone(res)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('ls::create -Dmf-name -under $p_ ;'),
            mock.call('ls::perform AddDmfMainflow $p_.Dmf 111 "Basic UDP"'),
        ])

    def test__configure_reservation(self):
        _reservation = copy.deepcopy(RESERVATIONS[0])
        self.ls_tcl_client.resolve_test_server_name = mock.Mock(
            return_value='4')
        res = self.ls_tcl_client._configure_reservation(_reservation)
        self.assertIsNone(res)
        self.assertEqual(4, self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('set reservation_ [ls::create Reservation -under $test_]'),
            mock.call('ls::config $reservation_ -TsIndex 0 -TsId 4 ' + \
                      '-TsName "TestServer_1"'),
            mock.call('set physubnet_ [ls::create PhySubnet -under $reservation_]'),
            mock.call('ls::config $physubnet_ -Name "eth1" ' + \
                      '-Base "10.42.32.100" -Mask "/24" -NumIps 20'),
        ])

    def test__configure_preresolved_arp(self):
        _arp = [{'StartingAddress': '10.81.1.10',
                 'NumNodes': 1}]
        res = self.ls_tcl_client._configure_preresolved_arp(_arp)
        self.mock_tcl_handler.execute.assert_called_once()
        self.assertIsNone(res)
        self.mock_tcl_handler.execute.assert_called_once_with(
            'ls::create PreResolvedArpAddress -under $tss_ ' + \
            '-StartingAddress "10.81.1.10" -NumNodes 1')

    def test__configure_preresolved_arp_none(self):
        res = self.ls_tcl_client._configure_preresolved_arp(None)
        self.assertIsNone(res)
        self.mock_tcl_handler.execute.assert_not_called()

    def test_delete_test_session(self):
        self.assertRaises(NotImplementedError,
                          self.ls_tcl_client.delete_test_session, {})

    def test__save_test_session(self):
        self.mock_tcl_handler.execute.side_effect = [TCL_SUCCESS_RESPONSE,
                                                     TCL_SUCCESS_RESPONSE]
        res = self.ls_tcl_client._save_test_session()
        self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
        self.assertIsNone(res)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('ls::perform Validate -TestSession $test_'),
            mock.call('ls::save $test_ -overwrite'),
        ])

    def test__save_test_session_invalid(self):
        self.mock_tcl_handler.execute.side_effect = ['Invalid', 'Errors']
        self.assertRaises(exceptions.LandslideTclException,
                          self.ls_tcl_client._save_test_session)
        self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call('ls::perform Validate -TestSession $test_'),
            mock.call('ls::get $test_ -ErrorsAndWarnings'),
        ])

    def test__get_library_id_system_lib(self):
        self.mock_tcl_handler.execute.return_value = '111'
        res = self.ls_tcl_client._get_library_id('name')
        self.mock_tcl_handler.execute.assert_called_once()
        self.assertEqual('111', res)
        self.mock_tcl_handler.execute.assert_called_with(
            'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id')

    def test__get_library_id_user_lib(self):
        self.mock_tcl_handler.execute.side_effect = ['Not found', '222']
        res = self.ls_tcl_client._get_library_id('name')
        self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
        self.assertEqual('222', res)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call(
                'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id'),
            mock.call(
                'ls::get [ls::query LibraryInfo -userLibraryName name] -Id'),
        ])

    def test__get_library_id_exception(self):
        self.mock_tcl_handler.execute.side_effect = ['Not found', 'Not found']
        self.assertRaises(exceptions.LandslideTclException,
                          self.ls_tcl_client._get_library_id,
                          'name')
        self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
        self.mock_tcl_handler.execute.assert_has_calls([
            mock.call(
                'ls::get [ls::query LibraryInfo -systemLibraryName name] -Id'),
            mock.call(
                'ls::get [ls::query LibraryInfo -userLibraryName name] -Id'),
        ])


class TestLsTclHandler(unittest.TestCase):

    def setUp(self):
        self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
        self.mock_lsapi.start()

        self.addCleanup(self._cleanup)

    def _cleanup(self):
        self.mock_lsapi.stop()

    def test___init__(self, *args):
        self.ls_tcl_handler = tg_landslide.LsTclHandler()
        self.assertEqual({}, self.ls_tcl_handler.tcl_cmds)
        self.ls_tcl_handler._ls.tcl.assert_called_once()

    def test_execute(self, *args):
        self.ls_tcl_handler = tg_landslide.LsTclHandler()
        self.ls_tcl_handler.execute('command')
        self.assertIn('command', self.ls_tcl_handler.tcl_cmds)