smoketest: tunnel: add individual tunnel verification logic

1 files changed, 261 insertions, 67 deletions

diff --git a/smoketest/scripts/cli/test_interfaces_tunnel.py b/smoketest/scripts/cli/test_interfaces_tunnel.py
index 7611ffe26..eb015e240 100755
--- a/smoketest/scripts/cli/test_interfaces_tunnel.py
+++ b/smoketest/scripts/cli/test_interfaces_tunnel.py
@@ -14,95 +14,289 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import os
 import unittest
+import json
 
 from vyos.configsession import ConfigSession
+from vyos.configsession import ConfigSessionError
+from vyos.util import cmd
 
 from base_interfaces_test import BasicInterfaceTest
 
-class TunnelInterfaceTest(BasicInterfaceTest.BaseTest):
-    # encoding, tunnel endpoint (v4/v6), address (v4/v6)
-    _valid = [
-        ('gre', 4, 4),
-        ('gre', 4, 6),
-        ('ip6gre', 6, 4),
-        ('ip6gre', 6, 6),
-        ('gre-bridge', 4, 4),
-        ('ipip', 4, 4),
-        ('ipip', 4, 6),
-        ('ipip6', 6, 4),
-        ('ipip6', 6, 6),
-        ('ip6ip6', 6, 6),
-        ('sit', 4, 6),
-    ]
-
-    local = {
-        4: '10.100.{}.1/24',
-        6:  '2001:db8:{}::1/64',
-    }
-
-    remote = {
-        4: '192.0.{}.1',
-        6: '2002::{}:1',
-    }
-
-    address = {
-        4: '10.100.{}.1/24',
-        6:  '2001:db8:{}::1/64',
-    }
+remote_ip4 = '192.0.2.100'
+remote_ip6 = '2001:db8::ffff'
+source_if = 'dum2222'
+
+def tunnel_conf(interface):
+    tmp = cmd(f'ip -d -j link show {interface}')
+    # {'address': '2.2.2.2',
+    #  'broadcast': '192.0.2.10',
+    #  'flags': ['POINTOPOINT', 'NOARP', 'UP', 'LOWER_UP'],
+    #  'group': 'default',
+    #  'gso_max_segs': 65535,
+    #  'gso_max_size': 65536,
+    #  'ifindex': 10,
+    #  'ifname': 'tun10',
+    #  'inet6_addr_gen_mode': 'none',
+    #  'link': None,
+    #  'link_pointtopoint': True,
+    #  'link_type': 'gre',
+    #  'linkinfo': {'info_data': {'local': '2.2.2.2',
+    #                             'pmtudisc': True,
+    #                             'remote': '192.0.2.10',
+    #                             'tos': '0x1',
+    #                             'ttl': 255},
+    #               'info_kind': 'gre'},
+    #  'linkmode': 'DEFAULT',
+    #  'max_mtu': 65511,
+    #  'min_mtu': 68,
+    #  'mtu': 1476,
+    #  'num_rx_queues': 1,
+    #  'num_tx_queues': 1,
+    #  'operstate': 'UNKNOWN',
+    #  'promiscuity': 0,
+    #  'qdisc': 'noqueue',
+    #  'txqlen': 1000}
+    return json.loads(tmp)[0]
 
+class TunnelInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
-        local = {}
-        remote = {}
-        address = {}
+        super().setUp()
 
-        self._intf_dummy = ['interfaces', 'dummy']
         self._base_path = ['interfaces', 'tunnel']
-        self._interfaces = ['tun{}'.format(n) for n in range(len(self._valid))]
+        self._test_ip = False
+        self._test_mtu = False
 
-        self._test_mtu = True
-        super().setUp()
+        self.local_v4 = '192.0.2.1'
+        self.local_v6 = '2001:db8::1'
+
+        self.session.set(['interfaces', 'dummy', source_if, 'address', self.local_v4 + '/32'])
+        self.session.set(['interfaces', 'dummy', source_if, 'address', self.local_v6 + '/128'])
+
+        self._options = {
+            'tun10': ['encapsulation ipip', 'remote-ip 192.0.2.10', 'local-ip ' + self.local_v4],
+            'tun20': ['encapsulation gre',  'remote-ip 192.0.2.20', 'local-ip ' + self.local_v4],
+        }
+
+        self._interfaces = list(self._options)
 
-        for number in range(len(self._valid)):
-            dum4 = 'dum4{}'.format(number)
-            dum6 = 'dum6{}'.format(number)
+    def tearDown(self):
+        self.session.delete(['interfaces', 'dummy', source_if])
+        super().tearDown()
 
