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)