This is an automated email from the ASF dual-hosted git repository. not-in-ldap pushed a commit to branch bschubert/standardize-source-tests in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit a2d1e16921df4e1382f601aa4fac8644308b7c9a Author: Benjamin Schubert <[email protected]> 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) [email protected](DATA_DIR) [email protected]("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. -# [email protected](DATA_DIR) [email protected]("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" - - [email protected](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() [email protected](DATA_DIR) [email protected]("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" [email protected](DATA_DIR) [email protected]("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") - - [email protected](DATA_DIR) [email protected]("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) [email protected](DATA_DIR) [email protected]("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()
