1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180
|
import os
import pytest
import tarfile
import tempfile
import subprocess
import shutil
from buildstream._exceptions import ErrorDomain
from buildstream import _yaml
from tempfile import TemporaryFile
from tests.testutils import cli
from tests.testutils.site import HAVE_ARPY
from . import list_dir_contents
DATA_DIR = os.path.join(
os.path.dirname(os.path.realpath(__file__)),
'deb',
)
deb_name = "a_deb.deb"
def generate_project(project_dir, tmpdir):
project_file = os.path.join(project_dir, "project.conf")
_yaml.dump({
'name': 'foo',
'aliases': {
'tmpdir': "file:///" + str(tmpdir)
}
}, project_file)
def _copy_deb(start_location, tmpdir):
source = os.path.join(start_location, deb_name)
destination = os.path.join(str(tmpdir), deb_name)
shutil.copyfile(source, destination)
# Test that without ref, consistency is set appropriately.
@pytest.mark.skipif(HAVE_ARPY is False, reason="arpy is not available")
@pytest.mark.datafiles(os.path.join(DATA_DIR, 'no-ref'))
def test_no_ref(cli, tmpdir, datafiles):
project = os.path.join(datafiles.dirname, datafiles.basename)
generate_project(project, tmpdir)
assert cli.get_element_state(project, 'target.bst') == 'no reference'
# Test that when I fetch a nonexistent URL, errors are handled gracefully and a retry is performed.
@pytest.mark.skipif(HAVE_ARPY is False, reason="arpy is not available")
@pytest.mark.datafiles(os.path.join(DATA_DIR, 'fetch'))
def test_fetch_bad_url(cli, tmpdir, datafiles):
project = os.path.join(datafiles.dirname, datafiles.basename)
generate_project(project, tmpdir)
# Try to fetch it
result = cli.run(project=project, args=[
'fetch', 'target.bst'
])
assert "FAILURE Try #" in result.stderr
result.assert_main_error(ErrorDomain.STREAM, None)
result.assert_task_error(ErrorDomain.SOURCE, None)
@pytest.mark.skipif(HAVE_ARPY is False, reason="arpy is not available")
@pytest.mark.datafiles(os.path.join(DATA_DIR, 'fetch'))
def test_fetch_bad_ref(cli, tmpdir, datafiles):
project = os.path.join(datafiles.dirname, datafiles.basename)
generate_project(project, tmpdir)
# Copy test deb to tmpdir
_copy_deb(DATA_DIR, tmpdir)
# Try to fetch it
result = cli.run(project=project, args=[
'fetch', 'target.bst'
])
result.assert_main_error(ErrorDomain.STREAM, None)
result.assert_task_error(ErrorDomain.SOURCE, None)
# Test that when tracking with a ref set, there is a warning
@pytest.mark.skipif(HAVE_ARPY is False, reason="arpy is not available")
@pytest.mark.datafiles(os.path.join(DATA_DIR, 'fetch'))
def test_track_warning(cli, tmpdir, datafiles):
project = os.path.join(datafiles.dirname, datafiles.basename)
generate_project(project, tmpdir)
# Copy test deb to tmpdir
_copy_deb(DATA_DIR, tmpdir)
# Track it
result = cli.run(project=project, args=[
'track', 'target.bst'
])
result.assert_success()
assert "Potential man-in-the-middle attack!" in result.stderr
# Test that a staged checkout matches what was tarred up, with the default first subdir
@pytest.mark.skipif(HAVE_ARPY is False, reason="arpy is not available")
@pytest.mark.datafiles(os.path.join(DATA_DIR, 'fetch'))
def test_stage_default_basedir(cli, tmpdir, datafiles):
project = os.path.join(datafiles.dirname, datafiles.basename)
generate_project(project, tmpdir)
checkoutdir = os.path.join(str(tmpdir), "checkout")
# Copy test deb to tmpdir
_copy_deb(DATA_DIR, tmpdir)
# Track, fetch, build, checkout
result = cli.run(project=project, args=['track', 'target.bst'])
result.assert_success()
result = cli.run(project=project, args=['fetch', 'target.bst'])
result.assert_success()
result = cli.run(project=project, args=['build', 'target.bst'])
result.assert_success()
result = cli.run(project=project, args=['checkout', 'target.bst', checkoutdir])
result.assert_success()
# Check that the content of the first directory is checked out (base-dir: '')
original_dir = os.path.join(str(datafiles), "content")
original_contents = list_dir_contents(original_dir)
checkout_contents = list_dir_contents(checkoutdir)
assert(checkout_contents == original_contents)
# Test that a staged checkout matches what was tarred up, with an empty base-dir
@pytest.mark.skipif(HAVE_ARPY is False, reason="arpy is not available")
@pytest.mark.datafiles(os.path.join(DATA_DIR, 'no-basedir'))
def test_stage_no_basedir(cli, tmpdir, datafiles):
project = os.path.join(datafiles.dirname, datafiles.basename)
generate_project(project, tmpdir)
checkoutdir = os.path.join(str(tmpdir), "checkout")
# Copy test deb to tmpdir
_copy_deb(DATA_DIR, tmpdir)
# Track, fetch, build, checkout
result = cli.run(project=project, args=['track', 'target.bst'])
result.assert_success()
result = cli.run(project=project, args=['fetch', 'target.bst'])
result.assert_success()
result = cli.run(project=project, args=['build', 'target.bst'])
result.assert_success()
result = cli.run(project=project, args=['checkout', 'target.bst', checkoutdir])
result.assert_success()
# Check that the full content of the tarball is checked out (base-dir: '')
original_dir = os.path.join(str(datafiles), "content")
original_contents = list_dir_contents(original_dir)
checkout_contents = list_dir_contents(checkoutdir)
assert(checkout_contents == original_contents)
# Test that a staged checkout matches what was tarred up, with an explicit basedir
@pytest.mark.skipif(HAVE_ARPY is False, reason="arpy is not available")
@pytest.mark.datafiles(os.path.join(DATA_DIR, 'explicit-basedir'))
def test_stage_explicit_basedir(cli, tmpdir, datafiles):
project = os.path.join(datafiles.dirname, datafiles.basename)
generate_project(project, tmpdir)
checkoutdir = os.path.join(str(tmpdir), "checkout")
# Copy test deb to tmpdir
_copy_deb(DATA_DIR, tmpdir)
# Track, fetch, build, checkout
result = cli.run(project=project, args=['track', 'target.bst'])
result.assert_success()
result = cli.run(project=project, args=['fetch', 'target.bst'])
result.assert_success()
result = cli.run(project=project, args=['build', 'target.bst'])
result.assert_success()
result = cli.run(project=project, args=['checkout', 'target.bst', checkoutdir])
result.assert_success()
# Check that the content of the first directory is checked out (base-dir: '')
original_dir = os.path.join(str(datafiles), "content")
original_contents = list_dir_contents(original_dir)
checkout_contents = list_dir_contents(checkoutdir)
assert(checkout_contents == original_contents)
|