Add tests for taskresults app

This commit adds tests for the taskresults management command
fetch_taskresults and completes the test coverage for
taskresults.models.
This commit is contained in:
Jan Dittberner 2015-12-06 18:47:11 +01:00
parent 4e54b6fcc5
commit e96aac82fc
4 changed files with 120 additions and 24 deletions

View file

@ -0,0 +1,73 @@
"""
This module provides tests for the
:py:mod:`taskresults.management.commands.fetch_taskresults` Django management
command.
"""
from __future__ import unicode_literals
try:
from unittest.mock import patch, MagicMock
except ImportError:
from mock import patch, MagicMock
from django.test import TestCase
from taskresults.models import TaskResult
from taskresults.management.commands.fetch_taskresults import Command
TEST_TASK_UUID = '3120f6a8-2665-4fa3-a785-79efd28bfe92'
TEST_TASK_NAME = 'test.task'
TEST_TASK_RESULT = '4ll y0ur b453 4r3 b3l0ng t0 u5'
@patch('taskresults.models.app.AsyncResult')
class FetchTaskResultsCommandTest(TestCase):
def test_handle_unfinished(self, asyncresult):
resultmock = MagicMock(task_id=TEST_TASK_UUID)
sigmock = MagicMock()
sigmock.apply_async.return_value = resultmock
tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, sigmock)
self.assertFalse(tr.finished)
self.assertEqual(tr.result, '')
self.assertEqual(tr.state, '')
aresult = asyncresult.return_value
aresult.state = 'PENDING'
aresult.ready.return_value = False
Command().handle()
tr = TaskResult.objects.get(task_id=TEST_TASK_UUID)
self.assertTrue(asyncresult.called_with(TEST_TASK_UUID))
self.assertTrue(aresult.ready.called_with())
self.assertFalse(tr.finished)
self.assertEqual(tr.result, '')
self.assertEqual(tr.state, 'PENDING')
def test_handle_finished(self, asyncresult):
resultmock = MagicMock(task_id=TEST_TASK_UUID)
sigmock = MagicMock()
sigmock.apply_async.return_value = resultmock
tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, sigmock)
self.assertFalse(tr.finished)
self.assertEqual(tr.result, '')
self.assertEqual(tr.state, '')
aresult = asyncresult.return_value
aresult.state = 'SUCCESS'
aresult.ready.return_value = True
aresult.get.return_value = TEST_TASK_RESULT
Command().handle()
tr = TaskResult.objects.get(task_id=TEST_TASK_UUID)
self.assertTrue(asyncresult.called_with(TEST_TASK_UUID))
self.assertTrue(aresult.ready.called_with())
self.assertTrue(aresult.get.called_with())
self.assertTrue(tr.finished)
self.assertEqual(tr.result, TEST_TASK_RESULT)
self.assertEqual(tr.state, 'SUCCESS')

View file

@ -1,7 +1,16 @@
"""
This module provides tests for :py:mod:`taskresults.models`.
"""
from __future__ import absolute_import, unicode_literals from __future__ import absolute_import, unicode_literals
try:
from unittest.mock import patch, MagicMock
except ImportError:
from mock import patch, MagicMock
from django.test import TestCase from django.test import TestCase
from mock import patch, MagicMock
from taskresults.models import TaskResult from taskresults.models import TaskResult
@ -11,38 +20,52 @@ TEST_TASK_RESULT = '4ll y0ur b453 4r3 b3l0ng t0 u5'
class TaskResultTest(TestCase): class TaskResultTest(TestCase):
@patch('taskresults.models.app') @patch('taskresults.models.app.AsyncResult')
def test_update_taskstatus_unfinished(self, app): def test_update_taskstatus_unfinished(self, asyncresult):
resultmock = MagicMock(task_id=TEST_TASK_UUID) resultmock = MagicMock(task_id=TEST_TASK_UUID)
resultmock.ready.return_value = False sigmock = MagicMock()
mock = MagicMock() sigmock.apply_async.return_value = resultmock
mock.apply_async.return_value = resultmock tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, sigmock)
tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, mock)
self.assertFalse(tr.finished) self.assertFalse(tr.finished)
mymock = app.AsyncResult(TEST_TASK_UUID) mymock = asyncresult.return_value
mymock.state = 'SUCCESS' mymock.state = 'PENDING'
mymock.get.return_value = TEST_RESULT mymock.ready.return_value = False
tr.fetch_result() tr.fetch_result()
mymock.get.assert_called_with() mymock.get.assert_not_called()
self.assertTrue(tr.finished) self.assertFalse(tr.finished)
@patch('taskresults.models.app') @patch('taskresults.models.app.AsyncResult')
def test_update_taskstatus_finished(self, app): def test_update_taskstatus_finished(self, asyncresult):
resultmock = MagicMock(task_id=TEST_TASK_UUID) resultmock = MagicMock(task_id=TEST_TASK_UUID)
resultmock.ready.return_value = True sigmock = MagicMock()
resultmock.state = 'SUCCESS' sigmock.apply_async.return_value = resultmock
resultmock.result = TEST_RESULT aresult = asyncresult.return_value
mock = MagicMock() tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, sigmock)
mock.apply_async.return_value = resultmock self.assertFalse(tr.finished)
mock.state = 'SUCCESS' aresult = asyncresult.return_value
mock.result = TEST_RESULT aresult.state = 'SUCCESS'
tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, mock) aresult.ready.return_value = True
aresult.get.return_value = TEST_TASK_RESULT
tr.fetch_result() tr.fetch_result()
self.assertTrue(aresult.get.called_with())
self.assertEqual(aresult.get.call_count, 1)
self.assertTrue(tr.finished) self.assertTrue(tr.finished)
mymock = app.AsyncResult(TEST_TASK_UUID) self.assertEqual(tr.result, str(TEST_TASK_RESULT))
tr.fetch_result() tr.fetch_result()
self.assertEqual(mymock.get.call_count, 1) self.assertEqual(aresult.get.call_count, 1)
self.assertTrue(tr.finished) self.assertTrue(tr.finished)
self.assertEqual(tr.result, str(TEST_TASK_RESULT))
def test___str__(self):
resultmock = MagicMock(task_id=TEST_TASK_UUID)
sigmock = MagicMock()
sigmock.apply_async.return_value = resultmock
tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, sigmock)
self.assertEqual(str(tr), "{name} ({taskid}): no".format(
name=TEST_TASK_NAME, taskid=TEST_TASK_UUID))
tr.finished = True
self.assertEqual(str(tr), "{name} ({taskid}): yes".format(
name=TEST_TASK_NAME, taskid=TEST_TASK_UUID))
TEST_RESULT = MagicMock() TEST_RESULT = MagicMock()