← Back to branch summary

~lutostag/ubuntu/trusty/maas/1.5.2+packagefix

« back to all changes in this revision

Viewing changes to src/maasserver/views/tests/test_rpc.py

  • Committer: Package Import Robot
  • Author(s): Andres Rodriguez
  • Date: 2014-03-28 10:43:53 UTC
  • mto: This revision was merged to the branch mainline in revision 57.
  • Revision ID: package-import@ubuntu.com-20140328104353-ekpolg0pm5xnvq2s
Tags: upstream-1.5+bzr2204
Import upstream version 1.5+bzr2204

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/maasserver/views/tests/test_rpc.py
16
16
 
17
17
import json
18
18
 
 
19
from crochet import run_in_reactor
19
20
from django.core.urlresolvers import reverse
20
21
from maasserver import eventloop
21
22
from maasserver.testing.testcase import MAASServerTestCase
30
31
    MatchesDict,
31
32
    MatchesListwise,
32
33
    )
 
34
from twisted.internet.defer import inlineCallbacks
 
35
from twisted.internet.threads import deferToThread
33
36
 
34
37
 
35
38
is_valid_port = MatchesAll(
45
48
        self.assertEqual({"eventloops": {}}, info)
46
49
 
47
50
    def test_rpc_info_when_rpc_running(self):
48
 
        eventloop.start()
49
 
        self.addCleanup(eventloop.stop)
 
51
        eventloop.start().wait(5)
 
52
        self.addCleanup(lambda: eventloop.stop().wait(5))
 
53
 
 
54
        getServiceNamed = eventloop.services.getServiceNamed
 
55
 
 
56
        @run_in_reactor
 
57
        @inlineCallbacks
 
58
        def wait_for_startup():
 
59
            # Wait for the rpc and the rpc-advertise services to start.
 
60
            yield getServiceNamed("rpc").starting
 
61
            yield getServiceNamed("rpc-advertise").starting
 
62
            # Force an update, because it's very hard to track when the
 
63
            # first iteration of the rpc-advertise service has completed.
 
64
            yield deferToThread(getServiceNamed("rpc-advertise").update)
 
65
        wait_for_startup().wait(5)
50
66
 
51
67
        response = self.client.get(reverse('rpc-info'))
 
68
 
52
69
        self.assertEqual("application/json", response["Content-Type"])
53
70
        info = json.loads(response.content)
54
71
        self.assertThat(info, KeysEqual("eventloops"))