1
from base64 import b64decode
3
from xmlrpclib import Fault
6
from twisted.internet.defer import fail, succeed
7
from twisted.web.error import Error
8
from twisted.web.xmlrpc import Proxy
10
from juju.lib.serializer import dump, load
11
from juju.lib.mocker import ANY, MATCH
12
from juju.providers.orchestra import MachineProvider
14
DATA_DIR = os.path.join(os.path.abspath(os.path.dirname(__file__)), "data")
16
CONFIG = {"type": "orchestra",
17
"juju-origin": "distro",
18
"orchestra-server": "somewhe.re",
19
"orchestra-user": "user",
20
"orchestra-pass": "pass",
21
"acquired-mgmt-class": "acquired",
22
"available-mgmt-class": "available",
23
"admin-secret": "SEEKRIT",
24
"storage-url": "http://somewhe.re/webdav",
25
"authorized-keys": "this-is-a-public-key",
26
"default-series": "cyclopean"}
29
class OrchestraTestMixin(object):
31
def get_provider(self):
32
return MachineProvider("tetrascape", CONFIG)
34
def setup_mocks(self):
35
self.proxy_m = self.mocker.mock(Proxy)
36
Proxy_m = self.mocker.replace(Proxy, spec=None)
37
Proxy_m("http://somewhe.re/cobbler_api")
38
self.mocker.result(self.proxy_m)
39
self.getPage = self.mocker.replace("twisted.web.client.getPage")
40
self.get_page_auth = self.mocker.replace(
41
"juju.providers.orchestra.digestauth.get_page_auth")
43
def mock_fs_get(self, url, code, content=None):
46
self.mocker.result(succeed(content))
48
self.mocker.result(fail(Error(str(code))))
50
def mock_fs_put(self, url, expect, code=201):
51
# NOTE: in some respects, it would be better to simulate the complete
52
# interaction with the webdav provider; the factors that work against
54
# 1) authentication is tested on DigestAuthenticator, on get_page_auth,
55
# and again on FileStorage; even if it were easy to do so, testing
56
# the same paths at yet another level starts to feel somewhat
58
# 2) it's not *easy* to do so: we'd have a unique storage URL base for
59
# every test, which involves a custom config dict for every test,
60
# and unique URLs to check for every test that hits webdav, and
61
# another base class to setUp/tearDown. that's not to say it's
62
# *hard* to do so, but it's time-consuming and costs more complexity
63
# -- in a large number of the orchestra tests -- than is warranted
64
# by whatever additional verification it might allow for.
65
self.get_page_auth(url, ANY, method="PUT", postdata=expect)
66
if code in (201, 204):
67
self.mocker.result(succeed(""))
69
self.mocker.result(fail(Error(str(code))))
71
def mock_find_zookeepers(self, existing=None):
72
url = "http://somewhe.re/webdav/provider-state"
74
self.mock_fs_get(url, 404)
77
content = dump({"zookeeper-instances": [uid]})
78
self.mock_fs_get(url, 200, content)
79
self.mock_describe_systems(succeed([{
80
"uid": uid, "name": name, "mgmt_classes": ["acquired"],
81
"netboot_enabled": True}]))
83
def mock_get_systems(self, acceptable=True, unacceptable=True,
84
mgmt_classes="available"):
85
self.proxy_m.callRemote("find_system",
86
{"netboot_enabled": "true",
87
"mgmt_classes": mgmt_classes})
90
systems.append("bad-system")
92
systems.append("good-system")
93
self.mocker.result(systems)
96
self.proxy_m.callRemote("get_system", "bad-system")
97
self.mocker.result({"mgmt_classes": ["available", "acquired"],
98
"profile": "funky-x86_64-blah",
99
"uid": "bad-system-uid"})
101
self.proxy_m.callRemote("get_system", "good-system")
102
self.mocker.result({"mgmt_classes": ["preserve_me", "available"],
103
"uid": "winston-uid",
104
"profile": "crazy-i386-blah-de-blah",
105
"netboot_enabled": True})
107
def mock_acquire_system(self, unexpected_auth_error=None):
108
self.proxy_m.callRemote("find_system", {"uid": "winston-uid"})
109
self.mocker.result(succeed(["winston"]))
110
self.proxy_m.callRemote("get_system_handle", "winston", "")
111
if unexpected_auth_error is not None:
112
self.mocker.result(fail(unexpected_auth_error))
114
self.mocker.result(fail(Fault("blah", "invalid token")))
115
self.proxy_m.callRemote("login", "user", "pass")
116
self.mocker.result(succeed("TOKEN"))
117
self.proxy_m.callRemote("get_system_handle", "winston", "TOKEN")
118
self.mocker.result(succeed("smith"))
119
self.proxy_m.callRemote(
120
"modify_system", "smith", "mgmt_classes",
121
["preserve_me", "acquired"], "TOKEN")
122
self.mocker.result(succeed(True))
123
self.proxy_m.callRemote("save_system", "smith", "TOKEN")
124
self.mocker.result(succeed(True))
126
def get_verify_ks_meta(self, machine_id, user_data_filename):
129
self.assertEquals(ks_meta["MACHINE_ID"], str(machine_id))
130
user_data = load(b64decode(ks_meta["USER_DATA_BASE64"]))
131
expect_path = os.path.join(DATA_DIR, user_data_filename)
132
with open(expect_path) as f:
133
expect_user_data = load(f.read())
134
self.assertEquals(user_data, expect_user_data)
138
def mock_start_system(
139
self, verify_ks_meta, fail_modify=False, fail_save=False,
140
expect_series="splendid"):
141
self.proxy_m.callRemote("get_systems")
142
self.mocker.result(succeed([{
143
"uid": "winston-uid", "mgmt_classes": ["acquired"],
144
"profile": "crazy-i386-blah-de-blah", "ks_meta": {}}])),
145
self.proxy_m.callRemote("find_system", {"uid": "winston-uid"})
146
self.mocker.result(succeed(["winston"]))
147
self.proxy_m.callRemote("get_system_handle", "winston", "TOKEN")
148
self.mocker.result(succeed("smith"))
150
match_ks_meta = MATCH(verify_ks_meta)
151
self.proxy_m.callRemote(
152
"modify_system", "smith", "ks_meta", match_ks_meta, "TOKEN")
154
self.mocker.result(succeed(False))
156
self.mocker.result(succeed(True))
158
self.proxy_m.callRemote(
159
"modify_system", "smith", "netboot_enabled", True, "TOKEN")
160
self.mocker.result(succeed(True))
161
profile = expect_series + "-i386-juju"
162
self.proxy_m.callRemote(
163
"modify_system", "smith", "profile", profile, "TOKEN")
164
self.mocker.result(succeed(True))
165
self.proxy_m.callRemote("save_system", "smith", "TOKEN")
167
self.mocker.result(succeed(False))
169
self.mocker.result(succeed(True))
171
self.proxy_m.callRemote(
172
"background_power_system",
173
{"power": "on", "systems": ["winston"]},
175
self.mocker.result(succeed("[some-timestamp]_power"))
177
def mock_describe_systems(self, result):
178
self.proxy_m.callRemote("get_systems")
179
self.mocker.result(result)
added
removed
juju/providers/orchestra/tests/common.py
