summaryrefslogtreecommitdiffstats
path: root/test/test_mpcap.py
blob: d77b543799f198464a075371fcbdffa183b78ed6 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
#!/usr/bin/env python3

import unittest

from framework import VppTestCase, VppTestRunner, running_extended_tests
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
import os


class TestMpcap(VppTestCase):
    """ Mpcap Unit Test Cases """

    @classmethod
    def setUpClass(cls):
        super(TestMpcap, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        super(TestMpcap, cls).tearDownClass()

    def setUp(self):
        super(TestMpcap, self).setUp()

    def tearDown(self):
        super(TestMpcap, self).tearDown()

    def test_mpcap_unittest(self):
        """ Mapped pcap file test """
        cmds = ["packet-generator new {\n"
                " name mpcap\n"
                " limit 15\n"
                " size 128-128\n"
                " interface local0\n"
                " node mpcap-unittest\n"
                " data {\n"
                "   IP6: 00:d0:2d:5e:86:85 -> 00:0d:ea:d0:00:00\n"
                "   ICMP: db00::1 -> db00::2\n"
                "   incrementing 30\n"
                "   }\n",
                "trace add pg-input 15",
                "pa en",
                "show trace",
                "show error"]

        for cmd in cmds:
            self.logger.info(self.vapi.cli(cmd))

        size = os.path.getsize("/tmp/mpcap_unittest.pcap")
        os.remove("/tmp/mpcap_unittest.pcap")
        if size != 2184:
            self.logger.critical("BUG: file size %d not 2184" % size)
            self.assertNotIn('WrongMPCAPFileSize', 'WrongMPCAPFileSize')

if __name__ == '__main__':
    unittest.main(testRunner=VppTestRunner)
cord.sw_if_index and \ self.source == r.record.input_source: return True return False def object_id(self): return ("qos-record-%s-%d" % (self.intf, self.source)) class VppQosStore(VppObject): """ QoS Store(ing) configuration """ def __init__(self, test, intf, source, value): self._test = test self.intf = intf self.source = source self.value = value def add_vpp_config(self): self._test.vapi.qos_store_enable_disable( enable=1, store={'sw_if_index': self.intf.sw_if_index, 'input_source': self.source, 'value': self.value}) self._test.registry.register(self, self._test.logger) return self def remove_vpp_config(self): self._test.vapi.qos_store_enable_disable( enable=0, store={'sw_if_index': self.intf.sw_if_index, 'input_source': self.source}) def query_vpp_config(self): rs = self._test.vapi.qos_store_dump() for r in rs: if self.intf.sw_if_index == r.store.sw_if_index and \ self.source == r.store.input_source and \ self.value == r.store.value: return True return False def object_id(self): return ("qos-store-%s-%d" % (self.intf, self.source)) class VppQosEgressMap(VppObject): """ QoS Egress Map(ping) configuration """ def __init__(self, test, id, rows): self._test = test self.id = id self.rows = rows def add_vpp_config(self): self._test.vapi.qos_egress_map_update( map={'id': self.id, 'rows': self.rows}) self._test.registry.register(self, self._test.logger) return self def remove_vpp_config(self): self._test.vapi.qos_egress_map_delete(id=self.id) def query_vpp_config(self): rs = self._test.vapi.qos_egress_map_dump() for r in rs: if self.id == r.map.id: return True return False def object_id(self): return ("qos-map-%d" % (self.id)) class VppQosMark(VppObject): """ QoS Mark(ing) configuration """ def __init__(self, test, intf, map, source): self._test = test self.intf = intf self.source = source self.map = map def add_vpp_config(self): self._test.vapi.qos_mark_enable_disable( enable=1, mark={'sw_if_index': self.intf.sw_if_index, 'map_id': self.map.id, 'output_source': self.source}) self._test.registry.register(self, self._test.logger) return self def remove_vpp_config(self): self._test.vapi.qos_mark_enable_disable( enable=0, mark={'sw_if_index': self.intf.sw_if_index, 'output_source': self.source}) def query_vpp_config(self): ms = self._test.vapi.qos_mark_dump() for m in ms: if self.intf.sw_if_index == m.mark.sw_if_index and \ self.source == m.mark.output_source and \ self.map.id == m.mark.map_id: return True return False def object_id(self): return ("qos-mark-%s-%d" % (self.intf, self.source))