-            ipv4 = self.local[4].format(number)
-            ipv6 = self.local[6].format(number)
+    def test_ipip(self):
+        interface = 'tun100'
+        encapsulation = 'ipip'
+        local_if_addr = '10.10.10.1/24'
 
-            local.setdefault(4, {})[number] = ipv4
-            local.setdefault(6, {})[number] = ipv6
+        self.session.set(self._base_path + [interface, 'address', local_if_addr])
 
-            ipv4 = self.remote[4].format(number)
-            ipv6 = self.remote[6].format(number)
+        # Must provide an "encapsulation" for tunnel  tun10
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'encapsulation', encapsulation])
 
-            remote.setdefault(4, {})[number] = ipv4
-            remote.setdefault(6, {})[number] = ipv6
+        # Must configure either local-ip or dhcp-interface for tunnel ipip tun100
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'local-ip', self.local_v4])
 
-            ipv4 = self.address[4].format(number)
-            ipv6 = self.address[6].format(number)
+        # missing required option remote for ipip
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'remote-ip', remote_ip4])
 
-            address.setdefault(4, {})[number] = ipv4
-            address.setdefault(6, {})[number] = ipv6
+        # Configure Tunnel Source interface
+        self.session.set(self._base_path + [interface, 'source-interface', source_if])
 
-            self.session.set(self._intf_dummy + [dum4, 'address', ipv4])
-            self.session.set(self._intf_dummy + [dum6, 'address', ipv6])
         self.session.commit()
 
-        for number, (encap, p2p, addr) in enumerate(self._valid):
-            intf = 'tun%d' % number
-            tunnel = {}
-            tunnel['encapsulation'] = encap
-            tunnel['local-ip'] = local[p2p][number].split('/')[0]
-            tunnel['remote-ip'] = remote[p2p][number].split('/')[0]
-            tunnel['address'] = address[addr][number]
-            for name in tunnel:
-                self.session.set(self._base_path + [intf, name, tunnel[name]])
+        conf = tunnel_conf(interface)
+        self.assertEqual(interface, conf['ifname'])
+        self.assertEqual(encapsulation, conf['link_type'])
+        self.assertEqual(1476, conf['mtu'])
+        self.assertEqual(source_if, conf['link'])
 
