From 21f919b14a9aeaa08aecbcdd41a530198315aad1 Mon Sep 17 00:00:00 2001 From: Michał Górny Date: Sun, 30 Aug 2020 16:24:10 +0200 Subject: tests: Use module-scope fixtures whenever possible MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Michał Górny --- tests/testutil.py | 58 ++++++++++++++++++++++++++++--------------------------- 1 file changed, 30 insertions(+), 28 deletions(-) (limited to 'tests/testutil.py') diff --git a/tests/testutil.py b/tests/testutil.py index c00ef75..092a4d2 100644 --- a/tests/testutil.py +++ b/tests/testutil.py @@ -3,7 +3,6 @@ # (c) 2017-2020 Michał Górny # Licensed under the terms of 2-clause BSD license -import collections import errno import functools import io @@ -88,31 +87,34 @@ class HKPServerRequestHandler(BaseHTTPRequestHandler): self.wfile.flush() -@pytest.fixture -def hkp_server(): - keys = {} - # try 10 randomly selected ports before giving up - for port in random.sample(range(1024, 32768), 10): - try: - server = HTTPServer( - ('127.0.0.1', port), - functools.partial(HKPServerRequestHandler, keys)) - except OSError as e: - if e.errno != errno.EADDRINUSE: - raise unittest.SkipTest('Unable to bind the HKP server: {}' - .format(e)) +class HKPServer: + def __init__(self): + self.keys = {} + self.addr = None + + def start(self): + # try 10 randomly selected ports before giving up + for port in random.sample(range(1024, 32768), 10): + try: + self.server = HTTPServer( + ('127.0.0.1', port), + functools.partial(HKPServerRequestHandler, self.keys)) + except OSError as e: + if e.errno != errno.EADDRINUSE: + pytest.skip( + f'Unable to bind the HKP server: {e}') + else: + break else: - break - else: - pytest.skip('Unable to find a free port for HKP server') - - server_addr = f'hkp://127.0.0.1:{port}' - server_thread = threading.Thread(target=server.serve_forever) - server_thread.start() - - yield collections.namedtuple('HKPServerTuple', ('addr', 'keys'))( - server_addr, keys) - - server.shutdown() - server.server_close() - server_thread.join() + pytest.skip('Unable to find a free port for HKP server') + + self.addr = f'hkp://127.0.0.1:{port}' + self.thread = threading.Thread(target=self.server.serve_forever) + self.thread.start() + + def stop(self): + assert self.addr is not None + self.server.shutdown() + self.server.server_close() + self.thread.join() + self.addr = None -- cgit v1.2.3