1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
import unittest
from testtools import TestCase
from testtools.content import text_content
from testtools.matchers import Contains
from utils import (
assertCommandReturnCode,
CLUSTER_CONTROLLER_IP,
run_command,
)
import zmq
tests_order = [
'test_preseed_updated_cluster_config',
'test_import_pxe_files',
'test_wait_for_region_controller',
]
def sorting_method(ignored, first_test, second_test):
return tests_order.index(first_test) - tests_order.index(second_test)
unittest.TestLoader.sortTestMethodsUsing = sorting_method
class ClusterControllerIntegration(TestCase):
def test_preseed_updated_cluster_config(self):
# Make sure the cluster config was updated by the seed file.
maas_fd = open("/etc/maas/maas_cluster.conf", "r+")
maas_file = maas_fd.read()
self.assertThat(maas_file, Contains(
'MAAS_URL="http://192.168.21.5/MAAS"'))
def test_import_pxe_files(self):
cmd = ['maas-import-pxe-files']
expected_output = 'Downloading to temporary location'
assertCommandReturnCode(self, cmd, expected_output)
def test_wait_for_region_controller(self):
"""Wait for the region controller to run the integration tests.
The region controller will send a message to the cluster controller
once all tests finish.
"""
context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind('tcp://%s:5555' % CLUSTER_CONTROLLER_IP)
msg = ''
while msg != "Region controller tests finished.":
msg = socket.recv()
self.addDetail(
"Waiting for region controller to signal end of tests",
text_content(
"Waiting signal from region controller: '%s'" % msg))
@classmethod
def tearDownClass(cls):
"""Power off the cluster controller VM after the test run."""
run_command(["sudo", "poweroff"])
|