77
80
self.assertRaises(ValueError, reader.read, 1)
80
class TestTFTPBackendRegex(MAASTestCase):
81
"""Tests for `provisioningserver.tftp.TFTPBackend.re_config_file`."""
84
def get_example_path_and_components():
85
"""Return a plausible path and its components.
87
The path is intended to match `re_config_file`, and the components are
88
the expected groups from a match.
90
components = {"mac": factory.getRandomMACAddress("-"),
93
config_path = compose_config_path(components["mac"])
94
return config_path, components
96
def test_re_config_file_is_compatible_with_config_path_generator(self):
97
# The regular expression for extracting components of the file path is
98
# compatible with the PXE config path generator.
99
regex = TFTPBackend.re_config_file
100
for iteration in range(10):
101
config_path, args = self.get_example_path_and_components()
102
match = regex.match(config_path)
103
self.assertIsNotNone(match, config_path)
104
self.assertEqual(args, match.groupdict())
106
def test_re_config_file_with_leading_slash(self):
107
# The regular expression for extracting components of the file path
108
# doesn't care if there's a leading forward slash; the TFTP server is
109
# easy on this point, so it makes sense to be also.
110
config_path, args = self.get_example_path_and_components()
111
# Ensure there's a leading slash.
112
config_path = "/" + config_path.lstrip("/")
113
match = TFTPBackend.re_config_file.match(config_path)
114
self.assertIsNotNone(match, config_path)
115
self.assertEqual(args, match.groupdict())
117
def test_re_config_file_without_leading_slash(self):
118
# The regular expression for extracting components of the file path
119
# doesn't care if there's no leading forward slash; the TFTP server is
120
# easy on this point, so it makes sense to be also.
121
config_path, args = self.get_example_path_and_components()
122
# Ensure there's no leading slash.
123
config_path = config_path.lstrip("/")
124
match = TFTPBackend.re_config_file.match(config_path)
125
self.assertIsNotNone(match, config_path)
126
self.assertEqual(args, match.groupdict())
128
def test_re_config_file_matches_classic_pxelinux_cfg(self):
129
# The default config path is simply "pxelinux.cfg" (without
130
# leading slash). The regex matches this.
131
mac = 'aa-bb-cc-dd-ee-ff'
132
match = TFTPBackend.re_config_file.match('pxelinux.cfg/01-%s' % mac)
133
self.assertIsNotNone(match)
134
self.assertEqual({'mac': mac, 'arch': None, 'subarch': None},
137
def test_re_config_file_matches_pxelinux_cfg_with_leading_slash(self):
138
mac = 'aa-bb-cc-dd-ee-ff'
139
match = TFTPBackend.re_config_file.match('/pxelinux.cfg/01-%s' % mac)
140
self.assertIsNotNone(match)
141
self.assertEqual({'mac': mac, 'arch': None, 'subarch': None},
144
def test_re_config_file_does_not_match_non_config_file(self):
146
TFTPBackend.re_config_file.match('pxelinux.cfg/kernel'))
148
def test_re_config_file_does_not_match_file_in_root(self):
150
TFTPBackend.re_config_file.match('01-aa-bb-cc-dd-ee-ff'))
152
def test_re_config_file_does_not_match_file_not_in_pxelinux_cfg(self):
154
TFTPBackend.re_config_file.match('foo/01-aa-bb-cc-dd-ee-ff'))
156
def test_re_config_file_with_default(self):
157
match = TFTPBackend.re_config_file.match('pxelinux.cfg/default')
158
self.assertIsNotNone(match)
160
{'mac': None, 'arch': None, 'subarch': None},
163
def test_re_config_file_with_default_arch(self):
164
arch = factory.make_name('arch', sep='')
165
match = TFTPBackend.re_config_file.match('pxelinux.cfg/default.%s' %
167
self.assertIsNotNone(match)
169
{'mac': None, 'arch': arch, 'subarch': None},
172
def test_re_config_file_with_default_arch_and_subarch(self):
173
arch = factory.make_name('arch', sep='')
174
subarch = factory.make_name('subarch', sep='')
175
match = TFTPBackend.re_config_file.match(
176
'pxelinux.cfg/default.%s-%s' % (arch, subarch))
177
self.assertIsNotNone(match)
179
{'mac': None, 'arch': arch, 'subarch': subarch},
183
83
class TestTFTPBackend(MAASTestCase):
184
84
"""Tests for `provisioningserver.tftp.TFTPBackend`."""
282
182
get_page_patch = self.patch(backend, "get_page")
283
183
get_page_patch.return_value = succeed(fake_get_page_result)
285
# Stub render_pxe_config to return the render parameters.
185
# Stub render_config to return the render parameters.
186
method = PXEBootMethod()
286
187
fake_render_result = factory.make_name("render")
287
render_patch = self.patch(backend, "render_pxe_config")
188
render_patch = self.patch(method, "render_config")
288
189
render_patch.return_value = fake_render_result
290
191
# Get the rendered configuration, which will actually be a JSON dump
291
192
# of the render-time parameters.
292
reader = yield backend.get_config_reader(fake_params)
193
reader = yield backend.get_config_reader(method, fake_params)
293
194
self.addCleanup(reader.finish)
294
195
self.assertIsInstance(reader, BytesReader)
295
196
output = reader.read(10000)
297
198
# The kernel parameters were fetched using `backend.get_page`.
298
backend.get_page.assert_called_once()
199
self.assertThat(backend.get_page, MockCalledOnceWith(mock.ANY))
300
# The result has been rendered by `backend.render_pxe_config`.
201
# The result has been rendered by `backend.render_config`.
301
202
self.assertEqual(fake_render_result.encode("utf-8"), output)
302
backend.render_pxe_config.assert_called_once_with(
303
kernel_params=fake_kernel_params, **fake_params)
203
self.assertThat(method.render_config, MockCalledOnceWith(
204
kernel_params=fake_kernel_params, **fake_params))
207
def test_get_config_reader_substitutes_armhf_in_params(self):
208
# get_config_reader() should substitute "arm" for "armhf" in the
209
# arch field of the parameters (mapping from pxe to maas
211
cluster_uuid = factory.getRandomUUID()
212
self.patch(tftp_module, 'get_cluster_uuid').return_value = (
214
config_path = "pxelinux.cfg/default-arm"
215
backend = TFTPBackend(self.make_dir(), b"http://example.com/")
216
# python-tx-tftp sets up call context so that backends can discover
217
# more about the environment in which they're running.
220
factory.getRandomIPAddress(),
221
factory.getRandomPort()),
223
factory.getRandomIPAddress(),
224
factory.getRandomPort()),
227
@partial(self.patch, backend, "get_config_reader")
228
def get_config_reader(boot_method, params):
229
params_json = json.dumps(params)
230
params_json_reader = BytesReader(params_json)
231
return succeed(params_json_reader)
233
reader = yield context.call(
234
call_context, backend.get_reader, config_path)
235
output = reader.read(10000)
236
observed_params = json.loads(output)
237
self.assertEqual("armhf", observed_params["arch"])
306
240
class TestTFTPService(MAASTestCase):