tests/sources/zip.py

# 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.testing import generate_project
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)


# 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, config={"aliases": {"tmpdir": "file:///" + str(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, config={"aliases": {"tmpdir": "file:///" + str(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, config={"aliases": {"tmpdir": "file:///" + str(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, config={"aliases": {"tmpdir": "file:///" + str(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, config={"aliases": {"tmpdir": "file:///" + str(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, config={"aliases": {"tmpdir": "file:///" + str(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, config={"aliases": {"tmpdir": "file:///" + str(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(project, config={"aliases": {"tmpdir": server.base_url()}})

        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