← Back to branch summary

~wgrant/ubuntu/natty/landscape-client/natty-updates-broken

« back to all changes in this revision

Viewing changes to landscape/monitor/tests/test_rebootrequired.py

  • Committer: Bazaar Package Importer
  • Author(s): Free Ekanayaka
  • Date: 2010-04-21 19:58:10 UTC
  • mfrom: (1.1.16 upstream)
  • Revision ID: james.westby@ubuntu.com-20100421195810-s30uv3s6i27lue38
Tags: 1.5.2-0ubuntu0.10.10.0
https://launchpad.net/bugs/594594
https://launchpad.net/bugs/272344
https://launchpad.net/bugs/538253
https://launchpad.net/bugs/574810
* New upstream version (LP: #594594):
  - A new includes information about active network devices and their
    IP address in sysinfo output (LP: #272344).
  - A new plugin collects information about network traffic (#LP :284662).
  - Report information about which packages requested a reboot (LP: #538253).
  - Fix breakage on Lucid AMIs having no ramdisk (LP: #574810).
  - Migrate the inter-process communication system from DBus to Twisted AMP.

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/monitor/tests/test_rebootrequired.py
1
 
import os
2
 
 
3
1
from landscape.monitor.rebootrequired import RebootRequired
4
 
from landscape.tests.helpers import LandscapeIsolatedTest
5
 
from landscape.tests.helpers import MonitorHelper, LogKeeperHelper
 
2
from landscape.tests.helpers import (
 
3
    LandscapeTest, MonitorHelper, LogKeeperHelper)
6
4
from landscape.tests.mocker import ANY
7
5
 
8
6
 
9
 
class RebootRequiredTest(LandscapeIsolatedTest):
 
7
class RebootRequiredTest(LandscapeTest):
10
8
 
11
9
    helpers = [MonitorHelper, LogKeeperHelper]
12
10
 
13
11
    def setUp(self):
14
12
        super(RebootRequiredTest, self).setUp()
15
 
        self.reboot_required_filename = self.makeFile("")
 
13
        self.reboot_required_filename = self.makeFile()
16
14
        self.plugin = RebootRequired(self.reboot_required_filename)
17
15
        self.monitor.add(self.plugin)
18
 
        self.mstore.set_accepted_types(["reboot-required"])
19
 
 
20
 
    def test_wb_check_reboot_required(self):
21
 
        """
22
 
        L{RebootRequired.check_reboot_required} should return C{True} if the
23
 
        reboot-required flag file is present, C{False} otherwise.
24
 
        """
25
 
        self.assertTrue(self.plugin._check_reboot_required())
26
 
        os.remove(self.reboot_required_filename)
27
 
        self.assertFalse(self.plugin._check_reboot_required())
 
16
        self.mstore.set_accepted_types(["reboot-required-info"])
 
17
 
 
18
    def test_wb_get_flag(self):
 
19
        """
 
20
        L{RebootRequired._get_flag} returns C{True} if the reboot-required
 
21
        flag file is present, C{False} otherwise.
 
22
        """
 
23
        self.assertFalse(self.plugin._get_flag())
 
24
        self.makeFile(path=self.reboot_required_filename, content="")
 
25
        self.assertTrue(self.plugin._get_flag())
 
26
 
 
27
    def test_wb_get_packages(self):
 
28
        """
 
29
        L{RebootRequired._get_packages} returns the packages listed in the
 
30
        reboot-required packages file if present, or an empty list otherwise.
 
31
        """
 
32
        self.assertEqual([], self.plugin._get_packages())
 
33
        self.makeFile(path=self.reboot_required_filename + ".pkgs",
 
34
                      content="foo\nbar\n")
 
35
        self.assertEqual(["bar", "foo"], self.plugin._get_packages())
 
36
 
 
37
    def test_wb_get_packages_with_duplicates(self):
 
38
        """
 
39
        The list of packages returned by L{RebootRequired._get_packages} does
 
40
        not contain duplicate values.
 
41
        """
 
42
        self.assertEqual([], self.plugin._get_packages())
 
43
        self.makeFile(path=self.reboot_required_filename + ".pkgs",
 
44
                      content="foo\nfoo\n")
 
45
        self.assertEqual(["foo"], self.plugin._get_packages())
 
46
 
 
47
    def test_wb_get_packages_with_blank_lines(self):
 
48
        """
 
49
        Blank lines are ignored by L{RebootRequired._get_packages}.
 
50
        """
 
51
        self.assertEqual([], self.plugin._get_packages())
 
52
        self.makeFile(path=self.reboot_required_filename + ".pkgs",
 
53
                      content="bar\n\nfoo\n")
 
54
        self.assertEqual(["bar", "foo"], self.plugin._get_packages())
28
55
 
29
56
    def test_wb_create_message(self):
30
57
        """
31
58
        A message should be created if and only if the reboot-required status
32
59
        of the system has changed.
33
60
        """
34
 
        self.assertEquals(self.plugin._create_message(), {"flag": True})
35
 
        self.assertEquals(self.plugin._create_message(), {})
 
61
        self.assertEquals({"flag": False, "packages": []},
 
62
                          self.plugin._create_message())
 
63
        self.makeFile(path=self.reboot_required_filename, content="")
 
64
        self.assertEquals({"flag": True},
 
65
                          self.plugin._create_message())
 
66
        self.makeFile(path=self.reboot_required_filename + ".pkgs",
 
67
                      content="foo\n")
 
68
        self.assertEquals({"packages": [u"foo"]},
 
69
                          self.plugin._create_message())
36
70
 
37
71
    def test_send_message(self):
38
72
        """
39
 
        A new C{"reboot-required"} message should be enqueued if and only
 
73
        A new C{"reboot-required-info"} message should be enqueued if and only
40
74
        if the reboot-required status of the system has changed.
41
75
        """
 
76
        self.makeFile(path=self.reboot_required_filename + ".pkgs",
 
77
                      content="foo\n")
 
78
        self.makeFile(path=self.reboot_required_filename, content="")
42
79
        self.plugin.send_message()
43
80
        self.assertIn("Queueing message with updated reboot-required status.",
44
81
                      self.logfile.getvalue())
45
82
        self.assertMessages(self.mstore.get_pending_messages(),
46
 
                            [{"type": "reboot-required", "flag": True}])
 
83
                            [{"type": "reboot-required-info",
 
84
                              "flag": True,
 
85
                              "packages": [u"foo"]}])
47
86
        self.mstore.delete_all_messages()
48
87
        self.plugin.send_message()
49
88
        self.assertMessages(self.mstore.get_pending_messages(), [])
50
89
 
 
90
    def test_run_interval(self):
 
91
        """
 
92
        The L{RebootRequired} plugin will be scheduled to run every 15 minutes.
 
93
        """
 
94
        self.assertEqual(900, self.plugin.run_interval)
 
95
 
 
96
    def test_run_immediately(self):
 
97
        """
 
98
        The L{RebootRequired} plugin will be run immediately at startup.
 
99
        """
 
100
        self.assertTrue(True, self.plugin.run_immediately)
 
101
 
51
102
    def test_run(self):
52
103
        """
53
104
        If the server can accept them, the plugin should send