-    def tearDown(self):
-        self.session.delete(self._intf_dummy)
-        super().tearDown()
+        self.assertEqual(self.local_v4, conf['linkinfo']['info_data']['local'])
+        self.assertEqual(remote_ip4,     conf['linkinfo']['info_data']['remote'])
+
+    def test_ipip6(self):
+        interface = 'tun110'
+        encapsulation = 'ipip6'
+        local_if_addr = '10.10.10.1/24'
+
+        self.session.set(self._base_path + [interface, 'address', local_if_addr])
+
+        # Must provide an "encapsulation" for tunnel  tun10
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'encapsulation', encapsulation])
+
+        # Must configure either local-ip or dhcp-interface for tunnel ipip tun100
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'local-ip', self.local_v6])
+
+        # missing required option remote for ipip
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'remote-ip', remote_ip6])
+
+        # Configure Tunnel Source interface
+        self.session.set(self._base_path + [interface, 'source-interface', source_if])
+
+        self.session.commit()
+
+        conf = tunnel_conf(interface)
+        self.assertEqual(interface, conf['ifname'])
+        self.assertEqual('tunnel6', conf['link_type'])
+        self.assertEqual(1476, conf['mtu'])
+        self.assertEqual(source_if, conf['link'])
+
+        self.assertEqual(self.local_v6, conf['linkinfo']['info_data']['local'])
+        self.assertEqual(remote_ip6,     conf['linkinfo']['info_data']['remote'])
+
+    def test_ip6ip6(self):
+        interface = 'tun120'
+        encapsulation = 'ip6ip6'
+        local_if_addr = '2001:db8:f00::1/24'
+
+        self.session.set(self._base_path + [interface, 'address', local_if_addr])
+
+        # Must provide an "encapsulation" for tunnel  tun10
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'encapsulation', encapsulation])
+
+        # Must configure either local-ip or dhcp-interface for tunnel ipip tun100
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'local-ip', self.local_v6])
+
+        # missing required option remote for ipip
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'remote-ip', remote_ip6])
+
+        # Configure Tunnel Source interface
+        self.session.set(self._base_path + [interface, 'source-interface', source_if])
+
+        self.session.commit()
+
+        conf = tunnel_conf(interface)
+        self.assertEqual(interface, conf['ifname'])
+        self.assertEqual('tunnel6', conf['link_type'])
+        self.assertEqual(1476, conf['mtu'])
+        self.assertEqual(source_if, conf['link'])
+
+        self.assertEqual(self.local_v6, conf['linkinfo']['info_data']['local'])
+        self.assertEqual(remote_ip6,     conf['linkinfo']['info_data']['remote'])
+
+    def test_gre_ipv4(self):
+        interface = 'tun200'
+        encapsulation = 'gre'
+        local_if_addr = '172.16.1.1/24'
+
+        self.session.set(self._base_path + [interface, 'address', local_if_addr])
+
+        # Must provide an "encapsulation" for tunnel  tun10
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'encapsulation', encapsulation])
+
+        # Must configure either local-ip or dhcp-interface
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'local-ip', self.local_v4])
+
+        # No assertion is raised for GRE remote-ip when missing
+        self.session.set(self._base_path + [interface, 'remote-ip', remote_ip4])
+
+        # Configure Tunnel Source interface
+        self.session.set(self._base_path + [interface, 'source-interface', source_if])
+
+        self.session.commit()
+
+        conf = tunnel_conf(interface)
+        self.assertEqual(interface, conf['ifname'])
+        self.assertEqual(encapsulation, conf['link_type'])
+        self.assertEqual(1476, conf['mtu'])
+        self.assertEqual(source_if, conf['link'])
+
+        self.assertEqual(self.local_v4, conf['linkinfo']['info_data']['local'])
+        self.assertEqual(remote_ip4,     conf['linkinfo']['info_data']['remote'])
+
+
+        def test_gre_ipv6(self):
+            interface = 'tun210'
+            encapsulation = 'ip6gre'
+            local_if_addr = '2001:db8:f01::1/24'
+
+            self.session.set(self._base_path + [interface, 'address', local_if_addr])
+
+            # Must provide an "encapsulation" for tunnel  tun10
+            with self.assertRaises(ConfigSessionError):
+                self.session.commit()
+            self.session.set(self._base_path + [interface, 'encapsulation', encapsulation])
+
+            # Must configure either local-ip or dhcp-interface
+            with self.assertRaises(ConfigSessionError):
+                self.session.commit()
+            self.session.set(self._base_path + [interface, 'local-ip', self.local_v6])
+
+            # No assertion is raised for GRE remote-ip when missing
+            self.session.set(self._base_path + [interface, 'remote-ip', remote_ip6])
+
+            # Configure Tunnel Source interface
+            self.session.set(self._base_path + [interface, 'source-interface', source_if])
+
+            self.session.commit()
+
+            conf = tunnel_conf(interface)
+            self.assertEqual(interface, conf['ifname'])
+            self.assertEqual(encapsulation, conf['link_type'])
+            self.assertEqual(1476, conf['mtu'])
+            self.assertEqual(source_if, conf['link'])
+
+            self.assertEqual(self.local_v6, conf['linkinfo']['info_data']['local'])
+            self.assertEqual(remote_ip6,     conf['linkinfo']['info_data']['remote'])
+
+
+    def test_sit(self):
+        interface = 'tun300'
+        encapsulation = 'sit'
+        local_if_addr = '172.16.2.1/24'
+
+        self.session.set(self._base_path + [interface, 'address', local_if_addr])
+
+        # Must provide an "encapsulation" for tunnel  tun10
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'encapsulation', encapsulation])
+
+        # Must configure either local-ip or dhcp-interface
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.set(self._base_path + [interface, 'local-ip', self.local_v4])
+
+        # No assertion is raised for GRE remote-ip when missing
+        self.session.set(self._base_path + [interface, 'remote-ip', remote_ip4])
+
+        # Source interface can not be used with si
+        self.session.set(self._base_path + [interface, 'source-interface', source_if])
+        with self.assertRaises(ConfigSessionError):
+            self.session.commit()
+        self.session.delete(self._base_path + [interface, 'source-interface'])
+
+        self.session.commit()
+
+        conf = tunnel_conf(interface)
+        self.assertEqual(interface, conf['ifname'])
+        self.assertEqual(encapsulation, conf['link_type'])
+        self.assertEqual(1476, conf['mtu'])
+
+        self.assertEqual(self.local_v4, conf['linkinfo']['info_data']['local'])
+        self.assertEqual(remote_ip4,     conf['linkinfo']['info_data']['remote'])
 
 
 if __name__ == '__main__':