| Total Complexity | 4 |
| Total Lines | 34 |
| Duplicated Lines | 0 % |
| Changes | 0 | ||
| 1 | """Module to test the scheduler.py""" |
||
| 2 | from unittest import TestCase |
||
| 3 | from unittest.mock import MagicMock |
||
| 4 | from napps.amlight.sdntrace_cp.scheduler import Scheduler |
||
| 5 | from apscheduler.jobstores.base import JobLookupError |
||
| 6 | |||
| 7 | |||
| 8 | class TestScheduler(TestCase): |
||
| 9 | """Test class Scheduler""" |
||
| 10 | |||
| 11 | def test_remove_job(self): |
||
| 12 | """Test remove_job with existent id""" |
||
| 13 | scheduler = Scheduler() |
||
| 14 | trigger_args = {'trigger': 'interval', 'seconds': 2} |
||
| 15 | scheduler.add_callable('mock_id', MagicMock, **trigger_args) |
||
| 16 | self.assertIsNotNone(scheduler.get_job('mock_id')) |
||
| 17 | scheduler.remove_job('mock_id') |
||
| 18 | self.assertIsNone(scheduler.get_job('mock_id')) |
||
| 19 | |||
| 20 | def test_remove_job_fail(self): |
||
| 21 | """Test remove_job with non existent id""" |
||
| 22 | scheduler = Scheduler() |
||
| 23 | try: |
||
| 24 | scheduler.remove_job('mock_id') |
||
| 25 | except JobLookupError: |
||
| 26 | self.fail("scheduler.remove_job() raised an error") |
||
| 27 | |||
| 28 | def test_get_job(self): |
||
| 29 | """Test get_job and add_callable""" |
||
| 30 | scheduler = Scheduler() |
||
| 31 | trigger_args = {'trigger': 'interval', 'seconds': 2} |
||
| 32 | scheduler.add_callable('mock_id', MagicMock, **trigger_args) |
||
| 33 | self.assertIsNotNone(scheduler.get_job('mock_id')) |
||
| 34 |