yardstick/tests/unit/network_services/traffic_profile/test_base.py

   1 # Copyright (c) 2016-2017 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 sys
  16
  17 import mock
  18 import unittest
  19
  20 from yardstick.common import exceptions
  21 from yardstick.network_services import traffic_profile as tprofile_package
  22 from yardstick.network_services.traffic_profile import base
  23 from yardstick import tests as y_tests
  24
  25
  26 class TestTrafficProfile(unittest.TestCase):
  27     TRAFFIC_PROFILE = {
  28         "schema": "isb:traffic_profile:0.1",
  29         "name": "fixed",
  30         "description": "Fixed traffic profile to run UDP traffic",
  31         "traffic_profile": {
  32             "traffic_type": "FixedTraffic",
  33             "frame_rate": 100,  # pps
  34             "flow_number": 10,
  35             "frame_size": 64}}
  36
  37     def _get_res_mock(self, **kw):
  38         _mock = mock.MagicMock()
  39         for k, v in kw.items():
  40             setattr(_mock, k, v)
  41             return _mock
  42
  43     def test___init__(self):
  44         traffic_profile = base.TrafficProfile(self.TRAFFIC_PROFILE)
  45         self.assertEqual(self.TRAFFIC_PROFILE, traffic_profile.params)
  46
  47     def test_execute_traffic(self):
  48         traffic_profile = base.TrafficProfile(self.TRAFFIC_PROFILE)
  49         self.assertRaises(NotImplementedError,
  50                           traffic_profile.execute_traffic, {})
  51
  52     def test_get_existing_traffic_profile(self):
  53         traffic_profile_list = [
  54             'RFC2544Profile', 'FixedProfile', 'TrafficProfileGenericHTTP',
  55             'IXIARFC2544Profile', 'ProxACLProfile', 'ProxBinSearchProfile',
  56             'ProxProfile', 'ProxRampProfile']
  57         with mock.patch.dict(sys.modules, y_tests.STL_MOCKS):
  58             tprofile_package.register_modules()
  59
  60             for tp in traffic_profile_list:
  61                 traffic_profile = base.TrafficProfile.get(
  62                     {'traffic_profile': {'traffic_type': tp}})
  63                 self.assertEqual(tp, traffic_profile.__class__.__name__)
  64
  65     def test_get_non_existing_traffic_profile(self):
  66         self.assertRaises(exceptions.TrafficProfileNotImplemented,
  67                           base.TrafficProfile.get, self.TRAFFIC_PROFILE)
  68
  69
  70 class TestDummyProfile(unittest.TestCase):
  71     def test_execute(self):
  72         tp_config = {'traffic_profile': {'duration': 15}}
  73         dummy_profile = base.DummyProfile(tp_config)
  74         self.assertIsNone(dummy_profile.execute({}))
  75
  76
  77 class TrafficProfileConfigTestCase(unittest.TestCase):
  78
  79     def test__init(self):
  80         tp_config = {'traffic_profile': {'packet_sizes': {'64B': 100}}}
  81         tp_config_obj = base.TrafficProfileConfig(tp_config)
  82         self.assertEqual({'64B': 100}, tp_config_obj.packet_sizes)
  83         self.assertEqual(base.TrafficProfileConfig.DEFAULT_DURATION,
  84                          tp_config_obj.duration)
  85
  86     def test__init_set_duration(self):
  87         tp_config = {'traffic_profile': {'duration': 15}}
  88         tp_config_obj = base.TrafficProfileConfig(tp_config)
  89         self.assertEqual(base.TrafficProfileConfig.DEFAULT_SCHEMA,
  90                          tp_config_obj.schema)
  91         self.assertEqual(float(base.TrafficProfileConfig.DEFAULT_FRAME_RATE),
  92                          tp_config_obj.frame_rate)
  93         self.assertEqual(15, tp_config_obj.duration)
  94
  95     def test__parse_rate(self):
  96         tp_config = {'traffic_profile': {'packet_sizes': {'64B': 100}}}
  97         tp_config_obj = base.TrafficProfileConfig(tp_config)
  98         self.assertEqual((100.0, 'fps'), tp_config_obj.parse_rate('100  '))
  99         self.assertEqual((200.5, 'fps'), tp_config_obj.parse_rate('200.5'))
 100         self.assertEqual((300.8, 'fps'), tp_config_obj.parse_rate('300.8fps'))
 101         self.assertEqual((400.2, 'fps'),
 102                          tp_config_obj.parse_rate('400.2 fps'))
 103         self.assertEqual((500.3, '%'), tp_config_obj.parse_rate('500.3%'))
 104         self.assertEqual((600.1, '%'), tp_config_obj.parse_rate('600.1 %'))
 105
 106     def test__parse_rate_exception(self):
 107         tp_config = {'traffic_profile': {'packet_sizes': {'64B': 100}}}
 108         tp_config_obj = base.TrafficProfileConfig(tp_config)
 109         with self.assertRaises(exceptions.TrafficProfileRate):
 110             tp_config_obj.parse_rate('100Fps')
 111         with self.assertRaises(exceptions.TrafficProfileRate):
 112             tp_config_obj.parse_rate('100 kbps')