1 | #!/usr/bin/python3 |
---|
2 | |
---|
3 | # Copyright 2019-2020 Fiona Klute |
---|
4 | # |
---|
5 | # Licensed under the Apache License, Version 2.0 (the "License"); |
---|
6 | # you may not use this file except in compliance with the License. |
---|
7 | # You may obtain a copy of the License at |
---|
8 | # |
---|
9 | # http://www.apache.org/licenses/LICENSE-2.0 |
---|
10 | # |
---|
11 | # Unless required by applicable law or agreed to in writing, software |
---|
12 | # distributed under the License is distributed on an "AS IS" BASIS, |
---|
13 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
---|
14 | # See the License for the specific language governing permissions and |
---|
15 | # limitations under the License. |
---|
16 | |
---|
17 | """Test objects and support functions for the mod_gnutls test |
---|
18 | suite. The classes defined in this module represent structures in the |
---|
19 | YAML test configuration files. |
---|
20 | |
---|
21 | The test configuration file defines either a TestConnection, or a list |
---|
22 | of them. Each connection contains a list of actions to run using this |
---|
23 | connection. The actions define their expected results, if an |
---|
24 | expectation is not met mgstest.TestExpectationFailed is raised. |
---|
25 | |
---|
26 | Example of a connection that runs two request actions, which are |
---|
27 | expected to succeed: |
---|
28 | |
---|
29 | ```yaml |
---|
30 | !connection |
---|
31 | # "host" defaults to $TEST_TARGET, so usually there's no need to set |
---|
32 | # it. You can use ${VAR} to substitute environment variables. |
---|
33 | host: 'localhost' |
---|
34 | # "port" defaults to $TEST_PORT, so usually there's no need to set it |
---|
35 | # it. You can use ${VAR} to substitute environment variables. |
---|
36 | port: '${TEST_PORT}' |
---|
37 | # All elements of gnutls_params will be prefixed with "--" and passed |
---|
38 | # to gnutls-cli on the command line. |
---|
39 | gnutls_params: |
---|
40 | - x509cafile=authority/x509.pem |
---|
41 | # The transport encryption. "Gnutls" is the default, "plain" can be |
---|
42 | # set to get an unencrypted connection (e.g. to test redirection to |
---|
43 | # HTTPS). |
---|
44 | transport: 'gnutls' |
---|
45 | description: 'This connection description will be logged.' |
---|
46 | actions: |
---|
47 | - !request |
---|
48 | # GET is the default. |
---|
49 | method: GET |
---|
50 | # The path part of the URL, required. |
---|
51 | path: /test.txt |
---|
52 | # "Expect" defines how the response must look to pass the test. |
---|
53 | expect: |
---|
54 | # 200 (OK) is the default. |
---|
55 | status: 200 |
---|
56 | # The response body is analyzed only if the "body" element |
---|
57 | # exists, otherwise any content is accepted. |
---|
58 | body: |
---|
59 | # The full response body must exactly match this string. |
---|
60 | exactly: | |
---|
61 | test |
---|
62 | - !request |
---|
63 | path: /status?auto |
---|
64 | expect: |
---|
65 | # The headers are analyzed only if the "headers" element exists. |
---|
66 | headers: |
---|
67 | # The Content-Type header must be present with exactly this |
---|
68 | # value. You can use ${VAR} to substitute environment |
---|
69 | # variables in the value. |
---|
70 | Content-Type: 'text/plain; charset=ISO-8859-1' |
---|
71 | # You can check the absence of a header by expecting null: |
---|
72 | X-Forbidden-Header: null |
---|
73 | body: |
---|
74 | # All strings in this list must occur in the body, in any |
---|
75 | # order. "Contains" may also contain a single string instead |
---|
76 | # of a list. |
---|
77 | contains: |
---|
78 | - 'Using GnuTLS version: ' |
---|
79 | - 'Current TLS session: (TLS1.3)' |
---|
80 | ``` |
---|
81 | |
---|
82 | Example of a connection that is expected to fail at the TLS level, in |
---|
83 | this case because the configured CA is not the one that issued the |
---|
84 | server certificate: |
---|
85 | |
---|
86 | ```yaml |
---|
87 | - !connection |
---|
88 | gnutls_params: |
---|
89 | - x509cafile=rogueca/x509.pem |
---|
90 | actions: |
---|
91 | - !request |
---|
92 | path: / |
---|
93 | expect: |
---|
94 | # The connection is expected to reset without an HTTP response. |
---|
95 | reset: yes |
---|
96 | ``` |
---|
97 | |
---|
98 | """ |
---|
99 | |
---|
100 | import os |
---|
101 | import re |
---|
102 | import select |
---|
103 | import subprocess |
---|
104 | import sys |
---|
105 | import yaml |
---|
106 | |
---|
107 | from enum import Enum, auto |
---|
108 | from http.client import HTTPConnection |
---|
109 | from string import Template |
---|
110 | |
---|
111 | from . import TestExpectationFailed |
---|
112 | from .http import HTTPSubprocessConnection |
---|
113 | |
---|
114 | class Transports(Enum): |
---|
115 | """Transports supported by TestConnection.""" |
---|
116 | GNUTLS = auto() |
---|
117 | PLAIN = auto() |
---|
118 | |
---|
119 | def __repr__(self): |
---|
120 | return f'{self.__class__.__name__!s}.{self.name}' |
---|
121 | |
---|
122 | class TestConnection(yaml.YAMLObject): |
---|
123 | """An HTTP connection in a test. It includes parameters for the |
---|
124 | transport, and the actions (e.g. sending requests) to take using |
---|
125 | this connection. |
---|
126 | |
---|
127 | Note that running one TestConnection object may result in multiple |
---|
128 | sequential network connections if the transport gets closed in a |
---|
129 | non-failure way (e.g. following a "Connection: close" request) and |
---|
130 | there are more actions, or (rarely) if an action requires its own |
---|
131 | transport. |
---|
132 | |
---|
133 | """ |
---|
134 | yaml_tag = '!connection' |
---|
135 | |
---|
136 | def __init__(self, actions, host=None, port=None, gnutls_params=[], |
---|
137 | transport='gnutls', description=None): |
---|
138 | self.gnutls_params = gnutls_params |
---|
139 | self.actions = actions |
---|
140 | self.transport = Transports[transport.upper()] |
---|
141 | self.description = description |
---|
142 | if host: |
---|
143 | self.host = subst_env(host) |
---|
144 | else: |
---|
145 | self.host = os.environ.get('TEST_TARGET', 'localhost') |
---|
146 | if port: |
---|
147 | self.port = int(subst_env(port)) |
---|
148 | else: |
---|
149 | self.port = int(os.environ.get('TEST_PORT', 8000)) |
---|
150 | |
---|
151 | def __repr__(self): |
---|
152 | return (f'{self.__class__.__name__!s}' |
---|
153 | f'(host={self.host!r}, port={self.port!r}, ' |
---|
154 | f'gnutls_params={self.gnutls_params!r}, ' |
---|
155 | f'actions={self.actions!r}, transport={self.transport!r}, ' |
---|
156 | f'description={self.description!r})') |
---|
157 | |
---|
158 | def run(self, timeout=5.0, conn_log=None, response_log=None): |
---|
159 | """Set up an HTTP connection and run the configured actions.""" |
---|
160 | |
---|
161 | # note: "--logfile" option requires GnuTLS version >= 3.6.7 |
---|
162 | command = ['gnutls-cli', '--logfile=/dev/stderr'] |
---|
163 | for s in self.gnutls_params: |
---|
164 | command.append('--' + s) |
---|
165 | command = command + ['-p', str(self.port), self.host] |
---|
166 | |
---|
167 | if self.transport == Transports.GNUTLS: |
---|
168 | conn = HTTPSubprocessConnection(command, self.host, self.port, |
---|
169 | output_filter=filter_cert_log, |
---|
170 | stderr_log=conn_log, |
---|
171 | timeout=timeout) |
---|
172 | elif self.transport == Transports.PLAIN: |
---|
173 | conn = HTTPConnection(self.host, port=self.port, |
---|
174 | timeout=timeout) |
---|
175 | |
---|
176 | try: |
---|
177 | for act in self.actions: |
---|
178 | if type(act) is TestRequest: |
---|
179 | act.run(conn, response_log) |
---|
180 | elif type(act) is TestReq10: |
---|
181 | act.run(command, timeout, conn_log, response_log) |
---|
182 | elif type(act) is Resume: |
---|
183 | act.run(conn, command) |
---|
184 | else: |
---|
185 | raise TypeError(f'Unsupported action requested: {act!r}') |
---|
186 | finally: |
---|
187 | conn.close() |
---|
188 | sys.stdout.flush() |
---|
189 | |
---|
190 | @classmethod |
---|
191 | def from_yaml(cls, loader, node): |
---|
192 | fields = loader.construct_mapping(node) |
---|
193 | conn = cls(**fields) |
---|
194 | return conn |
---|
195 | |
---|
196 | |
---|
197 | |
---|
198 | class TestRequest(yaml.YAMLObject): |
---|
199 | """Test action that sends an HTTP/1.1 request. |
---|
200 | |
---|
201 | The path must be specified in the configuration file, all other |
---|
202 | parameters (method, headers, expected response) have |
---|
203 | defaults. |
---|
204 | |
---|
205 | Options for checking the response currently are: |
---|
206 | * require a specific response status |
---|
207 | * require specific headers to be present with specific values |
---|
208 | * require the body to exactly match a specific string |
---|
209 | * require the body to contain all of a list of strings |
---|
210 | |
---|
211 | """ |
---|
212 | yaml_tag = '!request' |
---|
213 | def __init__(self, path, method='GET', body=None, headers=dict(), |
---|
214 | expect=dict(status=200)): |
---|
215 | self.method = method |
---|
216 | self.path = path |
---|
217 | self.body = body.encode('utf-8') if body else None |
---|
218 | self.headers = headers |
---|
219 | self.expect = expect |
---|
220 | |
---|
221 | def __repr__(self): |
---|
222 | return (f'{self.__class__.__name__!s}(path={self.path!r}, ' |
---|
223 | f'method={self.method!r}, body={self.body!r}, ' |
---|
224 | f'headers={self.headers!r}, expect={self.expect!r})') |
---|
225 | |
---|
226 | def run(self, conn, response_log=None): |
---|
227 | try: |
---|
228 | conn.request(self.method, self.path, body=self.body, |
---|
229 | headers=self.headers) |
---|
230 | resp = conn.getresponse() |
---|
231 | if self.expects_conn_reset(): |
---|
232 | raise TestExpectationFailed( |
---|
233 | 'Expected connection reset did not occur!') |
---|
234 | except (BrokenPipeError, ConnectionResetError) as err: |
---|
235 | if self.expects_conn_reset(): |
---|
236 | print('connection reset as expected.') |
---|
237 | return |
---|
238 | else: |
---|
239 | raise err |
---|
240 | body = resp.read().decode() |
---|
241 | log_str = format_response(resp, body) |
---|
242 | print(log_str) |
---|
243 | if response_log: |
---|
244 | print(log_str, file=response_log) |
---|
245 | self.check_response(resp, body) |
---|
246 | |
---|
247 | def check_headers(self, headers): |
---|
248 | """ |
---|
249 | >>> r1 = TestRequest(path='/test.txt', |
---|
250 | ... expect={ 'headers': {'X-Forbidden-Header': None, |
---|
251 | ... 'X-Required-Header': 'Hi!' }}) |
---|
252 | >>> r1.check_headers({ 'X-Required-Header': 'Hi!' }) |
---|
253 | >>> r1.check_headers({ 'X-Required-Header': 'Hello!' }) |
---|
254 | Traceback (most recent call last): |
---|
255 | ... |
---|
256 | mgstest.TestExpectationFailed: Unexpected value in header X-Required-Header: 'Hello!', expected 'Hi!' |
---|
257 | >>> r1.check_headers({ 'X-Forbidden-Header': 'Hi!' }) |
---|
258 | Traceback (most recent call last): |
---|
259 | ... |
---|
260 | mgstest.TestExpectationFailed: Unexpected value in header X-Forbidden-Header: 'Hi!', expected None |
---|
261 | """ |
---|
262 | for name, expected in self.expect['headers'].items(): |
---|
263 | value = headers.get(name) |
---|
264 | expected = subst_env(expected) |
---|
265 | if value != expected: |
---|
266 | raise TestExpectationFailed( |
---|
267 | f'Unexpected value in header {name}: {value!r}, ' |
---|
268 | f'expected {expected!r}') |
---|
269 | |
---|
270 | def check_body(self, body): |
---|
271 | """ |
---|
272 | >>> r1 = TestRequest(path='/test.txt', method='GET', headers={}, expect={'status': 200, 'body': {'exactly': 'test\\n'}}) |
---|
273 | >>> r1.check_body('test\\n') |
---|
274 | >>> r1.check_body('xyz\\n') |
---|
275 | Traceback (most recent call last): |
---|
276 | ... |
---|
277 | mgstest.TestExpectationFailed: Unexpected body: 'xyz\\n' != 'test\\n' |
---|
278 | >>> r2 = TestRequest(path='/test.txt', method='GET', headers={}, expect={'status': 200, 'body': {'contains': ['tes', 'est']}}) |
---|
279 | >>> r2.check_body('test\\n') |
---|
280 | >>> r2.check_body('est\\n') |
---|
281 | Traceback (most recent call last): |
---|
282 | ... |
---|
283 | mgstest.TestExpectationFailed: Unexpected body: 'est\\n' does not contain 'tes' |
---|
284 | >>> r3 = TestRequest(path='/test.txt', method='GET', headers={}, expect={'status': 200, 'body': {'contains': 'test'}}) |
---|
285 | >>> r3.check_body('test\\n') |
---|
286 | """ |
---|
287 | if 'exactly' in self.expect['body'] \ |
---|
288 | and body != self.expect['body']['exactly']: |
---|
289 | raise TestExpectationFailed( |
---|
290 | f'Unexpected body: {body!r} != ' |
---|
291 | f'{self.expect["body"]["exactly"]!r}') |
---|
292 | if 'contains' in self.expect['body']: |
---|
293 | if type(self.expect['body']['contains']) is str: |
---|
294 | self.expect['body']['contains'] = [ |
---|
295 | self.expect['body']['contains']] |
---|
296 | for s in self.expect['body']['contains']: |
---|
297 | if not s in body: |
---|
298 | raise TestExpectationFailed( |
---|
299 | f'Unexpected body: {body!r} does not contain ' |
---|
300 | f'{s!r}') |
---|
301 | |
---|
302 | def check_response(self, response, body): |
---|
303 | if self.expects_conn_reset(): |
---|
304 | raise TestExpectationFailed( |
---|
305 | 'Got a response, but connection should have failed!') |
---|
306 | if response.status != self.expect['status']: |
---|
307 | raise TestExpectationFailed( |
---|
308 | f'Unexpected status: {response.status} != ' |
---|
309 | f'{self.expect["status"]}') |
---|
310 | if 'headers' in self.expect: |
---|
311 | self.check_headers(dict(response.getheaders())) |
---|
312 | if 'body' in self.expect: |
---|
313 | self.check_body(body) |
---|
314 | |
---|
315 | def expects_conn_reset(self): |
---|
316 | """Returns True if running this request is expected to fail due to the |
---|
317 | connection being reset. That usually means the underlying TLS |
---|
318 | connection failed. |
---|
319 | |
---|
320 | >>> r1 = TestRequest(path='/test.txt', method='GET', headers={}, expect={'status': 200, 'body': {'contains': 'test'}}) |
---|
321 | >>> r1.expects_conn_reset() |
---|
322 | False |
---|
323 | >>> r2 = TestRequest(path='/test.txt', method='GET', headers={}, expect={'reset': True}) |
---|
324 | >>> r2.expects_conn_reset() |
---|
325 | True |
---|
326 | """ |
---|
327 | if 'reset' in self.expect: |
---|
328 | return self.expect['reset'] |
---|
329 | return False |
---|
330 | |
---|
331 | @classmethod |
---|
332 | def from_yaml(cls, loader, node): |
---|
333 | fields = loader.construct_mapping(node) |
---|
334 | req = cls(**fields) |
---|
335 | return req |
---|
336 | |
---|
337 | |
---|
338 | |
---|
339 | class TestReq10(TestRequest): |
---|
340 | """Test action that sends a request using a minimal (and likely |
---|
341 | incomplete) HTTP/1.0 test client for the one test case that |
---|
342 | strictly requires HTTP/1.0. |
---|
343 | |
---|
344 | TestReq10 objects use the same YAML parameters and defaults as |
---|
345 | TestRequest, but note that an empty "headers" parameter means that |
---|
346 | not even a "Host:" header will be sent. All headers must be |
---|
347 | specified in the test configuration file. |
---|
348 | |
---|
349 | """ |
---|
350 | yaml_tag = '!request10' |
---|
351 | status_re = re.compile(r'^HTTP/([\d\.]+) (\d+) (.*)$') |
---|
352 | header_re = re.compile(r'^([-\w]+):\s+(.*)$') |
---|
353 | |
---|
354 | def __init__(self, **kwargs): |
---|
355 | super().__init__(**kwargs) |
---|
356 | |
---|
357 | def run(self, command, timeout=None, conn_log=None, response_log=None): |
---|
358 | req = f'{self.method} {self.path} HTTP/1.0\r\n' |
---|
359 | for name, value in self.headers.items(): |
---|
360 | req = req + f'{name}: {value}\r\n' |
---|
361 | req = req.encode('utf-8') + b'\r\n' |
---|
362 | if self.body: |
---|
363 | req = req + self.body |
---|
364 | proc = subprocess.Popen(command, |
---|
365 | stdout=subprocess.PIPE, |
---|
366 | stderr=subprocess.PIPE, |
---|
367 | stdin=subprocess.PIPE, |
---|
368 | close_fds=True, |
---|
369 | bufsize=0) |
---|
370 | try: |
---|
371 | outs, errs = proc.communicate(input=req, timeout=timeout) |
---|
372 | except TimeoutExpired: |
---|
373 | proc.kill() |
---|
374 | outs, errs = proc.communicate() |
---|
375 | |
---|
376 | print(errs.decode()) |
---|
377 | if conn_log: |
---|
378 | print(errs.decode(), file=conn_log) |
---|
379 | |
---|
380 | if proc.returncode != 0: |
---|
381 | if len(outs) != 0: |
---|
382 | raise TestExpectationFailed( |
---|
383 | f'Connection failed, but got output: {outs!r}') |
---|
384 | if self.expects_conn_reset(): |
---|
385 | print('connection reset as expected.') |
---|
386 | return |
---|
387 | else: |
---|
388 | raise TestExpectationFailed( |
---|
389 | 'Connection failed unexpectedly!') |
---|
390 | else: |
---|
391 | if self.expects_conn_reset(): |
---|
392 | raise TestExpectationFailed( |
---|
393 | 'Expected connection reset did not occur!') |
---|
394 | |
---|
395 | # first line of the received data must be the status |
---|
396 | status, rest = outs.decode().split('\r\n', maxsplit=1) |
---|
397 | # headers and body are separated by double newline |
---|
398 | head, body = rest.split('\r\n\r\n', maxsplit=1) |
---|
399 | # log response for debugging |
---|
400 | print(f'{status}\n{head}\n\n{body}') |
---|
401 | if response_log: |
---|
402 | print(f'{status}\n{head}\n\n{body}', file=response_log) |
---|
403 | |
---|
404 | m = self.status_re.match(status) |
---|
405 | if m: |
---|
406 | status_code = int(m.group(2)) |
---|
407 | status_expect = self.expect.get('status') |
---|
408 | if status_expect and not status_code == status_expect: |
---|
409 | raise TestExpectationFailed('Unexpected status code: ' |
---|
410 | f'{status}, expected ' |
---|
411 | f'{status_expect}') |
---|
412 | else: |
---|
413 | raise TestExpectationFailed(f'Invalid status line: "{status}"') |
---|
414 | |
---|
415 | if 'headers' in self.expect: |
---|
416 | headers = dict() |
---|
417 | for line in head.splitlines(): |
---|
418 | m = self.header_re.fullmatch(line) |
---|
419 | if m: |
---|
420 | headers[m.group(1)] = m.group(2) |
---|
421 | self.check_headers(headers) |
---|
422 | |
---|
423 | if 'body' in self.expect: |
---|
424 | self.check_body(body) |
---|
425 | |
---|
426 | |
---|
427 | |
---|
428 | class Resume(yaml.YAMLObject): |
---|
429 | """Test action to close and resume the TLS session. |
---|
430 | |
---|
431 | Send the gnutls-cli inline command "^resume^" to close and resume |
---|
432 | the TLS session. "inline-commands" must be present in |
---|
433 | gnutls_params of the parent connection. This action does not need |
---|
434 | any arguments, but you must specify with an explicitly empty |
---|
435 | dictionary for YAML parsing to work, like this: |
---|
436 | |
---|
437 | !resume {} |
---|
438 | |
---|
439 | """ |
---|
440 | yaml_tag = '!resume' |
---|
441 | def run(self, conn, command): |
---|
442 | if not '--inline-commands' in command: |
---|
443 | raise ValueError('gnutls_params must include "inline-commands" ' |
---|
444 | 'to use the resume action!') |
---|
445 | if not type(conn) is HTTPSubprocessConnection: |
---|
446 | raise TypeError('Resume action works only with ' |
---|
447 | 'HTTPSubprocessConnection.') |
---|
448 | conn.sock.send(b'^resume^\n') |
---|
449 | |
---|
450 | |
---|
451 | |
---|
452 | def filter_cert_log(in_stream, out_stream): |
---|
453 | """Filter to stop an erroneous gnutls-cli log message. |
---|
454 | |
---|
455 | This function filters out a log line about loading client |
---|
456 | certificates that is mistakenly sent to stdout from gnutls-cli. My |
---|
457 | fix (https://gitlab.com/gnutls/gnutls/merge_requests/1125) has |
---|
458 | been merged, but buggy binaries will probably be around for a |
---|
459 | while. |
---|
460 | |
---|
461 | The filter is meant to run in a multiprocessing.Process or |
---|
462 | threading.Thread that receives the stdout of gnutls-cli as |
---|
463 | in_stream, and a connection for further processing as out_stream. |
---|
464 | |
---|
465 | """ |
---|
466 | # message to filter |
---|
467 | cert_log = b'Processed 1 client X.509 certificates...\n' |
---|
468 | |
---|
469 | # Set the input to non-blocking mode |
---|
470 | fd = in_stream.fileno() |
---|
471 | os.set_blocking(fd, False) |
---|
472 | |
---|
473 | # The poll object allows waiting for events on non-blocking IO |
---|
474 | # channels. |
---|
475 | poller = select.poll() |
---|
476 | poller.register(fd) |
---|
477 | |
---|
478 | init_done = False |
---|
479 | run_loop = True |
---|
480 | while run_loop: |
---|
481 | # The returned tuples are file descriptor and event, but |
---|
482 | # we're only listening on one stream anyway, so we don't |
---|
483 | # need to check it here. |
---|
484 | for x, event in poller.poll(): |
---|
485 | # Critical: "event" is a bitwise OR of the POLL* constants |
---|
486 | if event & select.POLLIN or event & select.POLLPRI: |
---|
487 | data = in_stream.read() |
---|
488 | if not init_done: |
---|
489 | # If the erroneous log line shows up it's the |
---|
490 | # first piece of data we receive. Just copy |
---|
491 | # everything after. |
---|
492 | init_done = True |
---|
493 | if cert_log in data: |
---|
494 | data = data.replace(cert_log, b'') |
---|
495 | out_stream.send(data) |
---|
496 | if event & select.POLLHUP or event & select.POLLRDHUP: |
---|
497 | # Stop the loop, but process any other events that |
---|
498 | # might be in the list returned by poll() first. |
---|
499 | run_loop = False |
---|
500 | |
---|
501 | in_stream.close() |
---|
502 | out_stream.close() |
---|
503 | |
---|
504 | |
---|
505 | |
---|
506 | def format_response(resp, body): |
---|
507 | """Format an http.client.HTTPResponse for logging.""" |
---|
508 | s = f'{resp.status} {resp.reason}\n' |
---|
509 | s = s + '\n'.join(f'{name}: {value}' for name, value in resp.getheaders()) |
---|
510 | s = s + '\n\n' + body |
---|
511 | return s |
---|
512 | |
---|
513 | |
---|
514 | |
---|
515 | def subst_env(text): |
---|
516 | """Use the parameter "text" as a template, substitute with environment |
---|
517 | variables. |
---|
518 | |
---|
519 | >>> os.environ['EXAMPLE_VAR'] = 'abc' |
---|
520 | >>> subst_env('${EXAMPLE_VAR}def') |
---|
521 | 'abcdef' |
---|
522 | |
---|
523 | Referencing undefined environment variables causes a KeyError. |
---|
524 | |
---|
525 | >>> subst_env('${EXAMPLE_UNSET}') |
---|
526 | Traceback (most recent call last): |
---|
527 | ... |
---|
528 | KeyError: 'EXAMPLE_UNSET' |
---|
529 | |
---|
530 | >>> subst_env(None) is None |
---|
531 | True |
---|
532 | |
---|
533 | """ |
---|
534 | if not text: |
---|
535 | return None |
---|
536 | t = Template(text) |
---|
537 | return t.substitute(os.environ) |
---|
538 | |
---|
539 | |
---|
540 | |
---|
541 | def run_test_conf(test_config, timeout=5.0, conn_log=None, response_log=None): |
---|
542 | """Load and run a test configuration. |
---|
543 | |
---|
544 | The test_conf parameter must be a YAML file, defining one or more |
---|
545 | TestConnections, to be run in order. The other three parameters |
---|
546 | are forwarded to TestConnection.run(). |
---|
547 | |
---|
548 | """ |
---|
549 | conns = None |
---|
550 | |
---|
551 | config = yaml.load(test_config, Loader=yaml.Loader) |
---|
552 | if type(config) is TestConnection: |
---|
553 | conns = [config] |
---|
554 | elif type(config) is list: |
---|
555 | # assume list elements are connections |
---|
556 | conns = config |
---|
557 | else: |
---|
558 | raise TypeError(f'Unsupported configuration: {config!r}') |
---|
559 | print(conns) |
---|
560 | sys.stdout.flush() |
---|
561 | |
---|
562 | for i, test_conn in enumerate(conns): |
---|
563 | if test_conn.description: |
---|
564 | print(f'Running test connection {i}: {test_conn.description}') |
---|
565 | else: |
---|
566 | print(f'Running test connection {i}.') |
---|
567 | sys.stdout.flush() |
---|
568 | test_conn.run(timeout=timeout, conn_log=conn_log, |
---|
569 | response_log=response_log) |
---|