| # Pylint doesn't play well with fixtures and dependency injection from pytest |
| # pylint: disable=redefined-outer-name |
| |
| import os |
| import zipfile |
| |
| import pytest |
| |
| from buildstream._exceptions import ErrorDomain |
| from buildstream import _yaml |
| from buildstream.testing import cli # pylint: disable=unused-import |
| from tests.testutils.file_server import create_file_server |
| from . import list_dir_contents |
| |
| DATA_DIR = os.path.join( |
| os.path.dirname(os.path.realpath(__file__)), |
| 'zip', |
| ) |
| |
| |
| def _assemble_zip(workingdir, dstfile): |
| old_dir = os.getcwd() |
| os.chdir(workingdir) |
| with zipfile.ZipFile(dstfile, "w") as zipfp: |
| for root, dirs, files in os.walk('.'): |
| names = dirs + files |
| names = [os.path.join(root, name) for name in names] |
| for name in names: |
| zipfp.write(name) |
| os.chdir(old_dir) |
| |
| |
| 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 generate_project_file_server(server, project_dir): |
| project_file = os.path.join(project_dir, "project.conf") |
| _yaml.dump({ |
| 'name': 'foo', |
| 'aliases': { |
| 'tmpdir': server.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): |
| project = str(datafiles) |
| 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.datafiles(os.path.join(DATA_DIR, 'fetch')) |
| def test_fetch_bad_url(cli, tmpdir, datafiles): |
| project = str(datafiles) |
| generate_project(project, tmpdir) |
| |
| # Try to fetch it |
| result = cli.run(project=project, args=[ |
| 'source', 'fetch', 'target.bst' |
| ]) |
| assert "FAILURE Try #" in result.stderr |
| result.assert_main_error(ErrorDomain.STREAM, None) |
| result.assert_task_error(ErrorDomain.SOURCE, None) |
| |
| |
| # Test that when I fetch with an invalid ref, it fails. |
| @pytest.mark.datafiles(os.path.join(DATA_DIR, 'fetch')) |
| def test_fetch_bad_ref(cli, tmpdir, datafiles): |
| project = str(datafiles) |
| generate_project(project, tmpdir) |
| |
| # Create a local tar |
| src_zip = os.path.join(str(tmpdir), "a.zip") |
| _assemble_zip(os.path.join(str(datafiles), "content"), src_zip) |
| |
| # Try to fetch it |
| result = cli.run(project=project, args=[ |
| 'source', '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.datafiles(os.path.join(DATA_DIR, 'fetch')) |
| def test_track_warning(cli, tmpdir, datafiles): |
| project = str(datafiles) |
| generate_project(project, tmpdir) |
| |
| # Create a local tar |
| src_zip = os.path.join(str(tmpdir), "a.zip") |
| _assemble_zip(os.path.join(str(datafiles), "content"), src_zip) |
| |
| # Track it |
| result = cli.run(project=project, args=[ |
| 'source', '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.datafiles(os.path.join(DATA_DIR, 'fetch')) |
| def test_stage_default_basedir(cli, tmpdir, datafiles): |
| project = str(datafiles) |
| generate_project(project, tmpdir) |
| checkoutdir = os.path.join(str(tmpdir), "checkout") |
| |
| # Create a local tar |
| src_zip = os.path.join(str(tmpdir), "a.zip") |
| _assemble_zip(os.path.join(str(datafiles), "content"), src_zip) |
| |
| # Track, fetch, build, checkout |
| result = cli.run(project=project, args=['source', 'track', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['source', 'fetch', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['build', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['artifact', 'checkout', 'target.bst', '--directory', 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", "a") |
| 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.datafiles(os.path.join(DATA_DIR, 'no-basedir')) |
| def test_stage_no_basedir(cli, tmpdir, datafiles): |
| project = str(datafiles) |
| generate_project(project, tmpdir) |
| checkoutdir = os.path.join(str(tmpdir), "checkout") |
| |
| # Create a local tar |
| src_zip = os.path.join(str(tmpdir), "a.zip") |
| _assemble_zip(os.path.join(str(datafiles), "content"), src_zip) |
| |
| # Track, fetch, build, checkout |
| result = cli.run(project=project, args=['source', 'track', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['source', 'fetch', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['build', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['artifact', 'checkout', 'target.bst', '--directory', 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.datafiles(os.path.join(DATA_DIR, 'explicit-basedir')) |
| def test_stage_explicit_basedir(cli, tmpdir, datafiles): |
| project = str(datafiles) |
| generate_project(project, tmpdir) |
| checkoutdir = os.path.join(str(tmpdir), "checkout") |
| |
| # Create a local tar |
| src_zip = os.path.join(str(tmpdir), "a.zip") |
| _assemble_zip(os.path.join(str(datafiles), "content"), src_zip) |
| |
| # Track, fetch, build, checkout |
| result = cli.run(project=project, args=['source', 'track', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['source', 'fetch', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['build', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['artifact', 'checkout', 'target.bst', '--directory', 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", "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_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, project) |
| |
| src_zip = os.path.join(file_server_files, 'a.zip') |
| _assemble_zip(os.path.join(str(datafiles), 'content'), src_zip) |
| |
| server.start() |
| |
| result = cli.run(project=project, args=['source', 'track', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['source', 'fetch', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['build', 'target.bst']) |
| result.assert_success() |
| result = cli.run(project=project, args=['artifact', 'checkout', 'target.bst', '--directory', 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 |