← Back to branch summary

~openstack-charmers/charm-helpers/stable-1604

« back to all changes in this revision

Viewing changes to tests/core/test_host.py

  • Committer: James Page
  • Date: 2015-10-22 13:25:47 UTC
  • Revision ID: james.page@ubuntu.com-20151022132547-6las4qai8fo93fol
Tags: 15.10
15.10 Charm release

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/core/test_host.py
35
35
    link/ether e4:11:5b:ab:a7:3c brd ff:ff:ff:ff:ff:ff
36
36
"""
37
37
 
 
38
IP_LINE_ETH100 = b"""
 
39
2: eth100: <BROADCAST,MULTICAST,SLAVE,UP,LOWER_UP> mtu 1500 qdisc mq master bond0 state UP qlen 1000
 
40
    link/ether e4:11:5b:ab:a7:3d brd ff:ff:ff:ff:ff:ff
 
41
"""
 
42
 
38
43
IP_LINE_ETH0_VLAN = b"""
39
44
6: eth0.10@eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue state UP group default
40
45
    link/ether 08:00:27:16:b9:5f brd ff:ff:ff:ff:ff:ff
47
52
 
48
53
IP_LINE_HWADDR = b"""2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc pfifo_fast state UP qlen 1000\    link/ether e4:11:5b:ab:a7:3c brd ff:ff:ff:ff:ff:ff"""
49
54
 
50
 
IP_LINES = IP_LINE_ETH0 + IP_LINE_ETH1 + IP_LINE_ETH0_VLAN
 
55
IP_LINES = IP_LINE_ETH0 + IP_LINE_ETH1 + IP_LINE_ETH0_VLAN + IP_LINE_ETH100
51
56
 
