Changeset View
Changeset View
Standalone View
Standalone View
swh/core/pytest_plugin.py
| Show All 21 Lines | |||||
| # Check get_local_factory function | # Check get_local_factory function | ||||
| # Maximum number of iteration checks to generate requests responses | # Maximum number of iteration checks to generate requests responses | ||||
| MAX_VISIT_FILES = 10 | MAX_VISIT_FILES = 10 | ||||
| def get_response_cb( | def get_response_cb( | ||||
| request: requests.Request, context, datadir, | request: requests.Request, | ||||
| context, | |||||
| datadir, | |||||
| ignore_urls: List[str] = [], | ignore_urls: List[str] = [], | ||||
| visits: Optional[Dict] = None): | visits: Optional[Dict] = None, | ||||
| ): | |||||
| """Mount point callback to fetch on disk the request's content. The request | """Mount point callback to fetch on disk the request's content. The request | ||||
| urls provided are url decoded first to resolve the associated file on disk. | urls provided are url decoded first to resolve the associated file on disk. | ||||
| This is meant to be used as 'body' argument of the requests_mock.get() | This is meant to be used as 'body' argument of the requests_mock.get() | ||||
| method. | method. | ||||
| It will look for files on the local filesystem based on the requested URL, | It will look for files on the local filesystem based on the requested URL, | ||||
| using the following rules: | using the following rules: | ||||
| Show All 34 Lines | Args: | ||||
| file exists | file exists | ||||
| visits: Dict of url, number of visits. If None, disable multi visit | visits: Dict of url, number of visits. If None, disable multi visit | ||||
| support (default) | support (default) | ||||
| Returns: | Returns: | ||||
| Optional[FileDescriptor] on disk file to read from the test context | Optional[FileDescriptor] on disk file to read from the test context | ||||
| """ | """ | ||||
| logger.debug('get_response_cb(%s, %s)', request, context) | logger.debug("get_response_cb(%s, %s)", request, context) | ||||
| logger.debug('url: %s', request.url) | logger.debug("url: %s", request.url) | ||||
| logger.debug('ignore_urls: %s', ignore_urls) | logger.debug("ignore_urls: %s", ignore_urls) | ||||
| unquoted_url = unquote(request.url) | unquoted_url = unquote(request.url) | ||||
| if unquoted_url in ignore_urls: | if unquoted_url in ignore_urls: | ||||
| context.status_code = 404 | context.status_code = 404 | ||||
| return None | return None | ||||
| url = urlparse(unquoted_url) | url = urlparse(unquoted_url) | ||||
| # http://pypi.org ~> http_pypi.org | # http://pypi.org ~> http_pypi.org | ||||
| # https://files.pythonhosted.org ~> https_files.pythonhosted.org | # https://files.pythonhosted.org ~> https_files.pythonhosted.org | ||||
| dirname = '%s_%s' % (url.scheme, url.hostname) | dirname = "%s_%s" % (url.scheme, url.hostname) | ||||
| # url.path: pypi/<project>/json -> local file: pypi_<project>_json | # url.path: pypi/<project>/json -> local file: pypi_<project>_json | ||||
| filename = url.path[1:] | filename = url.path[1:] | ||||
| if filename.endswith('/'): | if filename.endswith("/"): | ||||
| filename = filename[:-1] | filename = filename[:-1] | ||||
| filename = filename.replace('/', '_') | filename = filename.replace("/", "_") | ||||
| if url.query: | if url.query: | ||||
| filename += ',' + url.query.replace('&', ',') | filename += "," + url.query.replace("&", ",") | ||||
| filepath = path.join(datadir, dirname, filename) | filepath = path.join(datadir, dirname, filename) | ||||
| if visits is not None: | if visits is not None: | ||||
| visit = visits.get(url, 0) | visit = visits.get(url, 0) | ||||
| visits[url] = visit + 1 | visits[url] = visit + 1 | ||||
| if visit: | if visit: | ||||
| filepath = filepath + '_visit%s' % visit | filepath = filepath + "_visit%s" % visit | ||||
| if not path.isfile(filepath): | if not path.isfile(filepath): | ||||
| logger.debug('not found filepath: %s', filepath) | logger.debug("not found filepath: %s", filepath) | ||||
| context.status_code = 404 | context.status_code = 404 | ||||
| return None | return None | ||||
| fd = open(filepath, 'rb') | fd = open(filepath, "rb") | ||||
| context.headers['content-length'] = str(path.getsize(filepath)) | context.headers["content-length"] = str(path.getsize(filepath)) | ||||
| return fd | return fd | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def datadir(request): | def datadir(request): | ||||
| """By default, returns the test directory's data directory. | """By default, returns the test directory's data directory. | ||||
| This can be overridden on a per file tree basis. Add an override | This can be overridden on a per file tree basis. Add an override | ||||
| definition in the local conftest, for example:: | definition in the local conftest, for example:: | ||||
| import pytest | import pytest | ||||
| from os import path | from os import path | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def datadir(): | def datadir(): | ||||
| return path.join(path.abspath(path.dirname(__file__)), 'resources') | return path.join(path.abspath(path.dirname(__file__)), 'resources') | ||||
| """ | """ | ||||
| return path.join(path.dirname(str(request.fspath)), 'data') | return path.join(path.dirname(str(request.fspath)), "data") | ||||
| def requests_mock_datadir_factory(ignore_urls: List[str] = [], | def requests_mock_datadir_factory( | ||||
| has_multi_visit: bool = False): | ignore_urls: List[str] = [], has_multi_visit: bool = False | ||||
| ): | |||||
| """This factory generates fixture which allow to look for files on the | """This factory generates fixture which allow to look for files on the | ||||
| local filesystem based on the requested URL, using the following rules: | local filesystem based on the requested URL, using the following rules: | ||||
| - files are searched in the datadir/<hostname> directory | - files are searched in the datadir/<hostname> directory | ||||
| - the local file name is the path part of the URL with path hierarchy | - the local file name is the path part of the URL with path hierarchy | ||||
| markers (aka '/') replaced by '_' | markers (aka '/') replaced by '_' | ||||
| Show All 14 Lines | - requests_mock_datadir_factory(ignore_urls=['url1', 'url2']): | ||||
| returning 404. | returning 404. | ||||
| Args: | Args: | ||||
| ignore_urls: List of urls to always returns 404 (whether file | ignore_urls: List of urls to always returns 404 (whether file | ||||
| exists or not) | exists or not) | ||||
| has_multi_visit: Activate or not the multiple visits behavior | has_multi_visit: Activate or not the multiple visits behavior | ||||
| """ | """ | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def requests_mock_datadir(requests_mock, datadir): | def requests_mock_datadir(requests_mock, datadir): | ||||
| if not has_multi_visit: | if not has_multi_visit: | ||||
| cb = partial(get_response_cb, | cb = partial(get_response_cb, ignore_urls=ignore_urls, datadir=datadir) | ||||
| ignore_urls=ignore_urls, | requests_mock.get(re.compile("https?://"), body=cb) | ||||
| datadir=datadir) | |||||
| requests_mock.get(re.compile('https?://'), body=cb) | |||||
| else: | else: | ||||
| visits = {} | visits = {} | ||||
| requests_mock.get(re.compile('https?://'), body=partial( | requests_mock.get( | ||||
| get_response_cb, ignore_urls=ignore_urls, visits=visits, | re.compile("https?://"), | ||||
| datadir=datadir) | body=partial( | ||||
| get_response_cb, | |||||
| ignore_urls=ignore_urls, | |||||
| visits=visits, | |||||
| datadir=datadir, | |||||
| ), | |||||
| ) | ) | ||||
| return requests_mock | return requests_mock | ||||
| return requests_mock_datadir | return requests_mock_datadir | ||||
| # Default `requests_mock_datadir` implementation | # Default `requests_mock_datadir` implementation | ||||
| requests_mock_datadir = requests_mock_datadir_factory([]) | requests_mock_datadir = requests_mock_datadir_factory([]) | ||||
| # Implementation for multiple visits behavior: | # Implementation for multiple visits behavior: | ||||
| # - first time, it checks for a file named `filename` | # - first time, it checks for a file named `filename` | ||||
| # - second time, it checks for a file named `filename`_visit1 | # - second time, it checks for a file named `filename`_visit1 | ||||
| # etc... | # etc... | ||||
| requests_mock_datadir_visits = requests_mock_datadir_factory( | requests_mock_datadir_visits = requests_mock_datadir_factory(has_multi_visit=True) | ||||
| has_multi_visit=True) | |||||
| @pytest.fixture | @pytest.fixture | ||||
| def swh_rpc_client(swh_rpc_client_class, swh_rpc_adapter): | def swh_rpc_client(swh_rpc_client_class, swh_rpc_adapter): | ||||
| """This fixture generates an RPCClient instance that uses the class generated | """This fixture generates an RPCClient instance that uses the class generated | ||||
| by the rpc_client_class fixture as backend. | by the rpc_client_class fixture as backend. | ||||
| Since it uses the swh_rpc_adapter, HTTP queries will be intercepted and | Since it uses the swh_rpc_adapter, HTTP queries will be intercepted and | ||||
| routed directly to the current Flask app (as provided by the `app` | routed directly to the current Flask app (as provided by the `app` | ||||
| fixture). | fixture). | ||||
| So this stack of fixtures allows to test the RPCClient -> RPCServerApp | So this stack of fixtures allows to test the RPCClient -> RPCServerApp | ||||
| communication path using a real RPCClient instance and a real Flask | communication path using a real RPCClient instance and a real Flask | ||||
| (RPCServerApp) app instance. | (RPCServerApp) app instance. | ||||
| To use this fixture: | To use this fixture: | ||||
| - ensure an `app` fixture exists and generate a Flask application, | - ensure an `app` fixture exists and generate a Flask application, | ||||
| - implement an `swh_rpc_client_class` fixtures that returns the | - implement an `swh_rpc_client_class` fixtures that returns the | ||||
| RPCClient-based class to use as client side for the tests, | RPCClient-based class to use as client side for the tests, | ||||
| - implement your tests using this `swh_rpc_client` fixture. | - implement your tests using this `swh_rpc_client` fixture. | ||||
| See swh/core/api/tests/test_rpc_client_server.py for an example of usage. | See swh/core/api/tests/test_rpc_client_server.py for an example of usage. | ||||
| """ | """ | ||||
| url = 'mock://example.com' | url = "mock://example.com" | ||||
| cli = swh_rpc_client_class(url=url) | cli = swh_rpc_client_class(url=url) | ||||
| # we need to clear the list of existing adapters here so we ensure we | # we need to clear the list of existing adapters here so we ensure we | ||||
| # have one and only one adapter which is then used for all the requests. | # have one and only one adapter which is then used for all the requests. | ||||
| cli.session.adapters.clear() | cli.session.adapters.clear() | ||||
| cli.session.mount('mock://', swh_rpc_adapter) | cli.session.mount("mock://", swh_rpc_adapter) | ||||
| return cli | return cli | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def swh_rpc_adapter(app): | def swh_rpc_adapter(app): | ||||
| """Fixture that generates a requests.Adapter instance that | """Fixture that generates a requests.Adapter instance that | ||||
| can be used to test client/servers code based on swh.core.api classes. | can be used to test client/servers code based on swh.core.api classes. | ||||
| Show All 10 Lines | class RPCTestAdapter(BaseAdapter): | ||||
| def build_response(self, req, resp): | def build_response(self, req, resp): | ||||
| response = requests.Response() | response = requests.Response() | ||||
| # Fallback to None if there's no status_code, for whatever reason. | # Fallback to None if there's no status_code, for whatever reason. | ||||
| response.status_code = resp.status_code | response.status_code = resp.status_code | ||||
| # Make headers case-insensitive. | # Make headers case-insensitive. | ||||
| response.headers = CaseInsensitiveDict(getattr(resp, 'headers', {})) | response.headers = CaseInsensitiveDict(getattr(resp, "headers", {})) | ||||
| # Set encoding. | # Set encoding. | ||||
| response.encoding = get_encoding_from_headers(response.headers) | response.encoding = get_encoding_from_headers(response.headers) | ||||
| response.raw = resp | response.raw = resp | ||||
| response.reason = response.raw.status | response.reason = response.raw.status | ||||
| if isinstance(req.url, bytes): | if isinstance(req.url, bytes): | ||||
| response.url = req.url.decode('utf-8') | response.url = req.url.decode("utf-8") | ||||
| else: | else: | ||||
| response.url = req.url | response.url = req.url | ||||
| # Give the Response some context. | # Give the Response some context. | ||||
| response.request = req | response.request = req | ||||
| response.connection = self | response.connection = self | ||||
| response._content = resp.data | response._content = resp.data | ||||
| return response | return response | ||||
| def send(self, request, **kw): | def send(self, request, **kw): | ||||
| """ | """ | ||||
| Overrides ``requests.adapters.BaseAdapter.send`` | Overrides ``requests.adapters.BaseAdapter.send`` | ||||
| """ | """ | ||||
| resp = self._client.open( | resp = self._client.open( | ||||
| request.url, method=request.method, | request.url, | ||||
| method=request.method, | |||||
| headers=request.headers.items(), | headers=request.headers.items(), | ||||
| data=request.body, | data=request.body, | ||||
| ) | ) | ||||
| return self.build_response(request, resp) | return self.build_response(request, resp) | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def flask_app_client(app): | def flask_app_client(app): | ||||
| with app.test_client() as client: | with app.test_client() as client: | ||||
| yield client | yield client | ||||
| # stolen from pytest-flask, required to have url_for() working within tests | # stolen from pytest-flask, required to have url_for() working within tests | ||||
| # using flask_app_client fixture. | # using flask_app_client fixture. | ||||
| @pytest.fixture(autouse=True) | @pytest.fixture(autouse=True) | ||||
| def _push_request_context(request): | def _push_request_context(request): | ||||
| """During tests execution request context has been pushed, e.g. `url_for`, | """During tests execution request context has been pushed, e.g. `url_for`, | ||||
| `session`, etc. can be used in tests as is:: | `session`, etc. can be used in tests as is:: | ||||
| def test_app(app, client): | def test_app(app, client): | ||||
| assert client.get(url_for('myview')).status_code == 200 | assert client.get(url_for('myview')).status_code == 200 | ||||
| """ | """ | ||||
| if 'app' not in request.fixturenames: | if "app" not in request.fixturenames: | ||||
| return | return | ||||
| app = request.getfixturevalue('app') | app = request.getfixturevalue("app") | ||||
| ctx = app.test_request_context() | ctx = app.test_request_context() | ||||
| ctx.push() | ctx.push() | ||||
| def teardown(): | def teardown(): | ||||
| ctx.pop() | ctx.pop() | ||||
| request.addfinalizer(teardown) | request.addfinalizer(teardown) | ||||