tests/frontend/workspace.py

#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#
#  Authors: Tristan Van Berkom <tristan.vanberkom@codethink.co.uk>
#           Tristan Maat <tristan.maat@codethink.co.uk>
#           Chandan Singh <csingh43@bloomberg.net>
#           Phillip Smyth <phillip.smyth@codethink.co.uk>
#           Jonathan Maw <jonathan.maw@codethink.co.uk>
#           Richard Maw <richard.maw@codethink.co.uk>
#           William Salmon <will.salmon@codethink.co.uk>
#

# Pylint doesn't play well with fixtures and dependency injection from pytest
# pylint: disable=redefined-outer-name

import os
import stat
import shutil
import tempfile

import pytest

from buildstream._testing import create_repo, ALL_REPO_KINDS
from buildstream._testing import cli  # pylint: disable=unused-import
from buildstream import _yaml
from buildstream.exceptions import ErrorDomain, LoadErrorReason
from buildstream._workspaces import BST_WORKSPACE_FORMAT_VERSION

from tests.testutils import create_artifact_share, create_element_size, wait_for_cache_granularity

repo_kinds = ALL_REPO_KINDS

# Project directory
DATA_DIR = os.path.join(
    os.path.dirname(os.path.realpath(__file__)),
    "project",
)
BASE_FILENAME = os.path.basename(__file__)


class WorkspaceCreator:
    def __init__(self, cli, tmpdir, datafiles, project_path=None):
        self.cli = cli
        self.tmpdir = tmpdir
        self.datafiles = datafiles

        if not project_path:
            project_path = str(datafiles)
        else:
            shutil.copytree(str(datafiles), project_path)

        self.project_path = project_path
        self.bin_files_path = os.path.join(project_path, "files", "bin-files")

        self.workspace_cmd = os.path.join(self.project_path, "workspace_cmd")

    def create_workspace_element(self, kind, suffix="", workspace_dir=None, element_attrs=None):
        element_name = "workspace-test-{}{}.bst".format(kind, suffix)
        element_path = os.path.join(self.project_path, "elements")
        if not workspace_dir:
            workspace_dir = os.path.join(self.workspace_cmd, element_name)
            if workspace_dir[-4:] == ".bst":
                workspace_dir = workspace_dir[:-4]

        # Create our repo object of the given source type with
        # the bin files, and then collect the initial ref.
        # And ensure we store it in a suffix-specific directory, to avoid clashes
        # if using multiple times the same kind element here.
        repo = create_repo(kind, str(self.tmpdir), "repo-for-{}".format(element_name))

        with tempfile.TemporaryDirectory() as tempdir:
            dst_repo = os.path.join(tempdir, "repo")
            shutil.copytree(self.bin_files_path, dst_repo)
            # Touch a file with the element name in, to allow validating that this
            # is the correct repo
            # pylint: disable=consider-using-with
            open(os.path.join(dst_repo, element_name), "a", encoding="utf-8").close()

            ref = repo.create(os.path.join(tempdir, "repo"))

        # Write out our test target
        element = {"kind": "import", "sources": [repo.source_config(ref=ref)]}
        if element_attrs:
            element = {**element, **element_attrs}
        _yaml.roundtrip_dump(element, os.path.join(element_path, element_name))
        return element_name, element_path, workspace_dir

    def create_workspace_elements(self, kinds, suffixs=None, workspace_dir_usr=None, element_attrs=None):

        element_tuples = []

        if suffixs is None:
            suffixs = [
                "",
            ] * len(kinds)
        else:
            assert len(suffixs) == len(kinds)

        for suffix, kind in zip(suffixs, kinds):
            element_name, _, workspace_dir = self.create_workspace_element(
                kind, suffix, workspace_dir_usr, element_attrs
            )
            element_tuples.append((element_name, workspace_dir))

        # Assert that there is a fetch is needed
        states = self.cli.get_element_states(self.project_path, [e for e, _ in element_tuples])
        assert not any(states[e] != "fetch needed" for e, _ in element_tuples)

        return element_tuples

    def open_workspaces(self, kinds, suffixs=None, workspace_dir=None, element_attrs=None, no_checkout=False):

        element_tuples = self.create_workspace_elements(kinds, suffixs, workspace_dir, element_attrs)
        os.makedirs(self.workspace_cmd, exist_ok=True)

        # Now open the workspace, this should have the effect of automatically
        # tracking & fetching the source from the repo.
        args = ["workspace", "open"]
        if no_checkout:
            args.append("--no-checkout")
        if workspace_dir is not None:
            assert len(element_tuples) == 1, "test logic error"
            _, workspace_dir = element_tuples[0]
            args.extend(["--directory", workspace_dir])

        args.extend([element_name for element_name, workspace_dir_suffix in element_tuples])
        result = self.cli.run(cwd=self.workspace_cmd, project=self.project_path, args=args)

        result.assert_success()

        if not no_checkout:
            # Assert that we are now buildable because the source is now cached.
            states = self.cli.get_element_states(self.project_path, [e for e, _ in element_tuples])
            assert not any(states[e] != "buildable" for e, _ in element_tuples)

            # Check that the executable hello file is found in each workspace
            for _, workspace in element_tuples:
                filename = os.path.join(workspace, "usr", "bin", "hello")
                assert os.path.exists(filename)

        return element_tuples


def open_workspace(
    cli,
    tmpdir,
    datafiles,
    kind,
    suffix="",
    workspace_dir=None,
    project_path=None,
    element_attrs=None,
    no_checkout=False,
):
    workspace_object = WorkspaceCreator(cli, tmpdir, datafiles, project_path)
    workspaces = workspace_object.open_workspaces((kind,), (suffix,), workspace_dir, element_attrs, no_checkout)
    assert len(workspaces) == 1
    element_name, workspace = workspaces[0]
    return element_name, workspace_object.project_path, workspace


