diff options
| -rwxr-xr-x | scripts/cli/test_system_ntp.py | 43 |
1 files changed, 42 insertions, 1 deletions
diff --git a/scripts/cli/test_system_ntp.py b/scripts/cli/test_system_ntp.py index c3fff8b06..ede096376 100755 --- a/scripts/cli/test_system_ntp.py +++ b/scripts/cli/test_system_ntp.py @@ -18,6 +18,7 @@ import re import os import unittest +from ipaddress import ip_network from psutil import process_iter from vyos.configsession import ConfigSession, ConfigSessionError from vyos.util import read_file @@ -43,6 +44,7 @@ class TestSystemNTP(unittest.TestCase): del self.session def test_ntp_options(self): + """ Test basic NTP support with multiple servers and their options """ servers = ['192.0.2.1', '192.0.2.2'] options = ['noselect', 'preempt', 'prefer'] @@ -53,7 +55,7 @@ class TestSystemNTP(unittest.TestCase): # commit changes self.session.commit() - # Check configured port + # Check generated configuration tmp = get_config_value('server') for server in servers: test = f'{server} iburst ' + ' '.join(options) @@ -62,5 +64,44 @@ class TestSystemNTP(unittest.TestCase): # Check for running process self.assertTrue("ntpd" in (p.name() for p in process_iter())) + def test_ntp_clients(self): + """ Test the allowed-networks statement """ + listen_address = ['127.0.0.1', '::1'] + for listen in listen_address: + self.session.set(base_path + ['listen-address', listen]) + + networks = ['192.0.2.0/24', '2001:db8:1000::/64'] + for network in networks: + self.session.set(base_path + ['allow-clients', 'address', network]) + + # Verify "NTP server not configured" verify() statement + with self.assertRaises(ConfigSessionError): + self.session.commit() + + servers = ['192.0.2.1', '192.0.2.2'] + for server in servers: + self.session.set(base_path + ['server', server]) + + self.session.commit() + + # Check generated client address configuration + for network in networks: + network_address = ip_network(network).network_address + network_netmask = ip_network(network).netmask + + tmp = get_config_value(f'restrict {network_address}')[0] + test = f'mask {network_netmask} nomodify notrap nopeer' + self.assertTrue(tmp in test) + + # Check listen address + tmp = get_config_value('interface') + test = ['ignore wildcard'] + for listen in listen_address: + test.append(f'listen {listen}') + self.assertEqual(tmp, test) + + # Check for running process + self.assertTrue("ntpd" in (p.name() for p in process_iter())) + if __name__ == '__main__': unittest.main() |