smoketest: tftp: T4012: Add smoketest for TFTP in VRF context

1 files changed, 39 insertions, 1 deletions

diff --git a/smoketest/scripts/cli/test_service_tftp-server.py b/smoketest/scripts/cli/test_service_tftp-server.py
index aed4c6beb..b57c33f26 100755
--- a/smoketest/scripts/cli/test_service_tftp-server.py
+++ b/smoketest/scripts/cli/test_service_tftp-server.py
@@ -19,8 +19,8 @@ import unittest
 from psutil import process_iter
 from base_vyostest_shim import VyOSUnitTestSHIM
 
-from vyos.configsession import ConfigSession
 from vyos.configsession import ConfigSessionError
+from vyos.util import cmd
 from vyos.util import read_file
 from vyos.util import process_named_running
 from vyos.template import is_ipv6
@@ -30,6 +30,7 @@ base_path = ['service', 'tftp-server']
 dummy_if_path = ['interfaces', 'dummy', 'dum69']
 address_ipv4 = '192.0.2.1'
 address_ipv6 = '2001:db8::1'
+vrf = 'mgmt'
 
 class TestServiceTFTPD(VyOSUnitTestSHIM.TestCase):
     def setUp(self):
@@ -98,5 +99,42 @@ class TestServiceTFTPD(VyOSUnitTestSHIM.TestCase):
                 count += 1
         self.assertEqual(count, len(address))
 
+    def test_03_tftpd_vrf(self):
+        directory = '/tmp'
+        port = '69' # default port
+
+        self.cli_set(base_path + ['allow-upload'])
+        self.cli_set(base_path + ['directory', directory])
+        self.cli_set(base_path + ['listen-address', address_ipv4, 'vrf', vrf])
+
+        # VRF does yet not exist - an error must be thrown
+        with self.assertRaises(ConfigSessionError):
+            self.cli_commit()
+
+        self.cli_set(['vrf', 'name', vrf, 'table', '1338'])
+        self.cli_set(dummy_if_path + ['vrf', vrf])
+
+        # commit changes
+        self.cli_commit()
+
+        config = read_file('/etc/default/tftpd0')
+        # verify listen IP address
+        self.assertIn(f'{address_ipv4}:{port} -4', config)
+        # verify directory
+        self.assertIn(directory, config)
+        # verify upload
+        self.assertIn('--create --umask 000', config)
+
+        # Check for running process
+        self.assertTrue(process_named_running(PROCESS_NAME))
+
+        # Check for process in VRF
+        tmp = cmd(f'ip vrf pids {vrf}')
+        self.assertIn(PROCESS_NAME, tmp)
+
+        # delete VRF
+        self.cli_delete(dummy_if_path + ['vrf'])
+        self.cli_delete(['vrf', 'name', vrf])
+
 if __name__ == '__main__':
     unittest.main(verbosity=2)