You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@buildstream.apache.org by tv...@apache.org on 2021/02/04 07:25:53 UTC
[buildstream] 13/15: track.py: Adapt to use the standard source
tests
This is an automated email from the ASF dual-hosted git repository.
tvb pushed a commit to branch bschubert/standardize-source-tests
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit a2d1e16921df4e1382f601aa4fac8644308b7c9a
Author: Benjamin Schubert <co...@benschubert.me>
AuthorDate: Thu Oct 8 20:58:50 2020 +0000
track.py: Adapt to use the standard source tests
---
src/buildstream/testing/_sourcetests/__init__.py | 8 +-
src/buildstream/testing/_sourcetests/track.py | 551 +++++++++++------------
2 files changed, 282 insertions(+), 277 deletions(-)
diff --git a/src/buildstream/testing/_sourcetests/__init__.py b/src/buildstream/testing/_sourcetests/__init__.py
index 2a1f26a..e0b2e41 100644
--- a/src/buildstream/testing/_sourcetests/__init__.py
+++ b/src/buildstream/testing/_sourcetests/__init__.py
@@ -24,11 +24,17 @@ from .fetch import FetchSourceTests
from .mirror import MirrorSourceTests
from .source_determinism import SourceDeterminismTests
from .track_cross_junction import TrackCrossJunctionTests
+from .track import TrackSourceTests
__all__ = ["SourceTests"]
class SourceTests(
- BuildCheckoutSourceTests, FetchSourceTests, MirrorSourceTests, SourceDeterminismTests, TrackCrossJunctionTests
+ BuildCheckoutSourceTests,
+ FetchSourceTests,
+ MirrorSourceTests,
+ SourceDeterminismTests,
+ TrackCrossJunctionTests,
+ TrackSourceTests,
):
"""Definition of standardized tests that each source should pass."""
diff --git a/src/buildstream/testing/_sourcetests/track.py b/src/buildstream/testing/_sourcetests/track.py
index 38ef217..c804518 100644
--- a/src/buildstream/testing/_sourcetests/track.py
+++ b/src/buildstream/testing/_sourcetests/track.py
@@ -25,10 +25,9 @@ import pytest
from buildstream import _yaml
from buildstream.exceptions import ErrorDomain
from .._utils import generate_junction
-from .. import create_repo
from .. import cli # pylint: disable=unused-import
from .utils import update_project_configuration
-from .utils import kind # pylint: disable=unused-import
+from .base import BaseSourceTests
# Project directory
@@ -44,304 +43,226 @@ def generate_element(repo, element_path, dep_name=None):
_yaml.roundtrip_dump(element, element_path)
-@pytest.mark.datafiles(DATA_DIR)
-@pytest.mark.parametrize("ref_storage", ["inline", "project.refs"])
-def test_track(cli, tmpdir, datafiles, ref_storage, kind):
- project = str(datafiles)
- dev_files_path = os.path.join(project, "files", "dev-files")
- element_path = os.path.join(project, "elements")
- element_name = "track-test-{}.bst".format(kind)
+class TrackSourceTests(BaseSourceTests):
+ @pytest.mark.datafiles(DATA_DIR)
+ @pytest.mark.parametrize("ref_storage", ["inline", "project.refs"])
+ def test_track(self, cli, tmpdir, datafiles, ref_storage):
+ project = str(datafiles)
+ dev_files_path = os.path.join(project, "files", "dev-files")
+ element_path = os.path.join(project, "elements")
+ element_name = "track-test-{}.bst".format(self.KIND)
- update_project_configuration(project, {"ref-storage": ref_storage})
+ update_project_configuration(project, {"ref-storage": ref_storage})
- # Create our repo object of the given source type with
- # the dev files, and then collect the initial ref.
- #
- repo = create_repo(kind, str(tmpdir))
- repo.create(dev_files_path)
-
- # Generate the element
- generate_element(repo, os.path.join(element_path, element_name))
+ # Create our repo object of the given source type with
+ # the dev files, and then collect the initial ref.
+ #
+ repo = self.REPO(str(tmpdir))
+ repo.create(dev_files_path)
- # Assert that a fetch is needed
- assert cli.get_element_state(project, element_name) == "no reference"
+ # Generate the element
+ generate_element(repo, os.path.join(element_path, element_name))
- # Now first try to track it
- result = cli.run(project=project, args=["source", "track", element_name])
- result.assert_success()
+ # Assert that a fetch is needed
+ assert cli.get_element_state(project, element_name) == "no reference"
- # And now fetch it: The Source has probably already cached the
- # latest ref locally, but it is not required to have cached
- # the associated content of the latest ref at track time, that
- # is the job of fetch.
- result = cli.run(project=project, args=["source", "fetch", element_name])
- result.assert_success()
+ # Now first try to track it
+ result = cli.run(project=project, args=["source", "track", element_name])
+ result.assert_success()
- # Assert that we are now buildable because the source is
- # now cached.
- assert cli.get_element_state(project, element_name) == "buildable"
+ # And now fetch it: The Source has probably already cached the
+ # latest ref locally, but it is not required to have cached
+ # the associated content of the latest ref at track time, that
+ # is the job of fetch.
+ result = cli.run(project=project, args=["source", "fetch", element_name])
+ result.assert_success()
- # Assert there was a project.refs created, depending on the configuration
- if ref_storage == "project.refs":
- assert os.path.exists(os.path.join(project, "project.refs"))
- else:
- assert not os.path.exists(os.path.join(project, "project.refs"))
+ # Assert that we are now buildable because the source is
+ # now cached.
+ assert cli.get_element_state(project, element_name) == "buildable"
+ # Assert there was a project.refs created, depending on the configuration
+ if ref_storage == "project.refs":
+ assert os.path.exists(os.path.join(project, "project.refs"))
+ else:
+ assert not os.path.exists(os.path.join(project, "project.refs"))
-# NOTE:
-#
-# This test checks that recursive tracking works by observing
-# element states after running a recursive tracking operation.
-#
-# However, this test is ALSO valuable as it stresses the source
-# plugins in a situation where many source plugins are operating
-# at once on the same backing repository.
-#
-# Do not change this test to use a separate 'Repo' per element
-# as that would defeat the purpose of the stress test, otherwise
-# please refactor that aspect into another test.
-#
-@pytest.mark.datafiles(DATA_DIR)
-@pytest.mark.parametrize("amount", [1, 10])
-def test_track_recurse(cli, tmpdir, datafiles, kind, amount):
- project = str(datafiles)
- dev_files_path = os.path.join(project, "files", "dev-files")
- element_path = os.path.join(project, "elements")
-
- # Try to actually launch as many fetch jobs as possible at the same time
+ # NOTE:
#
- # This stresses the Source plugins and helps to ensure that
- # they handle concurrent access to the store correctly.
- cli.configure({"scheduler": {"fetchers": amount,}})
-
- # Create our repo object of the given source type with
- # the dev files, and then collect the initial ref.
+ # This test checks that recursive tracking works by observing
+ # element states after running a recursive tracking operation.
#
- repo = create_repo(kind, str(tmpdir))
- repo.create(dev_files_path)
-
- # Write out our test targets
- element_names = []
- last_element_name = None
- for i in range(amount + 1):
- element_name = "track-test-{}-{}.bst".format(kind, i + 1)
- filename = os.path.join(element_path, element_name)
-
- element_names.append(element_name)
-
- generate_element(repo, filename, dep_name=last_element_name)
- last_element_name = element_name
-
- # Assert that a fetch is needed
- states = cli.get_element_states(project, [last_element_name])
- for element_name in element_names:
- assert states[element_name] == "no reference"
-
- # Now first try to track it
- result = cli.run(project=project, args=["source", "track", "--deps", "all", last_element_name])
- result.assert_success()
-
- # And now fetch it: The Source has probably already cached the
- # latest ref locally, but it is not required to have cached
- # the associated content of the latest ref at track time, that
- # is the job of fetch.
- result = cli.run(project=project, args=["source", "fetch", "--deps", "all", last_element_name])
- result.assert_success()
-
- # Assert that the base is buildable and the rest are waiting
- states = cli.get_element_states(project, [last_element_name])
- for element_name in element_names:
- if element_name == element_names[0]:
- assert states[element_name] == "buildable"
- else:
- assert states[element_name] == "waiting"
-
-
-@pytest.mark.datafiles(DATA_DIR)
-def test_track_recurse_except(cli, tmpdir, datafiles, kind):
- project = str(datafiles)
- dev_files_path = os.path.join(project, "files", "dev-files")
- element_path = os.path.join(project, "elements")
- element_dep_name = "track-test-dep-{}.bst".format(kind)
- element_target_name = "track-test-target-{}.bst".format(kind)
-
- # Create our repo object of the given source type with
- # the dev files, and then collect the initial ref.
+ # However, this test is ALSO valuable as it stresses the source
+ # plugins in a situation where many source plugins are operating
+ # at once on the same backing repository.
#
- repo = create_repo(kind, str(tmpdir))
- repo.create(dev_files_path)
-
- # Write out our test targets
- generate_element(repo, os.path.join(element_path, element_dep_name))
- generate_element(repo, os.path.join(element_path, element_target_name), dep_name=element_dep_name)
-
- # Assert that a fetch is needed
- states = cli.get_element_states(project, [element_target_name])
- assert states[element_dep_name] == "no reference"
- assert states[element_target_name] == "no reference"
-
- # Now first try to track it
- result = cli.run(
- project=project, args=["source", "track", "--deps", "all", "--except", element_dep_name, element_target_name]
- )
- result.assert_success()
-
- # And now fetch it: The Source has probably already cached the
- # latest ref locally, but it is not required to have cached
- # the associated content of the latest ref at track time, that
- # is the job of fetch.
- result = cli.run(project=project, args=["source", "fetch", "--deps", "none", element_target_name])
- result.assert_success()
-
- # Assert that the dependency is buildable and the target is waiting
- states = cli.get_element_states(project, [element_target_name])
- assert states[element_dep_name] == "no reference"
- assert states[element_target_name] == "waiting"
+ # Do not change this test to use a separate 'Repo' per element
+ # as that would defeat the purpose of the stress test, otherwise
+ # please refactor that aspect into another test.
+ #
+ @pytest.mark.datafiles(DATA_DIR)
+ @pytest.mark.parametrize("amount", [1, 10])
+ def test_track_recurse(self, cli, tmpdir, datafiles, amount):
+ project = str(datafiles)
+ dev_files_path = os.path.join(project, "files", "dev-files")
+ element_path = os.path.join(project, "elements")
+
+ # Try to actually launch as many fetch jobs as possible at the same time
+ #
+ # This stresses the Source plugins and helps to ensure that
+ # they handle concurrent access to the store correctly.
+ cli.configure({"scheduler": {"fetchers": amount,}})
+
+ # Create our repo object of the given source type with
+ # the dev files, and then collect the initial ref.
+ #
+ repo = self.REPO(str(tmpdir))
+ repo.create(dev_files_path)
+
+ # Write out our test targets
+ element_names = []
+ last_element_name = None
+ for i in range(amount + 1):
+ element_name = "track-test-{}-{}.bst".format(self.KIND, i + 1)
+ filename = os.path.join(element_path, element_name)
+
+ element_names.append(element_name)
+
+ generate_element(repo, filename, dep_name=last_element_name)
+ last_element_name = element_name
+
+ # Assert that a fetch is needed
+ states = cli.get_element_states(project, [last_element_name])
+ for element_name in element_names:
+ assert states[element_name] == "no reference"
+
+ # Now first try to track it
+ result = cli.run(project=project, args=["source", "track", "--deps", "all", last_element_name])
+ result.assert_success()
+ # And now fetch it: The Source has probably already cached the
+ # latest ref locally, but it is not required to have cached
+ # the associated content of the latest ref at track time, that
+ # is the job of fetch.
+ result = cli.run(project=project, args=["source", "fetch", "--deps", "all", last_element_name])
+ result.assert_success()
-@pytest.mark.datafiles(DATA_DIR)
-@pytest.mark.parametrize("ref_storage", ["inline", "project.refs"])
-def test_cross_junction(cli, tmpdir, datafiles, ref_storage, kind):
- project = str(datafiles)
- subproject_path = os.path.join(project, "files", "sub-project")
- junction_path = os.path.join(project, "elements", "junction.bst")
- etc_files = os.path.join(subproject_path, "files", "etc-files")
- repo_element_path = os.path.join(subproject_path, "elements", "import-etc-repo.bst")
+ # Assert that the base is buildable and the rest are waiting
+ states = cli.get_element_states(project, [last_element_name])
+ for element_name in element_names:
+ if element_name == element_names[0]:
+ assert states[element_name] == "buildable"
+ else:
+ assert states[element_name] == "waiting"
+
+ @pytest.mark.datafiles(DATA_DIR)
+ def test_track_recurse_except(self, cli, tmpdir, datafiles):
+ project = str(datafiles)
+ dev_files_path = os.path.join(project, "files", "dev-files")
+ element_path = os.path.join(project, "elements")
+ element_dep_name = "track-test-dep-{}.bst".format(self.KIND)
+ element_target_name = "track-test-target-{}.bst".format(self.KIND)
+
+ # Create our repo object of the given source type with
+ # the dev files, and then collect the initial ref.
+ #
+ repo = self.REPO(str(tmpdir))
+ repo.create(dev_files_path)
+
+ # Write out our test targets
+ generate_element(repo, os.path.join(element_path, element_dep_name))
+ generate_element(repo, os.path.join(element_path, element_target_name), dep_name=element_dep_name)
+
+ # Assert that a fetch is needed
+ states = cli.get_element_states(project, [element_target_name])
+ assert states[element_dep_name] == "no reference"
+ assert states[element_target_name] == "no reference"
+
+ # Now first try to track it
+ result = cli.run(
+ project=project,
+ args=["source", "track", "--deps", "all", "--except", element_dep_name, element_target_name],
+ )
+ result.assert_success()
- update_project_configuration(project, {"ref-storage": ref_storage})
+ # And now fetch it: The Source has probably already cached the
+ # latest ref locally, but it is not required to have cached
+ # the associated content of the latest ref at track time, that
+ # is the job of fetch.
+ result = cli.run(project=project, args=["source", "fetch", "--deps", "none", element_target_name])
+ result.assert_success()
- repo = create_repo(kind, str(tmpdir.join("element_repo")))
- repo.create(etc_files)
+ # Assert that the dependency is buildable and the target is waiting
+ states = cli.get_element_states(project, [element_target_name])
+ assert states[element_dep_name] == "no reference"
+ assert states[element_target_name] == "waiting"
- generate_element(repo, repo_element_path)
+ @pytest.mark.datafiles(DATA_DIR)
+ @pytest.mark.parametrize("ref_storage", ["inline", "project.refs"])
+ def test_cross_junction(self, cli, tmpdir, datafiles, ref_storage):
+ project = str(datafiles)
+ subproject_path = os.path.join(project, "files", "sub-project")
+ junction_path = os.path.join(project, "elements", "junction.bst")
+ etc_files = os.path.join(subproject_path, "files", "etc-files")
+ repo_element_path = os.path.join(subproject_path, "elements", "import-etc-repo.bst")
- generate_junction(str(tmpdir.join("junction_repo")), subproject_path, junction_path, store_ref=False)
+ update_project_configuration(project, {"ref-storage": ref_storage})
- # Track the junction itself first.
- result = cli.run(project=project, args=["source", "track", "junction.bst"])
- result.assert_success()
+ repo = self.REPO(str(tmpdir.join("element_repo")))
+ repo.create(etc_files)
- assert cli.get_element_state(project, "junction.bst:import-etc-repo.bst") == "no reference"
+ generate_element(repo, repo_element_path)
- # Track the cross junction element. -J is not given, it is implied.
- result = cli.run(project=project, args=["source", "track", "junction.bst:import-etc-repo.bst"])
+ generate_junction(str(tmpdir.join("junction_repo")), subproject_path, junction_path, store_ref=False)
- if ref_storage == "inline":
- # This is not allowed to track cross junction without project.refs.
- result.assert_main_error(ErrorDomain.PIPELINE, "untrackable-sources")
- else:
+ # Track the junction itself first.
+ result = cli.run(project=project, args=["source", "track", "junction.bst"])
result.assert_success()
- assert cli.get_element_state(project, "junction.bst:import-etc-repo.bst") == "buildable"
-
- assert os.path.exists(os.path.join(project, "project.refs"))
-
+ assert cli.get_element_state(project, "junction.bst:import-etc-repo.bst") == "no reference"
-@pytest.mark.datafiles(DATA_DIR)
-@pytest.mark.parametrize("ref_storage", ["inline", "project.refs"])
-def test_track_include(cli, tmpdir, datafiles, ref_storage, kind):
- project = str(datafiles)
- dev_files_path = os.path.join(project, "files", "dev-files")
- element_path = os.path.join(project, "elements")
- element_name = "track-test-{}.bst".format(kind)
+ # Track the cross junction element. -J is not given, it is implied.
+ result = cli.run(project=project, args=["source", "track", "junction.bst:import-etc-repo.bst"])
- update_project_configuration(project, {"ref-storage": ref_storage})
+ if ref_storage == "inline":
+ # This is not allowed to track cross junction without project.refs.
+ result.assert_main_error(ErrorDomain.PIPELINE, "untrackable-sources")
+ else:
+ result.assert_success()
- # Create our repo object of the given source type with
- # the dev files, and then collect the initial ref.
- #
- repo = create_repo(kind, str(tmpdir))
- ref = repo.create(dev_files_path)
-
- # Generate the element
- element = {"kind": "import", "(@)": ["elements/sources.yml"]}
- sources = {"sources": [repo.source_config()]}
-
- _yaml.roundtrip_dump(element, os.path.join(element_path, element_name))
- _yaml.roundtrip_dump(sources, os.path.join(element_path, "sources.yml"))
-
- # Assert that a fetch is needed
- assert cli.get_element_state(project, element_name) == "no reference"
-
- # Now first try to track it
- result = cli.run(project=project, args=["source", "track", element_name])
- result.assert_success()
-
- # And now fetch it: The Source has probably already cached the
- # latest ref locally, but it is not required to have cached
- # the associated content of the latest ref at track time, that
- # is the job of fetch.
- result = cli.run(project=project, args=["source", "fetch", element_name])
- result.assert_success()
-
- # Assert that we are now buildable because the source is
- # now cached.
- assert cli.get_element_state(project, element_name) == "buildable"
-
- # Assert there was a project.refs created, depending on the configuration
- if ref_storage == "project.refs":
- assert os.path.exists(os.path.join(project, "project.refs"))
- else:
- assert not os.path.exists(os.path.join(project, "project.refs"))
-
- new_sources = _yaml.load(os.path.join(element_path, "sources.yml"), shortname="sources.yml")
-
- # Get all of the sources
- assert "sources" in new_sources
- sources_list = new_sources.get_sequence("sources")
- assert len(sources_list) == 1
-
- # Get the first source from the sources list
- new_source = sources_list.mapping_at(0)
- assert "ref" in new_source
- assert ref == new_source.get_str("ref")
-
-
-@pytest.mark.datafiles(DATA_DIR)
-@pytest.mark.parametrize("ref_storage", ["inline", "project.refs"])
-def test_track_include_junction(cli, tmpdir, datafiles, ref_storage, kind):
- project = str(datafiles)
- dev_files_path = os.path.join(project, "files", "dev-files")
- element_path = os.path.join(project, "elements")
- element_name = "track-test-{}.bst".format(kind)
- subproject_path = os.path.join(project, "files", "sub-project")
- sub_element_path = os.path.join(subproject_path, "elements")
- junction_path = os.path.join(element_path, "junction.bst")
-
- update_project_configuration(project, {"ref-storage": ref_storage})
-
- # Create our repo object of the given source type with
- # the dev files, and then collect the initial ref.
- #
- repo = create_repo(kind, str(tmpdir.join("element_repo")))
- repo.create(dev_files_path)
+ assert cli.get_element_state(project, "junction.bst:import-etc-repo.bst") == "buildable"
- # Generate the element
- element = {"kind": "import", "(@)": ["junction.bst:elements/sources.yml"]}
- sources = {"sources": [repo.source_config()]}
+ assert os.path.exists(os.path.join(project, "project.refs"))
- _yaml.roundtrip_dump(element, os.path.join(element_path, element_name))
- _yaml.roundtrip_dump(sources, os.path.join(sub_element_path, "sources.yml"))
+ @pytest.mark.datafiles(DATA_DIR)
+ @pytest.mark.parametrize("ref_storage", ["inline", "project.refs"])
+ def test_track_include(self, cli, tmpdir, datafiles, ref_storage):
+ project = str(datafiles)
+ dev_files_path = os.path.join(project, "files", "dev-files")
+ element_path = os.path.join(project, "elements")
+ element_name = "track-test-{}.bst".format(self.KIND)
- generate_junction(str(tmpdir.join("junction_repo")), subproject_path, junction_path, store_ref=True)
+ update_project_configuration(project, {"ref-storage": ref_storage})
- result = cli.run(project=project, args=["source", "track", "junction.bst"])
- result.assert_success()
+ # Create our repo object of the given source type with
+ # the dev files, and then collect the initial ref.
+ #
+ repo = self.REPO(str(tmpdir))
+ ref = repo.create(dev_files_path)
- # Assert that a fetch is needed
- assert cli.get_element_state(project, element_name) == "no reference"
+ # Generate the element
+ element = {"kind": "import", "(@)": ["elements/sources.yml"]}
+ sources = {"sources": [repo.source_config()]}
- # Now first try to track it
- result = cli.run(project=project, args=["source", "track", element_name])
+ _yaml.roundtrip_dump(element, os.path.join(element_path, element_name))
+ _yaml.roundtrip_dump(sources, os.path.join(element_path, "sources.yml"))
- # Assert there was a project.refs created, depending on the configuration
- if ref_storage == "inline":
- # FIXME: We should expect an error. But only a warning is emitted
- # result.assert_main_error(ErrorDomain.SOURCE, 'tracking-junction-fragment')
+ # Assert that a fetch is needed
+ assert cli.get_element_state(project, element_name) == "no reference"
- assert "junction.bst:elements/sources.yml: Cannot track source in a fragment from a junction" in result.stderr
- else:
- assert os.path.exists(os.path.join(project, "project.refs"))
+ # Now first try to track it
+ result = cli.run(project=project, args=["source", "track", element_name])
+ result.assert_success()
# And now fetch it: The Source has probably already cached the
# latest ref locally, but it is not required to have cached
@@ -354,18 +275,96 @@ def test_track_include_junction(cli, tmpdir, datafiles, ref_storage, kind):
# now cached.
assert cli.get_element_state(project, element_name) == "buildable"
+ # Assert there was a project.refs created, depending on the configuration
+ if ref_storage == "project.refs":
+ assert os.path.exists(os.path.join(project, "project.refs"))
+ else:
+ assert not os.path.exists(os.path.join(project, "project.refs"))
+
+ new_sources = _yaml.load(os.path.join(element_path, "sources.yml"), shortname="sources.yml")
+
+ # Get all of the sources
+ assert "sources" in new_sources
+ sources_list = new_sources.get_sequence("sources")
+ assert len(sources_list) == 1
+
+ # Get the first source from the sources list
+ new_source = sources_list.mapping_at(0)
+ assert "ref" in new_source
+ assert ref == new_source.get_str("ref")
+
+ @pytest.mark.datafiles(DATA_DIR)
+ @pytest.mark.parametrize("ref_storage", ["inline", "project.refs"])
+ def test_track_include_junction(self, cli, tmpdir, datafiles, ref_storage):
+ project = str(datafiles)
+ dev_files_path = os.path.join(project, "files", "dev-files")
+ element_path = os.path.join(project, "elements")
+ element_name = "track-test-{}.bst".format(self.KIND)
+ subproject_path = os.path.join(project, "files", "sub-project")
+ sub_element_path = os.path.join(subproject_path, "elements")
+ junction_path = os.path.join(element_path, "junction.bst")
+
+ update_project_configuration(project, {"ref-storage": ref_storage})
+
+ # Create our repo object of the given source type with
+ # the dev files, and then collect the initial ref.
+ #
+ repo = self.REPO(str(tmpdir.join("element_repo")))
+ repo.create(dev_files_path)
-@pytest.mark.datafiles(DATA_DIR)
-@pytest.mark.parametrize("ref_storage", ["inline", "project.refs"])
-def test_track_junction_included(cli, tmpdir, datafiles, ref_storage, kind):
- project = str(datafiles)
- element_path = os.path.join(project, "elements")
- subproject_path = os.path.join(project, "files", "sub-project")
- junction_path = os.path.join(element_path, "junction.bst")
+ # Generate the element
+ element = {"kind": "import", "(@)": ["junction.bst:elements/sources.yml"]}
+ sources = {"sources": [repo.source_config()]}
- update_project_configuration(project, {"ref-storage": ref_storage, "(@)": ["junction.bst:test.yml"]})
+ _yaml.roundtrip_dump(element, os.path.join(element_path, element_name))
+ _yaml.roundtrip_dump(sources, os.path.join(sub_element_path, "sources.yml"))
- generate_junction(str(tmpdir.join("junction_repo")), subproject_path, junction_path, store_ref=False)
+ generate_junction(str(tmpdir.join("junction_repo")), subproject_path, junction_path, store_ref=True)
- result = cli.run(project=project, args=["source", "track", "junction.bst"])
- result.assert_success()
+ result = cli.run(project=project, args=["source", "track", "junction.bst"])
+ result.assert_success()
+
+ # Assert that a fetch is needed
+ assert cli.get_element_state(project, element_name) == "no reference"
+
+ # Now first try to track it
+ result = cli.run(project=project, args=["source", "track", element_name])
+
+ # Assert there was a project.refs created, depending on the configuration
+ if ref_storage == "inline":
+ # FIXME: We should expect an error. But only a warning is emitted
+ # result.assert_main_error(ErrorDomain.SOURCE, 'tracking-junction-fragment')
+
+ assert (
+ "junction.bst:elements/sources.yml: Cannot track source in a fragment from a junction" in result.stderr
+ )
+ else:
+ assert os.path.exists(os.path.join(project, "project.refs"))
+
+ # And now fetch it: The Source has probably already cached the
+ # latest ref locally, but it is not required to have cached
+ # the associated content of the latest ref at track time, that
+ # is the job of fetch.
+ result = cli.run(project=project, args=["source", "fetch", element_name])
+ result.assert_success()
+
+ # Assert that we are now buildable because the source is
+ # now cached.
+ assert cli.get_element_state(project, element_name) == "buildable"
+
+ @pytest.mark.datafiles(DATA_DIR)
+ @pytest.mark.parametrize("ref_storage", ["inline", "project.refs"])
+ def test_track_junction_included(self, cli, tmpdir, datafiles, ref_storage):
+ # XXX: Not really sure what this is testing, an it doesn't seem to use
+ # the kind at any point?
+ project = str(datafiles)
+ element_path = os.path.join(project, "elements")
+ subproject_path = os.path.join(project, "files", "sub-project")
+ junction_path = os.path.join(element_path, "junction.bst")
+
+ update_project_configuration(project, {"ref-storage": ref_storage, "(@)": ["junction.bst:test.yml"]})
+
+ generate_junction(str(tmpdir.join("junction_repo")), subproject_path, junction_path, store_ref=False)
+
+ result = cli.run(project=project, args=["source", "track", "junction.bst"])
+ result.assert_success()