summaryrefslogtreecommitdiff
path: root/src/tests/test_task_scheduler.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/tests/test_task_scheduler.py')
-rw-r--r--src/tests/test_task_scheduler.py130
1 files changed, 130 insertions, 0 deletions
diff --git a/src/tests/test_task_scheduler.py b/src/tests/test_task_scheduler.py
new file mode 100644
index 000000000..084bd868c
--- /dev/null
+++ b/src/tests/test_task_scheduler.py
@@ -0,0 +1,130 @@
+#!/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 vyos import ConfigError
+try:
+ from src.conf_mode import task_scheduler
+except ModuleNotFoundError: # for unittest.main()
+ import sys
+ sys.path.append(os.path.join(os.path.dirname(__file__), '../..'))
+ from src.conf_mode import task_scheduler
+
+
+class TestUpdateCrontab(unittest.TestCase):
+
+ def test_verify(self):
+ tests = [
+ {'name': 'one_task',
+ 'tasks': [{'name': 'aaa', 'interval': '60m', 'spec': '', 'executable': '/bin/ls', 'args': '-l'}],
+ 'expected': None
+ },
+ {'name': 'has_interval_and_spec',
+ 'tasks': [{'name': 'aaa', 'interval': '60m', 'spec': '0 * * * *', 'executable': '/bin/ls', 'args': '-l'}],
+ 'expected': ConfigError
+ },
+ {'name': 'has_no_interval_and_spec',
+ 'tasks': [{'name': 'aaa', 'interval': '', 'spec': '', 'executable': '/bin/ls', 'args': '-l'}],
+ 'expected': ConfigError
+ },
+ {'name': 'invalid_interval',
+ 'tasks': [{'name': 'aaa', 'interval': '1y', 'spec': '', 'executable': '/bin/ls', 'args': '-l'}],
+ 'expected': ConfigError
+ },
+ {'name': 'invalid_interval_min',
+ 'tasks': [{'name': 'aaa', 'interval': '61m', 'spec': '', 'executable': '/bin/ls', 'args': '-l'}],
+ 'expected': ConfigError
+ },
+ {'name': 'invalid_interval_hour',
+ 'tasks': [{'name': 'aaa', 'interval': '25h', 'spec': '', 'executable': '/bin/ls', 'args': '-l'}],
+ 'expected': ConfigError
+ },
+ {'name': 'invalid_interval_day',
+ 'tasks': [{'name': 'aaa', 'interval': '32d', 'spec': '', 'executable': '/bin/ls', 'args': '-l'}],
+ 'expected': ConfigError
+ },
+ {'name': 'no_executable',
+ 'tasks': [{'name': 'aaa', 'interval': '60m', 'spec': '', 'executable': '', 'args': ''}],
+ 'expected': ConfigError
+ },
+ {'name': 'invalid_executable',
+ 'tasks': [{'name': 'aaa', 'interval': '60m', 'spec': '', 'executable': '/bin/aaa', 'args': ''}],
+ 'expected': ConfigError
+ }
+ ]
+ for t in tests:
+ with self.subTest(msg=t['name'], tasks=t['tasks'], expected=t['expected']):
+ if t['expected'] is not None:
+ with self.assertRaises(t['expected']):
+ task_scheduler.verify(t['tasks'])
+ else:
+ task_scheduler.verify(t['tasks'])
+
+ def test_generate(self):
+ tests = [
+ {'name': 'zero_task',
+ 'tasks': [],
+ 'expected': []
+ },
+ {'name': 'one_task',
+ 'tasks': [{'name': 'aaa', 'interval': '60m', 'spec': '', 'executable': '/bin/ls', 'args': '-l'}],
+ 'expected': [
+ '### Generated by vyos-update-crontab.py ###',
+ '*/60 * * * * root sg vyattacfg \"/bin/ls -l\"']
+ },
+ {'name': 'one_task_with_hour',
+ 'tasks': [{'name': 'aaa', 'interval': '10h', 'spec': '', 'executable': '/bin/ls', 'args': '-l'}],
+ 'expected': [
+ '### Generated by vyos-update-crontab.py ###',
+ '0 */10 * * * root sg vyattacfg \"/bin/ls -l\"']
+ },
+ {'name': 'one_task_with_day',
+ 'tasks': [{'name': 'aaa', 'interval': '10d', 'spec': '', 'executable': '/bin/ls', 'args': '-l'}],
+ 'expected': [
+ '### Generated by vyos-update-crontab.py ###',
+ '0 0 */10 * * root sg vyattacfg \"/bin/ls -l\"']
+ },
+ {'name': 'multiple_tasks',
+ 'tasks': [{'name': 'aaa', 'interval': '60m', 'spec': '', 'executable': '/bin/ls', 'args': '-l'},
+ {'name': 'bbb', 'interval': '', 'spec': '0 0 * * *', 'executable': '/bin/ls', 'args': '-ltr'}
+ ],
+ 'expected': [
+ '### Generated by vyos-update-crontab.py ###',
+ '*/60 * * * * root sg vyattacfg \"/bin/ls -l\"',
+ '0 0 * * * root sg vyattacfg \"/bin/ls -ltr\"']
+ }
+ ]
+ for t in tests:
+ with self.subTest(msg=t['name'], tasks=t['tasks'], expected=t['expected']):
+ task_scheduler.crontab_file = tempfile.mkstemp()[1]
+ task_scheduler.generate(t['tasks'])
+ if len(t['expected']) > 0:
+ self.assertTrue(os.path.isfile(task_scheduler.crontab_file))
+ with open(task_scheduler.crontab_file) as f:
+ actual = f.read()
+ self.assertEqual(t['expected'], actual.splitlines())
+ os.remove(task_scheduler.crontab_file)
+ else:
+ self.assertFalse(os.path.isfile(task_scheduler.crontab_file))
+
+
+if __name__ == "__main__":
+ unittest.main()