@pytest.mark.datafiles(DATA_DIR)
def test_open_multi(cli, tmpdir, datafiles):
    workspace_object = WorkspaceCreator(cli, tmpdir, datafiles)
    workspaces = workspace_object.open_workspaces(repo_kinds)

    for (elname, workspace), kind in zip(workspaces, repo_kinds):
        assert kind in elname
        workspace_lsdir = os.listdir(workspace)
        assert elname in workspace_lsdir


@pytest.mark.skipif(os.geteuid() == 0, reason="root may have CAP_DAC_OVERRIDE and ignore permissions")
@pytest.mark.datafiles(DATA_DIR)
def test_open_multi_unwritable(cli, tmpdir, datafiles):
    workspace_object = WorkspaceCreator(cli, tmpdir, datafiles)

    element_tuples = workspace_object.create_workspace_elements(repo_kinds, repo_kinds)
    os.makedirs(workspace_object.workspace_cmd, exist_ok=True)

    # Now open the workspace, this should have the effect of automatically
    # tracking & fetching the source from the repo.
    args = ["workspace", "open"]
    args.extend([element_name for element_name, workspace_dir_suffix in element_tuples])
    cli.configure({"workspacedir": workspace_object.workspace_cmd})

    cwdstat = os.stat(workspace_object.workspace_cmd)
    try:
        os.chmod(workspace_object.workspace_cmd, cwdstat.st_mode - stat.S_IWRITE)
        result = workspace_object.cli.run(project=workspace_object.project_path, args=args)
    finally:
        # Using this finally to make sure we always put thing back how they should be.
        os.chmod(workspace_object.workspace_cmd, cwdstat.st_mode)

    result.assert_main_error(ErrorDomain.STREAM, "workspace-directory-failure")
    # Normally we avoid checking stderr in favour of using the mechine readable result.assert_main_error
    # But Tristan was very keen that the names of the elements left needing workspaces were present in the out put
    assert " ".join([element_name for element_name, workspace_dir_suffix in element_tuples[1:]]) in result.stderr


@pytest.mark.datafiles(DATA_DIR)
def test_open_defaultlocation(cli, tmpdir, datafiles):
    workspace_object = WorkspaceCreator(cli, tmpdir, datafiles)

    # pylint: disable=unbalanced-tuple-unpacking
    ((element_name, workspace_dir),) = workspace_object.create_workspace_elements(["tar"], ["tar"])
    os.makedirs(workspace_object.workspace_cmd, exist_ok=True)

    # Now open the workspace, this should have the effect of automatically
    # tracking & fetching the source from the repo.
    args = ["workspace", "open"]
    args.append(element_name)

    # In the other tests we set the cmd to workspace_object.workspace_cmd with the optional
    # argument, cwd for the workspace_object.cli.run function. But hear we set the default
    # workspace location to workspace_object.workspace_cmd and run the cli.run function with
    # no cwd option so that it runs in the project directory.
    cli.configure({"workspacedir": workspace_object.workspace_cmd})
    result = workspace_object.cli.run(project=workspace_object.project_path, args=args)

    result.assert_success()

    assert cli.get_element_state(workspace_object.project_path, element_name) == "buildable"

    # Check that the executable hello file is found in the workspace
    # even though the cli.run function was not run with cwd = workspace_object.workspace_cmd
    # the workspace should be created in there as we used the 'workspacedir' configuration
    # option.
    filename = os.path.join(workspace_dir, "usr", "bin", "hello")
    assert os.path.exists(filename)


@pytest.mark.datafiles(DATA_DIR)
def test_open_defaultlocation_exists(cli, tmpdir, datafiles):
    workspace_object = WorkspaceCreator(cli, tmpdir, datafiles)

    # pylint: disable=unbalanced-tuple-unpacking
    ((element_name, workspace_dir),) = workspace_object.create_workspace_elements(["tar"], ["tar"])
    os.makedirs(workspace_object.workspace_cmd, exist_ok=True)

    with open(workspace_dir, "w", encoding="utf-8") as fl:
        fl.write("foo")

    # Now open the workspace, this should have the effect of automatically
    # tracking & fetching the source from the repo.
    args = ["workspace", "open"]
    args.append(element_name)

    # In the other tests we set the cmd to workspace_object.workspace_cmd with the optional
    # argument, cwd for the workspace_object.cli.run function. But hear we set the default
    # workspace location to workspace_object.workspace_cmd and run the cli.run function with
    # no cwd option so that it runs in the project directory.
    cli.configure({"workspacedir": workspace_object.workspace_cmd})
    result = workspace_object.cli.run(project=workspace_object.project_path, args=args)

    result.assert_main_error(ErrorDomain.STREAM, "bad-directory")


@pytest.mark.datafiles(DATA_DIR)
def test_open_track(cli, tmpdir, datafiles):
    open_workspace(cli, tmpdir, datafiles, "tar")


@pytest.mark.datafiles(DATA_DIR)
def test_open_noclose_open(cli, tmpdir, datafiles):
    # opening the same workspace twice without closing it should fail
    element_name, project, _ = open_workspace(cli, tmpdir, datafiles, "tar")

    result = cli.run(project=project, args=["workspace", "open", element_name])
    result.assert_main_error(ErrorDomain.STREAM, None)


@pytest.mark.datafiles(DATA_DIR)
def test_open_force(cli, tmpdir, datafiles):
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")

    # Close the workspace
    result = cli.run(project=project, args=["workspace", "close", element_name])
    result.assert_success()

    # Assert the workspace dir still exists
    assert os.path.exists(workspace)

    # Now open the workspace again with --force, this should happily succeed
    result = cli.run(project=project, args=["workspace", "open", "--force", "--directory", workspace, element_name])
    result.assert_success()


@pytest.mark.datafiles(DATA_DIR)
def test_open_force_open(cli, tmpdir, datafiles):
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")

    result = cli.run(project=project, args=["workspace", "close", element_name])
    result.assert_success()

    # Assert the workspace dir exists
    assert os.path.exists(workspace)

    # Now open the workspace again with --force, this should happily succeed
    result = cli.run(project=project, args=["workspace", "open", "--force", "--directory", workspace, element_name])
    result.assert_success()


