| # Pylint doesn't play well with fixtures and dependency injection from pytest |
| # pylint: disable=redefined-outer-name |
| |
| import stat |
| import os |
| import re |
| import pytest |
| |
| from buildstream.testing import create_repo, generate_project |
| from buildstream.testing import cli # pylint: disable=unused-import |
| from buildstream.testing._utils.site import have_subsecond_mtime |
| from buildstream.exceptions import ErrorDomain, LoadErrorReason |
| from buildstream import _yaml |
| from tests.testutils import generate_junction |
| from . import configure_project |
| |
| # Project directory |
| TOP_DIR = os.path.dirname(os.path.realpath(__file__)) |
| DATA_DIR = os.path.join(TOP_DIR, "project") |
| |
| |
| def generate_element(repo, element_path, dep_name=None): |
| element = {"kind": "import", "sources": [repo.source_config()]} |
| if dep_name: |
| element["depends"] = [dep_name] |
| |
| _yaml.roundtrip_dump(element, element_path) |
| |
| |
| @pytest.mark.datafiles(DATA_DIR) |
| def test_track_single(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" |
| element_target_name = "track-test-target.bst" |
| |
| # Create our repo object of the given source type with |
| # the dev files, and then collect the initial ref. |
| # |
| repo = create_repo("git", 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 tracking is needed for both elements |
| states = cli.get_element_states(project, [element_target_name]) |
| assert states == { |
| element_dep_name: "no reference", |
| element_target_name: "no reference", |
| } |
| |
| # Now first try to track only one element |
| result = cli.run(project=project, args=["source", "track", "--deps", "none", element_target_name]) |
| result.assert_success() |
| |
| # And now fetch it |
| result = cli.run(project=project, args=["source", "fetch", "--deps", "none", element_target_name]) |
| result.assert_success() |
| |
| # Assert that the dependency is waiting and the target has still never been tracked |
| states = cli.get_element_states(project, [element_target_name]) |
| assert states == { |
| element_dep_name: "no reference", |
| element_target_name: "waiting", |
| } |
| |
| |
| @pytest.mark.datafiles(os.path.join(TOP_DIR)) |
| @pytest.mark.parametrize("ref_storage", [("inline"), ("project-refs")]) |
| def test_track_optional(cli, tmpdir, datafiles, ref_storage): |
| project = os.path.join(datafiles.dirname, datafiles.basename, "track-optional-" + ref_storage) |
| dev_files_path = os.path.join(project, "files") |
| element_path = os.path.join(project, "target.bst") |
| |
| # Create our repo object of the given source type with |
| # the dev files, and then collect the initial ref. |
| # |
| repo = create_repo("git", str(tmpdir)) |
| repo.create(dev_files_path) |
| |
| # Now create an optional test branch and add a commit to that, |
| # so two branches with different heads now exist. |
| # |
| repo.branch("test") |
| repo.add_commit() |
| |
| # Substitute the {repo} for the git repo we created |
| with open(element_path) as f: |
| target_bst = f.read() |
| target_bst = target_bst.format(repo=repo.repo) |
| with open(element_path, "w") as f: |
| f.write(target_bst) |
| |
| # First track for both options |
| # |
| # We want to track and persist the ref separately in this test |
| # |
| result = cli.run(project=project, args=["--option", "test", "False", "source", "track", "target.bst"]) |
| result.assert_success() |
| result = cli.run(project=project, args=["--option", "test", "True", "source", "track", "target.bst"]) |
| result.assert_success() |
| |
| # Now fetch the key for both options |
| # |
| result = cli.run( |
| project=project, |
| args=["--option", "test", "False", "show", "--deps", "none", "--format", "%{key}", "target.bst"], |
| ) |
| result.assert_success() |
| master_key = result.output |
| |
| result = cli.run( |
| project=project, |
| args=["--option", "test", "True", "show", "--deps", "none", "--format", "%{key}", "target.bst"], |
| ) |
| result.assert_success() |
| test_key = result.output |
| |
| # Assert that the keys are different when having |
| # tracked separate branches |
| assert test_key != master_key |
| |
| |
| # Test all possible choices of the `--deps` option. |
| # |
| # NOTE: Elements used in this test must have sources that are trackable and do |
| # not have a reference already. The kinds of the sources do not matter so |
| # long as they can be tracked from somewhere. |
| # Currently we use remote sources for this purpose. |
| # |
| @pytest.mark.datafiles(os.path.join(TOP_DIR, "source-track")) |
| @pytest.mark.parametrize( |
| "deps, expected_states", |
| [ |
| ("build", ("no reference", "buildable", "no reference")), |
| ("none", ("waiting", "no reference", "no reference")), |
| ("run", ("waiting", "no reference", "buildable")), |
| ("all", ("waiting", "buildable", "buildable")), |
| ], |
| ) |
| def test_track_deps(cli, datafiles, deps, expected_states): |
| project = str(datafiles) |
| generate_project(project, {"aliases": {"project-root": "file:///" + project}}) |
| |
| target = "bananas.bst" |
| build_dep = "apples.bst" |
| runtime_dep = "oranges.bst" |
| |
| # Assert that none of the sources have a reference |
| states = cli.get_element_states(project, [target, build_dep, runtime_dep]) |
| assert all([state == "no reference" for state in states.values()]) |
| |
| # Now track the specified sources |
| result = cli.run(project=project, args=["source", "track", "--deps", deps, target]) |
| result.assert_success() |
| |
| # Finally assert that we have tracked _only_ the desired sources |
| states = cli.get_element_states(project, [target, build_dep, runtime_dep]) |
| states_flattened = (states[target], states[build_dep], states[runtime_dep]) |
| assert states_flattened == expected_states |
| |
| |
| @pytest.mark.datafiles(os.path.join(TOP_DIR, "track-cross-junction")) |
| @pytest.mark.parametrize("cross_junction", [("cross"), ("nocross")]) |
| @pytest.mark.parametrize("ref_storage", [("inline"), ("project.refs")]) |
| def test_track_cross_junction(cli, tmpdir, datafiles, cross_junction, ref_storage): |
| project = str(datafiles) |
| dev_files_path = os.path.join(project, "files") |
| target_path = os.path.join(project, "target.bst") |
| subtarget_path = os.path.join(project, "subproject", "subtarget.bst") |
| |
| # Create our repo object of the given source type with |
| # the dev files, and then collect the initial ref. |
| # |
| repo = create_repo("git", str(tmpdir)) |
| repo.create(dev_files_path) |
| |
| # Generate two elements using the git source, one in |
| # the main project and one in the subproject. |
| generate_element(repo, target_path, dep_name="subproject.bst") |
| generate_element(repo, subtarget_path) |
| |
| # Generate project.conf |
| # |
| project_conf = {"name": "test", "min-version": "2.0", "ref-storage": ref_storage} |
| _yaml.roundtrip_dump(project_conf, os.path.join(project, "project.conf")) |
| |
| # |
| # FIXME: This can be simplified when we have support |
| # for addressing of junctioned elements. |
| # |
| def get_subproject_element_state(): |
| result = cli.run(project=project, args=["show", "--deps", "all", "--format", "%{name}|%{state}", "target.bst"]) |
| result.assert_success() |
| |
| # Create two dimentional list of the result, |
| # first line should be the junctioned element |
| lines = [line.split("|") for line in result.output.splitlines()] |
| assert lines[0][0] == "subproject-junction.bst:subtarget.bst" |
| return lines[0][1] |
| |
| # |
| # Assert that we have no reference yet for the cross junction element |
| # |
| assert get_subproject_element_state() == "no reference" |
| |
| # Track recursively across the junction |
| args = ["source", "track", "--deps", "all"] |
| if cross_junction == "cross": |
| args += ["--cross-junctions"] |
| args += ["target.bst"] |
| |
| result = cli.run(project=project, args=args) |
| |
| if ref_storage == "inline": |
| |
| if cross_junction == "cross": |
| # |
| # Cross junction tracking is not allowed when the toplevel project |
| # is using inline ref storage. |
| # |
| result.assert_main_error(ErrorDomain.STREAM, "untrackable-sources") |
| else: |
| # |
| # No cross juction tracking was requested |
| # |
| result.assert_success() |
| assert get_subproject_element_state() == "no reference" |
| else: |
| # |
| # Tracking is allowed with project.refs ref storage |
| # |
| result.assert_success() |
| |
| # |
| # If cross junction tracking was enabled, we should now be buildable |
| # |
| if cross_junction == "cross": |
| assert get_subproject_element_state() == "buildable" |
| else: |
| assert get_subproject_element_state() == "no reference" |
| |
| |
| @pytest.mark.datafiles(os.path.join(TOP_DIR, "consistencyerror")) |
| def test_track_consistency_error(cli, datafiles): |
| project = str(datafiles) |
| |
| # Track the element causing a consistency error in `is_cached()` |
| result = cli.run(project=project, args=["source", "track", "error.bst"]) |
| |
| # We expect tracking to succeed as `is_cached()` is not required for tracking. |
| result.assert_success() |
| |
| |
| @pytest.mark.datafiles(os.path.join(TOP_DIR, "consistencyerror")) |
| def test_track_consistency_bug(cli, datafiles): |
| project = str(datafiles) |
| |
| # Track the element causing an unhandled exception in `is_cached()` |
| result = cli.run(project=project, args=["source", "track", "bug.bst"]) |
| |
| # We expect tracking to succeed as `is_cached()` is not required for tracking. |
| result.assert_success() |
| |
| |
| @pytest.mark.datafiles(DATA_DIR) |
| @pytest.mark.parametrize("ref_storage", [("inline"), ("project.refs")]) |
| def test_inconsistent_junction(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") |
| element_path = os.path.join(project, "elements", "junction-dep.bst") |
| |
| configure_project(project, {"ref-storage": ref_storage}) |
| |
| # Create a repo to hold the subproject and generate a junction element for it |
| generate_junction(tmpdir, subproject_path, junction_path, store_ref=False) |
| |
| # Create a stack element to depend on a cross junction element |
| # |
| element = {"kind": "stack", "depends": [{"junction": "junction.bst", "filename": "import-etc.bst"}]} |
| _yaml.roundtrip_dump(element, element_path) |
| |
| # Now try to track it, this will bail with the appropriate error |
| # informing the user to track the junction first |
| result = cli.run(project=project, args=["source", "track", "junction-dep.bst"]) |
| result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.SUBPROJECT_INCONSISTENT) |
| |
| # Assert that we have the expected provenance encoded into the error |
| element_node = _yaml.load(element_path, shortname="junction-dep.bst") |
| ref_node = element_node.get_sequence("depends").mapping_at(0) |
| provenance = ref_node.get_provenance() |
| assert str(provenance) in result.stderr |
| |
| |
| @pytest.mark.datafiles(DATA_DIR) |
| @pytest.mark.parametrize("ref_storage", [("inline"), ("project.refs")]) |
| def test_junction_element(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") |
| element_path = os.path.join(project, "elements", "junction-dep.bst") |
| |
| configure_project(project, {"ref-storage": ref_storage}) |
| |
| # Create a repo to hold the subproject and generate a junction element for it |
| generate_junction(tmpdir, subproject_path, junction_path, store_ref=False) |
| |
| # Create a stack element to depend on a cross junction element |
| # |
| element = {"kind": "stack", "depends": [{"junction": "junction.bst", "filename": "import-etc.bst"}]} |
| _yaml.roundtrip_dump(element, element_path) |
| |
| # First demonstrate that showing the pipeline yields an error |
| result = cli.run(project=project, args=["show", "junction-dep.bst"]) |
| result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.SUBPROJECT_INCONSISTENT) |
| |
| # Assert that we have the expected provenance encoded into the error |
| element_node = _yaml.load(element_path, shortname="junction-dep.bst") |
| ref_node = element_node.get_sequence("depends").mapping_at(0) |
| provenance = ref_node.get_provenance() |
| assert str(provenance) in result.stderr |
| |
| # Now track the junction itself |
| result = cli.run(project=project, args=["source", "track", "junction.bst"]) |
| result.assert_success() |
| |
| # Now assert element state (via bst show under the hood) of the dep again |
| assert cli.get_element_state(project, "junction-dep.bst") == "waiting" |
| |
| |
| @pytest.mark.datafiles(DATA_DIR) |
| def test_track_error_cannot_write_file(cli, tmpdir, datafiles): |
| if os.geteuid() == 0: |
| pytest.skip("This is not testable with root permissions") |
| |
| 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" |
| |
| configure_project(project, {"ref-storage": "inline"}) |
| |
| repo = create_repo("git", str(tmpdir)) |
| repo.create(dev_files_path) |
| |
| element_full_path = os.path.join(element_path, element_name) |
| generate_element(repo, element_full_path) |
| |
| st = os.stat(element_path) |
| try: |
| read_mask = stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH |
| os.chmod(element_path, stat.S_IMODE(st.st_mode) & ~read_mask) |
| |
| result = cli.run(project=project, args=["source", "track", element_name]) |
| result.assert_main_error(ErrorDomain.STREAM, None) |
| result.assert_task_error(ErrorDomain.SOURCE, "save-ref-error") |
| finally: |
| os.chmod(element_path, stat.S_IMODE(st.st_mode)) |
| |
| |
| @pytest.mark.datafiles(DATA_DIR) |
| def test_no_needless_overwrite(cli, tmpdir, datafiles): |
| project = os.path.join(datafiles.dirname, datafiles.basename) |
| dev_files_path = os.path.join(project, "files", "dev-files") |
| element_path = os.path.join(project, "elements") |
| target = "track-test-target.bst" |
| |
| # Skip this test if we do not have support for subsecond precision mtimes |
| # |
| if not have_subsecond_mtime(project): |
| pytest.skip("Filesystem does not support subsecond mtime precision: {}".format(project)) |
| |
| # Create our repo object of the given source type with |
| # the dev files, and then collect the initial ref. |
| # |
| repo = create_repo("git", str(tmpdir)) |
| repo.create(dev_files_path) |
| |
| # Write out our test target and assert it exists |
| generate_element(repo, os.path.join(element_path, target)) |
| path_to_target = os.path.join(element_path, target) |
| assert os.path.exists(path_to_target) |
| creation_mtime = os.path.getmtime(path_to_target) |
| |
| # Assert tracking is needed |
| states = cli.get_element_states(project, [target]) |
| assert states[target] == "no reference" |
| |
| # Perform the track |
| result = cli.run(project=project, args=["source", "track", target]) |
| result.assert_success() |
| |
| track1_mtime = os.path.getmtime(path_to_target) |
| |
| assert creation_mtime != track1_mtime |
| |
| # Now (needlessly) track again |
| result = cli.run(project=project, args=["source", "track", target]) |
| result.assert_success() |
| |
| track2_mtime = os.path.getmtime(path_to_target) |
| |
| assert track1_mtime == track2_mtime |
| |
| |
| # Regression test for https://gitlab.com/BuildStream/buildstream/-/issues/1265. |
| # Ensure that we can successfully track a `.bst` file that has comments inside |
| # one of our YAML directives (like list append, prepend etc). |
| @pytest.mark.datafiles(os.path.join(TOP_DIR, "source-track")) |
| def test_track_with_comments(cli, datafiles): |
| project = str(datafiles) |
| generate_project(project, {"aliases": {"project-root": "file:///" + project}}) |
| |
| target = "comments.bst" |
| |
| # Assert that it needs to be tracked |
| assert cli.get_element_state(project, target) == "no reference" |
| |
| # Track and fetch the sources |
| result = cli.run(project=project, args=["source", "track", target]) |
| result.assert_success() |
| result = cli.run(project=project, args=["source", "fetch", target]) |
| result.assert_success() |
| |
| # Assert that the sources are cached |
| assert cli.get_element_state(project, target) == "buildable" |
| |
| |
| # Test that elements which contain only sources which do not implement Source.track() |
| # produce a SKIP message in the logs instead of a SUCCESS message when tracking the |
| # element. |
| # |
| # Also test the same for an open workspace, which would be trackable if the |
| # workspace was not open. |
| # |
| # Also test that elements which do not have any sources do not produce any messages at all, |
| # as these jobs are discarded before ever processing. |
| # |
| @pytest.mark.datafiles(DATA_DIR) |
| def test_track_skip(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" |
| element_workspace_name = "track-test-workspace.bst" |
| element_target_name = "track-test-target.bst" |
| workspace_dir = os.path.join(str(tmpdir), "workspace") |
| |
| # Generate an import element with some local source plugins, these |
| # do not implement track() and thus can be skipped. |
| # |
| element = { |
| "kind": "import", |
| "sources": [ |
| {"kind": "local", "path": "files/dev-files", "directory": "/foo"}, |
| {"kind": "local", "path": "files/dev-files", "directory": "/bar"}, |
| ], |
| } |
| _yaml.roundtrip_dump(element, os.path.join(element_path, element_dep_name)) |
| |
| # Generate a regular import element which will have a workspace open |
| # |
| repo = create_repo("tar", str(tmpdir)) |
| repo.create(dev_files_path) |
| generate_element(repo, os.path.join(element_path, element_workspace_name)) |
| |
| # Generate a stack element which depends on the import of local files |
| # |
| # Stack elements do not have any sources, as such they are also skipped. |
| # |
| element = { |
| "kind": "stack", |
| "depends": [element_dep_name, element_workspace_name], |
| } |
| _yaml.roundtrip_dump(element, os.path.join(element_path, element_target_name)) |
| |
| # First track and fetch the workspace element |
| result = cli.run(project=project, args=["source", "track", "--deps", "none", element_workspace_name]) |
| result.assert_success() |
| result = cli.run(project=project, args=["source", "fetch", "--deps", "none", element_workspace_name]) |
| result.assert_success() |
| |
| # Open the workspace so it really is a workspace |
| result = cli.run(project=project, args=["workspace", "open", "--directory", workspace_dir, element_workspace_name]) |
| result.assert_success() |
| |
| # Now run track on the stack and all the deps |
| result = cli.run(project=project, args=["source", "track", "--deps", "all", element_target_name]) |
| result.assert_success() |
| |
| # Assert we got the expected skip messages |
| pattern = r"\[.*track:track-test-dep\.bst.*\] SKIPPED" |
| assert len(re.findall(pattern, result.stderr, re.MULTILINE)) == 1 |
| pattern = r"\[.*track:track-test-workspace\.bst.*\] SKIPPED" |
| assert len(re.findall(pattern, result.stderr, re.MULTILINE)) == 1 |
| |
| # For now, we expect to not see the job for stack elements |
| # |
| # This may be revisited, need to consider if we should emit |
| # START/SKIPPED message pairs for jobs which were assessed to |
| # be unneeded before ever processing. |
| # |
| pattern = r"\[.*track:track-test-target\.bst.*\]" |
| assert len(re.findall(pattern, result.stderr, re.MULTILINE)) == 0 |