Implement upd packet headers - Closes #45

open-traffic-generator · Sep 25, 2020 · d9def3b · d9def3b
1 parent bc47190
commit d9def3b
Show file tree

Hide file tree

Showing 4 changed files with 87 additions and 4 deletions.
diff --git a/ixnetwork_open_traffic_generator/trafficitem.py b/ixnetwork_open_traffic_generator/trafficitem.py
@@ -33,6 +33,7 @@ class TrafficItem(CustomField):
         'vlan': 'vlan',
         'ipv4': 'ipv4',
         "tcp" : "tcp",
+        "udp" : "udp",
         'custom': 'custom'
     }
 
@@ -42,6 +43,7 @@ class TrafficItem(CustomField):
         'vlan': 'vlan',
         'ipv4': 'ipv4',
         "tcp" : "tcp",
+        "udp" : "udp",
         'custom': 'custom'
     }
 
@@ -102,6 +104,13 @@ class TrafficItem(CustomField):
         "ctl_fin" : "tcp.header.controlBits.finBit",
     }
 
+    _UDP = {
+        "src_port" : "udp.header.srcPort",
+        "dst_port" : "udp.header.dstPort",
+        "length" : "udp.header.length",
+        "checksum" : "udp.header.checksum",
+    }
+
     _CUSTOM = '_custom_headers'
 
     def __init__(self, ixnetworkapi):

diff --git a/tests/conftest.py b/tests/conftest.py
@@ -141,3 +141,36 @@ def b2b_devices(tx_port, rx_port):
         )
     ]
     return [tx_port, rx_port]
+
+
+@pytest.fixture
+def b2b_simple_device(tx_port, rx_port):
+    """Returns a B2B tuple of tx port and rx port each with distinct device
+        groups of ethernet and ipv4 devices
+        """
+    from abstract_open_traffic_generator.device import Device, Ethernet, Vlan, Ipv4
+    from abstract_open_traffic_generator.device import Pattern
+
+    tx_port.devices = [
+        Device(name='Tx Devices Ipv4',
+               device_count=1,
+               choice=Ipv4(name='Tx Ipv4',
+                           address=Pattern('1.1.1.1'),
+                           prefix=Pattern('24'),
+                           gateway=Pattern('1.1.2.1'),
+                           ethernet=Ethernet(name='Tx Ipv4 Eth', vlans=[Vlan(name='Tx Ipv4 Vlan')])
+                           )
+               )
+    ]
+    rx_port.devices = [
+        Device(name='Rx Devices Ipv4',
+               device_count=1,
+               choice=Ipv4(name='Rx Ipv4',
+                           address=Pattern('1.1.1.1'),
+                           prefix=Pattern('24'),
+                           gateway=Pattern('1.1.2.1'),
+                           ethernet=Ethernet(name='Rx Ipv4 Eth', vlans=[Vlan(name='Rx Ipv4 Vlan')])
+                           )
+               ),
+    ]
+    return [tx_port, rx_port]
diff --git a/tests/test_flow_tcp.py b/tests/test_flow_tcp.py
@@ -2,12 +2,9 @@
 from abstract_open_traffic_generator.flow import *
 from abstract_open_traffic_generator.flow_ipv4 import *
 from abstract_open_traffic_generator.config import *
-from abstract_open_traffic_generator.control import FlowTransmit
-from abstract_open_traffic_generator.result import PortRequest, FlowRequest
-
 
 def test_flow_tcp(serializer, tx_port, rx_port, b2b_simple_device, api):
-    """Pfc pause lossless test traffic configuration
+    """TCP Flow test traffic configuration
     """
     tcp_endpoint = DeviceTxRx(tx_device_names=[b2b_simple_device[0].devices[0].name],
                                    rx_device_names=[b2b_simple_device[1].devices[0].name])

diff --git a/tests/test_flow_udp.py b/tests/test_flow_udp.py
@@ -0,0 +1,44 @@
+import pytest
+from abstract_open_traffic_generator.flow import *
+from abstract_open_traffic_generator.flow_ipv4 import *
+from abstract_open_traffic_generator.config import *
+
+
+def test_flow_udp(serializer, tx_port, rx_port, b2b_simple_device, api):
+    """UDP Flow test traffic configuration
+    """
+    udp_endpoint = DeviceTxRx(tx_device_names=[b2b_simple_device[0].devices[0].name],
+                              rx_device_names=[b2b_simple_device[1].devices[0].name])
+
+    test_dscp = Priority(Dscp(phb=Pattern(Dscp.PHB_CS7, ingress_result_name='phb')))
+    udp_header = Udp(src_port=Pattern(Counter(start="12001",step="2",count=100)),
+                     dst_port=Pattern("20", ingress_result_name="UDP dst port"))
+    udp_flow = Flow(name='UDP Flow',
+                    tx_rx=TxRx(udp_endpoint),
+                    packet=[
+                        Header(Ethernet()),
+                        Header(Vlan()),
+                        Header(Ipv4(priority=test_dscp)),
+                        Header(udp_header)
+                    ],
+                    size=Size(128),
+                    rate=Rate('line', 50),
+                    duration=Duration(Fixed(packets=0)))
+
+    config = Config(
+        ports=[
+            tx_port,
+            rx_port
+        ],
+        flows=[
+            udp_flow
+        ]
+    )
+    print(serializer.json(config))
+
+    api.set_config(None)
+    api.set_config(config)
+
+
+if __name__ == '__main__':
+    pytest.main(['-s', __file__])