# Regression test for #1086.
@pytest.mark.datafiles(DATA_DIR)
def test_open_force_open_no_checkout(cli, tmpdir, datafiles):
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")
    hello_path = os.path.join(workspace, "hello.txt")

    # Assert the workspace dir exists
    assert os.path.exists(workspace)

    # Create a new file in the workspace
    with open(hello_path, "w", encoding="utf-8") as f:
        f.write("hello")

    # Now open the workspace again with --force and --no-checkout
    result = cli.run(
        project=project, args=["workspace", "open", "--force", "--no-checkout", "--directory", workspace, element_name]
    )
    result.assert_success()

    # Ensure that our files were not overwritten
    assert os.path.exists(hello_path)
    with open(hello_path, encoding="utf-8") as f:
        assert f.read() == "hello"


@pytest.mark.datafiles(DATA_DIR)
def test_open_force_different_workspace(cli, tmpdir, datafiles):
    _, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar", "-alpha")

    # Assert the workspace dir exists
    assert os.path.exists(workspace)

    hello_path = os.path.join(workspace, "usr", "bin", "hello")
    hello1_path = os.path.join(workspace, "usr", "bin", "hello1")

    tmpdir = os.path.join(str(tmpdir), "-beta")
    shutil.move(hello_path, hello1_path)
    element_name2, _, workspace2 = open_workspace(cli, tmpdir, datafiles, "tar", "-beta")

    # Assert the workspace dir exists
    assert os.path.exists(workspace2)

    # Assert that workspace 1 contains the modified file
    assert os.path.exists(hello1_path)

    # Assert that workspace 2 contains the unmodified file
    assert os.path.exists(os.path.join(workspace2, "usr", "bin", "hello"))

    result = cli.run(project=project, args=["workspace", "close", element_name2])
    result.assert_success()

    # Now open the workspace again with --force, this should happily succeed
    result = cli.run(project=project, args=["workspace", "open", "--force", "--directory", workspace, element_name2])
    result.assert_success()

    # Assert that the file in workspace 1 has been replaced
    # With the file from workspace 2
    assert os.path.exists(hello_path)
    assert not os.path.exists(hello1_path)


@pytest.mark.datafiles(DATA_DIR)
def test_close(cli, tmpdir, datafiles):
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")

    # Close the workspace
    result = cli.run(project=project, args=["workspace", "close", "--remove-dir", element_name])
    result.assert_success()

    # Assert the workspace dir has been deleted
    assert not os.path.exists(workspace)


@pytest.mark.datafiles(DATA_DIR)
def test_close_external_after_move_project(cli, tmpdir, datafiles):
    workspace_dir = os.path.join(str(tmpdir), "workspace")
    project_path = os.path.join(str(tmpdir), "initial_project")
    element_name, _, _ = open_workspace(cli, tmpdir, datafiles, "tar", "", workspace_dir, project_path)
    assert os.path.exists(workspace_dir)
    moved_dir = os.path.join(str(tmpdir), "external_project")
    shutil.move(project_path, moved_dir)
    assert os.path.exists(moved_dir)

    # Close the workspace
    result = cli.run(project=moved_dir, args=["workspace", "close", "--remove-dir", element_name])
    result.assert_success()

    # Assert the workspace dir has been deleted
    assert not os.path.exists(workspace_dir)


@pytest.mark.datafiles(DATA_DIR)
def test_close_internal_after_move_project(cli, tmpdir, datafiles):
    initial_dir = os.path.join(str(tmpdir), "initial_project")
    initial_workspace = os.path.join(initial_dir, "workspace")
    element_name, _, _ = open_workspace(
        cli, tmpdir, datafiles, "tar", workspace_dir=initial_workspace, project_path=initial_dir
    )
    moved_dir = os.path.join(str(tmpdir), "internal_project")
    shutil.move(initial_dir, moved_dir)
    assert os.path.exists(moved_dir)

    # Close the workspace
    result = cli.run(project=moved_dir, args=["workspace", "close", "--remove-dir", element_name])
    result.assert_success()

    # Assert the workspace dir has been deleted
    workspace = os.path.join(moved_dir, "workspace")
    assert not os.path.exists(workspace)


@pytest.mark.datafiles(DATA_DIR)
def test_close_removed(cli, tmpdir, datafiles):
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")

    # Remove it first, closing the workspace should work
    shutil.rmtree(workspace)

    # Close the workspace
    result = cli.run(project=project, args=["workspace", "close", element_name])
    result.assert_success()

    # Assert the workspace dir has been deleted
    assert not os.path.exists(workspace)


@pytest.mark.datafiles(DATA_DIR)
def test_close_nonexistant_element(cli, tmpdir, datafiles):
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")
    element_path = os.path.join(datafiles.dirname, datafiles.basename, "elements", element_name)

    # First brutally remove the element.bst file, ensuring that
    # the element does not exist anymore in the project where
    # we want to close the workspace.
    os.remove(element_path)

    # Close the workspace
    result = cli.run(project=project, args=["workspace", "close", "--remove-dir", element_name])
    result.assert_success()

    # Assert the workspace dir has been deleted
    assert not os.path.exists(workspace)


@pytest.mark.datafiles(DATA_DIR)
def test_close_multiple(cli, tmpdir, datafiles):
    tmpdir_alpha = os.path.join(str(tmpdir), "alpha")
    tmpdir_beta = os.path.join(str(tmpdir), "beta")
    alpha, project, workspace_alpha = open_workspace(cli, tmpdir_alpha, datafiles, "tar", suffix="-alpha")
    beta, project, workspace_beta = open_workspace(cli, tmpdir_beta, datafiles, "tar", suffix="-beta")

    # Close the workspaces
    result = cli.run(project=project, args=["workspace", "close", "--remove-dir", alpha, beta])
    result.assert_success()

    # Assert the workspace dirs have been deleted
    assert not os.path.exists(workspace_alpha)
    assert not os.path.exists(workspace_beta)


