blob: d8a17eec8e8a5eca0e6670403467c80cc91a464f [file] [log] [blame]
# Pylint doesn't play well with fixtures and dependency injection from pytest
# pylint: disable=redefined-outer-name
import os
import pytest
from buildstream import _yaml
from buildstream.exceptions import ErrorDomain
from buildstream._testing import create_repo
from buildstream._testing import cli # pylint: disable=unused-import
from tests.testutils.repo.git import Git
from tests.testutils.site import pip_sample_packages # pylint: disable=unused-import
from tests.testutils.site import SAMPLE_PACKAGES_SKIP_REASON
# Project directory
TOP_DIR = os.path.dirname(os.path.realpath(__file__))
DATA_DIR = os.path.join(TOP_DIR, "project")
def generate_element(output_file):
element = {
"kind": "import",
"sources": [
{
"kind": "fetch_source",
"output-text": output_file,
"urls": ["foo:repo1", "bar:repo2"],
"fetch-succeeds": {
"FOO/repo1": True,
"BAR/repo2": False,
"OOF/repo1": False,
"RAB/repo2": True,
"OFO/repo1": False,
"RBA/repo2": False,
"ooF/repo1": False,
"raB/repo2": False,
},
}
],
}
return element
DEFAULT_MIRROR_LIST = [
{
"name": "middle-earth",
"aliases": {
"foo": ["OOF/"],
"bar": ["RAB/"],
},
},
{
"name": "arrakis",
"aliases": {
"foo": ["OFO/"],
"bar": ["RBA/"],
},
},
{
"name": "oz",
"aliases": {
"foo": ["ooF/"],
"bar": ["raB/"],
},
},
]
SUCCESS_MIRROR_LIST = [
{
"name": "middle-earth",
"aliases": {
"foo": ["OOF/"],
"bar": ["RAB/"],
},
},
{
"name": "arrakis",
"aliases": {
"foo": ["FOO/"],
"bar": ["RBA/"],
},
},
{
"name": "oz",
"aliases": {
"foo": ["ooF/"],
"bar": ["raB/"],
},
},
]
FAIL_MIRROR_LIST = [
{
"name": "middle-earth",
"aliases": {
"foo": ["pony/"],
"bar": ["horzy/"],
},
},
{
"name": "arrakis",
"aliases": {
"foo": ["donkey/"],
"bar": ["rabbit/"],
},
},
{
"name": "oz",
"aliases": {
"foo": ["bear/"],
"bar": ["buffalo/"],
},
},
]
class MirrorConfig:
NO_MIRRORS = 0
SUCCESS_MIRRORS = 1
FAIL_MIRRORS = 2
DEFAULT_MIRRORS = 3
def generate_project(mirror_config=MirrorConfig.DEFAULT_MIRRORS, base_alias_succeed=False):
aliases = {
"foo": "FOO/",
"bar": "BAR/",
}
if base_alias_succeed:
aliases["bar"] = "RAB/"
project = {
"name": "test",
"min-version": "2.0",
"element-path": "elements",
"aliases": aliases,
"plugins": [{"origin": "local", "path": "sources", "sources": ["fetch_source"]}],
}
if mirror_config == MirrorConfig.SUCCESS_MIRRORS:
project["mirrors"] = SUCCESS_MIRROR_LIST
elif mirror_config == MirrorConfig.FAIL_MIRRORS:
project["mirrors"] = FAIL_MIRROR_LIST
elif mirror_config == MirrorConfig.DEFAULT_MIRRORS:
project["mirrors"] = DEFAULT_MIRROR_LIST
return project
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("ref_storage", [("inline"), ("project.refs")])
@pytest.mark.parametrize("mirror", [("no-mirror"), ("mirror"), ("unrelated-mirror")])
def test_mirror_fetch_ref_storage(cli, tmpdir, datafiles, ref_storage, mirror):
bin_files_path = os.path.join(str(datafiles), "files", "bin-files", "usr")
dev_files_path = os.path.join(str(datafiles), "files", "dev-files", "usr")
upstream_repodir = os.path.join(str(tmpdir), "upstream")
mirror_repodir = os.path.join(str(tmpdir), "mirror")
project_dir = os.path.join(str(tmpdir), "project")
os.makedirs(project_dir)
element_dir = os.path.join(project_dir, "elements")
# Create repo objects of the upstream and mirror
upstream_repo = create_repo("tar", upstream_repodir)
upstream_repo.create(bin_files_path)
mirror_repo = upstream_repo.copy(mirror_repodir)
upstream_ref = upstream_repo.create(dev_files_path)
element = {
"kind": "import",
"sources": [upstream_repo.source_config(ref=upstream_ref if ref_storage == "inline" else None)],
}
element_name = "test.bst"
element_path = os.path.join(element_dir, element_name)
full_repo = element["sources"][0]["url"]
upstream_map, repo_name = os.path.split(full_repo)
alias = "foo"
aliased_repo = alias + ":" + repo_name
element["sources"][0]["url"] = aliased_repo
full_mirror = mirror_repo.source_config()["url"]
mirror_map, _ = os.path.split(full_mirror)
os.makedirs(element_dir)
_yaml.roundtrip_dump(element, element_path)
if ref_storage == "project.refs":
# Manually set project.refs to avoid caching the repo prematurely
project_refs = {"projects": {"test": {element_name: [{"ref": upstream_ref}]}}}
project_refs_path = os.path.join(project_dir, "project.refs")
_yaml.roundtrip_dump(project_refs, project_refs_path)
project = {
"name": "test",
"min-version": "2.0",
"element-path": "elements",
"aliases": {alias: upstream_map + "/"},
"ref-storage": ref_storage,
}
if mirror != "no-mirror":
mirror_data = [{"name": "middle-earth", "aliases": {alias: [mirror_map + "/"]}}]
if mirror == "unrelated-mirror":
mirror_data.insert(0, {"name": "narnia", "aliases": {"frob": ["http://www.example.com/repo"]}})
project["mirrors"] = mirror_data
project_file = os.path.join(project_dir, "project.conf")
_yaml.roundtrip_dump(project, project_file)
result = cli.run(project=project_dir, args=["source", "fetch", element_name])
result.assert_success()
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.usefixtures("datafiles")
@pytest.mark.parametrize(
"project_config,user_config,expect_success",
[
# User defined mirror configuration
(MirrorConfig.NO_MIRRORS, MirrorConfig.SUCCESS_MIRRORS, True),
# Project defined mirror configuration
(MirrorConfig.SUCCESS_MIRRORS, MirrorConfig.NO_MIRRORS, True),
# Both configurations active with success
(MirrorConfig.FAIL_MIRRORS, MirrorConfig.SUCCESS_MIRRORS, True),
# Both configurations active with failure, this ensures that
# the user configuration does not regress to extending project defined
# mirrors but properly overrides project defined mirrors.
(MirrorConfig.SUCCESS_MIRRORS, MirrorConfig.FAIL_MIRRORS, False),
],
ids=["user-config", "project-config", "override-success", "override-fail"],
)
def test_mirror_fetch_multi(cli, tmpdir, project_config, user_config, expect_success):
output_file = os.path.join(str(tmpdir), "output.txt")
project_dir = str(tmpdir)
element_dir = os.path.join(project_dir, "elements")
os.makedirs(element_dir, exist_ok=True)
element_name = "test.bst"
element_path = os.path.join(element_dir, element_name)
element = generate_element(output_file)
_yaml.roundtrip_dump(element, element_path)
project_file = os.path.join(project_dir, "project.conf")
project = generate_project(project_config)
_yaml.roundtrip_dump(project, project_file)
if user_config == MirrorConfig.SUCCESS_MIRRORS:
cli.configure({"projects": {"test": {"mirrors": SUCCESS_MIRROR_LIST}}})
elif user_config == MirrorConfig.FAIL_MIRRORS:
cli.configure({"projects": {"test": {"mirrors": FAIL_MIRROR_LIST}}})
result = cli.run(project=project_dir, args=["source", "fetch", element_name])
if expect_success:
result.assert_success()
with open(output_file, encoding="utf-8") as f:
contents = f.read()
assert "Fetch foo:repo1 succeeded from FOO/repo1" in contents
assert "Fetch bar:repo2 succeeded from RAB/repo2" in contents
else:
result.assert_main_error(ErrorDomain.STREAM, None)
result.assert_task_error(ErrorDomain.SOURCE, None)
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.usefixtures("datafiles")
@pytest.mark.parametrize(
"project_config,user_config,alias_success,expect_success,source",
[
#
# Test "alias" fetch source policy (aliases only)
#
# Test that we fail to fetch from the primary alias even if the user config defines a mirror
(MirrorConfig.NO_MIRRORS, MirrorConfig.SUCCESS_MIRRORS, False, False, "aliases"),
# Test that we fail to fetch from the primary alias even if the project config defines a mirror
(MirrorConfig.SUCCESS_MIRRORS, MirrorConfig.NO_MIRRORS, False, False, "aliases"),
# Test that we succeed to fetch from the primary alias even if the user config defines a failing mirror
(MirrorConfig.FAIL_MIRRORS, MirrorConfig.FAIL_MIRRORS, True, True, "aliases"),
#
# Test "mirrors" fetch source policy (mirrors only, no base aliases)
#
# Test that we fail to fetch from primary alias even if it is the only one configured to succeed
(MirrorConfig.FAIL_MIRRORS, MirrorConfig.FAIL_MIRRORS, True, False, "mirrors"),
# Test that we succeed to fetch from mirrors when primary alias is set to succeed
# (doesn't prove that primary alias is not consulted, but tests that we indeed consult
# mirrors when configued in mirror mode)
(MirrorConfig.SUCCESS_MIRRORS, MirrorConfig.NO_MIRRORS, True, True, "mirrors"),
(MirrorConfig.FAIL_MIRRORS, MirrorConfig.SUCCESS_MIRRORS, True, True, "mirrors"),
#
# Test "user" fetch source policy (only mirrors defined in user configuration)
#
# Test that we fail to fetch even if the alias is good and the project defined mirrors are good
(MirrorConfig.SUCCESS_MIRRORS, MirrorConfig.FAIL_MIRRORS, True, False, "user"),
# Test that we succeed to fetch when alias is bad and project mirrors are bad
# (this doesn't prove that project aliases and mirrors are not consulted, but here for completeness)
(MirrorConfig.FAIL_MIRRORS, MirrorConfig.SUCCESS_MIRRORS, False, True, "user"),
],
ids=[
"aliases-fail-user-config",
"aliases-fail-project-config",
"aliases-success-bad-mirrors",
"mirrors-fail-bad-mirrors",
"mirrors-success-project-config",
"mirrors-success-user-config",
"user-fail",
"user-succees",
],
)
def test_mirror_fetch_source(cli, tmpdir, project_config, user_config, alias_success, expect_success, source):
output_file = os.path.join(str(tmpdir), "output.txt")
project_dir = str(tmpdir)
element_dir = os.path.join(project_dir, "elements")
os.makedirs(element_dir, exist_ok=True)
element_name = "test.bst"
element_path = os.path.join(element_dir, element_name)
element = generate_element(output_file)
_yaml.roundtrip_dump(element, element_path)
project_file = os.path.join(project_dir, "project.conf")
project = generate_project(project_config, alias_success)
_yaml.roundtrip_dump(project, project_file)
# Configure the fetch source
cli.configure({"fetch": {"source": source}})
if user_config == MirrorConfig.SUCCESS_MIRRORS:
cli.configure({"projects": {"test": {"mirrors": SUCCESS_MIRROR_LIST}}})
elif user_config == MirrorConfig.FAIL_MIRRORS:
cli.configure({"projects": {"test": {"mirrors": FAIL_MIRROR_LIST}}})
result = cli.run(project=project_dir, args=["source", "fetch", element_name])
if expect_success:
result.assert_success()
with open(output_file, encoding="utf-8") as f:
contents = f.read()
assert "Fetch foo:repo1 succeeded from FOO/repo1" in contents
assert "Fetch bar:repo2 succeeded from RAB/repo2" in contents
else:
result.assert_main_error(ErrorDomain.STREAM, None)
result.assert_task_error(ErrorDomain.SOURCE, None)
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.usefixtures("datafiles")
def test_mirror_fetch_default_cmdline(cli, tmpdir):
output_file = os.path.join(str(tmpdir), "output.txt")
project_dir = str(tmpdir)
element_dir = os.path.join(project_dir, "elements")
os.makedirs(element_dir, exist_ok=True)
element_name = "test.bst"
element_path = os.path.join(element_dir, element_name)
element = generate_element(output_file)
_yaml.roundtrip_dump(element, element_path)
project_file = os.path.join(project_dir, "project.conf")
project = generate_project()
_yaml.roundtrip_dump(project, project_file)
result = cli.run(project=project_dir, args=["--default-mirror", "arrakis", "source", "fetch", element_name])
result.assert_success()
with open(output_file, encoding="utf-8") as f:
contents = f.read()
print(contents)
# Success if fetching from arrakis' mirror happened before middle-earth's
arrakis_str = "OFO/repo1"
arrakis_pos = contents.find(arrakis_str)
assert arrakis_pos != -1, "'{}' wasn't found".format(arrakis_str)
me_str = "OOF/repo1"
me_pos = contents.find(me_str)
assert me_pos != -1, "'{}' wasn't found".format(me_str)
assert arrakis_pos < me_pos, "'{}' wasn't found before '{}'".format(arrakis_str, me_str)
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.usefixtures("datafiles")
def test_mirror_fetch_default_userconfig(cli, tmpdir):
output_file = os.path.join(str(tmpdir), "output.txt")
project_dir = str(tmpdir)
element_dir = os.path.join(project_dir, "elements")
os.makedirs(element_dir, exist_ok=True)
element_name = "test.bst"
element_path = os.path.join(element_dir, element_name)
element = generate_element(output_file)
_yaml.roundtrip_dump(element, element_path)
project_file = os.path.join(project_dir, "project.conf")
project = generate_project()
_yaml.roundtrip_dump(project, project_file)
userconfig = {"projects": {"test": {"default-mirror": "oz"}}}
cli.configure(userconfig)
result = cli.run(project=project_dir, args=["source", "fetch", element_name])
result.assert_success()
with open(output_file, encoding="utf-8") as f:
contents = f.read()
print(contents)
# Success if fetching from Oz' mirror happened before middle-earth's
oz_str = "ooF/repo1"
oz_pos = contents.find(oz_str)
assert oz_pos != -1, "'{}' wasn't found".format(oz_str)
me_str = "OOF/repo1"
me_pos = contents.find(me_str)
assert me_pos != -1, "'{}' wasn't found".format(me_str)
assert oz_pos < me_pos, "'{}' wasn't found before '{}'".format(oz_str, me_str)
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.usefixtures("datafiles")
def test_mirror_fetch_default_cmdline_overrides_config(cli, tmpdir):
output_file = os.path.join(str(tmpdir), "output.txt")
project_dir = str(tmpdir)
element_dir = os.path.join(project_dir, "elements")
os.makedirs(element_dir, exist_ok=True)
element_name = "test.bst"
element_path = os.path.join(element_dir, element_name)
element = generate_element(output_file)
_yaml.roundtrip_dump(element, element_path)
project_file = os.path.join(project_dir, "project.conf")
project = generate_project()
_yaml.roundtrip_dump(project, project_file)
userconfig = {"projects": {"test": {"default-mirror": "oz"}}}
cli.configure(userconfig)
result = cli.run(project=project_dir, args=["--default-mirror", "arrakis", "source", "fetch", element_name])
result.assert_success()
with open(output_file, encoding="utf-8") as f:
contents = f.read()
print(contents)
# Success if fetching from arrakis' mirror happened before middle-earth's
arrakis_str = "OFO/repo1"
arrakis_pos = contents.find(arrakis_str)
assert arrakis_pos != -1, "'{}' wasn't found".format(arrakis_str)
me_str = "OOF/repo1"
me_pos = contents.find(me_str)
assert me_pos != -1, "'{}' wasn't found".format(me_str)
assert arrakis_pos < me_pos, "'{}' wasn't found before '{}'".format(arrakis_str, me_str)
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.skipif("not pip_sample_packages()", reason=SAMPLE_PACKAGES_SKIP_REASON)
def test_mirror_git_submodule_fetch(cli, tmpdir, datafiles):
# Test that it behaves as expected with submodules, both defined in config
# and discovered when fetching.
foo_file = os.path.join(str(datafiles), "files", "foo")
bar_file = os.path.join(str(datafiles), "files", "bar")
bin_files_path = os.path.join(str(datafiles), "files", "bin-files", "usr")
dev_files_path = os.path.join(str(datafiles), "files", "dev-files", "usr")
mirror_dir = os.path.join(str(datafiles), "mirror")
defined_subrepo = Git(str(tmpdir), "defined_subrepo")
defined_subrepo.create(bin_files_path)
defined_subrepo.copy(mirror_dir)
defined_subrepo.add_file(foo_file)
found_subrepo = Git(str(tmpdir), "found_subrepo")
found_subrepo.create(dev_files_path)
main_repo = Git(str(tmpdir))
main_mirror_ref = main_repo.create(bin_files_path)
main_repo.add_submodule("defined", "file://" + defined_subrepo.repo)
main_repo.add_submodule("found", "file://" + found_subrepo.repo)
main_mirror = main_repo.copy(mirror_dir)
main_repo.add_file(bar_file)
project_dir = os.path.join(str(tmpdir), "project")
os.makedirs(project_dir)
element_dir = os.path.join(project_dir, "elements")
os.makedirs(element_dir)
element = {"kind": "import", "sources": [main_repo.source_config(ref=main_mirror_ref)]}
element_name = "test.bst"
element_path = os.path.join(element_dir, element_name)
# Alias the main repo
full_repo = element["sources"][0]["url"]
_, repo_name = os.path.split(full_repo)
alias = "foo"
aliased_repo = alias + ":" + repo_name
element["sources"][0]["url"] = aliased_repo
# Hide the found subrepo
del element["sources"][0]["submodules"]["found"]
# Alias the defined subrepo
subrepo = element["sources"][0]["submodules"]["defined"]["url"]
_, repo_name = os.path.split(subrepo)
aliased_repo = alias + ":" + repo_name
element["sources"][0]["submodules"]["defined"]["url"] = aliased_repo
_yaml.roundtrip_dump(element, element_path)
full_mirror = main_mirror.source_config()["url"]
mirror_map, _ = os.path.split(full_mirror)
project = {
"name": "test",
"min-version": "2.0",
"element-path": "elements",
"aliases": {alias: "http://www.example.com/"},
"plugins": [
{
"origin": "pip",
"package-name": "sample-plugins",
"sources": ["git"],
}
],
"mirrors": [
{
"name": "middle-earth",
"aliases": {
alias: [mirror_map + "/"],
},
},
],
}
project_file = os.path.join(project_dir, "project.conf")
_yaml.roundtrip_dump(project, project_file)
result = cli.run(project=project_dir, args=["source", "fetch", element_name])
result.assert_success()
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.skipif("not pip_sample_packages()", reason=SAMPLE_PACKAGES_SKIP_REASON)
def test_mirror_fallback_git_only_submodules(cli, tmpdir, datafiles):
# Main repo has no mirror or alias.
# One submodule is overridden to use a mirror.
# There is another submodules not overriden.
# Upstream for overriden submodule is down.
#
# We expect:
# - overriden submodule is fetched from mirror.
# - other submodule is fetched.
bin_files_path = os.path.join(str(datafiles), "files", "bin-files", "usr")
dev_files_path = os.path.join(str(datafiles), "files", "dev-files", "usr")
upstream_bin_repodir = os.path.join(str(tmpdir), "bin-upstream")
mirror_bin_repodir = os.path.join(str(tmpdir), "bin-mirror")
upstream_bin_repo = Git(upstream_bin_repodir)
upstream_bin_repo.create(bin_files_path)
mirror_bin_repo = upstream_bin_repo.copy(mirror_bin_repodir)
dev_repodir = os.path.join(str(tmpdir), "dev-upstream")
dev_repo = Git(dev_repodir)
dev_repo.create(dev_files_path)
main_files = os.path.join(str(tmpdir), "main-files")
os.makedirs(main_files)
with open(os.path.join(main_files, "README"), "w", encoding="utf-8") as f:
f.write("TEST\n")
main_repodir = os.path.join(str(tmpdir), "main-upstream")
main_repo = Git(main_repodir)
main_repo.create(main_files)
upstream_url = "file://{}".format(upstream_bin_repo.repo)
main_repo.add_submodule("bin", url=upstream_url)
main_repo.add_submodule("dev", url="file://{}".format(dev_repo.repo))
# Unlist 'dev'.
del main_repo.submodules["dev"]
main_ref = main_repo.latest_commit()
upstream_map, repo_name = os.path.split(upstream_url)
alias = "foo"
aliased_repo = "{}:{}".format(alias, repo_name)
main_repo.submodules["bin"]["url"] = aliased_repo
full_mirror = mirror_bin_repo.source_config()["url"]
mirror_map, _ = os.path.split(full_mirror)
project_dir = os.path.join(str(tmpdir), "project")
os.makedirs(project_dir)
element_dir = os.path.join(project_dir, "elements")
element = {"kind": "import", "sources": [main_repo.source_config_extra(ref=main_ref, checkout_submodules=True)]}
element_name = "test.bst"
element_path = os.path.join(element_dir, element_name)
os.makedirs(element_dir)
_yaml.roundtrip_dump(element, element_path)
project = {
"name": "test",
"min-version": "2.0",
"element-path": "elements",
"aliases": {alias: upstream_map + "/"},
"plugins": [
{
"origin": "pip",
"package-name": "sample-plugins",
"sources": ["git"],
}
],
"mirrors": [
{
"name": "middle-earth",
"aliases": {
alias: [mirror_map + "/"],
},
}
],
}
project_file = os.path.join(project_dir, "project.conf")
_yaml.roundtrip_dump(project, project_file)
# Now make the upstream unavailable.
os.rename(upstream_bin_repo.repo, "{}.bak".format(upstream_bin_repo.repo))
result = cli.run(project=project_dir, args=["source", "fetch", element_name])
result.assert_success()
result = cli.run(project=project_dir, args=["build", element_name])
result.assert_success()
checkout = os.path.join(str(tmpdir), "checkout")
result = cli.run(project=project_dir, args=["artifact", "checkout", element_name, "--directory", checkout])
result.assert_success()
assert os.path.exists(os.path.join(checkout, "bin", "bin", "hello"))
assert os.path.exists(os.path.join(checkout, "dev", "include", "pony.h"))
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.skipif("not pip_sample_packages()", reason=SAMPLE_PACKAGES_SKIP_REASON)
def test_mirror_fallback_git_with_submodules(cli, tmpdir, datafiles):
# Main repo has mirror. But does not list submodules.
#
# We expect:
# - we will fetch submodules anyway
bin_files_path = os.path.join(str(datafiles), "files", "bin-files", "usr")
dev_files_path = os.path.join(str(datafiles), "files", "dev-files", "usr")
bin_repodir = os.path.join(str(tmpdir), "bin-repo")
bin_repo = Git(bin_repodir)
bin_repo.create(bin_files_path)
dev_repodir = os.path.join(str(tmpdir), "dev-repo")
dev_repo = Git(dev_repodir)
dev_repo.create(dev_files_path)
main_files = os.path.join(str(tmpdir), "main-files")
os.makedirs(main_files)
with open(os.path.join(main_files, "README"), "w", encoding="utf-8") as f:
f.write("TEST\n")
upstream_main_repodir = os.path.join(str(tmpdir), "main-upstream")
upstream_main_repo = Git(upstream_main_repodir)
upstream_main_repo.create(main_files)
upstream_main_repo.add_submodule("bin", url="file://{}".format(bin_repo.repo))
upstream_main_repo.add_submodule("dev", url="file://{}".format(dev_repo.repo))
# Unlist submodules.
del upstream_main_repo.submodules["bin"]
del upstream_main_repo.submodules["dev"]
upstream_main_ref = upstream_main_repo.latest_commit()
mirror_main_repodir = os.path.join(str(tmpdir), "main-mirror")
mirror_main_repo = upstream_main_repo.copy(mirror_main_repodir)
upstream_url = mirror_main_repo.source_config()["url"]
upstream_map, repo_name = os.path.split(upstream_url)
alias = "foo"
aliased_repo = "{}:{}".format(alias, repo_name)
full_mirror = mirror_main_repo.source_config()["url"]
mirror_map, _ = os.path.split(full_mirror)
project_dir = os.path.join(str(tmpdir), "project")
os.makedirs(project_dir)
element_dir = os.path.join(project_dir, "elements")
element = {
"kind": "import",
"sources": [upstream_main_repo.source_config_extra(ref=upstream_main_ref, checkout_submodules=True)],
}
element["sources"][0]["url"] = aliased_repo
element_name = "test.bst"
element_path = os.path.join(element_dir, element_name)
os.makedirs(element_dir)
_yaml.roundtrip_dump(element, element_path)
project = {
"name": "test",
"min-version": "2.0",
"element-path": "elements",
"aliases": {alias: upstream_map + "/"},
"plugins": [
{
"origin": "pip",
"package-name": "sample-plugins",
"sources": ["git"],
}
],
"mirrors": [
{
"name": "middle-earth",
"aliases": {
alias: [mirror_map + "/"],
},
}
],
}
project_file = os.path.join(project_dir, "project.conf")
_yaml.roundtrip_dump(project, project_file)
# Now make the upstream unavailable.
os.rename(upstream_main_repo.repo, "{}.bak".format(upstream_main_repo.repo))
result = cli.run(project=project_dir, args=["source", "fetch", element_name])
result.assert_success()
result = cli.run(project=project_dir, args=["build", element_name])
result.assert_success()
checkout = os.path.join(str(tmpdir), "checkout")
result = cli.run(project=project_dir, args=["artifact", "checkout", element_name, "--directory", checkout])
result.assert_success()
assert os.path.exists(os.path.join(checkout, "bin", "bin", "hello"))
assert os.path.exists(os.path.join(checkout, "dev", "include", "pony.h"))
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.usefixtures("datafiles")
def test_mirror_expand_project_and_toplevel_root(cli, tmpdir):
output_file = os.path.join(str(tmpdir), "output.txt")
project_dir = str(tmpdir)
element_dir = os.path.join(project_dir, "elements")
os.makedirs(element_dir, exist_ok=True)
element_name = "test.bst"
element_path = os.path.join(element_dir, element_name)
element = generate_element(output_file)
_yaml.roundtrip_dump(element, element_path)
project_file = os.path.join(project_dir, "project.conf")
project = {
"name": "test",
"min-version": "2.0",
"element-path": "elements",
"aliases": {
"foo": "FOO/",
"bar": "BAR/",
},
"mirrors": [
{
"name": "middle-earth",
"aliases": {
"foo": ["OOF/"],
"bar": ["RAB/"],
},
},
{
"name": "arrakis",
"aliases": {
"foo": ["%{project-root}/OFO/"],
"bar": ["%{project-root}/RBA/"],
},
},
{
"name": "oz",
"aliases": {
"foo": ["ooF/"],
"bar": ["raB/"],
},
},
],
"plugins": [{"origin": "local", "path": "sources", "sources": ["fetch_source"]}],
}
_yaml.roundtrip_dump(project, project_file)
result = cli.run(project=project_dir, args=["--default-mirror", "arrakis", "source", "fetch", element_name])
result.assert_success()
with open(output_file, encoding="utf-8") as f:
contents = f.read()
print(contents)
foo_str = os.path.join(project_dir, "OFO/repo1")
bar_str = os.path.join(project_dir, "RBA/repo2")
# Success if the expanded %{project-root} is found
assert foo_str in contents
assert bar_str in contents