X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_pnat.py;h=970249489d35894b532bd7c6ff66f6e84de1c8d9;hb=8a0a9d2600ef4da1da0b884e991a990644658963;hp=faf66e6ed534386ee443ab038831803d71f2d3e6;hpb=d9b0c6fbf7aa5bd9af84264105b39c82028a4a29;p=vpp.git

diff --git a/test/test_pnat.py b/test/test_pnat.py
index faf66e6ed53..970249489d3 100644
--- a/test/test_pnat.py
+++ b/test/test_pnat.py
@@ -242,6 +242,107 @@ class TestPNAT(VppTestCase):
             )
             self.vapi.pnat_binding_del(binding_index=binding_index[i])
 
+    def test_pnat_wildcard_proto(self):
+        """
+        PNAT test wildcard IP protocol, PNAT_PROTO for mask should be set by
+        handler
+        """
+
+        PNAT_IP4_INPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_INPUT
+        PNAT_IP4_OUTPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT
+
+        tests = [
+            {
+                "input": PNAT_IP4_INPUT,
+                "sw_if_index": self.pg0.sw_if_index,
+                "match": {"mask": 0x2, "dst": "10.10.10.10"},
+                "rewrite": {"mask": 0x2, "dst": self.pg1.remote_ip4},
+                "send": (IP(src=self.pg0.remote_ip4, dst="10.10.10.10")),
+                "reply": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)),
+            },
+            {
+                "input": PNAT_IP4_OUTPUT,
+                "sw_if_index": self.pg1.sw_if_index,
+                "match": {"mask": 0x1, "src": self.pg0.remote_ip4},
+                "rewrite": {"mask": 0x1, "src": "11.11.11.11"},
+                "send": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)),
+                "reply": (IP(src="11.11.11.11", dst=self.pg1.remote_ip4)),
+            },
+        ]
+
+        p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+        for t in tests:
+            rv = self.vapi.pnat_binding_add(match=t["match"], rewrite=t["rewrite"])
+            self.vapi.pnat_binding_attach(
+                sw_if_index=t["sw_if_index"],
+                attachment=t["input"],
+                binding_index=rv.binding_index,
+            )
+
+            reply = t["reply"]
+            reply[IP].ttl -= 1
+            rx = self.send_and_expect(self.pg0, p_ether / t["send"] * 1, self.pg1)
+            for p in rx:
+                self.validate(p[1], reply)
+
+            self.ping_check()
+
+            self.vapi.pnat_binding_detach(
+                sw_if_index=t["sw_if_index"],
+                attachment=t["input"],
+                binding_index=rv.binding_index,
+            )
+            self.vapi.pnat_binding_del(binding_index=rv.binding_index)
+
+    def test_pnat_wildcard_proto_v2(self):
+        """PNAT test wildcard IP protocol using pnat_binding_add_v2"""
+
+        PNAT_IP4_INPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_INPUT
+        PNAT_IP4_OUTPUT = VppEnum.vl_api_pnat_attachment_point_t.PNAT_IP4_OUTPUT
+
+        tests = [
+            {
+                "input": PNAT_IP4_INPUT,
+                "sw_if_index": self.pg0.sw_if_index,
+                "match": {"mask": 0x42, "dst": "10.10.10.10"},
+                "rewrite": {"mask": 0x42, "dst": self.pg1.remote_ip4},
+                "send": (IP(src=self.pg0.remote_ip4, dst="10.10.10.10")),
+                "reply": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)),
+            },
+            {
+                "input": PNAT_IP4_OUTPUT,
+                "sw_if_index": self.pg1.sw_if_index,
+                "match": {"mask": 0x41, "src": self.pg0.remote_ip4},
+                "rewrite": {"mask": 0x41, "src": "11.11.11.11"},
+                "send": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)),
+                "reply": (IP(src="11.11.11.11", dst=self.pg1.remote_ip4)),
+            },
+        ]
+
+        p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+        for t in tests:
+            rv = self.vapi.pnat_binding_add_v2(match=t["match"], rewrite=t["rewrite"])
+            self.vapi.pnat_binding_attach(
+                sw_if_index=t["sw_if_index"],
+                attachment=t["input"],
+                binding_index=rv.binding_index,
+            )
+
+            reply = t["reply"]
+            reply[IP].ttl -= 1
+            rx = self.send_and_expect(self.pg0, p_ether / t["send"] * 1, self.pg1)
+            for p in rx:
+                self.validate(p[1], reply)
+
+            self.ping_check()
+
+            self.vapi.pnat_binding_detach(
+                sw_if_index=t["sw_if_index"],
+                attachment=t["input"],
+                binding_index=rv.binding_index,
+            )
+            self.vapi.pnat_binding_del(binding_index=rv.binding_index)
+
 
 if __name__ == "__main__":
     unittest.main(testRunner=VppTestRunner)