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

Add unit tests for continuous.py

Browse Source 
Add unit tests about functions 'execute_audit'
and 'is_audit_inactive'

Change-Id: If9eb4d95166a486a59cc1535106944243bb7eb30
Closes-Bug: #1616759
This commit is contained in:
licanwei
2016-08-25 16:52:58 +08:00
parent eb421709d9
commit da67b246a5
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

@@ -76,7 +76,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 = [obj_utils.create_test_audit(
@@ -86,16 +86,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()
@@ -137,3 +142,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)