1 # Copyright (c) 2018 Intel Corporation
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
18 from yardstick.network_services.traffic_profile import landslide_profile
21 'schema': "nsb:traffic_profile:0.1",
22 'name': 'LandslideProfile',
23 'description': 'Spirent Landslide traffic profile (Data Message Flow)',
25 'traffic_type': 'LandslideProfile'
30 'name': 'Fireball UDP',
31 'description': "Basic data flow using UDP/IP (Fireball DMF)",
33 'dataProtocol': 'fb_udp',
37 'isClientPortRange': 'false'
41 'initiatingSide': 'Client',
42 'disconnectSide': 'Client',
43 'underlyingProtocol': 'none',
44 'persistentConnection': 'false'
47 'persistentConnection': 'false',
48 'transactionRate': 8.0,
50 'totalTransactions': 0,
52 'dataResponseTime': 60000,
60 'sizeDistribution': 'Fixed',
64 'intervalDistribution': 'Fixed',
65 'intervalDeviation': 10
74 'maxPacketsToForceAck': 0
79 'disableFinAckWait': 'false'
81 'disconnectType': 'FIN',
83 'connectOnly': 'false',
88 'sctpPayloadProtocolId': 0,
89 'billingIncludeSyn': 'true',
90 'billingIncludeSubflow': 'true',
91 'billingRecordPerTransaction': 'false',
93 'hostDataExpansionRatio': 1
106 class TestLandslideProfile(unittest.TestCase):
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)
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)
120 def test_execute(self):
121 ls_traffic_profile = landslide_profile.LandslideProfile(TP_CONFIG)
122 self.assertIsNone(ls_traffic_profile.execute(None))
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])
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']]))