diff --git a/swh/fuse/tests/conftest.py b/swh/fuse/tests/conftest.py index c565a6a..e5c2446 100644 --- a/swh/fuse/tests/conftest.py +++ b/swh/fuse/tests/conftest.py @@ -1,77 +1,76 @@ # Copyright (C) 2020 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import json from multiprocessing import Process import os from pathlib import Path import subprocess from tempfile import NamedTemporaryFile, TemporaryDirectory import time from click.testing import CliRunner import pytest import yaml import swh.fuse.cli as cli from swh.fuse.tests.data.api_data import API_URL, MOCK_ARCHIVE @pytest.fixture def web_api_mock(requests_mock): 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/") and not api_call.startswith("graph/"): data = json.dumps(data) requests_mock.get(f"{API_URL}/{api_call}", text=data) return requests_mock @pytest.fixture def fuse_mntdir(web_api_mock): tmpdir = TemporaryDirectory(suffix=".swh-fuse-test") - tmpfile = NamedTemporaryFile(suffix=".swh-fuse-test.yml") config = { "cache": {"metadata": {"in-memory": True}, "blob": {"in-memory": True},}, "web-api": {"url": API_URL, "auth-token": None}, "json-indent": None, } # Run FUSE in foreground mode but in a separate process, so it does not # block execution and remains easy to kill during teardown - def fuse_process(tmpdir, tmpfile): - with tmpdir as mntdir, tmpfile as config_path: - config_path = Path(config_path.name) + def fuse_process(mntdir: Path): + with NamedTemporaryFile(suffix=".swh-fuse-test.yml") as tmpfile: + config_path = Path(tmpfile.name) config_path.write_text(yaml.dump(config)) CliRunner().invoke( cli.fuse, args=[ "--config-file", str(config_path), "mount", - mntdir, + str(mntdir), "--foreground", ], ) - fuse = Process(target=fuse_process, args=[tmpdir, tmpfile]) + fuse = Process(target=fuse_process, args=[Path(tmpdir.name)]) fuse.start() # Wait max 3 seconds for the FUSE to correctly mount for i in range(30): try: root = os.listdir(tmpdir.name) if root: break except FileNotFoundError: pass time.sleep(0.1) else: raise FileNotFoundError(f"Could not mount FUSE in {tmpdir.name}") yield Path(tmpdir.name) subprocess.run(["fusermount", "-u", tmpdir.name], check=True) fuse.join()