External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-05-12 11:57:37 -02:30
Code Issues Packages Projects Releases Wiki Activity

Fix up unit and functional tests

Browse Source
This commit is contained in:
Wayne Witzel III
2017-09-20 13:15:54 -04:00
parent 2889df8013
commit 0f4523fabf
2 changed files with 6 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
awx/main/tests/functional/task_management/test_rampart_groups.py
View File

@@ -69,20 +69,10 @@ def test_multi_group_with_shared_dependency(instance_factory, default_instance_g
pu.save() pu.save()
with mock.patch("awx.main.scheduler.TaskManager.start_task"): with mock.patch("awx.main.scheduler.TaskManager.start_task"):
TaskManager().schedule() TaskManager().schedule()
TaskManager.start_task.assert_called_once_with(j1, ig1, [])
j1.finished = j1.created + timedelta(seconds=2) TaskManager.start_task.assert_any_call(j1, ig1, [])
j1.status = "successful" TaskManager.start_task.assert_any_call(j2, ig2, [])
j1.save() assert TaskManager.start_task.call_count == 2
with mock.patch("awx.main.scheduler.TaskManager.start_task"):
TaskManager().schedule()
pu = p.project_updates.last()
TaskManager.start_task.assert_called_once_with(pu, default_instance_group, [j2])
pu.finished = pu.created + timedelta(seconds=1)
pu.status = "successful"
pu.save()
with mock.patch("awx.main.scheduler.TaskManager.start_task"):
TaskManager().schedule()
TaskManager.start_task.assert_called_once_with(j2, ig2, [])
@pytest.mark.django_db @pytest.mark.django_db

4
awx/main/tests/unit/test_task_manager.py
View File

@@ -22,7 +22,7 @@ class TestCleanupInconsistentCeleryTasks():
@mock.patch.object(TaskManager, 'get_running_tasks', return_value=({'host1': [Job(id=2), Job(id=3),]}, [])) @mock.patch.object(TaskManager, 'get_running_tasks', return_value=({'host1': [Job(id=2), Job(id=3),]}, []))
@mock.patch.object(InstanceGroup.objects, 'prefetch_related', return_value=[]) @mock.patch.object(InstanceGroup.objects, 'prefetch_related', return_value=[])
@mock.patch.object(Instance.objects, 'filter', return_value=mock.MagicMock(first=lambda: None)) @mock.patch.object(Instance.objects, 'filter', return_value=mock.MagicMock(first=lambda: None))
@mock.patch('awx.main.scheduler.logger') @mock.patch('awx.main.scheduler.task_manager.logger')
def test_instance_does_not_exist(self, logger_mock, *args): def test_instance_does_not_exist(self, logger_mock, *args):
logger_mock.error = mock.MagicMock(side_effect=RuntimeError("mocked")) logger_mock.error = mock.MagicMock(side_effect=RuntimeError("mocked"))
tm = TaskManager() tm = TaskManager()
@@ -38,7 +38,7 @@ class TestCleanupInconsistentCeleryTasks():
@mock.patch.object(TaskManager, 'get_active_tasks', return_value=([], {'host1': []})) @mock.patch.object(TaskManager, 'get_active_tasks', return_value=([], {'host1': []}))
@mock.patch.object(InstanceGroup.objects, 'prefetch_related', return_value=[]) @mock.patch.object(InstanceGroup.objects, 'prefetch_related', return_value=[])
@mock.patch.object(TaskManager, 'get_running_tasks') @mock.patch.object(TaskManager, 'get_running_tasks')
@mock.patch('awx.main.scheduler.logger') @mock.patch('awx.main.scheduler.task_manager.logger')
def test_save_failed(self, logger_mock, get_running_tasks, *args): def test_save_failed(self, logger_mock, get_running_tasks, *args):
logger_mock.error = mock.MagicMock() logger_mock.error = mock.MagicMock()
job = Job(id=2, modified=tz_now(), status='running', celery_task_id='blah', execution_node='host1') job = Job(id=2, modified=tz_now(), status='running', celery_task_id='blah', execution_node='host1')