@pytest.mark.datafiles(DATA_DIR)
def test_close_all(cli, tmpdir, datafiles):
    tmpdir_alpha = os.path.join(str(tmpdir), "alpha")
    tmpdir_beta = os.path.join(str(tmpdir), "beta")
    _, project, workspace_alpha = open_workspace(cli, tmpdir_alpha, datafiles, "tar", suffix="-alpha")
    _, project, workspace_beta = open_workspace(cli, tmpdir_beta, datafiles, "tar", suffix="-beta")

    # Close the workspaces
    result = cli.run(project=project, args=["workspace", "close", "--remove-dir", "--all"])
    result.assert_success()

    # Assert the workspace dirs have been deleted
    assert not os.path.exists(workspace_alpha)
    assert not os.path.exists(workspace_beta)


@pytest.mark.datafiles(DATA_DIR)
def test_reset(cli, tmpdir, datafiles):
    # Open the workspace
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")

    # Modify workspace
    shutil.rmtree(os.path.join(workspace, "usr", "bin"))
    os.makedirs(os.path.join(workspace, "etc"))
    with open(os.path.join(workspace, "etc", "pony.conf"), "w", encoding="utf-8") as f:
        f.write("PONY='pink'")

    # Now reset the open workspace, this should have the
    # effect of reverting our changes.
    result = cli.run(project=project, args=["workspace", "reset", element_name])
    result.assert_success()
    assert os.path.exists(os.path.join(workspace, "usr", "bin", "hello"))
    assert not os.path.exists(os.path.join(workspace, "etc", "pony.conf"))


@pytest.mark.datafiles(DATA_DIR)
def test_reset_soft(cli, tmpdir, datafiles):
    # Open the workspace
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")

    assert cli.get_element_state(project, element_name) == "buildable"

    hello_path = os.path.join(workspace, "usr", "bin", "hello")
    pony_path = os.path.join(workspace, "etc", "pony.conf")

    assert os.path.exists(os.path.join(workspace, "usr", "bin"))
    assert os.path.exists(hello_path)
    assert not os.path.exists(pony_path)

    key_1 = cli.get_element_key(project, element_name)
    assert key_1 != "{:?<64}".format("")
    result = cli.run(project=project, args=["build", element_name])
    result.assert_success()
    assert cli.get_element_state(project, element_name) == "cached"
    key_2 = cli.get_element_key(project, element_name)
    assert key_2 != "{:?<64}".format("")

    # workspace keys are not recalculated
    assert key_1 == key_2

    wait_for_cache_granularity()

    # Modify workspace
    shutil.rmtree(os.path.join(workspace, "usr", "bin"))
    os.makedirs(os.path.join(workspace, "etc"))
    with open(os.path.join(workspace, "etc", "pony.conf"), "w", encoding="utf-8") as f:
        f.write("PONY='pink'")

    assert not os.path.exists(os.path.join(workspace, "usr", "bin"))
    assert os.path.exists(pony_path)

    # Now soft-reset the open workspace, this should not revert the changes
    result = cli.run(project=project, args=["workspace", "reset", "--soft", element_name])
    result.assert_success()
    # we removed this dir
    assert not os.path.exists(os.path.join(workspace, "usr", "bin"))
    # and added this one
    assert os.path.exists(os.path.join(workspace, "etc", "pony.conf"))

    assert cli.get_element_state(project, element_name) == "buildable"
    key_3 = cli.get_element_key(project, element_name)
    assert key_3 != "{:?<64}".format("")
    assert key_1 != key_3


@pytest.mark.datafiles(DATA_DIR)
def test_reset_multiple(cli, tmpdir, datafiles):
    # Open the workspaces
    tmpdir_alpha = os.path.join(str(tmpdir), "alpha")
    tmpdir_beta = os.path.join(str(tmpdir), "beta")
    alpha, project, workspace_alpha = open_workspace(cli, tmpdir_alpha, datafiles, "tar", suffix="-alpha")
    beta, project, workspace_beta = open_workspace(cli, tmpdir_beta, datafiles, "tar", suffix="-beta")

    # Modify workspaces
    shutil.rmtree(os.path.join(workspace_alpha, "usr", "bin"))
    os.makedirs(os.path.join(workspace_beta, "etc"))
    with open(os.path.join(workspace_beta, "etc", "pony.conf"), "w", encoding="utf-8") as f:
        f.write("PONY='pink'")

    # Now reset the open workspaces, this should have the
    # effect of reverting our changes.
    result = cli.run(
        project=project,
        args=[
            "workspace",
            "reset",
            alpha,
            beta,
        ],
    )
    result.assert_success()
    assert os.path.exists(os.path.join(workspace_alpha, "usr", "bin", "hello"))
    assert not os.path.exists(os.path.join(workspace_beta, "etc", "pony.conf"))


@pytest.mark.datafiles(DATA_DIR)
def test_reset_all(cli, tmpdir, datafiles):
    # Open the workspaces
    tmpdir_alpha = os.path.join(str(tmpdir), "alpha")
    tmpdir_beta = os.path.join(str(tmpdir), "beta")
    _, project, workspace_alpha = open_workspace(cli, tmpdir_alpha, datafiles, "tar", suffix="-alpha")
    _, project, workspace_beta = open_workspace(cli, tmpdir_beta, datafiles, "tar", suffix="-beta")

    # Modify workspaces
    shutil.rmtree(os.path.join(workspace_alpha, "usr", "bin"))
    os.makedirs(os.path.join(workspace_beta, "etc"))
    with open(os.path.join(workspace_beta, "etc", "pony.conf"), "w", encoding="utf-8") as f:
        f.write("PONY='pink'")

    # Now reset the open workspace, this should have the
    # effect of reverting our changes.
    result = cli.run(project=project, args=["workspace", "reset", "--all"])
    result.assert_success()
    assert os.path.exists(os.path.join(workspace_alpha, "usr", "bin", "hello"))
    assert not os.path.exists(os.path.join(workspace_beta, "etc", "pony.conf"))


