blob: 31eab76d30f55d29d937455ac067a4aa2effd77a [file] [log] [blame]
#
# Copyright (C) 2018 Codethink Limited
# Copyright (C) 2018 Bloomberg Finance LP
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library. If not, see <http://www.gnu.org/licenses/>.
#
# 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>
#
import os
import stat
import pytest
import shutil
import subprocess
from ruamel.yaml.comments import CommentedSet
from tests.testutils import create_repo, ALL_REPO_KINDS, wait_for_cache_granularity
from tests.testutils import create_artifact_share, create_element_size
from buildstream.plugintestutils import cli
from buildstream import _yaml
from buildstream._exceptions import ErrorDomain, LoadError, LoadErrorReason
from buildstream._workspaces import BST_WORKSPACE_FORMAT_VERSION
repo_kinds = [(kind) for kind in ALL_REPO_KINDS]
# Project directory
DATA_DIR = os.path.join(
os.path.dirname(os.path.realpath(__file__)),
"project",
)
class WorkspaceCreater():
def __init__(self, cli, tmpdir, datafiles, project_path=None):
self.cli = cli
self.tmpdir = tmpdir
self.datafiles = datafiles
if not project_path:
project_path = os.path.join(datafiles.dirname, datafiles.basename)
else:
shutil.copytree(os.path.join(datafiles.dirname, datafiles.basename), 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, track, 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.
repo = create_repo(kind, str(self.tmpdir))
ref = repo.create(self.bin_files_path)
if track:
ref = None
# Write out our test target
element = {
'kind': 'import',
'sources': [
repo.source_config(ref=ref)
]
}
if element_attrs:
element = {**element, **element_attrs}
_yaml.dump(element,
os.path.join(element_path,
element_name))
return element_name, element_path, workspace_dir
def create_workspace_elements(self, kinds, track, suffixes=None, workspace_dir_usr=None,
element_attrs=None):
element_tuples = []
if suffixes is None:
suffixes = ['', ] * len(kinds)
else:
if len(suffixes) != len(kinds):
raise "terable error"
for suffix, kind in zip(suffixes, kinds):
element_name, element_path, workspace_dir = \
self.create_workspace_element(kind, track, suffix, workspace_dir_usr,
element_attrs)
element_tuples.append((element_name, workspace_dir))
# Assert that there is no reference, a track & fetch is needed
states = self.cli.get_element_states(self.project_path, [
e for e, _ in element_tuples
])
if track:
assert not any(states[e] != 'no reference' for e, _ in element_tuples)
else:
assert not any(states[e] != 'fetch needed' for e, _ in element_tuples)
return element_tuples
def open_workspaces(self, kinds, track, suffixes=None, workspace_dir=None,
element_attrs=None, no_checkout=False, no_cache=False):
element_tuples = self.create_workspace_elements(kinds, track, suffixes, 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 track:
args.append('--track')
if no_checkout:
args.append('--no-checkout')
if no_cache:
args.append('--no-cache')
if workspace_dir is not None:
assert len(element_tuples) == 1, "test logic error"
_, workspace_dir = element_tuples[0]
args.extend(['--directory', workspace_dir])
print("element_tuples", element_tuples)
args.extend([element_name for element_name, workspace_dir_suffix in element_tuples])
print("args", args)
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 element_name, workspace_dir in element_tuples:
filename = os.path.join(workspace_dir, 'usr', 'bin', 'hello')
assert os.path.exists(filename)
return element_tuples, result
def open_workspace(cli, tmpdir, datafiles, kind, track, suffix='', workspace_dir=None,
project_path=None, element_attrs=None, no_checkout=False, no_cache=False):
workspace_object = WorkspaceCreater(cli, tmpdir, datafiles, project_path)
workspaces, _ = workspace_object.open_workspaces((kind, ), track, (suffix, ), workspace_dir,
element_attrs, no_checkout, no_cache)
assert len(workspaces) == 1
element_name, workspace = workspaces[0]
return element_name, workspace_object.project_path, workspace
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("kind", repo_kinds)
def test_open(cli, tmpdir, datafiles, kind):
open_workspace(cli, tmpdir, datafiles, kind, False)
@pytest.mark.datafiles(DATA_DIR)
def test_open_bzr_customize(cli, tmpdir, datafiles):
element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "bzr", False)
# Check that the .bzr dir exists
bzrdir = os.path.join(workspace, ".bzr")
assert(os.path.isdir(bzrdir))
# Check that the correct origin branch is set
element_config = _yaml.load(os.path.join(project, "elements", element_name))
source_config = element_config['sources'][0]
output = subprocess.check_output(["bzr", "info"], cwd=workspace)
stripped_url = source_config['url'].lstrip("file:///")
expected_output_str = ("checkout of branch: /{}/{}"
.format(stripped_url, source_config['track']))
assert(expected_output_str in str(output))
@pytest.mark.datafiles(DATA_DIR)
def test_open_multi(cli, tmpdir, datafiles):
workspace_object = WorkspaceCreater(cli, tmpdir, datafiles)
workspaces, _ = workspace_object.open_workspaces(repo_kinds, False)
for (elname, workspace), kind in zip(workspaces, repo_kinds):
assert kind in elname
workspace_lsdir = os.listdir(workspace)
if kind == 'git':
assert('.git' in workspace_lsdir)
elif kind == 'bzr':
assert('.bzr' in workspace_lsdir)
else:
assert not ('.git' in workspace_lsdir)
assert not ('.bzr' 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 = WorkspaceCreater(cli, tmpdir, datafiles)
element_tuples = workspace_object.create_workspace_elements(repo_kinds, False, 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, None)
# 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_multi_with_directory(cli, tmpdir, datafiles):
workspace_object = WorkspaceCreater(cli, tmpdir, datafiles)
element_tuples = workspace_object.create_workspace_elements(repo_kinds, False, 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(['--directory', 'any/dir/should/fail'])
args.extend([element_name for element_name, workspace_dir_suffix in element_tuples])
result = workspace_object.cli.run(cwd=workspace_object.workspace_cmd, project=workspace_object.project_path,
args=args)
result.assert_main_error(ErrorDomain.STREAM, 'directory-with-multiple-elements')
@pytest.mark.datafiles(DATA_DIR)
def test_open_defaultlocation(cli, tmpdir, datafiles):
workspace_object = WorkspaceCreater(cli, tmpdir, datafiles)
((element_name, workspace_dir), ) = workspace_object.create_workspace_elements(['git'], False, ['git'])
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 = WorkspaceCreater(cli, tmpdir, datafiles)
((element_name, workspace_dir), ) = workspace_object.create_workspace_elements(['git'], False, ['git'])
os.makedirs(workspace_object.workspace_cmd, exist_ok=True)
with open(workspace_dir, 'w') 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)
@pytest.mark.parametrize("kind", repo_kinds)
def test_open_track(cli, tmpdir, datafiles, kind):
open_workspace(cli, tmpdir, datafiles, kind, True)
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("kind", repo_kinds)
def test_open_force(cli, tmpdir, datafiles, kind):
element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, kind, False)
# 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)
@pytest.mark.parametrize("kind", repo_kinds)
def test_open_force_open(cli, tmpdir, datafiles, kind):
element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, kind, False)
# 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()
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("kind", repo_kinds)
def test_open_force_different_workspace(cli, tmpdir, datafiles, kind):
element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, kind, False, "-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, project2, workspace2 = open_workspace(cli, tmpdir, datafiles, kind, False, "-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'))
# Now open the workspace again with --force, this should happily succeed
result = cli.run(project=project, args=[
'workspace', 'open', '--force', '--directory', workspace, element_name2
])
# 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)
result.assert_success()
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("kind", repo_kinds)
def test_close(cli, tmpdir, datafiles, kind):
element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, kind, False)
# 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, 'git', False, "", 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, 'git', False,
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, 'git', False)
# 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, 'git', False)
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, 'git', False, suffix='-alpha')
beta, project, workspace_beta = open_workspace(
cli, tmpdir_beta, datafiles, 'git', False, 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')
alpha, project, workspace_alpha = open_workspace(
cli, tmpdir_alpha, datafiles, 'git', False, suffix='-alpha')
beta, project, workspace_beta = open_workspace(
cli, tmpdir_beta, datafiles, 'git', False, 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, 'git', False)
# 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') 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_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, 'git', False, suffix='-alpha')
beta, project, workspace_beta = open_workspace(
cli, tmpdir_beta, datafiles, 'git', False, 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') 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')
alpha, project, workspace_alpha = open_workspace(
cli, tmpdir_alpha, datafiles, 'git', False, suffix='-alpha')
beta, project, workspace_beta = open_workspace(
cli, tmpdir_beta, datafiles, 'git', False, 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') 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, 'git', False)
# Now list the workspaces
result = cli.run(project=project, args=[
'workspace', 'list'
])
result.assert_success()
loaded = _yaml.load_data(result.output)
assert isinstance(loaded.get('workspaces'), list)
workspaces = loaded['workspaces']
assert len(workspaces) == 1
space = workspaces[0]
assert space['element'] == element_name
assert space['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('')
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') 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'
assert cli.get_element_key(project, element_name) == "{:?<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'
assert cli.get_element_key(project, element_name) != "{:?<64}".format('')
# 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 = os.path.join(datafiles.dirname, datafiles.basename)
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('git', str(tmpdir))
element = {
'kind': 'import',
'sources': [
repo.source_config()
]
}
_yaml.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, 'git', False)
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'
assert cli.get_element_key(project, element_name) == "{:?<64}".format('')
result = cli.run(project=project, args=['build', element_name])
result.assert_success()
assert cli.get_element_state(project, element_name) == 'cached'
assert cli.get_element_key(project, element_name) != "{:?<64}".format('')
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') 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') 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'
assert cli.get_element_key(project, element_name) == "{:?<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'
assert cli.get_element_key(project, element_name) != "{:?<64}".format('')
# 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') 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 a future version
{"format-version": BST_WORKSPACE_FORMAT_VERSION + 1}
])
def test_list_unsupported_workspace(cli, tmpdir, datafiles, workspace_cfg):
project = os.path.join(datafiles.dirname, datafiles.basename)
bin_files_path = os.path.join(project, 'files', 'bin-files')
element_path = os.path.join(project, 'elements')
element_name = 'workspace-version.bst'
os.makedirs(os.path.join(project, '.bst'))
workspace_config_path = os.path.join(project, '.bst', 'workspaces.yml')
_yaml.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 0 without a dict
({
"alpha.bst": "/workspaces/bravo"
}, {
"format-version": BST_WORKSPACE_FORMAT_VERSION,
"workspaces": {
"alpha.bst": {
"prepared": False,
"path": "/workspaces/bravo",
"running_files": {},
"cached_build": False
}
}
}),
# Test loading version 0 with only one source
({
"alpha.bst": {
0: "/workspaces/bravo"
}
}, {
"format-version": BST_WORKSPACE_FORMAT_VERSION,
"workspaces": {
"alpha.bst": {
"prepared": False,
"path": "/workspaces/bravo",
"running_files": {},
"cached_build": False
}
}
}),
# Test loading version 1
({
"format-version": 1,
"workspaces": {
"alpha.bst": {
"path": "/workspaces/bravo"
}
}
}, {
"format-version": BST_WORKSPACE_FORMAT_VERSION,
"workspaces": {
"alpha.bst": {
"prepared": False,
"path": "/workspaces/bravo",
"running_files": {},
"cached_build": False
}
}
}),
# Test loading version 2
({
"format-version": 2,
"workspaces": {
"alpha.bst": {
"path": "/workspaces/bravo",
"last_successful": "some_key",
"running_files": {
"beta.bst": ["some_file"]
}
}
}
}, {
"format-version": BST_WORKSPACE_FORMAT_VERSION,
"workspaces": {
"alpha.bst": {
"prepared": False,
"path": "/workspaces/bravo",
"last_successful": "some_key",
"running_files": {
"beta.bst": ["some_file"]
},
"cached_build": False
}
}
}),
# Test loading version 3
({
"format-version": 3,
"workspaces": {
"alpha.bst": {
"prepared": True,
"path": "/workspaces/bravo",
"running_files": {}
}
}
}, {
"format-version": BST_WORKSPACE_FORMAT_VERSION,
"workspaces": {
"alpha.bst": {
"prepared": True,
"path": "/workspaces/bravo",
"running_files": {},
"cached_build": False
}
}
}),
# Test loading version 4
({
"format-version": 4,
"workspaces": {
"alpha.bst": {
"prepared": False,
"path": "/workspaces/bravo",
"running_files": {},
"cached_build": True
}
}
}, {
"format-version": BST_WORKSPACE_FORMAT_VERSION,
"workspaces": {
"alpha.bst": {
"prepared": False,
"path": "/workspaces/bravo",
"running_files": {},
"cached_build": True
}
}
})
])
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.dump(node, tempfile)
return _yaml.node_sanitize(_yaml.load(tempfile))
project = os.path.join(datafiles.dirname, datafiles.basename)
os.makedirs(os.path.join(project, '.bst'))
workspace_config_path = os.path.join(project, '.bst', 'workspaces.yml')
_yaml.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.node_sanitize(_yaml.load(workspace_config_path))
# 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('git', str(tmpdir))
ref = repo.create(bin_files_path)
# Write out our test target
element = {
'kind': 'import',
'sources': [
repo.source_config(ref=ref)
]
}
_yaml.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.node_sanitize(_yaml.load(workspace_config_path))
assert loaded_config == parse_dict_as_yaml(expected)
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("kind", repo_kinds)
def test_inconsitent_pipeline_message(cli, tmpdir, datafiles, kind):
element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, kind, False)
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, 'git', False)
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.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') 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'
assert cli.get_element_key(project, element_name) == "{:?<64}".format('')
assert cli.get_element_state(project, back_dep_element_name) == 'waiting'
assert cli.get_element_key(project, back_dep_element_name) == "{:?<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'
assert cli.get_element_key(project, element_name) != "{:?<64}".format('')
assert cli.get_element_state(project, back_dep_element_name) == 'cached'
assert cli.get_element_key(project, back_dep_element_name) != "{:?<64}".format('')
result = cli.run(project=project, args=['build', back_dep_element_name])
result.assert_success()
# 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,
"git", False, 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('')
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, "git", False, 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'] + 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('')
element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "git", False)
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': {'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()
@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('')
element_name, project, workspace = open_workspace(cli, tmpdir, datafiles, "git", False)
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
element_contents = _yaml.load(element_file)
del element_contents['sources'][0]['ref']
_yaml.dump(_yaml.node_sanitize(element_contents), element_file)
result = cli.run(project=project, args=['-C', workspace, 'source', 'track'] + arg_elm)
result.assert_success()
# Element is tracked now
element_contents = _yaml.load(element_file)
assert 'ref' in element_contents['sources'][0]
@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('')
tmpdir2 = tmpdir_factory.mktemp('')
# Making use of the assumption that it's the same project in both invocations of open_workspace
alpha_element, project, alpha_workspace = open_workspace(cli, tmpdir1, datafiles, "git", False, suffix="-alpha")
beta_element, _, beta_workspace = open_workspace(cli, tmpdir2, datafiles, "git", False, 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_close_other(cli, datafiles, tmpdir_factory):
# From inside an external workspace, close the other workspace
tmpdir1 = tmpdir_factory.mktemp('')
tmpdir2 = tmpdir_factory.mktemp('')
# Making use of the assumption that it's the same project in both invocations of open_workspace
alpha_element, project, alpha_workspace = open_workspace(cli, tmpdir1, datafiles, "git", False, suffix="-alpha")
beta_element, _, beta_workspace = open_workspace(cli, tmpdir2, datafiles, "git", False, suffix="-beta")
result = cli.run(project=project, args=['-C', alpha_workspace, 'workspace', 'close', beta_element])
result.assert_success()
@pytest.mark.datafiles(DATA_DIR)
@pytest.mark.parametrize("guess_element", [True, False], ids=["guess", "no-guess"])
def test_external_close_self(cli, datafiles, tmpdir_factory, guess_element):
# From inside an external workspace, close it
tmpdir1 = tmpdir_factory.mktemp('')
tmpdir2 = tmpdir_factory.mktemp('')
# Making use of the assumption that it's the same project in both invocations of open_workspace
alpha_element, project, alpha_workspace = open_workspace(cli, tmpdir1, datafiles, "git", False, suffix="-alpha")
beta_element, _, beta_workspace = open_workspace(cli, tmpdir2, datafiles, "git", False, suffix="-beta")
arg_elm = [alpha_element] if not guess_element else []
result = cli.run(project=project, args=['-C', alpha_workspace, 'workspace', 'close'] + arg_elm)
result.assert_success()
@pytest.mark.datafiles(DATA_DIR)
def test_external_reset_other(cli, datafiles, tmpdir_factory):
tmpdir1 = tmpdir_factory.mktemp('')
tmpdir2 = tmpdir_factory.mktemp('')
# Making use of the assumption that it's the same project in both invocations of open_workspace
alpha_element, project, alpha_workspace = open_workspace(cli, tmpdir1, datafiles, "git", False, suffix="-alpha")
beta_element, _, beta_workspace = open_workspace(cli, tmpdir2, datafiles, "git", False, 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, "git", False)
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('')
# Making use of the assumption that it's the same project in both invocations of open_workspace
element, project, workspace = open_workspace(cli, tmpdir, datafiles, "git", False)
result = cli.run(project=project, args=['-C', workspace, 'workspace', 'list'])
result.assert_success()
@pytest.mark.datafiles(DATA_DIR)
def test_nocache_open_messages(cli, tmpdir, datafiles):
workspace_object = WorkspaceCreater(cli, tmpdir, datafiles)
_, result = workspace_object.open_workspaces(('git', ), False)
# cli default WARN for source dropback possibility when no-cache flag is not passed
assert "WARNING: Workspace will be opened without the cached buildtree if not cached locally" in result.output
# cli WARN for source dropback happening when no-cache flag not given, but buildtree not available
assert "workspace will be opened with source checkout" in result.stderr
# cli default WARN for source dropback possibilty not given when no-cache flag is passed
tmpdir = os.path.join(str(tmpdir), "2")
workspace_object = WorkspaceCreater(cli, tmpdir, datafiles)
_, result = workspace_object.open_workspaces(('git', ), False, suffixes='1', no_cache=True)
assert "WARNING: Workspace will be opened without the cached buildtree if not cached locally" not in result.output
@pytest.mark.datafiles(DATA_DIR)
def test_nocache_reset_messages(cli, tmpdir, datafiles):
workspace_object = WorkspaceCreater(cli, tmpdir, datafiles)
workspaces, result = workspace_object.open_workspaces(('git', ), False)
element_name, workspace = workspaces[0]
project = workspace_object.project_path
# Modify workspace, without building so the artifact is not cached
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') as f:
f.write("PONY='pink'")
# Now reset the open workspace, this should have the
# effect of reverting our changes to the original source, as it
# was not originally opened with a cached buildtree and as such
# should not notify the user
result = cli.run(cwd=workspace_object.workspace_cmd, project=project, args=[
'workspace', 'reset', element_name
])
result.assert_success()
assert "original buildtree artifact not available" not in result.output
assert os.path.exists(os.path.join(workspace, 'usr', 'bin', 'hello'))
assert not os.path.exists(os.path.join(workspace, 'etc', 'pony.conf'))
# Close the workspace
result = cli.run(cwd=workspace_object.workspace_cmd, project=project, args=[
'workspace', 'close', '--remove-dir', element_name
])
result.assert_success()
# Build the workspace so we have a cached buildtree artifact for the element
assert cli.get_element_state(project, element_name) == 'buildable'
result = cli.run(project=project, args=['build', element_name])
result.assert_success()
# Opening the workspace after a build should lead to the cached buildtree being
# staged by default
result = cli.run(cwd=workspace_object.workspace_cmd, project=project, args=[
'workspace', 'open', element_name
])
result.assert_success()
result = cli.run(cwd=workspace_object.workspace_cmd, project=project, args=[
'workspace', 'list'
])
result.assert_success()
# Now reset the workspace and ensure that a warning is not given about the artifact
# buildtree not being available
result = cli.run(cwd=workspace_object.workspace_cmd, project=project, args=[
'workspace', 'reset', element_name
])
result.assert_success()
assert "original buildtree artifact not available" not in result.output