source: mod_gnutls/test/runtest.py @ 05984a0

proxy-ticket
Last change on this file since 05984a0 was 05984a0, checked in by Fiona Klute <fiona.klute@…>, 7 months ago

Replace "runtests" with "runtest.py"

This is the next step from handling HTTP requests and responses in
Python. In particular error handling is a lot easier to do in Python
than using Bash trap functions.

  • Property mode set to 100755
File size: 7.7 KB
Line 
1#!/usr/bin/python3
2# PYTHON_ARGCOMPLETE_OK
3
4# Copyright 2019 Fiona Klute
5#
6# Licensed under the Apache License, Version 2.0 (the "License");
7# you may not use this file except in compliance with the License.
8# You may obtain a copy of the License at
9#
10#     http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17
18import contextlib
19import os
20import os.path
21import subprocess
22import sys
23
24from mgstest import lockfile, TestExpectationFailed
25from mgstest.services import ApacheService, TestService
26from mgstest.tests import run_test_conf
27
28
29
30def find_testdir(number, dir):
31    """Find the configuration directory for a test based on its
32    number. The given directory must contain exactly one directory
33    with a name matching "NUMBER_*", otherwise a LookupError is
34    raised.
35
36    """
37    with os.scandir(dir) as it:
38        found = None
39        for entry in it:
40            if entry.is_dir():
41                num = int(entry.name.split('_', maxsplit=1)[0])
42                if number == num:
43                    if found:
44                        # duplicate numbers are an error
45                        raise LookupError('Multiple directories found for '
46                                          f'test number {args.test_number}: '
47                                          f'{found.name} and {entry.name}')
48                    else:
49                        found = entry
50        if found == None:
51            raise LookupError('No test directory found for test number '
52                              f'{args.test_number}!')
53        else:
54            return (found.path, found.name)
55
56
57
58def check_ocsp_responder():
59    # Check if OCSP responder works
60    issuer_cert = 'authority/x509.pem'
61    check_cert = 'authority/server/x509.pem'
62    command = ['ocsptool', '--ask', '--nonce',
63               '--load-issuer', issuer_cert,
64               '--load-cert', check_cert]
65    return subprocess.run(command).returncode == 0
66
67def check_msva():
68    # Check if MSVA is up
69    cert_file = 'authority/client/x509.pem'
70    uid_file = 'authority/client/uid'
71    with open(uid_file, 'r') as file:
72        uid = file.read().strip()
73        command = ['msva-query-agent', 'https', uid, 'x509pem', 'client']
74        with open(cert_file, 'r') as cert:
75            return subprocess.run(command, stdin=cert).returncode == 0
76
77
78
79if __name__ == "__main__":
80    import argparse
81    parser = argparse.ArgumentParser(
82        description='Run a mod_gnutls server test')
83    parser.add_argument('--test-number', type=int,
84                        required=True, help='load YAML test configuration')
85    parser.add_argument('--log-connection', type=str, default=None,
86                        help='write connection log to this file')
87    parser.add_argument('--log-responses', type=str, default=None,
88                        help='write HTTP responses to this file')
89
90    # enable bash completion if argcomplete is available
91    try:
92        import argcomplete
93        argcomplete.autocomplete(parser)
94    except ImportError:
95        pass
96
97    args = parser.parse_args()
98
99    # The Automake environment always provides srcdir, the default is
100    # for manual use.
101    srcdir = os.path.realpath(os.environ.get('srcdir', '.'))
102    # ensure environment srcdir is absolute
103    os.environ['srcdir'] = srcdir
104
105    # Find the configuration directory for the test in
106    # ${srcdir}/tests/, based on the test number.
107    testdir, testname = find_testdir(args.test_number,
108                                     os.path.join(srcdir, 'tests'))
109    print(f'Found test {testname}, test dir is {testdir}')
110    os.environ['TEST_NAME'] = testname
111
112    # PID file name varies depending on whether we're using
113    # namespaces.
114    #
115    # TODO: Check if having the different names is really necessary.
116    pidaffix = ''
117    if 'USE_TEST_NAMESPACE' in os.environ:
118        pidaffix = f'-{testname}'
119
120    # Define the available services
121    apache = ApacheService(config=os.path.join(testdir, 'apache.conf'),
122                           pidfile=f'apache2{pidaffix}.pid')
123    backend = ApacheService(config=os.path.join(testdir, 'backend.conf'),
124                            pidfile=f'backend{pidaffix}.pid')
125    ocsp = ApacheService(config=os.path.join(testdir, 'ocsp.conf'),
126                         pidfile=f'ocsp{pidaffix}.pid',
127                         check=check_ocsp_responder)
128    msva = TestService(start=['monkeysphere-validation-agent'],
129                       env={'GNUPGHOME': 'msva.gnupghome',
130                            'MSVA_KEYSERVER_POLICY': 'never'},
131                       condition=lambda: 'USE_MSVA' in os.environ,
132                       check=check_msva)
133
134    # background services: must be ready before the main apache
135    # instance is started
136    bg_services = [backend, ocsp, msva]
137
138    # TODO: check extra requirements (e.g. specific modules)
139
140    # TODO: add hook to modify environment (unless made obsolete by
141    # parameters)
142
143    # If VERBOSE is enabled, log the HTTPD build configuration
144    if 'VERBOSE' in os.environ:
145        apache2 = os.environ.get('APACHE2', 'apache2')
146        subprocess.run([apache2, '-f', f'{srcdir}/base_apache.conf', '-V'],
147                       check=True)
148
149    if 'USE_MSVA' in os.environ:
150        os.environ['MONKEYSPHERE_VALIDATION_AGENT_SOCKET'] = \
151            f'http://127.0.0.1:{os.environ["MSVA_PORT"]}'
152
153    with contextlib.ExitStack() as service_stack:
154        service_stack.enter_context(lockfile('test.lock', nolock='MGS_NETNS_ACTIVE' in os.environ))
155        service_stack.enter_context(ocsp.run())
156        service_stack.enter_context(backend.run())
157        service_stack.enter_context(msva.run())
158
159        # TEST_SERVICE_MAX_WAIT is in milliseconds
160        wait_timeout = \
161            int(os.environ.get('TEST_SERVICE_MAX_WAIT', 10000)) / 1000
162        for s in bg_services:
163            if s.condition():
164                s.wait_ready(timeout=wait_timeout)
165
166        # special case: expected to fail in a few cases
167        try:
168            service_stack.enter_context(apache.run())
169            if os.path.exists(os.path.join(testdir, 'fail.server')):
170                raise TestExpectationFailed(
171                    'Server start did not fail as expected!')
172            apache.wait_ready()
173        except subprocess.CalledProcessError as e:
174            if os.path.exists(os.path.join(testdir, 'fail.server')):
175                print('Apache server failed to start as expected',
176                      file=sys.stderr)
177            else:
178                raise e
179
180        # Set TEST_TARGET for the request. Might be replaced with a
181        # parameter later.
182        if 'TARGET_IP' in os.environ:
183            os.environ['TEST_TARGET'] = os.environ['TARGET_IP']
184        else:
185            os.environ['TEST_TARGET'] = os.environ['TEST_HOST']
186
187        # Run the test connections
188        with contextlib.ExitStack() as stack:
189            test_conf = stack.enter_context(
190                open(os.path.join(testdir, 'test.yml'), 'r'))
191            log_file = None
192            output_file = None
193            if args.log_connection:
194                log_file = stack.enter_context(open(args.log_connection, 'w'))
195            if args.log_responses:
196                output_file = stack.enter_context(open(args.log_responses, 'w'))
197
198            run_test_conf(test_conf,
199                          float(os.environ.get('TEST_QUERY_TIMEOUT', 5.0)),
200                          conn_log=log_file, response_log=output_file)
201
202        # TODO: add hook to replace the test request, e.g. for curl
203
204        # TODO: add hook for extra checks
Note: See TracBrowser for help on using the repository browser.