@pytest.mark.datafiles(DATA_DIR)
def test_list(cli, tmpdir, datafiles):
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")

    # Now list the workspaces
    result = cli.run(project=project, args=["workspace", "list"])
    result.assert_success()

    loaded = _yaml.load_data(result.output)
    workspaces = loaded.get_sequence("workspaces")
    assert len(workspaces) == 1

    space = workspaces.mapping_at(0)
    assert space.get_str("element") == element_name
    assert space.get_str("directory") == workspace


@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("kind", repo_kinds)
@pytest.mark.parametrize("strict", [("strict"), ("non-strict")])
@pytest.mark.parametrize(
    "from_workspace,guess_element",
    [(False, False), (True, True), (True, False)],
    ids=["project-no-guess", "workspace-guess", "workspace-no-guess"],
)
def test_build(cli, tmpdir_factory, datafiles, kind, strict, from_workspace, guess_element):
    tmpdir = tmpdir_factory.mktemp(BASE_FILENAME)
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, kind, False)
    checkout = os.path.join(str(tmpdir), "checkout")
    args_dir = ["-C", workspace] if from_workspace else []
    args_elm = [element_name] if not guess_element else []

    # Modify workspace
    shutil.rmtree(os.path.join(workspace, "usr", "bin"))
    os.makedirs(os.path.join(workspace, "etc"))
    with open(os.path.join(workspace, "etc", "pony.conf"), "w", encoding="utf-8") as f:
        f.write("PONY='pink'")

    # Configure strict mode
    strict_mode = True
    if strict != "strict":
        strict_mode = False
    cli.configure({"projects": {"test": {"strict": strict_mode}}})

    # Build modified workspace
    assert cli.get_element_state(project, element_name) == "buildable"
    key_1 = cli.get_element_key(project, element_name)
    assert key_1 != "{:?<64}".format("")
    result = cli.run(project=project, args=args_dir + ["build", *args_elm])
    result.assert_success()
    assert cli.get_element_state(project, element_name) == "cached"
    key_2 = cli.get_element_key(project, element_name)
    assert key_2 != "{:?<64}".format("")

    # workspace keys are not recalculated
    assert key_1 == key_2

    # Checkout the result
    result = cli.run(project=project, args=args_dir + ["artifact", "checkout", "--directory", checkout, *args_elm])
    result.assert_success()

    # Check that the pony.conf from the modified workspace exists
    filename = os.path.join(checkout, "etc", "pony.conf")
    assert os.path.exists(filename)

    # Check that the original /usr/bin/hello is not in the checkout
    assert not os.path.exists(os.path.join(checkout, "usr", "bin", "hello"))


@pytest.mark.datafiles(DATA_DIR)
def test_buildable_no_ref(cli, tmpdir, datafiles):
    project = str(datafiles)
    element_name = "workspace-test-no-ref.bst"
    element_path = os.path.join(project, "elements")

    # Write out our test target without any source ref
    repo = create_repo("tar", str(tmpdir))
    element = {"kind": "import", "sources": [repo.source_config()]}
    _yaml.roundtrip_dump(element, os.path.join(element_path, element_name))

    # Assert that this target is not buildable when no workspace is associated.
    assert cli.get_element_state(project, element_name) == "no reference"

    # Now open the workspace. We don't need to checkout the source though.
    workspace = os.path.join(str(tmpdir), "workspace-no-ref")
    os.makedirs(workspace)
    args = ["workspace", "open", "--no-checkout", "--directory", workspace, element_name]
    result = cli.run(project=project, args=args)
    result.assert_success()

    # Assert that the target is now buildable.
    assert cli.get_element_state(project, element_name) == "buildable"


@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("modification", [("addfile"), ("removefile"), ("modifyfile")])
@pytest.mark.parametrize("strict", [("strict"), ("non-strict")])
def test_detect_modifications(cli, tmpdir, datafiles, modification, strict):
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")
    checkout = os.path.join(str(tmpdir), "checkout")

    # Configure strict mode
    strict_mode = True
    if strict != "strict":
        strict_mode = False
    cli.configure({"projects": {"test": {"strict": strict_mode}}})

    # Build clean workspace
    assert cli.get_element_state(project, element_name) == "buildable"
    key_1 = cli.get_element_key(project, element_name)
    assert key_1 != "{:?<64}".format("")
    result = cli.run(project=project, args=["build", element_name])
    result.assert_success()
    assert cli.get_element_state(project, element_name) == "cached"
    key_2 = cli.get_element_key(project, element_name)
    assert key_2 != "{:?<64}".format("")

    # workspace keys are not recalculated
    assert key_1 == key_2

    wait_for_cache_granularity()

    # Modify the workspace in various different ways, ensuring we
    # properly detect the changes.
    #
    if modification == "addfile":
        os.makedirs(os.path.join(workspace, "etc"))
        with open(os.path.join(workspace, "etc", "pony.conf"), "w", encoding="utf-8") as f:
            f.write("PONY='pink'")
    elif modification == "removefile":
        os.remove(os.path.join(workspace, "usr", "bin", "hello"))
    elif modification == "modifyfile":
        with open(os.path.join(workspace, "usr", "bin", "hello"), "w", encoding="utf-8") as f:
            f.write("cookie")
    else:
        # This cannot be reached
        assert 0

    # First assert that the state is properly detected
    assert cli.get_element_state(project, element_name) == "buildable"
    key_3 = cli.get_element_key(project, element_name)
    assert key_3 != "{:?<64}".format("")

    # Since there are different things going on at `bst build` time
    # than `bst show` time, we also want to build / checkout again,
    # and ensure that the result contains what we expect.
    result = cli.run(project=project, args=["build", element_name])
    result.assert_success()
    assert cli.get_element_state(project, element_name) == "cached"
    key_4 = cli.get_element_key(project, element_name)
    assert key_4 != "{:?<64}".format("")

    # workspace keys are not recalculated
    assert key_3 == key_4
    # workspace keys are determined by the files
    assert key_1 != key_3

    # Checkout the result
    result = cli.run(project=project, args=["artifact", "checkout", element_name, "--directory", checkout])
    result.assert_success()

    # Check the result for the changes we made
    #
    if modification == "addfile":
        filename = os.path.join(checkout, "etc", "pony.conf")
        assert os.path.exists(filename)
    elif modification == "removefile":
        assert not os.path.exists(os.path.join(checkout, "usr", "bin", "hello"))
    elif modification == "modifyfile":
        with open(os.path.join(workspace, "usr", "bin", "hello"), "r", encoding="utf-8") as f:
            data = f.read()
            assert data == "cookie"
    else:
        # This cannot be reached
        assert 0


