Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/tests/server_testing.py
| # Copyright (C) 2015-2018 The Software Heritage developers | # Copyright (C) 2015-2018 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import abc | import abc | ||||
| import multiprocessing | import multiprocessing | ||||
| import os | import os | ||||
| import socket | |||||
| import time | import time | ||||
| from urllib.request import urlopen | from urllib.request import urlopen | ||||
| import aiohttp | import aiohttp | ||||
| import aiohttp.test_utils | |||||
| class ServerTestFixtureBaseClass(metaclass=abc.ABCMeta): | class ServerTestFixtureBaseClass(metaclass=abc.ABCMeta): | ||||
| """Base class for http client/server testing implementations. | """Base class for http client/server testing implementations. | ||||
| Override this class to implement the following methods: | Override this class to implement the following methods: | ||||
| - process_config: to do something needed for the server | - process_config: to do something needed for the server | ||||
| configuration (e.g propagate the configuration to other part) | configuration (e.g propagate the configuration to other part) | ||||
| Show All 36 Lines | def define_worker_function(self, app, port): | ||||
| pass | pass | ||||
| def start_server(self): | def start_server(self): | ||||
| """ Spawn the API server using multiprocessing. | """ Spawn the API server using multiprocessing. | ||||
| """ | """ | ||||
| self.process = None | self.process = None | ||||
| self.process_config() | self.process_config() | ||||
| self.port = aiohttp.test_utils.unused_port() | |||||
| # Get an available port number | |||||
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |||||
| sock.bind(('127.0.0.1', 0)) | |||||
| self.port = sock.getsockname()[1] | |||||
| sock.close() | |||||
| worker_fn = self.define_worker_function() | worker_fn = self.define_worker_function() | ||||
| self.process = multiprocessing.Process( | self.process = multiprocessing.Process( | ||||
| target=worker_fn, args=(self.app, self.port) | target=worker_fn, args=(self.app, self.port) | ||||
| ) | ) | ||||
| self.process.start() | self.process.start() | ||||
| ▲ Show 20 Lines • Show All 70 Lines • Show Last 20 Lines | |||||