52
57
IP_LINE_BONDS = b"""
53
58
6: bond0.10@bond0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue state UP group default
103
108
        service.assert_called_with('restart', service_name)
104
109
 
105
110
    @patch.object(host, 'service')
106
 
    def test_pauses_a_service(self, service):
 
111
    def test_pauses_an_upstart_service(self, service):
107
112
        service_name = 'foo-service'
108
113
        service.side_effect = [True]
109
 
        tempdir = mkdtemp(prefix="test_pauses_a_service")
 
114
        tempdir = mkdtemp(prefix="test_pauses_an_upstart_service")
 
115
        conf_path = os.path.join(tempdir, "{}.conf".format(service_name))
 
116
        # Just needs to exist
 
117
        with open(conf_path, "w") as fh:
 
118
            fh.write("")
110
119
        self.addCleanup(rmtree, tempdir)
111
120
        self.assertTrue(host.service_pause(service_name, init_dir=tempdir))
112
121
 
117
126
            override_contents = fh.read()
118
127
        self.assertEqual("manual\n", override_contents)
119
128
 
120
 
    @patch.object(host, 'service')
121
 
    def test_resumes_a_service(self, service):
122
 
        service_name = 'foo-service'
123
 
        service.side_effect = [True]
124
 
        tempdir = mkdtemp(prefix="test_resumes_a_service")
 
129
    @patch('subprocess.check_call')
 
130
    @patch.object(host, 'service')
 
131
    def test_pauses_a_sysv_service(self, service, check_call):
 
132
        service_name = 'foo-service'
 
133
        service.side_effect = [True]
 
134
        tempdir = mkdtemp(prefix="test_pauses_a_sysv_service")
 
135
        sysv_path = os.path.join(tempdir, service_name)
 
136
        # Just needs to exist
 
137
        with open(sysv_path, "w") as fh:
 
138
            fh.write("")
 
139
        self.addCleanup(rmtree, tempdir)
 
140
        self.assertTrue(host.service_pause(
 
141
            service_name, init_dir=tempdir, initd_dir=tempdir))
 
142
 
 
143
        service.assert_called_with('stop', service_name)
 
144
        check_call.assert_called_with(["update-rc.d", service_name, "disable"])
 
145
 
 
146
    @patch.object(host, 'service')
 
147
    def test_pause_with_unknown_service(self, service):
 
148
        service_name = 'foo-service'
 
149
        service.side_effect = [True]
 
150
        tempdir = mkdtemp(prefix="test_pauses_with_unknown_service")
 
151
        self.addCleanup(rmtree, tempdir)
 
152
        exception = self.assertRaises(
 
153
            ValueError, host.service_pause,
 
154
            service_name, init_dir=tempdir, initd_dir=tempdir)
 
155
        self.assertIn(
 
156
            "Unable to detect {0}".format(service_name), str(exception))
 
157
        self.assertIn(tempdir, str(exception))
 
158
 
 
159
    @patch.object(host, 'service')
 
160
    def test_resumes_an_upstart_service(self, service):
 
161
        service_name = 'foo-service'
 
162
        service.side_effect = [True]
 
163
        tempdir = mkdtemp(prefix="test_resumes_an_upstart_service")
 
164
        conf_path = os.path.join(tempdir, "{}.conf".format(service_name))
 
165
        with open(conf_path, "w") as fh:
 
166
            fh.write("")
125
167
        self.addCleanup(rmtree, tempdir)
126
168
        self.assertTrue(host.service_resume(service_name, init_dir=tempdir))
127
169
 
130
172
            tempdir, "{}.override".format(service_name))
131
173
        self.assertFalse(os.path.exists(override_path))
132
174
 
 
175
    @patch('subprocess.check_call')
 
176
    @patch.object(host, 'service')
 
177
    def test_resumes_a_sysv_service(self, service, check_call):
 
178
        service_name = 'foo-service'
 
179
        service.side_effect = [True]
 
180
        tempdir = mkdtemp(prefix="test_resumes_a_sysv_service")
 
181
        sysv_path = os.path.join(tempdir, service_name)
 
182
        # Just needs to exist
 
183
        with open(sysv_path, "w") as fh:
 
184
            fh.write("")
 
185
        self.addCleanup(rmtree, tempdir)
 
186
        self.assertTrue(host.service_resume(
 
187
            service_name, init_dir=tempdir, initd_dir=tempdir))
 
188
 
 
189
        service.assert_called_with('start', service_name)
 
190
        check_call.assert_called_with(["update-rc.d", service_name, "enable"])
 
191
 
 
192
    @patch.object(host, 'service')
 
193
    def test_resume_with_unknown_service(self, service):
 
194
        service_name = 'foo-service'
 
195
        service.side_effect = [True]
 
196
        tempdir = mkdtemp(prefix="test_resumes_with_unknown_service")
 
197
        self.addCleanup(rmtree, tempdir)
 
198
        exception = self.assertRaises(
 
199
            ValueError, host.service_resume,
 
200
            service_name, init_dir=tempdir, initd_dir=tempdir)
 
201
        self.assertIn(
 
202
            "Unable to detect {0}".format(service_name), str(exception))
 
203
        self.assertIn(tempdir, str(exception))
 
204
 
133
205
    @patch.object(host, 'service')
134
206
    def test_reloads_a_service(self, service):
135
207
        service_name = 'foo-service'
303
375
        ])
304
376
        getpwnam.assert_called_with(username)
305
377
 
 
378
    @patch('pwd.getpwnam')
 
379
    def test_user_exists_true(self, getpwnam):
 
380
        getpwnam.side_effect = 'pw info'
 
381
        self.assertTrue(host.user_exists('bob'))
 
382
 
 
383
    @patch('pwd.getpwnam')
 
384
    def test_user_exists_false(self, getpwnam):
 
385
        getpwnam.side_effect = KeyError('user not found')
 
386
        self.assertFalse(host.user_exists('bob'))
 
387
 
306
388
    @patch('subprocess.check_call')
307
389
    @patch.object(host, 'log')
308
390
    def test_adds_a_user_to_a_group(self, log, check_call):
666
748
        '/etc/missing.conf': None
667
749
    }
668
750
 
 
751
    @patch('subprocess.check_output')
 
752
    @patch.object(host, 'log')
 
753
    def test_fstab_mount(self, log, check_output):
 
754
        self.assertTrue(host.fstab_mount('/mnt/mymntpnt'))
 
755
        check_output.assert_called_with(['mount', '/mnt/mymntpnt'])
 
756
 
 
757
    @patch('subprocess.check_output')
 
758
    @patch.object(host, 'log')
 
759
    def test_fstab_mount_fail(self, log, check_output):
 
760
        error = subprocess.CalledProcessError(123, 'mount it', 'Oops...')
 
761
        check_output.side_effect = error
 
762
        self.assertFalse(host.fstab_mount('/mnt/mymntpnt'))
 
763
        check_output.assert_called_with(['mount', '/mnt/mymntpnt'])
 
764
 
669
765
    @patch('hashlib.md5')
670
766
    @patch('os.path.exists')
671
767
    def test_file_hash_exists(self, exists, md5):
938
1034
        pw2 = host.pwgen(10)
939
1035
        self.assertNotEqual(pw, pw2, 'Duplicated password')
940
1036
 
 
1037
    @patch.object(host, 'glob')
 
1038
    @patch('os.path.realpath')
 
1039
    @patch('os.path.isdir')
 
1040
    def test_is_phy_iface(self, mock_isdir, mock_realpath, mock_glob):
 
1041
        mock_isdir.return_value = True
 
1042
        mock_glob.glob.return_value = ['/sys/class/net/eth0',
 
1043
                                       '/sys/class/net/veth0']
 
1044
 
 
1045
        def fake_realpath(soft):
 
1046
            if soft.endswith('/eth0'):
 
1047
                hard = \
 
1048
                    '/sys/devices/pci0000:00/0000:00:1c.4/0000:02:00.1/net/eth0'
 
1049
            else:
 
1050
                hard = '/sys/devices/virtual/net/veth0'
 
1051
 
 
1052
            return hard
 
1053
 
 
1054
        mock_realpath.side_effect = fake_realpath
 
1055
        self.assertTrue(host.is_phy_iface('eth0'))
 
1056
        self.assertFalse(host.is_phy_iface('veth0'))
 
1057
 
 
1058
    @patch('os.path.exists')
 
1059
    @patch('os.path.realpath')
 
1060
    @patch('os.path.isdir')
 
1061
    def test_get_bond_master(self, mock_isdir, mock_realpath, mock_exists):
 
1062
        mock_isdir.return_value = True
 
1063
 
 
1064
        def fake_realpath(soft):
 
1065
            if soft.endswith('/eth0'):
 
1066
                return \
 
1067
                    '/sys/devices/pci0000:00/0000:00:1c.4/0000:02:00.1/net/eth0'
 
1068
            elif soft.endswith('/br0'):
 
1069
                return '/sys/devices/virtual/net/br0'
 
1070
            elif soft.endswith('/master'):
 
1071
                return '/sys/devices/virtual/net/bond0'
 
1072
 
 
1073
            return None
 
1074
 
 
1075
        def fake_exists(path):
 
1076
            return True
 
1077
 
 
1078
        mock_exists.side_effect = fake_exists
 
1079
        mock_realpath.side_effect = fake_realpath
 
1080
        self.assertEqual(host.get_bond_master('eth0'), 'bond0')
 
1081
        self.assertIsNone(host.get_bond_master('br0'))
 
1082
 
941
1083
    @patch('subprocess.check_output')
942
1084
    def test_list_nics(self, check_output):
943
1085
        check_output.return_value = IP_LINES
 
1086
        nics = host.list_nics()
 
1087
        self.assertEqual(nics, ['eth0', 'eth1', 'eth0.10', 'eth100'])
944
1088
        nics = host.list_nics('eth')
945
 
        self.assertEqual(nics, ['eth0', 'eth1', 'eth0.10'])
 
1089
        self.assertEqual(nics, ['eth0', 'eth1', 'eth0.10', 'eth100'])
946
1090
        nics = host.list_nics(['eth'])
947
 
        self.assertEqual(nics, ['eth0', 'eth1', 'eth0.10'])
 
1091
        self.assertEqual(nics, ['eth0', 'eth1', 'eth0.10', 'eth100'])
948
1092
 
949
1093
    @patch('subprocess.check_output')
950
1094
    def test_list_nics_with_bonds(self, check_output):