# Ensure that various versions that should not be accepted raise a
# LoadError.INVALID_DATA
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize(
    "workspace_cfg",
    [
        # Test loading a negative workspace version
        {"format-version": -1},
        # Test loading version 0 with two sources
        {
            "format-version": 0,
            "alpha.bst": {
                0: "/workspaces/bravo",
                1: "/workspaces/charlie",
            },
        },
        # Test loading a version with decimals
        {"format-version": 0.5},
        # Test loading an unsupported old version
        {"format-version": 3},
        # Test loading a future version
        {"format-version": BST_WORKSPACE_FORMAT_VERSION + 1},
    ],
)
def test_list_unsupported_workspace(cli, datafiles, workspace_cfg):
    project = str(datafiles)
    os.makedirs(os.path.join(project, ".bst2"))
    workspace_config_path = os.path.join(project, ".bst2", "workspaces.yml")

    _yaml.roundtrip_dump(workspace_cfg, workspace_config_path)

    result = cli.run(project=project, args=["workspace", "list"])
    result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.INVALID_DATA)


# Ensure that various versions that should be accepted are parsed
# correctly.
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize(
    "workspace_cfg,expected",
    [
        # Test loading version 4
        (
            {
                "format-version": 4,
                "workspaces": {"alpha.bst": {"path": "/workspaces/bravo"}},
            },
            {
                "format-version": BST_WORKSPACE_FORMAT_VERSION,
                "workspaces": {"alpha.bst": {"path": "/workspaces/bravo"}},
            },
        ),
    ],
)
def test_list_supported_workspace(cli, tmpdir, datafiles, workspace_cfg, expected):
    def parse_dict_as_yaml(node):
        tempfile = os.path.join(str(tmpdir), "yaml_dump")
        _yaml.roundtrip_dump(node, tempfile)
        return _yaml.load(tempfile, shortname=None).strip_node_info()

    project = str(datafiles)
    os.makedirs(os.path.join(project, ".bst"))
    workspace_config_path = os.path.join(project, ".bst", "workspaces.yml")

    _yaml.roundtrip_dump(workspace_cfg, workspace_config_path)

    # Check that we can still read workspace config that is in old format
    result = cli.run(project=project, args=["workspace", "list"])
    result.assert_success()

    loaded_config = _yaml.load(workspace_config_path, shortname=None).strip_node_info()

    # Check that workspace config remains the same if no modifications
    # to workspaces were made
    assert loaded_config == parse_dict_as_yaml(workspace_cfg)

    # Create a test bst file
    bin_files_path = os.path.join(project, "files", "bin-files")
    element_path = os.path.join(project, "elements")
    element_name = "workspace-test.bst"
    workspace = os.path.join(str(tmpdir), "workspace")

    # Create our repo object of the given source type with
    # the bin files, and then collect the initial ref.
    #
    repo = create_repo("tar", str(tmpdir))
    ref = repo.create(bin_files_path)

    # Write out our test target
    element = {"kind": "import", "sources": [repo.source_config(ref=ref)]}
    _yaml.roundtrip_dump(element, os.path.join(element_path, element_name))

    # Make a change to the workspaces file
    result = cli.run(project=project, args=["workspace", "open", "--directory", workspace, element_name])
    result.assert_success()
    result = cli.run(project=project, args=["workspace", "close", "--remove-dir", element_name])
    result.assert_success()

    # Check that workspace config is converted correctly if necessary
    loaded_config = _yaml.load(workspace_config_path, shortname=None).strip_node_info()
    assert loaded_config == parse_dict_as_yaml(expected)


@pytest.mark.datafiles(DATA_DIR)
def test_inconsitent_pipeline_message(cli, tmpdir, datafiles):
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")

    shutil.rmtree(workspace)

    result = cli.run(project=project, args=["build", element_name])
    result.assert_main_error(ErrorDomain.PIPELINE, "inconsistent-pipeline-workspaced")


@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("strict", [("strict"), ("non-strict")])
def test_cache_key_workspace_in_dependencies(cli, tmpdir, datafiles, strict):
    checkout = os.path.join(str(tmpdir), "checkout")
    element_name, project, workspace = open_workspace(cli, os.path.join(str(tmpdir), "repo-a"), datafiles, "tar")

    element_path = os.path.join(project, "elements")
    back_dep_element_name = "workspace-test-back-dep.bst"

    # Write out our test target
    element = {"kind": "compose", "depends": [{"filename": element_name, "type": "build"}]}
    _yaml.roundtrip_dump(element, os.path.join(element_path, back_dep_element_name))

    # Modify workspace
    shutil.rmtree(os.path.join(workspace, "usr", "bin"))
    os.makedirs(os.path.join(workspace, "etc"))
    with open(os.path.join(workspace, "etc", "pony.conf"), "w", encoding="utf-8") as f:
        f.write("PONY='pink'")

    # Configure strict mode
    strict_mode = True
    if strict != "strict":
        strict_mode = False
    cli.configure({"projects": {"test": {"strict": strict_mode}}})

    # Build artifact with dependency's modified workspace
    assert cli.get_element_state(project, element_name) == "buildable"
    key_a1 = cli.get_element_key(project, element_name)
    assert key_a1 != "{:?<64}".format("")
    assert cli.get_element_state(project, back_dep_element_name) == "waiting"
    key_b1 = cli.get_element_key(project, back_dep_element_name)
    assert key_b1 != "{:?<64}".format("")
    result = cli.run(project=project, args=["build", back_dep_element_name])
    result.assert_success()
    assert cli.get_element_state(project, element_name) == "cached"
    key_a2 = cli.get_element_key(project, element_name)
    assert key_a2 != "{:?<64}".format("")
    assert cli.get_element_state(project, back_dep_element_name) == "cached"
    key_b2 = cli.get_element_key(project, back_dep_element_name)
    assert key_b2 != "{:?<64}".format("")
    result = cli.run(project=project, args=["build", back_dep_element_name])
    result.assert_success()

    # workspace keys are not recalculated
    assert key_a1 == key_a2
    assert key_b1 == key_b2

    # Checkout the result
    result = cli.run(project=project, args=["artifact", "checkout", back_dep_element_name, "--directory", checkout])
    result.assert_success()

    # Check that the pony.conf from the modified workspace exists
    filename = os.path.join(checkout, "etc", "pony.conf")
    assert os.path.exists(filename)

    # Check that the original /usr/bin/hello is not in the checkout
    assert not os.path.exists(os.path.join(checkout, "usr", "bin", "hello"))


