← Back to branch summary

~rvb/maas/parent-child-relationship

« back to all changes in this revision

Viewing changes to src/maastesting/tests/test_celery.py

  • Committer: MaaS Lander
  • Author(s): Gavin Panella
  • Date: 2015-01-13 20:47:12 UTC
  • mfrom: (3373.4.45 celery-in-dns-removal)
  • Revision ID: maas_lander-20150113204712-h7axs50qohgoc0k8
[r=allenap,jtv][bug=][author=allenap] Remove all remaining Celery dependencies.

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/maastesting/tests/test_celery.py
1
 
# Copyright 2012, 2013 Canonical Ltd.  This software is licensed under the
2
 
# GNU Affero General Public License version 3 (see the file LICENSE).
3
 
 
4
 
"""Test matchers."""
5
 
 
6
 
from __future__ import (
7
 
    absolute_import,
8
 
    print_function,
9
 
    unicode_literals,
10
 
    )
11
 
 
12
 
str = None
13
 
 
14
 
__metaclass__ = type
15
 
__all__ = []
16
 
 
17
 
import random
18
 
 
19
 
from celery.decorators import task
20
 
from celery.result import EagerResult
21
 
from maastesting.celery import CeleryFixture
22
 
from maastesting.testcase import MAASTestCase
23
 
 
24
 
 
25
 
@task()
26
 
def task_add(x, y):
27
 
    return x + y
28
 
 
29
 
 
30
 
@task()
31
 
def task_exception(x, y):
32
 
    raise RuntimeError()
33
 
 
34
 
 
35
 
class TestCeleryFixture(MAASTestCase):
36
 
    """Tests `CeleryFixture`."""
37
 
 
38
 
    def setUp(self):
39
 
        super(TestCeleryFixture, self).setUp()
40
 
        self.celery = self.useFixture(CeleryFixture())
41
 
 
42
 
    def test_celery_eagerresult_contains_result(self):
43
 
        # The result is an instance of EagerResult and it contains the actual
44
 
        # result.
45
 
        x = random.randrange(100)
46
 
        y = random.randrange(100)
47
 
        result = task_add.delay(x, y)
48
 
        self.assertIsInstance(result, EagerResult)
49
 
        self.assertEqual(x + y, result.result)
50
 
 
51
 
    def test_celery_exception_raised(self):
52
 
        self.assertRaises(RuntimeError, task_exception.delay, 1, 2)
53
 
 
54
 
    def test_celery_records_tasks(self):
55
 
        x = random.randrange(100)
56
 
        y = random.randrange(100)
57
 
        task_add.delay(x=x, y=y)
58
 
        z = random.randrange(100)
59
 
        t = random.randrange(100)
60
 
        task_add.delay(x=z, y=t)
61
 
        tasks = self.celery.tasks
62
 
        self.assertEqual(2, len(tasks))
63
 
        self.assertEqual(
64
 
            ['maastesting.tests.test_celery.task_add'] * 2,
65
 
            [task['task'].name for task in tasks])
66
 
        self.assertEqual({'x': x, 'y': y}, tasks[0]['kwargs'])
67
 
        self.assertEqual({'x': z, 'y': t}, tasks[1]['kwargs'])