path: root/yardstick/tests/unit/network_services/traffic_profile/test_rfc2544.py

# Copyright (c) 2016-2017 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import datetime

import mock
from trex_stl_lib import api as Pkt
from trex_stl_lib import trex_stl_client
from trex_stl_lib import trex_stl_packet_builder_scapy
from trex_stl_lib import trex_stl_streams

from yardstick.common import constants
from yardstick.network_services.traffic_profile import rfc2544
from yardstick.tests.unit import base


class TestRFC2544Profile(base.BaseUnitTestCase):
    TRAFFIC_PROFILE = {
        "schema": "isb:traffic_profile:0.1",
        "name": "fixed",
        "description": "Fixed traffic profile to run UDP traffic",
        "traffic_profile": {
            "traffic_type": "FixedTraffic",
            "frame_rate": 100,
            "flow_number": 10,
            "frame_size": 64}}

    PROFILE = {'description': 'Traffic profile to run RFC2544 latency',
               'name': 'rfc2544',
               'traffic_profile': {'traffic_type': 'RFC2544Profile',
                                   'frame_rate': 100},
               'downlink_0':
                   {'ipv4':
                        {'outer_l2':
                             {'framesize':
                                  {'64B': '100', '1518B': '0',
                                   '128B': '0', '1400B': '0',
                                   '256B': '0', '373b': '0',
                                   '570B': '0'}},
                         'outer_l3v4':
                             {'dstip4': '1.1.1.1-1.15.255.255',
                              'proto': 'udp',
                              'srcip4': '90.90.1.1-90.105.255.255',
                              'dscp': 0, 'ttl': 32, 'count': 1},
                         'outer_l4':
                             {'srcport': '2001',
                              'dsrport': '1234', 'count': 1}}},
               'uplink_0':
                   {'ipv4':
                        {'outer_l2':
                             {'framesize':
                                  {'64B': '100', '1518B': '0',
                                   '128B': '0', '1400B': '0',
                                   '256B': '0', '373b': '0',
                                   '570B': '0'}},
                         'outer_l3v4':
                             {'dstip4': '9.9.1.1-90.105.255.255',
                              'proto': 'udp',
                              'srcip4': '1.1.1.1-1.15.255.255',
                              'dscp': 0, 'ttl': 32, 'count': 1},
                         'outer_l4':
                             {'dstport': '2001',
                              'srcport': '1234', 'count': 1}}},
               'schema': 'isb:traffic_profile:0.1'}

    def test___init__(self):
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        self.assertEqual(rfc2544_profile.max_rate, rfc2544_profile.rate)
        self.assertEqual(0, rfc2544_profile.min_rate)

    def test_stop_traffic(self):
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        mock_generator = mock.Mock()
        rfc2544_profile.stop_traffic(traffic_generator=mock_generator)
        mock_generator.client.stop.assert_called_once()
        mock_generator.client.reset.assert_called_once()
        mock_generator.client.remove_all_streams.assert_called_once()

    def test_execute_traffic(self):
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        mock_generator = mock.Mock()
        mock_generator.networks = {
            'downlink_0': ['xe0', 'xe1'],
            'uplink_0': ['xe2', 'xe3'],
            'downlink_1': []}
        mock_generator.port_num.side_effect = [10, 20, 30, 40]
        mock_generator.rfc2544_helper.correlated_traffic = False
        rfc2544_profile.params = {
            'downlink_0': 'profile1',
            'uplink_0': 'profile2'}

        with mock.patch.object(rfc2544_profile, '_create_profile') as \
                mock_create_profile:
            rfc2544_profile.execute_traffic(traffic_generator=mock_generator)
        mock_create_profile.assert_has_calls([
            mock.call('profile1', rfc2544_profile.rate, mock.ANY, False),
            mock.call('profile1', rfc2544_profile.rate, mock.ANY, False),
            mock.call('profile2', rfc2544_profile.rate, mock.ANY, False),
            mock.call('profile2', rfc2544_profile.rate, mock.ANY, False)])
        mock_generator.client.add_streams.assert_has_calls([
            mock.call(mock.ANY, ports=[10]),
            mock.call(mock.ANY, ports=[20]),
            mock.call(mock.ANY, ports=[30]),
            mock.call(mock.ANY, ports=[40])])
        mock_generator.client.start(ports=[10, 20, 30, 40],
                                    duration=rfc2544_profile.config.duration,
                                    force=True)

    @mock.patch.object(trex_stl_streams, 'STLProfile')
    def test__create_profile(self, mock_stl_profile):
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        port_pg_id = mock.ANY
        profile_data = {'packetid_1': {'outer_l2': {'framesize': 'imix_info'}}}
        rate = 100
        with mock.patch.object(rfc2544_profile, '_create_imix_data') as \
                mock_create_imix, \
                mock.patch.object(rfc2544_profile, '_create_vm') as \
                mock_create_vm, \
                mock.patch.object(rfc2544_profile, '_create_streams') as \
                mock_create_streams:
            mock_create_imix.return_value = 'imix_data'
            mock_create_streams.return_value = ['stream1']
            rfc2544_profile._create_profile(profile_data, rate, port_pg_id,
                                            True)

        mock_create_imix.assert_called_once_with('imix_info')
        mock_create_vm.assert_called_once_with(
            {'outer_l2': {'framesize': 'imix_info'}})
        mock_create_streams.assert_called_once_with('imix_data', 100,
                                                    port_pg_id, True)
        mock_stl_profile.assert_called_once_with(['stream1'])

    def test__create_imix_data_mode_DIP(self):
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        data = {'64B': 50, '128B': 50}
        self.assertEqual(
            {'64': 50.0, '128': 50.0},
            rfc2544_profile._create_imix_data(
                data, weight_mode=constants.DISTRIBUTION_IN_PACKETS))
        data = {'64B': 1, '128b': 3}
        self.assertEqual(
            {'64': 25.0, '128': 75.0},
            rfc2544_profile._create_imix_data(
                data, weight_mode=constants.DISTRIBUTION_IN_PACKETS))
        data = {}
        self.assertEqual(
            {},
            rfc2544_profile._create_imix_data(
                data, weight_mode=constants.DISTRIBUTION_IN_PACKETS))

    def test__create_imix_data_mode_DIB(self):
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        data = {'64B': 25, '128B': 25, '512B': 25, '1518B': 25}
        byte_total = 64 * 25 + 128 * 25 + 512 * 25 + 1518 * 25
        self.assertEqual(
            {'64': 64 * 25.0 * 100 / byte_total,
             '128': 128 * 25.0 * 100 / byte_total,
             '512': 512 * 25.0 * 100 / byte_total,
             '1518': 1518 * 25.0 * 100/ byte_total},
            rfc2544_profile._create_imix_data(
                data, weight_mode=constants.DISTRIBUTION_IN_BYTES))
        data = {}
        self.assertEqual(
            {},
            rfc2544_profile._create_imix_data(
                data, weight_mode=constants.DISTRIBUTION_IN_BYTES))
        data = {'64B': 100}
        self.assertEqual(
            {'64': 100.0},
            rfc2544_profile._create_imix_data(
                data, weight_mode=constants.DISTRIBUTION_IN_BYTES))

    def test__create_vm(self):
        packet = {'outer_l2': 'l2_definition'}
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        with mock.patch.object(rfc2544_profile, '_set_outer_l2_fields') as \
                mock_l2_fileds:
            rfc2544_profile._create_vm(packet)
        mock_l2_fileds.assert_called_once_with('l2_definition')

    @mock.patch.object(trex_stl_packet_builder_scapy, 'STLPktBuilder',
                       return_value='packet')
    def test__create_single_packet(self, mock_pktbuilder):
        size = 128
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        rfc2544_profile.ether_packet = Pkt.Eth()
        rfc2544_profile.ip_packet = Pkt.IP()
        rfc2544_profile.udp_packet = Pkt.UDP()
        rfc2544_profile.trex_vm = 'trex_vm'
        base_pkt = (rfc2544_profile.ether_packet / rfc2544_profile.ip_packet /
                    rfc2544_profile.udp_packet)
        pad = (size - len(base_pkt)) * 'x'
        output = rfc2544_profile._create_single_packet(size=size)
        mock_pktbuilder.assert_called_once_with(pkt=base_pkt / pad,
                                                vm='trex_vm')
        self.assertEqual(output, 'packet')

    @mock.patch.object(trex_stl_packet_builder_scapy, 'STLPktBuilder',
                       return_value='packet')
    def test__create_single_packet_qinq(self, mock_pktbuilder):
        size = 128
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        rfc2544_profile.ether_packet = Pkt.Eth()
        rfc2544_profile.ip_packet = Pkt.IP()
        rfc2544_profile.udp_packet = Pkt.UDP()
        rfc2544_profile.trex_vm = 'trex_vm'
        rfc2544_profile.qinq = True
        rfc2544_profile.qinq_packet = Pkt.Dot1Q(vlan=1) / Pkt.Dot1Q(vlan=2)
        base_pkt = (rfc2544_profile.ether_packet /
                    rfc2544_profile.qinq_packet / rfc2544_profile.ip_packet /
                    rfc2544_profile.udp_packet)
        pad = (size - len(base_pkt)) * 'x'
        output = rfc2544_profile._create_single_packet(size=size)
        mock_pktbuilder.assert_called_once_with(pkt=base_pkt / pad,
                                                vm='trex_vm')
        self.assertEqual(output, 'packet')

    @mock.patch.object(trex_stl_streams, 'STLFlowLatencyStats')
    @mock.patch.object(trex_stl_streams, 'STLTXCont')
    @mock.patch.object(trex_stl_client, 'STLStream')
    def test__create_streams(self, mock_stream, mock_txcont, mock_latency):
        imix_data = {'64': 25, '512': 75}
        rate = 35
        port_pg_id = rfc2544.PortPgIDMap()
        port_pg_id.add_port(10)
        mock_stream.side_effect = ['stream1', 'stream2']
        mock_txcont.side_effect = ['txcont1', 'txcont2']
        mock_latency.side_effect = ['latency1', 'latency2']
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        with mock.patch.object(rfc2544_profile, '_create_single_packet'):
            output = rfc2544_profile._create_streams(imix_data, rate,
                                                     port_pg_id, True)
        self.assertEqual(['stream1', 'stream2'], output)
        mock_latency.assert_has_calls([
            mock.call(pg_id=1), mock.call(pg_id=2)])
        mock_txcont.assert_has_calls([
            mock.call(percentage=float(25 * 35) / 100),
            mock.call(percentage=float(75 * 35) / 100)], any_order=True)

    def test_get_drop_percentage(self):
        rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
        samples = [
            {'xe1': {'tx_throughput_fps': 110,
                     'rx_throughput_fps': 101,
                     'out_packets': 2100,
                     'in_packets': 2010,
                     'out_bytes': 134400,
                     'in_bytes': 128640,
                     'timestamp': datetime.datetime(2000, 1, 1, 1, 1, 1, 1)},
             'xe2': {'tx_throughput_fps': 210,
                     'rx_throughput_fps': 201,
                     'out_packets': 4100,
                     'in_packets': 4010,
                     'out_bytes': 262400,
                     'in_bytes': 256640,
                     'timestamp': datetime.datetime(2000, 1, 1, 1, 1, 1, 1)}},
            {'xe1': {'tx_throughput_fps': 156,
                     'rx_throughput_fps': 108,
                     'out_packets': 2110,
                     'in_packets': 2040,
                     'out_bytes': 135040,
                     'in_bytes': 130560,
                     'latency': 'Latency1',
                     'timestamp': datetime.datetime(2000, 1, 1, 1, 1, 1, 31)},
             'xe2': {'tx_throughput_fps': 253,
                     'rx_throughput_fps': 215,
                     'out_packets': 4150,
                     'in_packets': 4010,
                     'out_bytes': 265600,
                     'in_bytes': 256640,
                     'latency': 'Latency2',
                     'timestamp': datetime.datetime(2000, 1, 1, 1, 1, 1, 31)}}
        ]
        completed, output = rfc2544_profile.get_drop_percentage(
            samples, 0, 0, False, 0.1)
        expected = {'DropPercentage': 50.0,
                    'Latency': {'xe1': 'Latency1', 'xe2': 'Latency2'},
                    'RxThroughput': 1000000.0,
                    'TxThroughput': 2000000.0,
                    'RxThroughputBps': 64000000.0,
                    'TxThroughputBps': 128000000.0,
                    'CurrentDropPercentage': 50.0,
                    'Rate': 100.0,
                    'Throughput': 1000000.0}
        self.assertEqual(expected, output)
        self.assertFalse(completed)