@pytest.mark.datafiles(DATA_DIR)
def test_multiple_failed_builds(cli, tmpdir, datafiles):
    element_config = {"kind": "manual", "config": {"configure-commands": ["unknown_command_that_will_fail"]}}
    element_name, project, _ = open_workspace(cli, tmpdir, datafiles, "tar", element_attrs=element_config)

    for _ in range(2):
        result = cli.run(project=project, args=["build", element_name])
        assert "BUG" not in result.stderr
        assert cli.get_element_state(project, element_name) != "cached"


@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("subdir", [True, False], ids=["subdir", "no-subdir"])
@pytest.mark.parametrize("guess_element", [True, False], ids=["guess", "no-guess"])
def test_external_fetch(cli, datafiles, tmpdir_factory, subdir, guess_element):
    # An element with an open workspace can't be fetched, but we still expect fetches
    # to fetch any dependencies
    tmpdir = tmpdir_factory.mktemp(BASE_FILENAME)
    depend_element = "fetchable.bst"

    # Create an element to fetch (local sources do not need to fetch)
    create_element_size(depend_element, str(datafiles), "elements", [], 1024)

    element_name, project, workspace = open_workspace(
        cli, tmpdir, datafiles, "tar", no_checkout=True, element_attrs={"depends": [depend_element]}
    )
    arg_elm = [element_name] if not guess_element else []

    if subdir:
        call_dir = os.path.join(workspace, "usr")
        os.makedirs(call_dir, exist_ok=True)
    else:
        call_dir = workspace

    # Assert that the depended element is not fetched yet
    assert cli.get_element_state(str(datafiles), depend_element) == "fetch needed"

    # Fetch the workspaced element
    result = cli.run(project=project, args=["-C", call_dir, "source", "fetch", "--deps", "all", *arg_elm])
    result.assert_success()

    # Assert that the depended element has now been fetched
    assert cli.get_element_state(str(datafiles), depend_element) == "buildable"


@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("guess_element", [True, False], ids=["guess", "no-guess"])
def test_external_push_pull(cli, datafiles, tmpdir_factory, guess_element):
    # Pushing and pulling to/from an artifact cache works from an external workspace
    tmpdir = tmpdir_factory.mktemp(BASE_FILENAME)
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")
    arg_elm = [element_name] if not guess_element else []

    with create_artifact_share(os.path.join(str(tmpdir), "artifactshare")) as share:
        result = cli.run(project=project, args=["-C", workspace, "build", element_name])
        result.assert_success()

        cli.configure({"artifacts": {"servers": [{"url": share.repo, "push": True}]}})

        result = cli.run(project=project, args=["-C", workspace, "artifact", "push", *arg_elm])
        result.assert_success()

        result = cli.run(project=project, args=["-C", workspace, "artifact", "pull", "--deps", "all", *arg_elm])
        result.assert_success()


# Attempting to track in an open workspace is not a sensible thing and it's not compatible with workspaces as plugin
# sources: The new ref (if it differed from the old) would have been ignored regardless.
# The user should be expected to simply close the workspace before tracking.
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("guess_element", [True, False], ids=["guess", "no-guess"])
def test_external_track(cli, datafiles, tmpdir_factory, guess_element):
    tmpdir = tmpdir_factory.mktemp(BASE_FILENAME)
    element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")
    element_file = os.path.join(str(datafiles), "elements", element_name)
    arg_elm = [element_name] if not guess_element else []

    # Delete the ref from the source so that we can detect if the
    # element has been tracked after closing the workspace
    element_contents = _yaml.load(element_file, shortname=None)
    ref1 = element_contents.get_sequence("sources").mapping_at(0).get_str("ref")
    del element_contents.get_sequence("sources").mapping_at(0)["ref"]
    _yaml.roundtrip_dump(element_contents, element_file)

    result = cli.run(project=project, args=["-C", workspace, "source", "track", *arg_elm])
    result.assert_success()

    # Element is not tracked now
    element_contents = _yaml.load(element_file, shortname=None)
    assert "ref" not in element_contents.get_sequence("sources").mapping_at(0)

    # close the workspace
    result = cli.run(project=project, args=["-C", workspace, "workspace", "close", *arg_elm])
    result.assert_success()

    # and retrack the element
    result = cli.run(project=project, args=["source", "track", element_name])
    result.assert_success()

    element_contents = _yaml.load(element_file, shortname=None)
    ref2 = element_contents.get_sequence("sources").mapping_at(0).get_str("ref")
    # these values should be equivalent
    assert ref1 == ref2


