summaryrefslogtreecommitdiff
path: root/src/tests/test_ntp.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/tests/test_ntp.py')
-rw-r--r--src/tests/test_ntp.py262
1 files changed, 262 insertions, 0 deletions
diff --git a/src/tests/test_ntp.py b/src/tests/test_ntp.py
new file mode 100644
index 000000000..08e1b665e
--- /dev/null
+++ b/src/tests/test_ntp.py
@@ -0,0 +1,262 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2018 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# 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 tempfile
+import unittest
+from unittest import TestCase, mock
+import ipaddress
+from contextlib import ExitStack
+import textwrap
+
+from vyos import ConfigError
+from vyos.config import Config
+try:
+ from src.conf_mode import ntp
+except ModuleNotFoundError: # for unittest.main()
+ import sys
+ sys.path.append(os.path.join(os.path.dirname(__file__), '../..'))
+ from src.conf_mode import ntp
+
+
+class TestNtp(TestCase):
+
+ def test_get_config(self):
+ tests = [
+ {
+ 'name': 'empty',
+ 'config': {
+ 'system ntp': None,
+ },
+ 'expected': None,
+ },
+ {
+ 'name': 'full-options',
+ 'config': {
+ 'system ntp': 'yes',
+ 'allow-clients address': ['192.0.2.0/24'],
+ 'listen-address': ['198.51.100.0/24'],
+ 'server': ['example.com'],
+ 'server example.com noselect': 'yes',
+ 'server example.com preempt': 'yes',
+ 'server example.com prefer': 'yes',
+ },
+ 'expected': {
+ 'allowed_networks': [{
+ 'address': ipaddress.ip_address('192.0.2.0'),
+ 'netmask': ipaddress.ip_address('255.255.255.0'),
+ 'network': '192.0.2.0/24',
+ }],
+ 'listen_address': ['198.51.100.0/24'],
+ 'servers': [
+ {'name': 'example.com', 'options': ['noselect', 'preempt', 'prefer']}
+ ]
+ },
+ },
+ {
+ 'name': 'non-options',
+ 'config': {
+ 'system ntp': 'yes',
+ 'allow-clients address': ['192.0.2.0/24'],
+ 'listen-address': ['198.51.100.0/24'],
+ 'server': ['example.com'],
+ },
+ 'expected': {
+ 'allowed_networks': [{
+ 'address': ipaddress.ip_address('192.0.2.0'),
+ 'netmask': ipaddress.ip_address('255.255.255.0'),
+ 'network': '192.0.2.0/24',
+ }],
+ 'listen_address': ['198.51.100.0/24'],
+ 'servers': [
+ {'name': 'example.com', 'options': []}
+ ]
+ },
+ },
+ ]
+ for case in tests:
+ def mocked_fn(path):
+ return case['config'].get(path)
+
+ with self.subTest(msg = case['name']):
+ m = {
+ 'return_value': mock.Mock(side_effect = mocked_fn),
+ 'return_values': mock.Mock(side_effect = mocked_fn),
+ 'list_nodes': mock.Mock(side_effect = mocked_fn),
+ 'exists': mock.Mock(side_effect = mocked_fn),
+ }
+ with mock.patch.multiple(Config, **m):
+ actual = ntp.get_config()
+ self.assertEqual(actual, case['expected'])
+
+ def test_verify(self):
+ tests = [
+ {
+ 'name': 'none',
+ 'config': None,
+ 'expected': None
+ },
+ {
+ 'name': 'valid',
+ 'config': {
+ 'allowed_networks': [{
+ 'address': ipaddress.ip_address('192.0.2.1'),
+ 'netmask': ipaddress.ip_address('255.255.255.0'),
+ 'network': '192.0.2.0/24',
+ }],
+ 'listen_address': ['198.51.100.0/24'],
+ 'servers': [
+ {'name': 'example.com', 'options': ['noselect', 'preempt', 'prefer']}
+ ]
+ },
+ 'expected': None,
+ },
+ {
+ 'name': 'not configure servers',
+ 'config': {
+ 'allowed_networks': [{
+ 'address': ipaddress.ip_address('192.0.2.1'),
+ 'netmask': ipaddress.ip_address('255.255.255.0'),
+ 'network': '192.0.2.0/24',
+ }],
+ 'servers': []
+ },
+ 'expected': ConfigError,
+ },
+ {
+ 'name': 'does not exist in the network',
+ 'config': {
+ 'allowed_networks': [{
+ 'address': ipaddress.ip_address('192.0.2.1'),
+ 'netmask': ipaddress.ip_address('255.255.255.0'),
+ 'network': '192.0.2.0/50', # invalid netmask
+ }],
+ 'listen_address': ['198.51.100.0/24'],
+ 'servers': [
+ {'name': 'example.com', 'options': []}
+ ]
+ },
+ 'expected': ConfigError,
+ },
+ ]
+ for case in tests:
+ with self.subTest(msg = case['name']):
+ if case['expected'] is not None:
+ with self.assertRaises(case['expected']):
+ ntp.verify(case['config'])
+ else:
+ ntp.verify(case['config'])
+
+ def test_generate(self):
+ tests = [
+ {
+ 'name': 'empty',
+ 'config': None,
+ 'expected': '',
+ },
+ {
+ 'name': 'valid',
+ 'config': {
+ 'allowed_networks': [
+ {
+ 'address': ipaddress.ip_address('192.0.2.1'),
+ 'netmask': ipaddress.ip_address('255.255.255.0'),
+ 'network': '192.0.2.0/24',
+ },
+ {
+ 'address': ipaddress.ip_address('198.51.100.1'),
+ 'netmask': ipaddress.ip_address('255.255.255.0'),
+ 'network': '198.51.100.0/24',
+ },
+ ],
+ 'listen_address': ['198.51.100.0/24'],
+ 'servers': [
+ {'name': '1.example.com', 'options': ['noselect', 'preempt', 'prefer']},
+ {'name': '2.example.com', 'options': []},
+ ]
+ },
+ 'expected': textwrap.dedent('''
+ ### Autogenerated by ntp.py ###
+
+ #
+ # Non-configurable defaults
+ #
+ driftfile /var/lib/ntp/ntp.drift
+ # By default, only allow ntpd to query time sources, ignore any incoming requests
+ restrict default noquery nopeer notrap nomodify
+ # Local users have unrestricted access, allowing reconfiguration via ntpdc
+ restrict 127.0.0.1
+ restrict -6 ::1
+
+
+ #
+ # Configurable section
+ #
+
+ # Server configuration for: 1.example.com
+ server 1.example.com iburst noselect preempt prefer
+
+ # Server configuration for: 2.example.com
+ server 2.example.com iburst
+
+
+
+ # Client configuration for network: 192.0.2.0/24
+ restrict 192.0.2.1 mask 255.255.255.0 nomodify notrap nopeer
+
+ # Client configuration for network: 198.51.100.0/24
+ restrict 198.51.100.1 mask 255.255.255.0 nomodify notrap nopeer
+
+
+
+ # NTP should listen on configured addresses only
+ interface ignore wildcard
+ interface listen 198.51.100.0/24
+
+ '''),
+ },
+ ]
+
+ for case in tests:
+ with self.subTest(msg = case['name']):
+ with tempfile.NamedTemporaryFile() as fp:
+ ntp.config_file = fp.name
+
+ ntp.generate(case['config'])
+ actual = fp.file.read().decode('ascii')
+ # print(actual)
+ self.assertEqual(case['expected'], actual)
+
+ def test_apply(self):
+ with tempfile.NamedTemporaryFile(delete = False) as fp:
+ ntp.config_file = fp.name
+ with mock.patch('os.system') as os_system:
+ ntp.apply({}) # some configure
+ os_system.assert_has_calls([
+ mock.call('sudo systemctl restart ntp.service'),
+ ])
+ self.assertTrue(os.path.exists(fp.name))
+
+ ntp.apply(None) # empty configure
+ os_system.assert_has_calls([
+ mock.call('sudo systemctl stop ntp.service'),
+ ])
+ self.assertFalse(os.path.exists(fp.name))
+
+if __name__ == "__main__":
+ unittest.main()
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-14 07:04:00 +0000