tests: Use errno value rather than a specific int

[vpp.git] / test / test_nat64.py

diff --git a/test/test_nat64.py b/test/test_nat64.py
index 214072a..f650b8d 100644 (file)
--- a/test/test_nat64.py
+++ b/test/test_nat64.py
@@ -9,8 +9,13 @@ from io import BytesIO
 
 import scapy.compat
 from config import config
-from framework import tag_fixme_vpp_workers
-from framework import VppTestCase, VppTestRunner
+from framework import VppTestCase
+from asfframework import (
+    tag_fixme_vpp_workers,
+    tag_fixme_ubuntu2204,
+    is_distro_ubuntu2204,
+    VppTestRunner,
+)
 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
 from scapy.data import IP_PROTOS
 from scapy.layers.inet import IP, TCP, UDP, ICMP
@@ -34,6 +39,7 @@ from vpp_papi import VppEnum
 
 
 @tag_fixme_vpp_workers
+@tag_fixme_ubuntu2204
 class TestNAT64(VppTestCase):
     """NAT64 Test Cases"""
 
@@ -49,6 +55,8 @@ class TestNAT64(VppTestCase):
     def setUpClass(cls):
         super(TestNAT64, cls).setUpClass()
 
+        if is_distro_ubuntu2204 == True and not hasattr(cls, "vpp"):
+            return
         cls.tcp_port_in = 6303
         cls.tcp_port_out = 6303
         cls.udp_port_in = 6304
@@ -459,9 +467,10 @@ class TestNAT64(VppTestCase):
         # natEvent
         self.assertEqual(scapy.compat.orb(record[230]), 13)
         # natQuotaExceededEvent
-        self.assertEqual(struct.pack("I", 2), record[466])
+        self.assertEqual(struct.pack("!I", 2), record[466])
         # maxBIBEntries
-        self.assertEqual(struct.pack("I", limit), record[472])
+        self.assertEqual(struct.pack("!I", limit), record[472])
+        return len(data)
 
     def verify_ipfix_bib(self, data, is_create, src_addr):
         """
@@ -619,9 +628,10 @@ class TestNAT64(VppTestCase):
         # natEvent
         self.assertEqual(scapy.compat.orb(record[230]), 13)
         # natQuotaExceededEvent
-        self.assertEqual(struct.pack("I", 1), record[466])
+        self.assertEqual(struct.pack("!I", 1), record[466])
         # maxSessionEntries
-        self.assertEqual(struct.pack("I", limit), record[471])
+        self.assertEqual(struct.pack("!I", limit), record[471])
+        return len(data)
 
     def test_nat64_inside_interface_handles_neighbor_advertisement(self):
         """NAT64 inside interface handles Neighbor Advertisement"""
@@ -646,7 +656,7 @@ class TestNAT64(VppTestCase):
         capture = self.pg5.get_capture(len(pkts))
         packet = capture[0]
         try:
-            self.assertEqual(packet[IPv6].src, self.pg5.local_ip6)
+            self.assertEqual(packet[IPv6].src, self.pg5.local_ip6_ll)
             self.assertEqual(packet.haslayer(ICMPv6ND_NS), 1)
             tgt = packet[ICMPv6ND_NS].tgt
         except:
@@ -1856,7 +1866,7 @@ class TestNAT64(VppTestCase):
             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
             / IPv6(src=src, dst=remote_host_ip6)
             / TCP(sport=12345, dport=25)
-        )
+        ) * 3
         self.pg0.add_stream(p)
         self.pg_enable_capture(self.pg_interfaces)
         self.pg_start()
@@ -1875,16 +1885,18 @@ class TestNAT64(VppTestCase):
             if p.haslayer(Template):
                 ipfix.add_template(p.getlayer(Template))
         # verify events in data set
+        event_count = 0
         for p in capture:
             if p.haslayer(Data):
                 data = ipfix.decode_data_set(p.getlayer(Set))
-                self.verify_ipfix_max_sessions(data, max_sessions)
+                event_count += self.verify_ipfix_max_sessions(data, max_sessions)
+        self.assertEqual(event_count, 1)
 
         p = (
             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
             / IPv6(src=self.pg0.remote_ip6, dst=remote_host_ip6)
             / TCP(sport=12345, dport=80)
-        )
+        ) * 3
         self.pg0.add_stream(p)
         self.pg_enable_capture(self.pg_interfaces)
         self.pg_start()
@@ -1892,6 +1904,7 @@ class TestNAT64(VppTestCase):
         self.vapi.ipfix_flush()
         capture = self.pg3.get_capture(1)
         # verify events in data set
+        event_count = 0
         for p in capture:
             self.assertTrue(p.haslayer(IPFIX))
             self.assertEqual(p[IP].src, self.pg3.local_ip4)
@@ -1901,7 +1914,8 @@ class TestNAT64(VppTestCase):
             self.assertEqual(p[IPFIX].observationDomainID, self.ipfix_domain_id)
             if p.haslayer(Data):
                 data = ipfix.decode_data_set(p.getlayer(Set))
-                self.verify_ipfix_max_bibs(data, max_bibs)
+                event_count += self.verify_ipfix_max_bibs(data, max_bibs)
+        self.assertEqual(event_count, 1)
 
     def test_ipfix_bib_ses(self):
         """IPFIX logging NAT64 BIB/session create and delete events"""