Differential D4235 Diff 14953 swh/fuse/tests/conftest.py

Changeset View

Standalone View

swh/fuse/tests/conftest.py

	# Copyright (C) 2020 The Software Heritage developers			# Copyright (C) 2020 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 json
	from multiprocessing import Process			from multiprocessing import Process
	from os import listdir			from os import listdir
	from pathlib import Path			from pathlib import Path
	import subprocess			import subprocess
	from tempfile import NamedTemporaryFile, TemporaryDirectory			from tempfile import NamedTemporaryFile, TemporaryDirectory
	import time			import time

	from click.testing import CliRunner			from click.testing import CliRunner
	import pytest			import pytest
	import yaml			import yaml

	from swh.fuse import cli			from swh.fuse import cli
				from swh.fuse.tests.data.api_data import API_URL, MOCK_ARCHIVE
	from .api_data import API_URL, MOCK_ARCHIVE, ROOTDIR_SWHID, ROOTREV_SWHID			from swh.fuse.tests.data.config import MOUNTPOINTS


	@pytest.fixture			@pytest.fixture
	def web_api_mock(requests_mock):			def web_api_mock(requests_mock):
	for api_call, data in MOCK_ARCHIVE.items():			for api_call, data in MOCK_ARCHIVE.items():
				# Convert Python dict JSON into a string (only for non-raw API call)
				if not api_call.endswith("raw/"):
				data = json.dumps(data)
	requests_mock.get(f"{API_URL}/{api_call}", text=data)			requests_mock.get(f"{API_URL}/{api_call}", text=data)
	return requests_mock			return requests_mock


	@pytest.fixture			@pytest.fixture
	def fuse_mntdir(web_api_mock):			def fuse_mntdir(web_api_mock):
	tmpdir = TemporaryDirectory(suffix=".swh-fuse-test")			tmpdir = TemporaryDirectory(suffix=".swh-fuse-test")
	tmpfile = NamedTemporaryFile(suffix=".swh-fuse-test.yml")			tmpfile = NamedTemporaryFile(suffix=".swh-fuse-test.yml")

	config = {			config = {
	"cache": {"metadata": {"in-memory": True}, "blob": {"in-memory": True}},			"cache": {"metadata": {"in-memory": True}, "blob": {"in-memory": True}},
	"web-api": {"url": API_URL, "auth-token": None},			"web-api": {"url": API_URL, "auth-token": None},
	}			}

	# Run FUSE in foreground mode but in a separate process, so it does not			# Run FUSE in foreground mode but in a separate process, so it does not
	# block execution and remains easy to kill during teardown			# block execution and remains easy to kill during teardown
	def fuse_process(tmpdir, tmpfile):			def fuse_process(tmpdir, tmpfile):
	with tmpdir as mntdir, tmpfile as config_path:			with tmpdir as mntdir, tmpfile as config_path:
	config_path = Path(config_path.name)			config_path = Path(config_path.name)
	config_path.write_text(yaml.dump(config))			config_path.write_text(yaml.dump(config))
	CliRunner().invoke(			CliRunner().invoke(
	cli.mount,			cli.mount,
	args=[			args=[
	mntdir,			mntdir,
	ROOTDIR_SWHID,			*MOUNTPOINTS,
	ROOTREV_SWHID,
	"--foreground",			"--foreground",
	"--config-file",			"--config-file",
	config_path,			config_path,
	],			],
	)			)

	fuse = Process(target=fuse_process, args=[tmpdir, tmpfile])			fuse = Process(target=fuse_process, args=[tmpdir, tmpfile])
	fuse.start()			fuse.start()
	# Wait max 3 seconds for the FUSE to correctly mount			# Wait max 3 seconds for the FUSE to correctly mount
	for i in range(30):			for i in range(30):
	try:			try:
	root = listdir(tmpdir.name)			root = listdir(tmpdir.name)
	if root:			if root:
	break			break
	except FileNotFoundError:			except FileNotFoundError:
	pass			pass
	time.sleep(0.1)			time.sleep(0.1)
	else:			else:
	raise FileNotFoundError(f"Could not mount FUSE in {tmpdir.name}")			raise FileNotFoundError(f"Could not mount FUSE in {tmpdir.name}")

	yield tmpdir.name			yield Path(tmpdir.name)
				seirlUnsubmitted Done Inline Actions Why don't you yield a Path() here? that would allow you to use the slash notation everywhere else, instead of reconstructing paths all the time. seirl: Why don't you yield a Path() here? that would allow you to use the slash notation everywhere…

	subprocess.run(["fusermount", "-u", tmpdir.name], check=True)			subprocess.run(["fusermount", "-u", tmpdir.name], check=True)
	fuse.join()			fuse.join()