-# Copyright (c) 2018 Intel Corporation
+# Copyright (c) 2018-2019 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
class LandslideTrafficGen(sample_vnf.SampleVNFTrafficGen):
APP_NAME = 'LandslideTG'
- def __init__(self, name, vnfd, task_id, setup_env_helper_type=None,
+ def __init__(self, name, vnfd, setup_env_helper_type=None,
resource_helper_type=None):
if resource_helper_type is None:
resource_helper_type = LandslideResourceHelper
- super(LandslideTrafficGen, self).__init__(name, vnfd, task_id,
+ super(LandslideTrafficGen, self).__init__(name, vnfd,
setup_env_helper_type,
resource_helper_type)
self.session_profile['reservePorts'] = 'true'
self.session_profile['reservations'] = [reservation]
+ def _update_session_library_name(self, test_session):
+ """Update DMF library name in session profile"""
+ for _ts_group in test_session['tsGroups']:
+ for _tc in _ts_group['testCases']:
+ try:
+ for _mainflow in _tc['parameters']['Dmf']['mainflows']:
+ _mainflow['library'] = \
+ self.vnfd_helper.mgmt_interface['user']
+ except KeyError:
+ pass
+
@staticmethod
def _update_session_tc_params(tc_options, testcase):
for _param_key in tc_options:
_testcase_idx].update(
self._update_session_tc_params(tc_options, _testcase))
+ self._update_session_library_name(self.session_profile)
+
class LandslideResourceHelper(sample_vnf.ClientResourceHelper):
"""Landslide TG helper class"""
self._terminated.value = 1
def create_dmf(self, dmf):
- if isinstance(dmf, list):
- for _dmf in dmf:
- self._tcl.create_dmf(_dmf)
- else:
- self._tcl.create_dmf(dmf)
+ if isinstance(dmf, dict):
+ dmf = [dmf]
+ for _dmf in dmf:
+ # Update DMF library name in traffic profile
+ _dmf['dmf'].update(
+ {'library': self.vnfd_helper.mgmt_interface['user']})
+ # Create DMF on Landslide server
+ self._tcl.create_dmf(_dmf)
def delete_dmf(self, dmf):
if isinstance(dmf, list):
def create_test_session(self, test_session):
# Use tcl client to create session
test_session['library'] = self._user_id
+
+ # If no traffic duration set in test case, use predefined default value
+ # in session profile
+ test_session['duration'] = self.scenario_helper.all_options.get(
+ 'traffic_duration',
+ test_session['duration'])
+
LOG.debug("Creating session='%s'", test_session['name'])
self._tcl.create_test_session(test_session)