conftest.py (476 lines of code) (raw):

# Copyright 2022-2025 The Ramble Authors # # Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or # https://www.apache.org/licenses/LICENSE-2.0> or the MIT license # <LICENSE-MIT or https://opensource.org/licenses/MIT>, at your # option. This file may not be copied, modified, or distributed # except according to those terms. import builtins import collections import io import os import os.path import pathlib import shutil import py import pytest from llnl.util.filesystem import remove_linked_tree import ramble.config import ramble.paths import ramble.repository import ramble.stage from ramble.fetch_strategy import FetchError, FetchStrategyComposite, URLFetchStrategy from ramble.util.file_util import is_dry_run_path import spack.platforms import spack.util.executable import spack.util.spack_yaml as syaml def _can_access(path, perms): return False # Hooks to add command line options or set other custom behaviors. # They must be placed here to be found by pytest. See: # # https://docs.pytest.org/en/latest/writing_plugins.html # def pytest_addoption(parser): group = parser.getgroup("Ramble specific command line options") group.addoption( "--fast", action="store_true", default=False, help='runs only "fast" unit tests, instead of the whole suite', ) group.addoption( "--repo-path", default=None, help="runs only tests under the given Ramble object repo path", ) def pytest_configure(config): repo_path = config.getoption("--repo-path") if not repo_path: return path = pathlib.Path(repo_path) # Define testpaths config.args = [p for p in path.rglob("test") if p.is_dir()] def pytest_collection_modifyitems(config, items): if not config.getoption("--fast"): # --fast not given, run all the tests return slow_tests = ["db", "network", "maybeslow", "long"] skip_as_slow = pytest.mark.skip(reason="skipped slow test [--fast command line option given]") for item in items: if any(x in item.keywords for x in slow_tests): item.add_marker(skip_as_slow) # # These fixtures are applied to all tests # @pytest.fixture(scope="function", autouse=True) def no_chdir(): """Ensure that no test changes Ramble's working directory. This prevents Ramble tests (and therefore Ramble commands) from changing the working directory and causing other tests to fail mysteriously. Tests should use ``working_dir`` or ``py.path``'s ``.as_cwd()`` instead of ``os.chdir`` to avoid failing this check. We assert that the working directory hasn't changed, unless the original wd somehow ceased to exist. """ original_wd = os.getcwd() yield if os.path.isdir(original_wd): assert os.getcwd() == original_wd def remove_whatever_it_is(path): """Type-agnostic remove.""" if os.path.isfile(path): os.remove(path) elif os.path.islink(path): remove_linked_tree(path) else: shutil.rmtree(path) @pytest.fixture def working_env(): saved_env = os.environ.copy() yield # os.environ = saved_env doesn't work # it causes module_parsing::test_module_function to fail # when it's run after any test using this fixture os.environ.clear() os.environ.update(saved_env) # # Note on context managers used by fixtures # # Because these context managers modify global state, they should really # ONLY be used persistently (i.e., around yield statements) in # function-scoped fixtures, OR in autouse session- or module-scoped # fixtures. # # If they're used in regular tests or in module-scoped fixtures that are # then injected as function arguments, weird things can happen, because # the original state won't be restored until *after* the fixture is # destroyed. This makes sense for an autouse fixture, where you know # everything in the module/session is going to need the modified # behavior, but modifying global state for one function in a way that # won't be restored until after the module or session is done essentially # leaves garbage behind for other tests. # # In general, we should module- or session-scope the *STATE* required for # these global objects, but we shouldn't module- or session-scope their # *USE*, or things can get really confusing. # # # Test-specific fixtures # @pytest.fixture(scope="function") def mock_apps_repo_path(): obj_type = ramble.repository.ObjectTypes.applications yield ramble.repository.Repo(ramble.paths.mock_builtin_path, obj_type) @pytest.fixture(scope="function") def mock_mods_repo_path(): obj_type = ramble.repository.ObjectTypes.modifiers yield ramble.repository.Repo(ramble.paths.mock_builtin_path, obj_type) @pytest.fixture(scope="function") def mock_pkg_mans_repo_path(): obj_type = ramble.repository.ObjectTypes.package_managers yield ramble.repository.Repo(ramble.paths.mock_builtin_path, obj_type) @pytest.fixture(scope="function") def mock_wms_repo_path(): obj_type = ramble.repository.ObjectTypes.workflow_managers yield ramble.repository.Repo(ramble.paths.mock_builtin_path, obj_type) def _get_obj_repo_path(obj_type, extra_repo_path): repos = [] # extra_repo_path takes precedence if extra_repo_path is not None: try: repo = ramble.repository.Repo(extra_repo_path, obj_type) repos.append(repo) except ramble.repository.BadRepoError: pass repos.append(ramble.repository.Repo(ramble.paths.builtin_path, obj_type)) yield ramble.repository.RepoPath(*repos, object_type=obj_type) @pytest.fixture(scope="function") def mutable_apps_repo_path(pytestconfig): obj_type = ramble.repository.ObjectTypes.applications extra_repo_path = pytestconfig.getoption("--repo-path") yield from _get_obj_repo_path(obj_type, extra_repo_path) @pytest.fixture(scope="function") def mutable_mods_repo_path(pytestconfig): obj_type = ramble.repository.ObjectTypes.modifiers extra_repo_path = pytestconfig.getoption("--repo-path") yield from _get_obj_repo_path(obj_type, extra_repo_path) @pytest.fixture(scope="function") def mutable_pkg_mans_repo_path(pytestconfig): obj_type = ramble.repository.ObjectTypes.package_managers extra_repo_path = pytestconfig.getoption("--repo-path") yield from _get_obj_repo_path(obj_type, extra_repo_path) @pytest.fixture(scope="function") def mutable_wms_repo_path(pytestconfig): obj_type = ramble.repository.ObjectTypes.workflow_managers extra_repo_path = pytestconfig.getoption("--repo-path") yield from _get_obj_repo_path(obj_type, extra_repo_path) @pytest.fixture(scope="function") def mock_applications(mock_apps_repo_path): """Use the 'builtin.mock' repository for applications instead of 'builtin'""" obj_type = ramble.repository.ObjectTypes.applications with ramble.repository.use_repositories( mock_apps_repo_path, object_type=obj_type ) as mock_apps_repo: yield mock_apps_repo @pytest.fixture(scope="function") def mock_modifiers(mock_mods_repo_path): """Use the 'builtin.mock' repository for modifiersinstead of 'builtin'""" obj_type = ramble.repository.ObjectTypes.modifiers with ramble.repository.use_repositories( mock_mods_repo_path, object_type=obj_type ) as mock_mods_repo: yield mock_mods_repo @pytest.fixture(scope="function") def mock_package_managers(mock_pkg_mans_repo_path): """Use the 'builtin.mock' repository for package managers of 'builtin'""" obj_type = ramble.repository.ObjectTypes.package_managers with ramble.repository.use_repositories( mock_pkg_mans_repo_path, object_type=obj_type ) as mock_pkg_mans_repo: yield mock_pkg_mans_repo @pytest.fixture(scope="function") def mock_workflow_managers(mock_wms_repo_path): """Use the 'builtin.mock' repository for package managers of 'builtin'""" obj_type = ramble.repository.ObjectTypes.workflow_managers with ramble.repository.use_repositories(mock_wms_repo_path, object_type=obj_type) as mock_repo: yield mock_repo @pytest.fixture(scope="function") def mutable_applications(mutable_apps_repo_path): obj_type = ramble.repository.ObjectTypes.applications with ramble.repository.use_repositories( mutable_apps_repo_path, object_type=obj_type ) as apps_repo: yield apps_repo @pytest.fixture(scope="function") def mutable_modifiers(mutable_mods_repo_path): obj_type = ramble.repository.ObjectTypes.modifiers with ramble.repository.use_repositories( mutable_mods_repo_path, object_type=obj_type ) as mods_repo: yield mods_repo def mutable_package_managers(mutable_pkg_mans_repo_path): obj_type = ramble.repository.ObjectTypes.package_managers with ramble.repository.use_repositories( mutable_pkg_mans_repo_path, object_type=obj_type ) as pkg_mans_repo: yield pkg_mans_repo @pytest.fixture(scope="function") def mutable_workflow_managers(mutable_wms_repo_path): obj_type = ramble.repository.ObjectTypes.workflow_managers with ramble.repository.use_repositories( mutable_wms_repo_path, object_type=obj_type ) as wms_repo: yield wms_repo @pytest.fixture(scope="function") def mutable_mock_apps_repo(mock_apps_repo_path): """Function-scoped mock applications, for tests that need to modify them.""" obj_type = ramble.repository.ObjectTypes.applications mock_repo = ramble.repository.Repo(ramble.paths.mock_builtin_path, object_type=obj_type) with ramble.repository.use_repositories(mock_repo, object_type=obj_type) as mock_repo_path: yield mock_repo_path @pytest.fixture(scope="function") def mutable_mock_mods_repo(mock_mods_repo_path): """Function-scoped mock modifiers, for tests that need to modify them.""" obj_type = ramble.repository.ObjectTypes.modifiers mock_repo = ramble.repository.Repo(ramble.paths.mock_builtin_path, object_type=obj_type) with ramble.repository.use_repositories(mock_repo, object_type=obj_type) as mock_repo_path: yield mock_repo_path @pytest.fixture(scope="function") def mutable_mock_pkg_mans_repo(mock_pkg_mans_repo_path): """Function-scoped mock package managers, for tests that need to modify them.""" obj_type = ramble.repository.ObjectTypes.package_managers mock_repo = ramble.repository.Repo(ramble.paths.mock_builtin_path, object_type=obj_type) with ramble.repository.use_repositories(mock_repo, object_type=obj_type) as mock_repo_path: yield mock_repo_path @pytest.fixture(scope="function") def mutable_mock_wms_repo(mock_wms_repo_path): """Function-scoped mock package managers, for tests that need to modify them.""" obj_type = ramble.repository.ObjectTypes.workflow_managers mock_repo = ramble.repository.Repo(ramble.paths.mock_builtin_path, object_type=obj_type) with ramble.repository.use_repositories(mock_repo, object_type=obj_type) as mock_repo_path: yield mock_repo_path @pytest.fixture(scope="function") def default_config(): """Isolates the default configuration from the user configs. This ensures we can test the real default configuration without having tests fail when the user overrides the defaults that we test against.""" defaults_path = os.path.join(ramble.paths.etc_path, "ramble", "defaults") with ramble.config.use_configuration(defaults_path) as defaults_config: yield defaults_config @pytest.fixture(scope="session") def configuration_dir(tmpdir_factory, linux_os): """Copies mock configuration files in a temporary directory. Returns the directory path. """ tmpdir = tmpdir_factory.mktemp("configurations") # <test_path>/data/config has mock config yaml files in it # copy these to the site config. test_config = py.path.local(ramble.paths.test_path).join("data", "config") test_config.copy(tmpdir.join("site")) # Create temporary 'defaults', 'site' and 'user' folders tmpdir.ensure("user", dir=True) # Slightly modify config.yaml solver = os.environ.get("SPACK_TEST_SOLVER", "original") config_yaml = test_config.join("config.yaml") modules_root = tmpdir_factory.mktemp("share") tcl_root = modules_root.ensure("modules", dir=True) lmod_root = modules_root.ensure("lmod", dir=True) content = "".join(config_yaml.read()).format(solver, str(tcl_root), str(lmod_root)) t = tmpdir.join("site", "config.yaml") t.write(content) yield tmpdir # Once done, cleanup the directory shutil.rmtree(str(tmpdir)) @pytest.fixture(scope="session") def linux_os(): """Returns a named tuple with attributes 'name' and 'version' representing the OS. """ platform = spack.platforms.host() name, version = "debian", "6" if platform.name == "linux": current_os = platform.operating_system("default_os") name, version = current_os.name, current_os.version LinuxOS = collections.namedtuple("LinuxOS", ["name", "version"]) return LinuxOS(name=name, version=version) @pytest.fixture(scope="session") def mock_configuration_scopes(configuration_dir): """Create a persistent Configuration object from the configuration_dir.""" defaults = ramble.config.InternalConfigScope("_builtin", ramble.config.config_defaults) test_scopes = [defaults] test_scopes += [ ramble.config.ConfigScope(name, str(configuration_dir.join(name))) for name in ["site", "system", "user"] ] test_scopes.append(ramble.config.InternalConfigScope("command_line")) yield test_scopes @pytest.fixture(scope="function") def config(mock_configuration_scopes): """This fixture activates/deactivates the mock configuration.""" with ramble.config.use_configuration(*mock_configuration_scopes) as config: yield config @pytest.fixture(scope="function") def mutable_config(tmpdir_factory, configuration_dir): """Like config, but tests can modify the configuration.""" mutable_dir = tmpdir_factory.mktemp("mutable_config").join("tmp") configuration_dir.copy(mutable_dir) defaults = ramble.config.InternalConfigScope("_builtin", ramble.config.config_defaults) scopes = [defaults] scopes += [ ramble.config.ConfigScope(name, str(mutable_dir.join(name))) for name in ["site", "system", "user"] ] with ramble.config.use_configuration(*scopes) as cfg: yield cfg @pytest.fixture(scope="function") def mutable_empty_config(tmpdir_factory, configuration_dir): """Empty configuration that can be modified by the tests.""" mutable_dir = tmpdir_factory.mktemp("mutable_config").join("tmp") scopes = [ ramble.config.ConfigScope(name, str(mutable_dir.join(name))) for name in ["site", "system", "user"] ] with ramble.config.use_configuration(*scopes) as cfg: yield cfg @pytest.fixture() def mock_low_high_config(tmpdir): """Mocks two configuration scopes: 'low' and 'high'.""" scopes = [ramble.config.ConfigScope(name, str(tmpdir.join(name))) for name in ["low", "high"]] with ramble.config.use_configuration(*scopes) as config: yield config @pytest.fixture(scope="session") def _store_dir_and_cache(tmpdir_factory): """Returns the directory where to build the mock database and where to cache it. """ store = tmpdir_factory.mktemp("mock_store") cache = tmpdir_factory.mktemp("mock_store_cache") return store, cache class MockLayout: def __init__(self, root): self.root = root def path_for_spec(self, spec): return "/".join([self.root, spec.name]) def check_installed(self, spec): return True @pytest.fixture() def gen_mock_layout(tmpdir): # Generate a MockLayout in a temporary directory. In general the prefixes # specified by MockLayout should never be written to, but this ensures # that even if they are, that it causes no harm def create_layout(root): subroot = tmpdir.mkdir(root) return MockLayout(str(subroot)) yield create_layout class MockConfig: def __init__(self, configuration, writer_key): self._configuration = configuration self.writer_key = writer_key def configuration(self): return self._configuration def writer_configuration(self): return self.configuration()[self.writer_key] class ConfigUpdate: def __init__(self, root_for_conf, writer_mod, writer_key, monkeypatch): self.root_for_conf = root_for_conf self.writer_mod = writer_mod self.writer_key = writer_key self.monkeypatch = monkeypatch def __call__(self, filename): file = os.path.join(self.root_for_conf, filename + ".yaml") with open(file) as f: mock_config = MockConfig(syaml.load_config(f), self.writer_key) self.monkeypatch.setattr(ramble.modules.common, "configuration", mock_config.configuration) self.monkeypatch.setattr( self.writer_mod, "configuration", mock_config.writer_configuration ) self.monkeypatch.setattr(self.writer_mod, "configuration_registry", {}) ########## # Class and fixture to work around problems raising exceptions in directives, # which cause tests like test_from_list_url to hang for Python 2.x metaclass # processing. # # At this point only version and patch directive handling has been addressed. ########## class MockBundle: has_code = False name = "mock-bundle" versions = {} @pytest.fixture def mock_directive_bundle(): """Return a mock bundle package for directive tests.""" return MockBundle() @pytest.fixture def clear_directive_functions(): """Clear all overridden directive functions for subsequent tests.""" yield # Make sure any directive functions overridden by tests are cleared before # proceeding with subsequent tests that may depend on the original # functions. ramble.directives.DirectiveMeta._directives_to_be_executed = [] @pytest.fixture def mock_executable(tmpdir): """Factory to create a mock executable in a temporary directory that output a custom string when run. """ import jinja2 def _factory(name, output, subdir=("bin",)): f = tmpdir.ensure(*subdir, dir=True).join(name) t = jinja2.Template("#!/bin/bash\n{{ output }}\n") f.write(t.render(output=output)) f.chmod(0o755) return str(f) return _factory @pytest.fixture(scope="function") def mutable_mock_workspace_path(tmpdir_factory, mutable_config): """Fixture for mocking the internal ramble workspaces directory.""" mock_path = tmpdir_factory.mktemp("mock-workspace-path") with ramble.config.override("config:workspace_dirs", str(mock_path)): yield mock_path @pytest.fixture def no_path_access(monkeypatch): monkeypatch.setattr(os, "access", _can_access) @pytest.fixture(scope="function", autouse=True) def print_all_logs(monkeypatch): import ramble.util.logger monkeypatch.setattr(ramble.util.logger.logger, "msg", ramble.util.logger.logger.all_msg) ########## # Fake archives and repositories ########## @pytest.fixture(scope="session", params=[(".tar.gz", "z")]) def mock_archive(request, tmpdir_factory): """Creates a very simple archive directory with a configure script and a makefile that installs to a prefix. Tars it up into an archive. """ tar = spack.util.executable.which("tar", required=True) tmpdir = tmpdir_factory.mktemp("mock-archive-dir") tmpdir.ensure(ramble.stage._input_subdir, dir=True) repodir = tmpdir.join(ramble.stage._input_subdir) # Create the configure script configure_path = str(tmpdir.join(ramble.stage._input_subdir, "configure")) with open(configure_path, "w") as f: f.write( "#!/bin/sh\n" "prefix=$(echo $1 | sed 's/--prefix=//')\n" "cat > Makefile <<EOF\n" "all:\n" "\techo Building...\n\n" "install:\n" "\tmkdir -p $prefix\n" "\ttouch $prefix/dummy_file\n" "EOF\n" ) os.chmod(configure_path, 0o755) # Archive it with tmpdir.as_cwd(): archive_name = f"{ramble.stage._input_subdir}{request.param[0]}" tar(f"-c{request.param[1]}f", archive_name, ramble.stage._input_subdir) Archive = collections.namedtuple( "Archive", ["url", "path", "archive_file", "expanded_archive_basedir"] ) archive_file = str(tmpdir.join(archive_name)) url = "file://" + archive_file # Return the url yield Archive( url=url, archive_file=archive_file, path=str(repodir), expanded_archive_basedir=ramble.stage._input_subdir, ) @pytest.fixture(scope="function") def install_mockery_mutable_config(mutable_config, mock_applications): """Hooks fake applications and config directory into Ramble. This is specifically for tests which want to use 'install_mockery' but also need to modify configuration (and hence would want to use 'mutable config'): 'install_mockery' does not support this. """ # We use a fake package, so temporarily disable checksumming with ramble.config.override("config:checksum", False): yield class MockCache: def store(self, copy_cmd, relative_dest): pass def fetcher(self, target_path, digest, **kwargs): return MockCacheFetcher() class MockCacheFetcher: def fetch(self): raise FetchError("Mock cache always fails for tests") def __str__(self): return "[mock fetch cache]" @pytest.fixture(autouse=True) def mock_fetch_cache(monkeypatch): """Substitutes ramble.paths.fetch_cache with a mock object that does nothing and raises on fetch. """ monkeypatch.setattr(ramble.caches, "fetch_cache", MockCache()) @pytest.fixture() def mock_fetch(mock_archive, monkeypatch): """Fake the URL for an input so it downloads from a file.""" mock_fetcher = FetchStrategyComposite() mock_fetcher.append(URLFetchStrategy(mock_archive.url)) yield mock_fetcher @pytest.fixture() def mock_file_auto_create(monkeypatch): builtin_open = builtins.open def open_or_create_inmem(path, *args, **kwargs): if not os.path.exists(path) and is_dry_run_path(path): if path.endswith(".yaml") or path.endswith(".yml"): content = "{}" else: content = "" inmem = io.StringIO(content) return inmem return builtin_open(path, *args, **kwargs) monkeypatch.setattr(builtins, "open", open_or_create_inmem) def pytest_generate_tests(metafunc): import re name_regex = re.compile(r"\s*(?P<name>[a-z0-9\-\_]+)\s*$") if "application" in metafunc.fixturenames: from ramble.main import RambleCommand list_cmd = RambleCommand("list") all_applications = [] repo_apps = list_cmd().split("\n") for app_str in repo_apps: m = name_regex.match(app_str) if m: all_applications.append(m.group("name")) metafunc.parametrize("application", all_applications) if "modifier" in metafunc.fixturenames: from ramble.main import RambleCommand list_cmd = RambleCommand("list") all_modifiers = [] repo_mods = list_cmd("--type", "modifiers").split("\n") for mod_str in repo_mods: m = name_regex.match(mod_str) if m: all_modifiers.append(m.group("name")) metafunc.parametrize("modifier", all_modifiers) if "mock_modifier" in metafunc.fixturenames: obj_type = ramble.repository.ObjectTypes.modifiers repo_path = ramble.repository.Repo(ramble.paths.mock_builtin_path, obj_type) all_modifiers = [] for mod_name in repo_path.all_object_names(): all_modifiers.append(mod_name) metafunc.parametrize("mock_modifier", all_modifiers) if "package_manager" in metafunc.fixturenames: from ramble.main import RambleCommand list_cmd = RambleCommand("list") all_package_managers = ["None"] repo_pms = list_cmd("--type", "package_managers").split("\n") for pm_str in repo_pms: m = name_regex.match(pm_str) if m: all_package_managers.append(m.group("name")) metafunc.parametrize("package_manager", all_package_managers) if "mock_package_managers" in metafunc.fixturenames: obj_type = ramble.repository.ObjectTypes.package_managers repo_path = ramble.repository.Repo(ramble.paths.mock_builtin_path, obj_type) all_package_managers = ["None"] for mod_name in repo_path.all_object_names(): all_package_managers.append(mod_name) metafunc.parametrize("mock_package_managers", all_package_managers) if "workflow_manager" in metafunc.fixturenames: from ramble.main import RambleCommand list_cmd = RambleCommand("list") all_workflow_managers = ["None"] repo_pms = list_cmd("--type", "workflow_managers").split("\n") for pm_str in repo_pms: m = name_regex.match(pm_str) if m: all_workflow_managers.append(m.group("name")) metafunc.parametrize("workflow_manager", all_workflow_managers) if "mock_workflow_managers" in metafunc.fixturenames: obj_type = ramble.repository.ObjectTypes.workflow_managers repo_path = ramble.repository.Repo(ramble.paths.mock_builtin_path, obj_type) all_workflow_managers = ["None"] for mod_name in repo_path.all_object_names(): all_workflow_managers.append(mod_name) metafunc.parametrize("mock_workflow_managers", all_workflow_managers) if "config_section" in metafunc.fixturenames: from ramble.main import RambleCommand config_cmd = RambleCommand("config") all_sections = [] config_sections = config_cmd("list").split(" ") for section_str in config_sections: if section_str != "": all_sections.append(section_str.strip()) metafunc.parametrize("config_section", all_sections)