@pytest.mark.datafiles(DATA_DIR)
def test_external_open_other(cli, datafiles, tmpdir_factory):
    # From inside an external workspace, open another workspace
    tmpdir1 = tmpdir_factory.mktemp(BASE_FILENAME)
    tmpdir2 = tmpdir_factory.mktemp(BASE_FILENAME)
    # Making use of the assumption that it's the same project in both invocations of open_workspace
    _, project, alpha_workspace = open_workspace(cli, tmpdir1, datafiles, "tar", suffix="-alpha")
    beta_element, _, beta_workspace = open_workspace(cli, tmpdir2, datafiles, "tar", suffix="-beta")

    # Closing the other element first, because I'm too lazy to create an
    # element without opening it
    result = cli.run(project=project, args=["workspace", "close", beta_element])
    result.assert_success()

    result = cli.run(
        project=project,
        args=["-C", alpha_workspace, "workspace", "open", "--force", "--directory", beta_workspace, beta_element],
    )
    result.assert_success()


@pytest.mark.datafiles(DATA_DIR)
def test_external_reset_other(cli, datafiles, tmpdir_factory):
    tmpdir1 = tmpdir_factory.mktemp(BASE_FILENAME)
    tmpdir2 = tmpdir_factory.mktemp(BASE_FILENAME)
    # Making use of the assumption that it's the same project in both invocations of open_workspace
    _, project, alpha_workspace = open_workspace(cli, tmpdir1, datafiles, "tar", suffix="-alpha")
    beta_element, _, _ = open_workspace(cli, tmpdir2, datafiles, "tar", suffix="-beta")

    result = cli.run(project=project, args=["-C", alpha_workspace, "workspace", "reset", beta_element])
    result.assert_success()


@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("guess_element", [True, False], ids=["guess", "no-guess"])
def test_external_reset_self(cli, datafiles, tmpdir, guess_element):
    element, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")
    arg_elm = [element] if not guess_element else []

    # Command succeeds
    result = cli.run(project=project, args=["-C", workspace, "workspace", "reset", *arg_elm])
    result.assert_success()

    # Successive commands still work (i.e. .bstproject.yaml hasn't been deleted)
    result = cli.run(project=project, args=["-C", workspace, "workspace", "list"])
    result.assert_success()


@pytest.mark.datafiles(DATA_DIR)
def test_external_list(cli, datafiles, tmpdir_factory):
    tmpdir = tmpdir_factory.mktemp(BASE_FILENAME)
    # Making use of the assumption that it's the same project in both invocations of open_workspace
    _, project, workspace = open_workspace(cli, tmpdir, datafiles, "tar")

    result = cli.run(project=project, args=["-C", workspace, "workspace", "list"])
    result.assert_success()


@pytest.mark.datafiles(DATA_DIR)
def test_multisource_workspace(cli, datafiles, tmpdir):
    # checks that if an element has multiple sources, then the opened workspace
    # will contain them
    project = str(datafiles)
    element_name = "multisource.bst"
    element = {
        "kind": "import",
        "sources": [{"kind": "local", "path": "files/bin-files"}, {"kind": "local", "path": "files/dev-files"}],
    }
    element_path = os.path.join(project, "elements", element_name)
    _yaml.roundtrip_dump(element, element_path)

    workspace_dir = os.path.join(str(tmpdir), "multisource")
    res = cli.run(project=project, args=["workspace", "open", "multisource.bst", "--directory", workspace_dir])
    res.assert_success()

    directories = os.listdir(os.path.join(workspace_dir, "usr"))
    assert "bin" in directories and "include" in directories


# This strange test tests against a regression raised in issue #919,
# where opening a workspace on a runtime dependency of a build only
# dependency causes `bst build` to not build the specified target
# but just successfully builds the workspaced element and happily
# exits without completing the build.
#
TEST_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)))


@pytest.mark.datafiles(TEST_DIR)
@pytest.mark.parametrize(
    ["case", "non_workspaced_elements_state"],
    [
        ("workspaced-build-dep", ["waiting", "waiting", "waiting", "waiting", "waiting"]),
        ("workspaced-runtime-dep", ["buildable", "buildable", "waiting", "waiting", "waiting"]),
    ],
)
@pytest.mark.parametrize("strict", [("strict"), ("non-strict")])
def test_build_all(cli, tmpdir, datafiles, case, strict, non_workspaced_elements_state):
    project = os.path.join(str(datafiles), case)
    workspace = os.path.join(str(tmpdir), "workspace")
    non_leaf_elements = ["elem2.bst", "elem3.bst", "stack.bst", "elem4.bst", "elem5.bst"]
    all_elements = ["elem1.bst", *non_leaf_elements]

    # Configure strict mode
    strict_mode = True
    if strict != "strict":
        strict_mode = False
    cli.configure({"projects": {"test": {"strict": strict_mode}}})

    # First open the workspace
    result = cli.run(project=project, args=["workspace", "open", "--directory", workspace, "elem1.bst"])
    result.assert_success()

    # Ensure all elements are waiting build the first
    assert cli.get_element_states(project, all_elements) == dict(
        zip(all_elements, ["buildable", *non_workspaced_elements_state])
    )

    # Now build the targets elem4.bst and elem5.bst
    result = cli.run(project=project, args=["build", "elem4.bst", "elem5.bst"])
    result.assert_success()

    # Assert that the target is built
    assert cli.get_element_states(project, all_elements) == {elem: "cached" for elem in all_elements}


@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("strict", ["strict", "non-strict"])
def test_show_workspace_logs(cli, tmpdir, datafiles, strict):
    project = str(datafiles)
    workspace = os.path.join(str(tmpdir), "workspace")
    target = "manual.bst"

    # Configure strict mode
    strict_mode = True
    if strict != "strict":
        strict_mode = False
    cli.configure({"projects": {"test": {"strict": strict_mode}}})

    # First open the workspace
    result = cli.run(project=project, args=["workspace", "open", "--directory", workspace, target])
    result.assert_success()

    # Build the element
    result = cli.run(project=project, args=["build", target])
    result.assert_task_error(ErrorDomain.SANDBOX, "missing-command")

    result = cli.run(project=project, args=["artifact", "log", target])
    result.assert_success()

    # Assert that the log is not empty
    assert result.output != ""