Arnike/watcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge "Add unit tests for continuous.py"

Browse Source
This commit is contained in:
Jenkins
2016-08-26 17:44:30 +00:00
committed by Gerrit Code Review
parent 359debb6b0 da67b246a5
commit f42106ca34
1 changed files with 13 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
watcher/tests/decision_engine/audit/test_audit_handlers.py
View File

@@ -81,7 +81,7 @@ class TestOneShotAuditHandler(base.DbTestCase):
class TestContinuousAuditHandler(base.DbTestCase):
def setUp(self):
super(TestContinuousAuditHandler, self).setUp()
obj_utils.create_test_goal(self.context, id=1, name="DUMMY")
obj_utils.create_test_goal(self.context, id=1, name="dummy")
audit_template = obj_utils.create_test_audit_template(
self.context)
self.audits = [
@@ -92,16 +92,21 @@ class TestContinuousAuditHandler(base.DbTestCase):
audit_type=audit_objects.AuditType.CONTINUOUS.value)
for i in range(2)]
@mock.patch.object(manager.CollectorManager, "get_cluster_model_collector")
@mock.patch.object(background.BackgroundScheduler, 'add_job')
@mock.patch.object(background.BackgroundScheduler, 'get_jobs')
@mock.patch.object(audit_objects.Audit, 'list')
def test_launch_audits_periodically(self, mock_list,
mock_jobs, mock_add_job):
def test_launch_audits_periodically(self, mock_list, mock_jobs,
mock_add_job, mock_collector):
audit_handler = continuous.ContinuousAuditHandler(mock.MagicMock())
audits = [audit_objects.Audit.get_by_uuid(self.context,
self.audits[0].uuid)]
mock_list.return_value = audits
mock_jobs.return_value = mock.MagicMock()
mock_add_job.return_value = audit_handler.execute_audit(
self.audits[0], self.context)
mock_collector.return_value = faker.FakerModelCollector()
audit_handler.launch_audits_periodically()
mock_add_job.assert_called()
@@ -143,3 +148,8 @@ class TestContinuousAuditHandler(base.DbTestCase):
next_run_time=mock.ANY)]
audit_handler.launch_audits_periodically()
mock_add_job.assert_has_calls(calls)
audit_handler.update_audit_state(self.context, audits[1],
audit_objects.State.CANCELLED)
is_inactive = audit_handler._is_audit_inactive(audits[1])
self.assertTrue(is_inactive)