aboutsummaryrefslogtreecommitdiffstats
path: root/test/test_pnat.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_pnat.py')
-rw-r--r--test/test_pnat.py242
1 files changed, 143 insertions, 99 deletions
diff --git a/test/test_pnat.py b/test/test_pnat.py
index d5b60050691..faf66e6ed53 100644
--- a/test/test_pnat.py
+++ b/test/test_pnat.py
@@ -8,7 +8,8 @@ from vpp_papi import VppEnum
class TestPNAT(VppTestCase):
- """ PNAT Test Case """
+ """PNAT Test Case"""
+
maxDiff = None
@classmethod
@@ -42,147 +43,187 @@ class TestPNAT(VppTestCase):
self.assertEqual(rx, expected)
def ping_check(self):
- """ Verify non matching traffic works. """
+ """Verify non matching traffic works."""
p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
- icmpecho = (IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
- ICMP())
- reply = (IP(src=self.pg0.local_ip4, dst=self.pg0.remote_ip4) /
- ICMP(type='echo-reply'))
- rx = self.send_and_expect(self.pg0, p_ether/icmpecho * 1, self.pg0)
+ icmpecho = IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / ICMP()
+ reply = IP(src=self.pg0.local_ip4, dst=self.pg0.remote_ip4) / ICMP(
+ type="echo-reply"
+ )
+ rx = self.send_and_expect(self.pg0, p_ether / icmpecho * 1, self.pg0)
for p in rx:
reply[IP].id = p[IP].id
self.validate(p[1], reply)
def test_pnat(self):
- """ PNAT test """
+ """PNAT test"""
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
+ 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': 0xa, 'dst': '10.10.10.10', 'proto': 17,
- 'dport': 6871},
- 'rewrite': {'mask': 0x2, 'dst': self.pg1.remote_ip4},
- 'send': (IP(src=self.pg0.remote_ip4, dst='10.10.10.10') /
- UDP(dport=6871)),
- 'reply': (IP(src=self.pg0.remote_ip4,
- dst=self.pg1.remote_ip4) /
- UDP(dport=6871))
+ "input": PNAT_IP4_INPUT,
+ "sw_if_index": self.pg0.sw_if_index,
+ "match": {
+ "mask": 0xA,
+ "dst": "10.10.10.10",
+ "proto": 17,
+ "dport": 6871,
+ },
+ "rewrite": {"mask": 0x2, "dst": self.pg1.remote_ip4},
+ "send": (
+ IP(src=self.pg0.remote_ip4, dst="10.10.10.10") / UDP(dport=6871)
+ ),
+ "reply": (
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
+ / UDP(dport=6871)
+ ),
},
{
- 'input': PNAT_IP4_OUTPUT,
- 'sw_if_index': self.pg1.sw_if_index,
- 'match': {'mask': 0x9, 'src': self.pg0.remote_ip4, 'proto': 17,
- 'dport': 6871},
- 'rewrite': {'mask': 0x1, 'src': '11.11.11.11'},
- 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- UDP(dport=6871)),
- 'reply': (IP(src='11.11.11.11', dst=self.pg1.remote_ip4) /
- UDP(dport=6871))
+ "input": PNAT_IP4_OUTPUT,
+ "sw_if_index": self.pg1.sw_if_index,
+ "match": {
+ "mask": 0x9,
+ "src": self.pg0.remote_ip4,
+ "proto": 17,
+ "dport": 6871,
+ },
+ "rewrite": {"mask": 0x1, "src": "11.11.11.11"},
+ "send": (
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
+ / UDP(dport=6871)
+ ),
+ "reply": (
+ IP(src="11.11.11.11", dst=self.pg1.remote_ip4) / UDP(dport=6871)
+ ),
},
{
- 'input': PNAT_IP4_INPUT,
- 'sw_if_index': self.pg0.sw_if_index,
- 'match': {'mask': 0xa, 'dst': '10.10.10.10', 'proto': 17,
- 'dport': 6871},
- 'rewrite': {'mask': 0xa, 'dst': self.pg1.remote_ip4,
- 'dport': 5555},
- 'send': (IP(src=self.pg0.remote_ip4, dst='10.10.10.10') /
- UDP(sport=65530, dport=6871)),
- 'reply': (IP(src=self.pg0.remote_ip4,
- dst=self.pg1.remote_ip4) /
- UDP(sport=65530, dport=5555))
+ "input": PNAT_IP4_INPUT,
+ "sw_if_index": self.pg0.sw_if_index,
+ "match": {
+ "mask": 0xA,
+ "dst": "10.10.10.10",
+ "proto": 17,
+ "dport": 6871,
+ },
+ "rewrite": {"mask": 0xA, "dst": self.pg1.remote_ip4, "dport": 5555},
+ "send": (
+ IP(src=self.pg0.remote_ip4, dst="10.10.10.10")
+ / UDP(sport=65530, dport=6871)
+ ),
+ "reply": (
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
+ / UDP(sport=65530, dport=5555)
+ ),
},
{
- 'input': PNAT_IP4_INPUT,
- 'sw_if_index': self.pg0.sw_if_index,
- 'match': {'mask': 0xa, 'dst': self.pg1.remote_ip4, 'proto': 17,
- 'dport': 6871},
- 'rewrite': {'mask': 0x8, 'dport': 5555},
- 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- UDP(dport=6871, chksum=0)),
- 'reply': (IP(src=self.pg0.remote_ip4,
- dst=self.pg1.remote_ip4) /
- UDP(dport=5555, chksum=0))
+ "input": PNAT_IP4_INPUT,
+ "sw_if_index": self.pg0.sw_if_index,
+ "match": {
+ "mask": 0xA,
+ "dst": self.pg1.remote_ip4,
+ "proto": 17,
+ "dport": 6871,
+ },
+ "rewrite": {"mask": 0x8, "dport": 5555},
+ "send": (
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
+ / UDP(dport=6871, chksum=0)
+ ),
+ "reply": (
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
+ / UDP(dport=5555, chksum=0)
+ ),
},
{
- 'input': PNAT_IP4_INPUT,
- 'sw_if_index': self.pg0.sw_if_index,
- 'match': {'mask': 0x2, 'dst': self.pg1.remote_ip4, 'proto': 1},
- 'rewrite': {'mask': 0x1, 'src': '8.8.8.8'},
- 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- ICMP()),
- 'reply': IP(src='8.8.8.8', dst=self.pg1.remote_ip4)/ICMP(),
+ "input": PNAT_IP4_INPUT,
+ "sw_if_index": self.pg0.sw_if_index,
+ "match": {"mask": 0x2, "dst": self.pg1.remote_ip4, "proto": 1},
+ "rewrite": {"mask": 0x1, "src": "8.8.8.8"},
+ "send": (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / ICMP()),
+ "reply": IP(src="8.8.8.8", dst=self.pg1.remote_ip4) / ICMP(),
},
]
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']
+ 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)
+ rx = self.send_and_expect(self.pg0, p_ether / t["send"] * 1, self.pg1)
for p in rx:
# p.show2()
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_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_show(self):
- """ PNAT show tests """
+ """PNAT show tests"""
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
+ 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': 0xa, 'dst': '10.10.10.10', 'proto': 17,
- 'dport': 6871},
- 'rewrite': {'mask': 0x2, 'dst': self.pg1.remote_ip4},
- 'send': (IP(src=self.pg0.remote_ip4, dst='10.10.10.10') /
- UDP(dport=6871)),
- 'reply': (IP(src=self.pg0.remote_ip4,
- dst=self.pg1.remote_ip4) /
- UDP(dport=6871))
+ "input": PNAT_IP4_INPUT,
+ "sw_if_index": self.pg0.sw_if_index,
+ "match": {
+ "mask": 0xA,
+ "dst": "10.10.10.10",
+ "proto": 17,
+ "dport": 6871,
+ },
+ "rewrite": {"mask": 0x2, "dst": self.pg1.remote_ip4},
+ "send": (
+ IP(src=self.pg0.remote_ip4, dst="10.10.10.10") / UDP(dport=6871)
+ ),
+ "reply": (
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
+ / UDP(dport=6871)
+ ),
},
{
- 'input': PNAT_IP4_OUTPUT,
- 'sw_if_index': self.pg1.sw_if_index,
- 'match': {'mask': 0x9, 'src': self.pg0.remote_ip4, 'proto': 17,
- 'dport': 6871},
- 'rewrite': {'mask': 0x1, 'src': '11.11.11.11'},
- 'send': (IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- UDP(dport=6871)),
- 'reply': (IP(src='11.11.11.11', dst=self.pg1.remote_ip4) /
- UDP(dport=6871))
+ "input": PNAT_IP4_OUTPUT,
+ "sw_if_index": self.pg1.sw_if_index,
+ "match": {
+ "mask": 0x9,
+ "src": self.pg0.remote_ip4,
+ "proto": 17,
+ "dport": 6871,
+ },
+ "rewrite": {"mask": 0x1, "src": "11.11.11.11"},
+ "send": (
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
+ / UDP(dport=6871)
+ ),
+ "reply": (
+ IP(src="11.11.11.11", dst=self.pg1.remote_ip4) / UDP(dport=6871)
+ ),
},
]
binding_index = []
for t in tests:
- rv = self.vapi.pnat_binding_add(match=t['match'],
- rewrite=t['rewrite'])
+ rv = self.vapi.pnat_binding_add(match=t["match"], rewrite=t["rewrite"])
binding_index.append(rv.binding_index)
- self.vapi.pnat_binding_attach(sw_if_index=t['sw_if_index'],
- attachment=t['input'],
- binding_index=rv.binding_index)
+ self.vapi.pnat_binding_attach(
+ sw_if_index=t["sw_if_index"],
+ attachment=t["input"],
+ binding_index=rv.binding_index,
+ )
rv, l = self.vapi.pnat_bindings_get()
self.assertEqual(len(l), len(tests))
@@ -194,10 +235,13 @@ class TestPNAT(VppTestCase):
self.logger.info(self.vapi.cli("show pnat interfaces"))
for i, t in enumerate(tests):
- self.vapi.pnat_binding_detach(sw_if_index=t['sw_if_index'],
- attachment=t['input'],
- binding_index=binding_index[i])
+ self.vapi.pnat_binding_detach(
+ sw_if_index=t["sw_if_index"],
+ attachment=t["input"],
+ binding_index=binding_index[i],
+ )
self.vapi.pnat_binding_del(binding_index=binding_index[i])
-if __name__ == '__main__':
+
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)