diff options
Diffstat (limited to 'tests/sources/tar.py')
| -rw-r--r-- | tests/sources/tar.py | 86 |
1 files changed, 86 insertions, 0 deletions
diff --git a/tests/sources/tar.py b/tests/sources/tar.py index 1fd79f10b..1a1f54f87 100644 --- a/tests/sources/tar.py +++ b/tests/sources/tar.py @@ -3,11 +3,13 @@ import pytest import tarfile import tempfile import subprocess +import urllib.parse from shutil import copyfile, rmtree from buildstream._exceptions import ErrorDomain from buildstream import _yaml from tests.testutils import cli +from tests.testutils.file_server import create_file_server from tests.testutils.site import HAVE_LZIP from . import list_dir_contents @@ -49,6 +51,16 @@ def generate_project(project_dir, tmpdir): }, project_file) +def generate_project_file_server(base_url, project_dir): + project_file = os.path.join(project_dir, "project.conf") + _yaml.dump({ + 'name': 'foo', + 'aliases': { + 'tmpdir': base_url + } + }, project_file) + + # Test that without ref, consistency is set appropriately. @pytest.mark.datafiles(os.path.join(DATA_DIR, 'no-ref')) def test_no_ref(cli, tmpdir, datafiles): @@ -302,3 +314,77 @@ def test_read_only_dir(cli, tmpdir, datafiles): else: os.remove(path) rmtree(str(tmpdir), onerror=make_dir_writable) + + +@pytest.mark.parametrize('server_type', ('FTP', 'HTTP')) +@pytest.mark.datafiles(os.path.join(DATA_DIR, 'fetch')) +def test_use_netrc(cli, datafiles, server_type, tmpdir): + file_server_files = os.path.join(str(tmpdir), 'file_server') + fake_home = os.path.join(str(tmpdir), 'fake_home') + os.makedirs(file_server_files, exist_ok=True) + os.makedirs(fake_home, exist_ok=True) + project = str(datafiles) + checkoutdir = os.path.join(str(tmpdir), 'checkout') + + os.environ['HOME'] = fake_home + with open(os.path.join(fake_home, '.netrc'), 'wb') as f: + os.fchmod(f.fileno(), 0o700) + f.write(b'machine 127.0.0.1\n') + f.write(b'login testuser\n') + f.write(b'password 12345\n') + + with create_file_server(server_type) as server: + server.add_user('testuser', '12345', file_server_files) + generate_project_file_server(server.base_url(), project) + + src_tar = os.path.join(file_server_files, 'a.tar.gz') + _assemble_tar(os.path.join(str(datafiles), 'content'), 'a', src_tar) + + server.start() + + 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() + + original_dir = os.path.join(str(datafiles), 'content', 'a') + original_contents = list_dir_contents(original_dir) + checkout_contents = list_dir_contents(checkoutdir) + assert(checkout_contents == original_contents) + + +@pytest.mark.parametrize('server_type', ('FTP', 'HTTP')) +@pytest.mark.datafiles(os.path.join(DATA_DIR, 'fetch')) +def test_netrc_already_specified_user(cli, datafiles, server_type, tmpdir): + file_server_files = os.path.join(str(tmpdir), 'file_server') + fake_home = os.path.join(str(tmpdir), 'fake_home') + os.makedirs(file_server_files, exist_ok=True) + os.makedirs(fake_home, exist_ok=True) + project = str(datafiles) + checkoutdir = os.path.join(str(tmpdir), 'checkout') + + os.environ['HOME'] = fake_home + with open(os.path.join(fake_home, '.netrc'), 'wb') as f: + os.fchmod(f.fileno(), 0o700) + f.write(b'machine 127.0.0.1\n') + f.write(b'login testuser\n') + f.write(b'password 12345\n') + + with create_file_server(server_type) as server: + server.add_user('otheruser', '12345', file_server_files) + parts = urllib.parse.urlsplit(server.base_url()) + base_url = urllib.parse.urlunsplit([parts[0]] + ['otheruser@{}'.format(parts[1])] + list(parts[2:])) + generate_project_file_server(base_url, project) + + src_tar = os.path.join(file_server_files, 'a.tar.gz') + _assemble_tar(os.path.join(str(datafiles), 'content'), 'a', src_tar) + + server.start() + + result = cli.run(project=project, args=['track', 'target.bst']) + result.assert_main_error(ErrorDomain.STREAM, None) + result.assert_task_error(ErrorDomain.SOURCE, None) |