Differential D2772 Diff 9864 swh/core/pytest_plugin.py

Changeset 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 20 Lines • Show All 89 Lines • ▼ Show 20 Lines	definition in the local conftest, for example::
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,
datadir=datadir)
requests_mock.get(re.compile('https?://'), body=cb)		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
▲ Show 20 Lines • Show All 64 Lines • ▼ Show 20 Lines	def build_response(self, req, resp):

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

Show All 22 Lines