class PortPgIDMapTestCase(base.BaseUnitTestCase):

    def test_add_port(self):
        port_pg_id_map = rfc2544.PortPgIDMap()
        port_pg_id_map.add_port(10)
        self.assertEqual(10, port_pg_id_map._last_port)
        self.assertEqual([], port_pg_id_map._port_pg_id_map[10])

    def test_get_pg_ids(self):
        port_pg_id_map = rfc2544.PortPgIDMap()
        port_pg_id_map.add_port(10)
        port_pg_id_map.increase_pg_id()
        port_pg_id_map.increase_pg_id()
        port_pg_id_map.add_port(20)
        port_pg_id_map.increase_pg_id()
        self.assertEqual([1, 2], port_pg_id_map.get_pg_ids(10))
        self.assertEqual([3], port_pg_id_map.get_pg_ids(20))
        self.assertEqual([], port_pg_id_map.get_pg_ids(30))

    def test_increase_pg_id_no_port(self):
        port_pg_id_map = rfc2544.PortPgIDMap()
        self.assertIsNone(port_pg_id_map.increase_pg_id())

    def test_increase_pg_id_last_port(self):
        port_pg_id_map = rfc2544.PortPgIDMap()
        port_pg_id_map.add_port(10)
        self.assertEqual(1, port_pg_id_map.increase_pg_id())
        self.assertEqual([1], port_pg_id_map.get_pg_ids(10))
        self.assertEqual(10, port_pg_id_map._last_port)

    def test_increase_pg_id(self):
        port_pg_id_map = rfc2544.PortPgIDMap()
        port_pg_id_map.add_port(10)
        port_pg_id_map.increase_pg_id()
        self.assertEqual(2, port_pg_id_map.increase_pg_id(port=20))
        self.assertEqual([1], port_pg_id_map.get_pg_ids(10))
        self.assertEqual([2], port_pg_id_map.get_pg_ids(20))
        self.assertEqual(20, port_pg_id_map._last_port)