# Copyright 2013-2024 Lawrence Livermore National Security, LLC and other
# Spack Project Developers. See the top-level COPYRIGHT file for details.
#
# SPDX-License-Identifier: (Apache-2.0 OR MIT)
import os
import pytest
import spack.cmd.mirror
import spack.config
import spack.environment as ev
import spack.spec
import spack.util.url as url_util
from spack.main import SpackCommand, SpackCommandError
mirror = SpackCommand("mirror")
env = SpackCommand("env")
add = SpackCommand("add")
concretize = SpackCommand("concretize")
install = SpackCommand("install")
buildcache = SpackCommand("buildcache")
uninstall = SpackCommand("uninstall")
pytestmark = pytest.mark.not_on_windows("does not run on windows")
@pytest.mark.disable_clean_stage_check
@pytest.mark.regression("8083")
def test_regression_8083(tmpdir, capfd, mock_packages, mock_fetch, config):
with capfd.disabled():
output = mirror("create", "-d", str(tmpdir), "externaltool")
assert "Skipping" in output
assert "as it is an external spec" in output
@pytest.mark.regression("12345")
def test_mirror_from_env(tmp_path, mock_packages, mock_fetch, config, mutable_mock_env_path):
mirror_dir = str(tmp_path / "mirror")
env_name = "test"
env("create", env_name)
with ev.read(env_name):
add("trivial-install-test-package")
add("git-test")
concretize()
with spack.config.override("config:checksum", False):
mirror("create", "-d", mirror_dir, "--all")
e = ev.read(env_name)
assert set(os.listdir(mirror_dir)) == set([s.name for s in e.user_specs])
for spec in e.specs_by_hash.values():
mirror_res = os.listdir(os.path.join(mirror_dir, spec.name))
expected = ["%s.tar.gz" % spec.format("{name}-{version}")]
assert mirror_res == expected
@pytest.fixture
def source_for_pkg_with_hash(mock_packages, tmpdir):
s = spack.spec.Spec("trivial-pkg-with-valid-hash").concretized()
local_url_basename = os.path.basename(s.package.url)
local_path = os.path.join(str(tmpdir), local_url_basename)
with open(local_path, "w") as f:
f.write(s.package.hashed_content)
local_url = url_util.path_to_file_url(local_path)
s.package.versions[spack.version.Version("1.0")]["url"] = local_url
def test_mirror_skip_unstable(tmpdir_factory, mock_packages, config, source_for_pkg_with_hash):
mirror_dir = str(tmpdir_factory.mktemp("mirror-dir"))
specs = [spack.spec.Spec(x).concretized() for x in ["git-test", "trivial-pkg-with-valid-hash"]]
spack.mirror.create(mirror_dir, specs, skip_unstable_versions=True)
assert set(os.listdir(mirror_dir)) - set(["_source-cache"]) == set(
["trivial-pkg-with-valid-hash"]
)
class MockMirrorArgs:
def __init__(
self,
specs=None,
all=False,
file=None,
versions_per_spec=None,
dependencies=False,
exclude_file=None,
exclude_specs=None,
directory=None,
):
self.specs = specs or []
self.all = all
self.file = file
self.versions_per_spec = versions_per_spec
self.dependencies = dependencies
self.exclude_file = exclude_file
self.exclude_specs = exclude_specs
self.directory = directory
def test_exclude_specs(mock_packages, config):
args = MockMirrorArgs(
specs=["mpich"], versions_per_spec="all", exclude_specs="mpich@3.0.1:3.0.2 mpich@1.0"
)
mirror_specs = spack.cmd.mirror.concrete_specs_from_user(args)
expected_include = set(
spack.spec.Spec(x).concretized() for x in ["mpich@3.0.3", "mpich@3.0.4", "mpich@3.0"]
)
expected_exclude = set(spack.spec.Spec(x) for x in ["mpich@3.0.1", "mpich@3.0.2", "mpich@1.0"])
assert expected_include <= set(mirror_specs)
assert not any(spec.satisfies(y) for spec in mirror_specs for y in expected_exclude)
def test_exclude_file(mock_packages, tmpdir, config):
exclude_path = os.path.join(str(tmpdir), "test-exclude.txt")
with open(exclude_path, "w") as exclude_file:
exclude_file.write(
"""\
mpich@3.0.1:3.0.2
mpich@1.0
"""
)
args = MockMirrorArgs(specs=["mpich"], versions_per_spec="all", exclude_file=exclude_path)
mirror_specs = spack.cmd.mirror.concrete_specs_from_user(args)
expected_include = set(
spack.spec.Spec(x).concretized() for x in ["mpich@3.0.3", "mpich@3.0.4", "mpich@3.0"]
)
expected_exclude = set(spack.spec.Spec(x) for x in ["mpich@3.0.1", "mpich@3.0.2", "mpich@1.0"])
assert expected_include <= set(mirror_specs)
assert not any(spec.satisfies(y) for spec in mirror_specs for y in expected_exclude)
def test_mirror_crud(mutable_config, capsys):
with capsys.disabled():
mirror("add", "mirror", "http://spack.io")
output = mirror("remove", "mirror")
assert "Removed mirror" in output
mirror("add", "mirror", "http://spack.io")
# no-op
output = mirror("set-url", "mirror", "http://spack.io")
assert "No changes made" in output
output = mirror("set-url", "--push", "mirror", "s3://spack-public")
assert not output
# no-op
output = mirror("set-url", "--push", "mirror", "s3://spack-public")
assert "No changes made" in output
output = mirror("remove", "mirror")
assert "Removed mirror" in output
# Test S3 connection info token
mirror("add", "--s3-access-token", "aaaaaazzzzz", "mirror", "s3://spack-public")
output = mirror("remove", "mirror")
assert "Removed mirror" in output
# Test S3 connection info id/key
mirror(
"add",
"--s3-access-key-id",
"foo",
"--s3-access-key-secret",
"bar",
"mirror",
"s3://spack-public",
)
output = mirror("remove", "mirror")
assert "Removed mirror" in output
# Test S3 connection info with endpoint URL
mirror(
"add",
"--s3-access-token",
"aaaaaazzzzz",
"--s3-endpoint-url",
"http://localhost/",
"mirror",
"s3://spack-public",
)
output = mirror("remove", "mirror")
assert "Removed mirror" in output
output = mirror("list")
assert "No mirrors configured" in output
# Test GCS Mirror
mirror("add", "mirror", "gs://spack-test")
output = mirror("remove", "mirror")
assert "Removed mirror" in output
def test_mirror_nonexisting(mutable_config):
with pytest.raises(SpackCommandError):
mirror("remove", "not-a-mirror")
with pytest.raises(SpackCommandError):
mirror("set-url", "not-a-mirror", "http://spack.io")
def test_mirror_name_collision(mutable_config):
mirror("add", "first", "1")
with pytest.raises(SpackCommandError):
mirror("add", "first", "1")
def test_mirror_destroy(
install_mockery_mutable_config,
mock_packages,
mock_fetch,
mock_archive,
mutable_config,
monkeypatch,
tmpdir,
):
# Create a temp mirror directory for buildcache usage
mirror_dir = tmpdir.join("mirror_dir")
mirror_url = "file://{0}".format(mirror_dir.strpath)
mirror("add", "atest", mirror_url)
spec_name = "libdwarf"
# Put a binary package in a buildcache
install("--no-cache", spec_name)
buildcache("push", "-u", "-f", mirror_dir.strpath, spec_name)
contents = os.listdir(mirror_dir.strpath)
assert "build_cache" in contents
# Destroy mirror by name
mirror("destroy", "-m", "atest")
assert not os.path.exists(mirror_dir.strpath)
buildcache("push", "-u", "-f", mirror_dir.strpath, spec_name)
contents = os.listdir(mirror_dir.strpath)
assert "build_cache" in contents
# Destroy mirror by url
mirror("destroy", "--mirror-url", mirror_url)
assert not os.path.exists(mirror_dir.strpath)
uninstall("-y", spec_name)
mirror("remove", "atest")
@pytest.mark.usefixtures("mock_packages")
class TestMirrorCreate:
@pytest.mark.regression("31736", "31985")
def test_all_specs_with_all_versions_dont_concretize(self):
args = MockMirrorArgs(exclude_file=None, exclude_specs=None)
specs = spack.cmd.mirror.all_specs_with_all_versions(
selection_fn=spack.cmd.mirror.not_excluded_fn(args)
)
assert all(not s.concrete for s in specs)
@pytest.mark.parametrize(
"cli_args,error_str",
[
# Passed more than one among -f --all and specs
({"specs": "hdf5", "file": None, "all": True}, "cannot specify specs on command line"),
(
{"specs": None, "file": "input.txt", "all": True},
"cannot specify specs with a file if",
),
(
{"specs": "hdf5", "file": "input.txt", "all": False},
"cannot specify specs with a file AND",
),
({"specs": None, "file": None, "all": False}, "no packages were specified"),
# Passed -n along with --all
(
{"specs": None, "file": None, "all": True, "versions_per_spec": 2},
"cannot specify '--versions_per-spec'",
),
],
)
def test_error_conditions(self, cli_args, error_str):
args = MockMirrorArgs(**cli_args)
with pytest.raises(spack.error.SpackError, match=error_str):
spack.cmd.mirror.mirror_create(args)
@pytest.mark.parametrize(
"cli_args,not_expected",
[
(
{
"specs": "boost bowtie callpath",
"exclude_specs": "bowtie",
"dependencies": False,
},
["bowtie"],
),
(
{
"specs": "boost bowtie callpath",
"exclude_specs": "bowtie callpath",
"dependencies": False,
},
["bowtie", "callpath"],
),
(
{
"specs": "boost bowtie callpath",
"exclude_specs": "bowtie",
"dependencies": True,
},
["bowtie"],
),
],
)
def test_exclude_specs_from_user(self, cli_args, not_expected, config):
specs = spack.cmd.mirror.concrete_specs_from_user(MockMirrorArgs(**cli_args))
assert not any(s.satisfies(y) for s in specs for y in not_expected)
@pytest.mark.parametrize("abstract_specs", [("bowtie", "callpath")])
def test_specs_from_cli_are_the_same_as_from_file(self, abstract_specs, config, tmpdir):
args = MockMirrorArgs(specs=" ".join(abstract_specs))
specs_from_cli = spack.cmd.mirror.concrete_specs_from_user(args)
input_file = tmpdir.join("input.txt")
input_file.write("\n".join(abstract_specs))
args = MockMirrorArgs(file=str(input_file))
specs_from_file = spack.cmd.mirror.concrete_specs_from_user(args)
assert specs_from_cli == specs_from_file
@pytest.mark.parametrize(
"input_specs,nversions",
[("callpath", 1), ("mpich", 4), ("callpath mpich", 3), ("callpath mpich", "all")],
)
def test_versions_per_spec_produces_concrete_specs(self, input_specs, nversions, config):
args = MockMirrorArgs(specs=input_specs, versions_per_spec=nversions)
specs = spack.cmd.mirror.concrete_specs_from_user(args)
assert all(s.concrete for s in specs)
def test_mirror_type(mutable_config):
"""Test the mirror set command"""
mirror("add", "example", "--type", "binary", "http://example.com")
assert spack.config.get("mirrors:example") == {
"url": "http://example.com",
"source": False,
"binary": True,
}
mirror("set", "example", "--type", "source")
assert spack.config.get("mirrors:example") == {
"url": "http://example.com",
"source": True,
"binary": False,
}
mirror("set", "example", "--type", "binary")
assert spack.config.get("mirrors:example") == {
"url": "http://example.com",
"source": False,
"binary": True,
}
mirror("set", "example", "--type", "binary", "--type", "source")
assert spack.config.get("mirrors:example") == {
"url": "http://example.com",
"source": True,
"binary": True,
}
def test_mirror_set_2(mutable_config):
"""Test the mirror set command"""
mirror("add", "example", "http://example.com")
mirror(
"set",
"example",
"--push",
"--url",
"http://example2.com",
"--s3-access-key-id",
"username",
"--s3-access-key-secret",
"password",
)
assert spack.config.get("mirrors:example") == {
"url": "http://example.com",
"push": {"url": "http://example2.com", "access_pair": ["username", "password"]},
}
def test_mirror_add_set_signed(mutable_config):
mirror("add", "--signed", "example", "http://example.com")
assert spack.config.get("mirrors:example") == {"url": "http://example.com", "signed": True}
mirror("set", "--unsigned", "example")
assert spack.config.get("mirrors:example") == {"url": "http://example.com", "signed": False}
mirror("set", "--signed", "example")
assert spack.config.get("mirrors:example") == {"url": "http://example.com", "signed": True}