yardstick/tests/unit/network_services/traffic_profile/test_landslide_profile.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 unittest
  17
  18 from yardstick.network_services.traffic_profile import landslide_profile
  19
  20 TP_CONFIG = {
  21     'schema': "nsb:traffic_profile:0.1",
  22     'name': 'LandslideProfile',
  23     'description': 'Spirent Landslide traffic profile (Data Message Flow)',
  24     'traffic_profile': {
  25         'traffic_type': 'LandslideProfile'
  26     },
  27     'dmf_config': {
  28         'dmf': {
  29             'library': 'test',
  30             'name': 'Fireball UDP',
  31             'description': "Basic data flow using UDP/IP (Fireball DMF)",
  32             'keywords': 'UDP ',
  33             'dataProtocol': 'fb_udp',
  34             'burstCount': 1,
  35             'clientPort': {
  36                 'clientPort': 2002,
  37                 'isClientPortRange': 'false'
  38             },
  39             'serverPort': 2003,
  40             'connection': {
  41                 'initiatingSide': 'Client',
  42                 'disconnectSide': 'Client',
  43                 'underlyingProtocol': 'none',
  44                 'persistentConnection': 'false'
  45             },
  46             'protocolId': 0,
  47             'persistentConnection': 'false',
  48             'transactionRate': 8.0,
  49             'transactions': {
  50                 'totalTransactions': 0,
  51                 'retries': 0,
  52                 'dataResponseTime': 60000,
  53                 'packetSize': 64
  54             },
  55             'segment': {
  56                 'segmentSize': 64000,
  57                 'maxSegmentSize': 0
  58             },
  59             'size': {
  60                 'sizeDistribution': 'Fixed',
  61                 'sizeDeviation': 10
  62             },
  63             'interval': {
  64                 'intervalDistribution': 'Fixed',
  65                 'intervalDeviation': 10
  66             },
  67             'ipHeader': {
  68                 'typeOfService': 0,
  69                 'timeToLive': 64
  70             },
  71             'tcpConnection': {
  72                 'force3Way': 'false',
  73                 'fixedRetryTime': 0,
  74                 'maxPacketsToForceAck': 0
  75             },
  76             'tcp': {
  77                 'windowSize': 32768,
  78                 'windowScaling': -1,
  79                 'disableFinAckWait': 'false'
  80             },
  81             'disconnectType': 'FIN',
  82             'slowStart': 'false',
  83             'connectOnly': 'false',
  84             'vtag': {
  85                 'VTagMask': '0x0',
  86                 'VTagValue': '0x0'
  87             },
  88             'sctpPayloadProtocolId': 0,
  89             'billingIncludeSyn': 'true',
  90             'billingIncludeSubflow': 'true',
  91             'billingRecordPerTransaction': 'false',
  92             'tcpPush': 'false',
  93             'hostDataExpansionRatio': 1
  94         }
  95     }
  96 }
  97 DMF_OPTIONS = {
  98     'dmf': {
  99         'transactionRate': 5,
 100         'packetSize': 512,
 101         'burstCount': 1
 102     }
 103 }
 104
 105
 106 class TestLandslideProfile(unittest.TestCase):
 107
 108     def test___init__(self):
 109         ls_traffic_profile = landslide_profile.LandslideProfile(TP_CONFIG)
 110         self.assertListEqual([TP_CONFIG["dmf_config"]],
 111                              ls_traffic_profile.dmf_config)
 112
 113     def test___init__config_not_a_dict(self):
 114         _tp_config = copy.deepcopy(TP_CONFIG)
 115         _tp_config['dmf_config'] = [_tp_config['dmf_config']]
 116         ls_traffic_profile = landslide_profile.LandslideProfile(_tp_config)
 117         self.assertListEqual(_tp_config['dmf_config'],
 118                              ls_traffic_profile.dmf_config)
 119
 120     def test_execute(self):
 121         ls_traffic_profile = landslide_profile.LandslideProfile(TP_CONFIG)
 122         self.assertIsNone(ls_traffic_profile.execute(None))
 123
 124     def test_update_dmf_options_dict(self):
 125         ls_traffic_profile = landslide_profile.LandslideProfile(TP_CONFIG)
 126         ls_traffic_profile.update_dmf(DMF_OPTIONS)
 127         self.assertDictContainsSubset(DMF_OPTIONS['dmf'],
 128                                       ls_traffic_profile.dmf_config[0])
 129
 130     def test_update_dmf_options_list(self):
 131         ls_traffic_profile = landslide_profile.LandslideProfile(TP_CONFIG)
 132         _dmf_options = copy.deepcopy(DMF_OPTIONS)
 133         _dmf_options['dmf'] = [_dmf_options['dmf']]
 134         ls_traffic_profile.update_dmf(_dmf_options)
 135         self.assertTrue(all([x in ls_traffic_profile.dmf_config[0]
 136                              for x in DMF_OPTIONS['dmf']]))