X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_gtpu.py;h=23f1bd8f69435aa67c25849827a6f10c775599ae;hb=548f75744915c8e1c5e59fb866af0d912d1173a1;hp=f5b1089757e229529d90caded4e9f4ea9f69f91f;hpb=a5b2eec0535f9025319a752891d77ff9948ae0df;p=vpp.git diff --git a/test/test_gtpu.py b/test/test_gtpu.py index f5b1089757e..23f1bd8f694 100644 --- a/test/test_gtpu.py +++ b/test/test_gtpu.py @@ -8,10 +8,97 @@ from template_bd import BridgeDomain from scapy.layers.l2 import Ether, Raw from scapy.layers.inet import IP, UDP +from scapy.layers.inet6 import IPv6 from scapy.contrib.gtp import GTP_U_Header from scapy.utils import atol +class TestGtpuUDP(VppTestCase): + """ GTPU UDP ports Test Case """ + + def setUp(self): + super(TestGtpuUDP, self).setUp() + + self.dport = 2152 + + self.ip4_err = 0 + self.ip6_err = 0 + + self.create_pg_interfaces(range(1)) + for pg in self.pg_interfaces: + pg.admin_up() + self.pg0.config_ip4() + self.pg0.config_ip6() + + def _check_udp_port_ip4(self, enabled=True): + + pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) / + IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / + UDP(sport=self.dport, dport=self.dport, chksum=0)) + + self.pg0.add_stream(pkt) + self.pg_start() + + err = self.statistics.get_counter( + '/err/ip4-udp-lookup/no listener for dst port')[0] + + if enabled: + self.assertEqual(err, self.ip4_err) + else: + self.assertEqual(err, self.ip4_err + 1) + + self.ip4_err = err + + def _check_udp_port_ip6(self, enabled=True): + + pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) / + IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) / + UDP(sport=self.dport, dport=self.dport, chksum=0)) + + self.pg0.add_stream(pkt) + self.pg_start() + + err = self.statistics.get_counter( + '/err/ip6-udp-lookup/no listener for dst port')[0] + + if enabled: + self.assertEqual(err, self.ip6_err) + else: + self.assertEqual(err, self.ip6_err + 1) + + self.ip6_err = err + + def test_udp_port(self): + """ test UDP ports + Check if there are no udp listeners before gtpu is enabled + """ + + # UDP ports should be disabled unless a tunnel is configured + self._check_udp_port_ip4(False) + self._check_udp_port_ip6(False) + + r = self.vapi.gtpu_add_del_tunnel(src_addr=self.pg0.local_ip4n, + dst_addr=self.pg0.remote_ip4n) + + # UDP port 2152 enabled for ip4 + self._check_udp_port_ip4() + + r = self.vapi.gtpu_add_del_tunnel(is_ipv6=1, + src_addr=self.pg0.local_ip6n, + dst_addr=self.pg0.remote_ip6n) + + # UDP port 2152 enabled for ip6 + self._check_udp_port_ip6() + + r = self.vapi.gtpu_add_del_tunnel(is_add=0, + src_addr=self.pg0.local_ip4n, + dst_addr=self.pg0.remote_ip4n) + + r = self.vapi.gtpu_add_del_tunnel(is_add=0, is_ipv6=1, + src_addr=self.pg0.local_ip6n, + dst_addr=self.pg0.remote_ip6n) + + class TestGtpu(BridgeDomain, VppTestCase): """ GTPU Test Case """ @@ -86,7 +173,7 @@ class TestGtpu(BridgeDomain, VppTestCase): self.pg_start() - # Pick first received frame and check if it's corectly encapsulated. + # Pick first received frame and check if it's correctly encapsulated. out = self.pg0.get_capture(1) pkt = out[0] self.check_encapsulation(pkt, self.single_tunnel_bd) @@ -105,7 +192,7 @@ class TestGtpu(BridgeDomain, VppTestCase): self.pg_start() - # Get packet from each tunnel and assert it's corectly encapsulated. + # Get packet from each tunnel and assert it's correctly encapsulated. out = self.pg0.get_capture(self.n_ucast_tunnels) for pkt in out: self.check_encapsulation(pkt, self.ucast_flood_bd, True) @@ -123,7 +210,7 @@ class TestGtpu(BridgeDomain, VppTestCase): self.pg_start() - # Pick first received frame and check if it's corectly encapsulated. + # Pick first received frame and check if it's correctly encapsulated. out = self.pg0.get_capture(1) pkt = out[0] self.check_encapsulation(pkt, self.mcast_flood_bd, @@ -168,7 +255,7 @@ class TestGtpu(BridgeDomain, VppTestCase): teid=teid, is_add=is_add) if r.sw_if_index == 0xffffffff: - raise "bad sw_if_index" + raise ValueError("bad sw_if_index: ~0") @classmethod def add_shared_mcast_dst_load(cls): @@ -278,18 +365,23 @@ class TestGtpu(BridgeDomain, VppTestCase): super(TestGtpu, cls).tearDownClass() raise + @classmethod + def tearDownClass(cls): + super(TestGtpu, cls).tearDownClass() + # Method to define VPP actions before tear down of the test case. # Overrides tearDown method in VppTestCase class. # @param self The object pointer. def tearDown(self): super(TestGtpu, self).tearDown() - if not self.vpp_dead: - self.logger.info(self.vapi.cli("show bridge-domain 11 detail")) - self.logger.info(self.vapi.cli("show bridge-domain 12 detail")) - self.logger.info(self.vapi.cli("show bridge-domain 13 detail")) - self.logger.info(self.vapi.cli("show int")) - self.logger.info(self.vapi.cli("show gtpu tunnel")) - self.logger.info(self.vapi.cli("show trace")) + + def show_commands_at_teardown(self): + self.logger.info(self.vapi.cli("show bridge-domain 11 detail")) + self.logger.info(self.vapi.cli("show bridge-domain 12 detail")) + self.logger.info(self.vapi.cli("show bridge-domain 13 detail")) + self.logger.info(self.vapi.cli("show int")) + self.logger.info(self.vapi.cli("show gtpu tunnel")) + self.logger.info(self.vapi.cli("show trace")) if __name__ == '__main__':