~allenap/maas/async-use-fixture

« back to all changes in this revision

Viewing changes to src/provisioningserver/dhcp/tests/test_dhcp_probe_service.py

  • Committer: MaaS Lander
  • Author(s): Graham Binns
  • Date: 2014-08-19 08:34:03 UTC
  • mfrom: (2756.1.1 revert-r2750)
  • Revision ID: maas_lander-20140819083403-b800oag7c99ect0w
[r=gmb][bug=][author=gmb] Revert r2750, which has several large problems that need fixing.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright 2014 Canonical Ltd.  This software is licensed under the
2
 
# GNU Affero General Public License version 3 (see the file LICENSE).
3
 
 
4
 
"""Tests for periodic DHCP prober."""
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
 
from maastesting.matchers import get_mock_calls
18
 
from maastesting.testcase import MAASTestCase
19
 
from provisioningserver.dhcp import detect
20
 
from provisioningserver.dhcp.dhcp_probe_service import (
21
 
    PeriodicDHCPProbeService,
22
 
    )
23
 
from testtools.deferredruntest import AsynchronousDeferredRunTest
24
 
from twisted.internet.task import Clock
25
 
 
26
 
 
27
 
class TestDHCPProbeService(MAASTestCase):
28
 
 
29
 
    run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=5)
30
 
 
31
 
    def setUp(self):
32
 
        super(TestDHCPProbeService, self).setUp()
33
 
 
34
 
    def test_is_called_every_interval(self):
35
 
        clock = Clock()
36
 
        # Avoid actually probing
37
 
        probe_task = self.patch(detect, "periodic_probe_task")
38
 
        service = PeriodicDHCPProbeService(clock)
39
 
 
40
 
        # Until the service has started, periodic_probe_task() won't
41
 
        # be called.
42
 
        self.assertEqual(0, len(get_mock_calls(probe_task)))
43
 
 
44
 
        # Avoid actual downloads:
45
 
        service.startService()
46
 
 
47
 
        # The first call is issued at startup.
48
 
        self.assertEqual(1, len(get_mock_calls(probe_task)))
49
 
 
50
 
        # Wind clock forward one second less than the desired interval.
51
 
        clock.advance(service.check_interval - 1)
52
 
        # No more periodic calls made.
53
 
        self.assertEqual(1, len(get_mock_calls(probe_task)))
54
 
 
55
 
        # Wind clock forward one second, past the interval.
56
 
        clock.advance(1)
57
 
 
58
 
        # Now there were two calls.
59
 
        self.assertEqual(2, len(get_mock_calls(probe_task)))