<file_sep>import os

import docker
from docker.errors import BuildError

from patchworkdocker.errors import PatchworkDockerError


class DockerBuildError(PatchworkDockerError):
    """
    Docker build error.
    """


def build_docker_image(image_name: str, context: str, dockerfile: str):
    """
    Builds a Docker image with the given tag from the given Dockerfile in the given context.
    :param image_name: image tag (can optionally include a version tag)
    :param context: context to build the image in (absolute file path)
    :param dockerfile: Dockerfile to build the image from (absolute file path)
    :raises BuildFailedError: raised if an error occurs during the build
    """
    if not os.path.isabs(context):
        raise ValueError(f"Context location must be absolute: {context}")
    if not os.path.isabs(dockerfile):
        raise ValueError(f"Dockerfile location must be absolute: {dockerfile}")

    client = docker.from_env()
    try:
        client.images.build(path=context, dockerfile=dockerfile, tag=image_name)
    except BuildError as e:
        raise DockerBuildError(f"Error building image: {image_name}") from e
<file_sep>ARG baseImage=python:3.8
FROM ${baseImage}

ENV INSTALL_DIRECTORY=/usr/local/src/patchwork-docker
ENV PYTHONPATH=${INSTALL_DIRECTORY}

COPY requirements.txt /tmp/requirements.txt
RUN pip install -r /tmp/requirements.txt

COPY . "${INSTALL_DIRECTORY}"

# TODO: Install into CLI
ENTRYPOINT ["python", "/usr/local/src/patchwork-docker/patchworkdocker/cli.py"]

CMD ["--help"]
<file_sep>#!/usr/bin/env bash
set -eu -o pipefail
shopt -s expand_aliases

WORKDIR="/root"

BASE_IMAGE="${BASE_IMAGE:-python:3.7}"
DOCKER_VERSION="${DOCKER_VERSION:-18.06.1-ce}"
userArguments="$@"

# Not assuming jq is installed on the machine
alias djq="docker run --rm -i endeveit/docker-jq jq"

# Bash implementation of `readlink` (which is not on OSX)
function readlinkx {
    echo "$(cd "$(dirname "$1")"; pwd)/$(basename "$1")"
}

# Creates mount Docker CLI setting for the given location
function createMountSetting {
    location="$1"
    if [[ "${location}" != /* ]]; then
        mountLocation="${WORKDIR}/$(echo "${location}" | sed -e 's/^[\.\/]*//g')"
    else
        mountLocation="${location}"
    fi
    echo "-v $(readlinkx ${location}):${mountLocation}"
}

>&2 echo "Building Docker image for patchworkdocker (BASE_IMAGE=${BASE_IMAGE}; DOCKER_VERSION=${DOCKER_VERSION})"
docker build -t patchworkdocker \
    --build-arg baseImage="${BASE_IMAGE}" --build-arg "dockerVersion=${DOCKER_VERSION}" . > /dev/null

>&2 echo "Determining configuration"
configuration=$(docker run --rm -i patchworkdocker --dry-run ${userArguments})
if [[ ! "${configuration}" =~ ^{.* ]]; then
    # Dry run not written JSON (probably written help)
    >&2 echo "${configuration}"
    exit 0
fi
>&2 echo "Configuration: ${configuration}"

>&2 echo "Preparing inputs"
inputLocations="$(echo ${configuration} | djq -r '[.additional_files, .patches | keys] | flatten | .[]')"
echo "${inputLocations}" | while read location; do
    if [[ ! -e "${location}" ]]; then
        >&2 echo "Input location does not exist: ${location}"
        exit 1
    fi
done
>&2 echo "Inputs: $(echo ${inputLocations} | sed 's/\n/ /g')"

>&2 echo "Preparing outputs"
buildLocation=$(echo "${configuration}" | djq -r '.build_location')
if [[ "${buildLocation}" == "null" ]]; then
    buildLocation=$(mktemp -d /tmp/patchworkdocker.XXXXXX)
    trap "rm -rf ${buildLocation}" EXIT
    userArguments="${userArguments} --build-location ${buildLocation}"
fi
outputLocations="${buildLocation}"
>&2 echo "Outputs: $(echo ${outputLocations} | sed 's/\n/ /g')"

>&2 echo "Running patchworkdocker..."
docker run --rm -i -v /var/run/docker.sock:/var/run/docker.sock:ro \
    $(echo "${outputLocations}" | while read location; do echo "$(createMountSetting ${location}) "; done) \
    $(echo "${inputLocations}" | while read location; do echo "$(createMountSetting ${location}):ro "; done) \
    patchworkdocker ${userArguments}
<file_sep>from setuptools import setup, find_packages

from patchworkdocker.meta import VERSION, DESCRIPTION, PACKAGE_NAME, EXECUTABLE_NAME

setup(
    name=PACKAGE_NAME,
    version=VERSION,
    author="<NAME>",
    author_email="<EMAIL>",
    packages=find_packages(exclude=["tests"]),
    install_requires=open("requirements.txt", "r").readlines(),
    url="https://github.com/colin-nolan/patchwork-docker",
    license="MIT",
    description=DESCRIPTION,
    long_description=open("README.md", "r").read(),
    long_description_content_type="text/markdown",
    entry_points={
        "console_scripts": [
            f"{EXECUTABLE_NAME}={PACKAGE_NAME}.cli:entrypoint"
        ]
    },
    zip_safe=True
)
<file_sep>../../subrepos/colin-nolan/key_value_string_parser.py/key_value_string_parser.py<file_sep>import os
import shutil
from typing import Dict, Optional

from frozendict import frozendict
from logzero import logger

from patchworkdocker.docker_images import build_docker_image
from patchworkdocker.importers import ImporterFactory
from patchworkdocker.modifiers import copy_file, apply_patch, change_base_image

_importer_factory = ImporterFactory()


class PatchworkDocker:
    """
    Builds patchwork Docker images.
    """
    @property
    def dockerfile_location(self) -> str:
        return self._dockerfile_location

    @dockerfile_location.setter
    def dockerfile_location(self, location: str):
        if os.path.isabs(location):
            raise ValueError(f"Dockerfile location must be relative to the context root: {location}")
        self._dockerfile_location = location

    def __init__(self, import_repository_from: str, *, additional_files: Dict[str, Optional[str]]=(),
                 patches: Dict[str, str]=frozendict(), dockerfile_location: str="Dockerfile", base_image: str=None):
        """
        Constructor.
        :param import_repository_from: where to import the starting materials for the image from
        :param additional_files: files to add to the build context, where the key is the location on the host and the
        value (if given) is the relative location in the build context (added in the order given, overwrites possible)
        :param patches: patches to apply to files in the build context, where the key is the location of the patch file
        and the value is the location of the file to apply it to, relative to the build context root (applied in the
        order given)
        :param dockerfile_location: location of the Dockerfile to build, relative to the root of the repository
        :param base_image: Docker base image to change to
        """
        self._dockerfile_location = None
        self.import_repository_from = import_repository_from
        self.additional_files = additional_files
        self.patches = patches
        self.dockerfile_location = dockerfile_location
        self.base_image = base_image

    def build(self, image_name: str, build_directory: str=None):
        """
        Builds the patchworked Docker image.
        :param image_name: image tag (can optionally include a version tag)
        :param build_directory: directory to build in
        """
        repository_location = self.prepare(build_directory)
        dockerfile_location = os.path.join(repository_location, self.dockerfile_location)
        try:
            build_docker_image(image_name, repository_location, dockerfile_location)
        finally:
            if build_directory is None:
                logger.info(f"Removing temp build directory: {repository_location}")
                shutil.rmtree(repository_location)
            else:
                logger.info(f"Not removing build directory as directory was given by the user: {repository_location}")

    def prepare(self, build_directory: str=None) -> str:
        """
        Prepare a directory with the patched build materials.
        :param build_directory: the directory to load the patched build context in
        :return: the location of the build directory
        """
        if build_directory is not None:
            build_directory = os.path.abspath(build_directory)
            if len(os.listdir(path=build_directory)) > 0:
                raise ValueError(f"Build directory {build_directory} is not empty")

        repository_location = _importer_factory.create(self.import_repository_from).load(
            self.import_repository_from, build_directory)
        logger.info(f"Imported repository at {self.import_repository_from} to {repository_location}")

        for src, dest in self.additional_files.items():
            src = os.path.abspath(src)
            if dest is None:
                dest = os.path.basename(src)
            if os.path.isabs(dest):
                raise ValueError(f"Destination must be relative to the root of the context: {dest}")
            dest = os.path.join(repository_location, dest)
            os.path.exists(src), os.path.exists(dest)
            logger.info(f"{'Overwriting' if os.path.exists(dest) else 'Creating'} {dest} with {src}")
            copy_file(src, dest)

        for src, dest in self.patches.items():
            src = os.path.abspath(src)
            dest = os.path.join(repository_location, dest)
            os.path.exists(src), os.path.exists(dest)
            logger.info(f"Patching {dest} with {src}")
            apply_patch(src, dest)

        if self.base_image is not None:
            logger.info(f"Setting base image to {self.base_image}")
            dockerfile_location = os.path.join(repository_location, self.dockerfile_location)
            change_base_image(dockerfile_location, self.base_image)

        return repository_location
<file_sep>[![Build Status](https://travis-ci.org/colin-nolan/patchwork-docker.svg?branch=master)](https://travis-ci.org/colin-nolan/patchwork-docker)
[![codecov](https://codecov.io/gh/colin-nolan/patchwork-docker/branch/master/graph/badge.svg)](https://codecov.io/gh/colin-nolan/patchwork-docker)

# Patchwork Docker

## Purpose
Allows a Docker build to be easily changed, without having to clone and modify an existing build repository. 


## Features
All with a single command, the tool can:
- Build images from a different base image.
- Add/override files in the build context.
- Apply patches to the Dockerfile or other files in the build context.
- Define a different Dockerfile.


## Use Cases
A few basic use cases (`./docker-run.sh` can be used instead of `patchworkdocker`):
- To change a base image from `debian:stretch` to `python:3.7-stretch` so as to get the latest version of Python in the 
  image with no hassle:
  ```bash
  patchworkdocker build \
      --base-image python:3.7-stretch \
      example:1.0.0 https://github.com/example/docker-example.git
  ```
- Add an alternate Dockerfile to a pre-existing context:
  ```bash
  patchworkdocker build \
      --additional-file Dockerfile:Dockerfile.example \
      --dockerfile Dockerfile.example \
      https://github.com/example/docker-example.git example:1.0.0
  ```
- Change the URL of a piece of software that gets installed into a Docker image:
  ```bash
  patchworkdocker build \
      --dockerfile 3.7/stretch/Dockerfile \
      --patch change-install-url.patch:Dockerfile \
      https://github.com/docker-library/python.git example:1.0.0
  ```

Raspberry Pi users may find this tool particularly useful as, outside the official images, there is often need to 
change image build in order to get them to work on the non-standard rpi architectures. 


## Installation
Prerequisites
- Docker
- Python 3.7+ (optional, as can be run entirely from Docker if you don't have Python 3.7)

The tool can be installed directly from GitHub:
```bash
pip install git+https://github.com/colin-nolan/patchwork-docker@master#egg=patchworkdocker
```


## Usage
Up-to-date usage information can be seen with:
```bash
patchworkdocker --help
patchworkdocker build --help
```


## Legal
This work is in no way related to the company that I work for.
<file_sep>import os
import shutil
import unittest
from pathlib import Path

from patchworkdocker.modifiers import copy_file, apply_patch
from patchworkdocker.tests._common import TestWithTempFiles

_RESOURCES_LOCATION = os.path.join(os.path.dirname(os.path.abspath(__file__)), "resources")

_EXAMPLE_FILE_NAME = "test-file"
_EXAMPLE_FILE_NAME_2 = "test-file-2"


class TestCopyFile(TestWithTempFiles):
    """
    Tests for `copy_file`.
    """
    def setUp(self):
        super().setUp()
        self.src_directory = self.temp_manager.create_temp_directory()
        self.dest_directory = self.temp_manager.create_temp_directory()

    def test_empty_src_to_empty_dest(self):
        copy_file(self.src_directory, self.dest_directory)
        self.assertEquals(0, len(os.listdir(self.dest_directory)))

    def test_src_to_empty_dest(self):
        Path(os.path.join(self.src_directory, _EXAMPLE_FILE_NAME)).touch()
        copy_file(self.src_directory, self.dest_directory)
        self.assertTrue(os.path.exists(os.path.join(self.dest_directory, _EXAMPLE_FILE_NAME)))
        self.assertEquals(1, len(os.listdir(self.dest_directory)))

    def test_src_to_dest(self):
        Path(os.path.join(self.dest_directory, _EXAMPLE_FILE_NAME_2)).touch()
        Path(os.path.join(self.src_directory, _EXAMPLE_FILE_NAME)).touch()
        copy_file(self.src_directory, self.dest_directory)
        self.assertTrue(os.path.exists(os.path.join(self.dest_directory, _EXAMPLE_FILE_NAME)))
        self.assertEquals(2, len(os.listdir(self.dest_directory)))

    def test_src_to_dest_with_directory_merge(self):
        Path(os.path.join(self.dest_directory, _EXAMPLE_FILE_NAME_2)).touch()
        Path(os.path.join(self.src_directory, _EXAMPLE_FILE_NAME)).touch()
        copy_file(self.src_directory, self.dest_directory)
        self.assertTrue(os.path.exists(os.path.join(self.dest_directory, _EXAMPLE_FILE_NAME)))
        self.assertEquals(2, len(os.listdir(self.dest_directory)))

    def test_src_file_to_empty_dest(self):
        file_location = os.path.join(self.src_directory, _EXAMPLE_FILE_NAME)
        Path(file_location).touch()
        copy_file(file_location, self.dest_directory)
        self.assertTrue(os.path.exists(os.path.join(self.dest_directory, _EXAMPLE_FILE_NAME)))
        self.assertEquals(1, len(os.listdir(self.dest_directory)))

    def test_src_file_to_dest_with_overwrite(self):
        file_location = os.path.join(self.src_directory, _EXAMPLE_FILE_NAME)
        print("1", file=open(os.path.join(self.dest_directory, _EXAMPLE_FILE_NAME), "w"), end="")
        print("1", file=open(os.path.join(self.dest_directory, _EXAMPLE_FILE_NAME_2), "w"), end="")
        print("2", file=open(file_location, "w"), end="")
        copy_file(file_location, self.dest_directory)
        self.assertEquals("2", open(os.path.join(self.dest_directory, _EXAMPLE_FILE_NAME), "r").read())
        self.assertEquals(2, len(os.listdir(self.dest_directory)))


class TestApplyPatch(TestWithTempFiles):
    """
    Tests for `apply_patch`.
    """
    _DOCKERFILE_NAME = "Dockerfile"
    _EXAMPLE_DOCKERFILE_LOCATION = os.path.join(_RESOURCES_LOCATION, "patching", _DOCKERFILE_NAME)

    def setUp(self):
        super().setUp()
        temp_directory = self.temp_manager.create_temp_directory()
        self._dockerfile_location = os.path.join(temp_directory, TestApplyPatch._DOCKERFILE_NAME)
        shutil.copyfile(TestApplyPatch._EXAMPLE_DOCKERFILE_LOCATION, self._dockerfile_location)

    def test_change_from(self):
        patched_content = self._apply(f"{_RESOURCES_LOCATION}/patching/from-change.patch")
        self.assertTrue(patched_content.startswith("FROM arm32v7/ubuntu:16.04"))

    def test_add_and_remove(self):
        patched_content = self._apply(f"{_RESOURCES_LOCATION}/patching/add-and-remove.patch")
        self.assertTrue("RUN /other.sh" in patched_content)
        self.assertTrue("COPY . /data" not in patched_content)

    def _apply(self, patch_location: str) -> str:
        """
        Applies the given patch to the example Docker file.
        :param patch_location: the location of the patch to apply
        :return: the contents of the example Docker file after the patch has been applied
        """
        apply_patch(patch_location, self._dockerfile_location)
        with open(self._dockerfile_location, "r") as file:
            return file.read()


if __name__ == "__main__":
    unittest.main()
<file_sep>import os
from abc import ABCMeta, abstractmethod
from distutils import dir_util
from tempfile import mkdtemp
from urllib.parse import urldefrag, urlparse

from git import Repo


class Importer(metaclass=ABCMeta):
    """
    Imports a Docker build directory.
    """
    @abstractmethod
    def _load(self, origin: str, load_directory: str) -> str:
        """
        Loads the build directory from the given origin into the given directory.
        :param origin: where to import materials from
        :param load_directory: the directory in which imported materials should be saved
        :return: directory containing the loaded content
        """

    def load(self, origin: str, load_directory: str=None) -> str:
        """
        Loads build directory from the given origin into a load directory, which can be specified or is a generated
        temp directory if `None`.

        The returned directory is not cleaned up automatically.
        :param origin: where to import materials from
        :param load_directory: the directory in which imported materials should be saved
        :return: directory containing the loaded content
        """
        if load_directory is None:
            load_directory = mkdtemp()
        return self._load(origin, load_directory)


class GitImporter(Importer):
    """
    Imports content from a git repository.
    
    For a specific commit, branch or tag, set the fragment, e.g. http://example.com/repo.git#branch_tag_or_commit.
    """
    def _load(self, origin: str, load_directory: str) -> str:
        origin, branch = urldefrag(origin)
        repository = Repo.clone_from(url=origin, to_path=load_directory)

        if branch != "":
            if branch not in repository.heads:
                branch_reference = None
                for reference in repository.refs:
                    if reference.name == f"origin/{branch}":
                        branch_reference = reference
                        break
                if branch_reference is not None:
                    commit = branch_reference.commit
                else:
                    commit = repository.commit(branch)
                repository.create_head(path=branch, commit=commit)
            repository.heads[branch].checkout()

        return load_directory


class FileSystemImporter(Importer):
    """
    Imports content from the local file system.
    """
    def _load(self, origin: str, load_directory: str) -> str:
        dir_util.copy_tree(origin, load_directory)
        return load_directory


class ImporterFactory:
    """
    Importer factory, which can create the correct importer for an origin.
    """
    def create(self, origin: str) -> Importer:
        """
        Create an importer determined by analysis of the given origin.
        :param origin: where to import materials from
        :return: importer for the given origin
        """
        if os.path.exists(origin):
            return FileSystemImporter()

        parsed_origin = urlparse(origin)
        if parsed_origin.scheme == "git" or parsed_origin.path.endswith(".git"):
            # XXX: it is possible that there's a Git repo at a location that does not have these attributes...
            return GitImporter()

        raise NotImplementedError(f"No importer implemented to work with: {origin}")
<file_sep>class PatchworkDockerError(Exception):
    """
    Base error for all exceptions raised by this module.
    """
<file_sep># Copyright (c) 2018 Genome Research Limited
#
# Permission is hereby granted, free of charge, to any person obtaining a
# copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be included
# in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

import json
from argparse import Action
from json import JSONDecodeError


class KeyValueStringParserAction(Action):
    """
    Parses input in the form `xxx:yyy` or a stringified JSON map as a key value pair.

    Accepts multiple key-value pairs in stringified JSON map.

    e.g.
    ```
    parser.add_argument(f"-{MY_SHORT_PARAMETER}", f"--{MY_LONG_PARAMETER}", action=StringDictParseAction)
    ```
    """
    def __call__(self, parser, namespace, values, option_string=None):
        try:
            value_as_json = json.loads(values)
            if type(value_as_json) is not dict:
                raise ValueError(f"Not an acceptable JSON value: {values}")
            getattr(namespace, self.dest, {}).update(value_as_json)
        except JSONDecodeError:
            if ":" not in values:
                raise ValueError(f"Unable to parse: {values}. Must be in form \"xxx:yyy\"")
            key, value = values.split(":")
            getattr(namespace, self.dest, {})[key] = value
<file_sep>import os
import shutil
import unittest

import docker
from capturewrap import CaptureWrapBuilder, CaptureResult

from patchworkdocker.cli import main
from patchworkdocker.tests._common import TestWithTempFiles, EXAMPLE_BUILD_DIRECTORY, create_image_name


class CliTest(TestWithTempFiles):
    """
    TODO
    """
    def setUp(self) -> None:
        super().setUp()
        self.capture_wrap_builder = CaptureWrapBuilder(capture_stdout=True, capture_stderr=True)
        self.capture_wrap_builder.capture_exceptions = lambda e: isinstance(e, SystemExit)

    def test_no_arguments(self):
        self.capture_wrap_builder.capture_exceptions = True
        result = self._call_wrapped_main([])
        self.assertNotEqual(result.exception.code, 0)
        self.assertTrue(result.stderr.startswith("usage:"))

    def test_help(self):
        result = self._call_wrapped_main(["--help"])
        self.assertEqual(result.exception.code, 0)
        self.assertTrue(result.stdout.startswith("usage:"))

    def test_invalid_action(self):
        result = self._call_wrapped_main(["invalid"])
        self.assertNotEqual(result.exception.code, 0)

    def test_basic_prepare(self):
        result = self._call_wrapped_main(["prepare", EXAMPLE_BUILD_DIRECTORY])
        directory = result.stdout.strip()
        try:
            self.assertTrue(os.path.exists(directory))
        finally:
            shutil.rmtree(directory, ignore_errors=True)

    def test_basic_build(self):
        image_name = create_image_name()
        client = docker.from_env()
        expected_output = open(os.path.join(EXAMPLE_BUILD_DIRECTORY, "hello-world.txt"), "rb").read()
        try:
            self._call_wrapped_main(["build", EXAMPLE_BUILD_DIRECTORY, image_name])
            output = client.containers.run(image_name)
            self.assertEqual(expected_output, output)
        finally:
            client.images.remove(image_name, force=True)

    def _call_wrapped_main(self, *args, **kwargs) -> CaptureResult:
        wrapped_main = self.capture_wrap_builder.build(main)
        return wrapped_main(*args, **kwargs)


if __name__ == "__main__":
    unittest.main()
<file_sep>#!/usr/bin/env bash
set -euf -o pipefail

# Remove old test coverage data
python -m coverage erase

# Run tests
PYTHONPATH=. python -m coverage run -m unittest discover -v -s patchworkdocker/tests
PYTHONPATH=. python -m coverage run patchworkdocker/cli.py -h
python -m coverage run setup.py -q install

# Generate coverage reports
python -m coverage combine -a
python -m coverage report
python -m coverage xml
<file_sep>import os
import unittest
from abc import abstractmethod
from pathlib import Path
from typing import TypeVar, Generic, Optional

from patchworkdocker.importers import GitImporter, Importer, FileSystemImporter
from patchworkdocker.tests._common import TestWithTempFiles, EXAMPLE_GIT_REPOSITORY

ImporterType = TypeVar("ImporterType", bound=Importer)


class _TestImporter(Generic[ImporterType], TestWithTempFiles):
    """
    Tests for `ImporterType`.
    """

    @property
    @abstractmethod
    def importer(self) -> ImporterType:
        """
        Gets an instance of the importer under test.
        :return: the importer to test
        """

    def setUp(self):
        super().setUp()
        self._importer: Optional[Importer] = None
        self._paths = []

    def load(self, origin: str) -> str:
        """
        Uses the importer under test and calls loads with the given origin. The returned path is removed on tear down.
        :param origin: the origin to load from
        :return: the path of where the import has been loaded to
        """
        path = self.importer.load(origin)
        self.temp_manager._temp_directories.add(path)
        return path


class TestGitImporter(_TestImporter[GitImporter]):
    """
    Tests for `GitImporter`.
    """

    @property
    def importer(self) -> ImporterType:
        if self._importer is None:
            self._importer = GitImporter()
        return self._importer

    def test_load(self):
        path = self.load(EXAMPLE_GIT_REPOSITORY)
        self.assertTrue(os.path.exists(os.path.join(path, "a/d.txt")))

    def test_load_commit(self):
        path = self.load(f"{EXAMPLE_GIT_REPOSITORY}#e22fcb940d5356f8dc57fa99d7a6cb4ecdc04b66")
        self.assertTrue(os.path.exists(os.path.join(path, "b.txt")))

    def test_load_branch(self):
        path = self.load(f"{EXAMPLE_GIT_REPOSITORY}#develop")
        self.assertTrue(os.path.exists(os.path.join(path, "develop.txt")))

    def test_load_tag(self):
        path = self.load(f"{EXAMPLE_GIT_REPOSITORY}#1.0")
        self.assertTrue(os.path.exists(os.path.join(path, "b.txt")))


class TestFileSystemImporter(_TestImporter[GitImporter]):
    """
    Tests for `FileSystemImporter`.
    """
    _EXAMPLE_FILE = "test.txt"

    @property
    def importer(self) -> ImporterType:
        if self._importer is None:
            self._importer = FileSystemImporter()
        return self._importer

    def setUp(self):
        super().setUp()
        self.test_directory = self.temp_manager.create_temp_directory()
        Path(os.path.join(self.test_directory, TestFileSystemImporter._EXAMPLE_FILE)).touch()

    def test_load(self):
        path = self.load(self.test_directory)
        self.assertTrue(os.path.exists(os.path.join(path, TestFileSystemImporter._EXAMPLE_FILE)))


if __name__ == "__main__":
    unittest.main()
<file_sep>GitPython
patch==1.*
logzero
frozendict
docker
<file_sep>PACKAGE_NAME = "patchworkdocker"
VERSION = "0.0.0"
DESCRIPTION = "TODO"
EXECUTABLE_NAME = "patchworkdocker"
<file_sep>import os
from abc import ABCMeta
from unittest import TestCase
from temphelpers import TempManager
from uuid import uuid4
from patchworkdocker.meta import PACKAGE_NAME

EXAMPLE_GIT_REPOSITORY = "https://github.com/colin-nolan/test-repository.git"
EXAMPLE_BUILD_DIRECTORY = os.path.join(os.path.dirname(os.path.abspath(__file__)), "resources", "building")


class TestWithTempFiles(TestCase, metaclass=ABCMeta):
    """
    Base class for tests that use temp files.
    """
    def setUp(self):
        self.temp_manager = TempManager()

    def tearDown(self):
        self.temp_manager.tear_down()


def create_image_name():
    """
    TODO
    :return:
    """
    return f"{PACKAGE_NAME}-test:{str(uuid4())}"
<file_sep>#!/usr/bin/env bash
set -euf -o pipefail

scriptDirectory="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
testDirectory="${scriptDirectory}"

# Note: `mktemp` creates temps in directory that cannot be mounted by default on Mac
tempDirectory="/tmp/patchwork-docker-${RANDOM}"
trap "echo ${tempDirectory} && rm -rf ${tempDirectory}" INT TERM HUP EXIT
mkdir "${tempDirectory}"

# XXX: could use docker-compose to build
docker build -t patchworkdocker -f Dockerfile .
docker build -t patchworkdocker-tests -f Dockerfile.test .
docker run --rm -it \
            -v /var/run/docker.sock:/var/run/docker.sock:ro \
            -v "${testDirectory}:${testDirectory}" \
            -v "${tempDirectory}:${tempDirectory}" \
            -e TMPDIR="${tempDirectory}" \
            --entrypoint=bash \
            -w "${testDirectory}" \
        patchworkdocker-tests ./run-tests.sh
<file_sep>import os
import unittest
import uuid

import docker

from patchworkdocker.docker_images import build_docker_image
from patchworkdocker.tests._common import TestWithTempFiles, EXAMPLE_BUILD_DIRECTORY


class TestBuildDockerImage(TestWithTempFiles):
    """
    Test `build_docker_image`.
    """
    def setUp(self):
        super().setUp()
        self._docker_client = docker.from_env()
        self._docker_image = f"{__name__}-{uuid.uuid4()}"

    def tearDown(self):
        super().tearDown()
        self._docker_client.images.remove(self._docker_image)

    def test_build_docker_image(self):
        build_docker_image(self._docker_image, EXAMPLE_BUILD_DIRECTORY, os.path.join(EXAMPLE_BUILD_DIRECTORY, "Dockerfile"))
        contents = self._docker_client.containers.run(self._docker_image, "cat /test.txt", remove=True).decode("UTF8")
        self.assertEqual(contents, open(os.path.join(EXAMPLE_BUILD_DIRECTORY, "hello-world.txt"), "r").read())


if __name__ == "__main__":
    unittest.main()
<file_sep>FROM ubuntu:16.04

COPY . /data
WORKDIR /data
RUN ./setup-1.sh \
    && ./setup-2.sh

ENTRYPOINT ["./entrypoint.sh"]
CMD ["--help"]
<file_sep>import dataclasses
import json
import logging
import sys
from argparse import ArgumentParser
from dataclasses import dataclass
from enum import Enum, unique
from typing import List, Dict, Optional, Union

import logzero
from logzero import logger

from patchworkdocker._external.key_value_string_parser import KeyValueStringParserAction
from patchworkdocker._external.verbosity_argument_parser import verbosity_parser_configuration, VERBOSE_PARAMETER_KEY, \
    get_verbosity, DEFAULT_LOG_VERBOSITY_KEY
from patchworkdocker.core import PatchworkDocker
from patchworkdocker.meta import EXECUTABLE_NAME, DESCRIPTION, VERSION

ACTION_PARAMETER = "action"
IMPORT_REPOSITORY_FROM_PARAMETER = "context"
ADDITIONAL_FILES_LONG_PARAMETER = "additional-file"
ADDITIONAL_FILES_SHORT_PARAMETER = "f"
PATCHES_LONG_PARAMETER = "patch"
PATCHES_SHORT_PARAMETER = "p"
DOCKERFILE_LOCATION_LONG_PARAMETER = "dockerfile"
DOCKERFILE_LOCATION_SHORT_PARAMETER = "d"
IMAGE_NAME_PARAMETER = "image-name"
BUILD_LOCATION_LONG_PARAMETER = "build-location"
BUILD_LOCATION_SHORT_PARAMETER = "b"
VERBOSITY_SHORT_PARAMETER = verbosity_parser_configuration[VERBOSE_PARAMETER_KEY]
DRY_RUN_LONG_PARAMETER = "dry-run"
BASE_IMAGE_SHORT_PARAMETER = "i"
BASE_IMAGE_LONG_PARAMETER = "base-image"

DEFAULT_ADDITIONAL_FILES = {}
DEFAULT_PATCHES = {}
DEFAULT_DOCKERFILE_LOCATION = "Dockerfile"
DEFAULT_VERBOSITY = verbosity_parser_configuration[DEFAULT_LOG_VERBOSITY_KEY]


@unique
class ActionValue(Enum):
    """
    Program action.
    """
    BUILD = "build"
    PREPARE = "prepare"


@dataclass
class BaseCliConfiguration:
    """
    Base CLI configuration.
    """
    log_verbosity: int
    dry_run: bool


@dataclass
class SubcommandCliConfiguration(BaseCliConfiguration):
    """
    CLI configuration for subcommands.
    """
    additional_files: Dict[str, str]
    patches: Dict[str, str]
    dockerfile_location: str
    build_location: Optional[str]
    import_from: str
    base_image: str


@dataclass
class PrepareCliConfiguration(SubcommandCliConfiguration):
    """
    CLI configuration for preparing a patchwork Docker build.
    """


@dataclass
class BuildCliConfiguration(SubcommandCliConfiguration):
    """
    CLI configuration for building a patchwork Docker image.
    """
    image_name: str


def _create_parser() -> ArgumentParser:
    """
    Creates an argument parser.
    :return: the created parser
    """
    parser = ArgumentParser(prog=EXECUTABLE_NAME, description=f"{DESCRIPTION} (v{VERSION})")
    parser.add_argument(f"-{VERBOSITY_SHORT_PARAMETER}", action="count", default=0,
                        help="increase the level of log verbosity (add multiple increase further)")
    parser.add_argument(f"--{DRY_RUN_LONG_PARAMETER}", action="store_true", default=False, help="")
    subparsers = parser.add_subparsers(dest=ACTION_PARAMETER, help="TODO")

    def take_context_arguments(parser: ArgumentParser):
        parser.add_argument(IMPORT_REPOSITORY_FROM_PARAMETER, help="TODO")

    def take_common_arguments(parser: ArgumentParser):
        parser.add_argument(f"-{ADDITIONAL_FILES_SHORT_PARAMETER}", f"--{ADDITIONAL_FILES_LONG_PARAMETER}",
                            action=KeyValueStringParserAction, default=DEFAULT_ADDITIONAL_FILES,
                            help="Files to add to the build context (will override existing files). Input in the form:")
        parser.add_argument(f"-{PATCHES_SHORT_PARAMETER}", f"--{PATCHES_LONG_PARAMETER}",
                            action=KeyValueStringParserAction, help="TODO", default=DEFAULT_PATCHES)
        parser.add_argument(f"-{DOCKERFILE_LOCATION_SHORT_PARAMETER}", f"--{DOCKERFILE_LOCATION_LONG_PARAMETER}",
                            help="TODO", default=DEFAULT_DOCKERFILE_LOCATION)
        parser.add_argument(f"-{BUILD_LOCATION_SHORT_PARAMETER}", f"--{BUILD_LOCATION_LONG_PARAMETER}",
                            help="TODO", default=None)
        parser.add_argument(f"-{BASE_IMAGE_SHORT_PARAMETER}", f"--{BASE_IMAGE_LONG_PARAMETER}",
                            help="TODO", default=None)

    build_parser = subparsers.add_parser(ActionValue.BUILD.value, help="TODO")
    take_context_arguments(build_parser)
    build_parser.add_argument(IMAGE_NAME_PARAMETER, help="TODO")
    take_common_arguments(build_parser)

    prepare_parser = subparsers.add_parser(ActionValue.PREPARE.value, help="TODO")
    take_context_arguments(prepare_parser)
    take_common_arguments(prepare_parser)

    return parser


def parse_cli_configuration(arguments: List[str]) -> BaseCliConfiguration:
    """
    Parses the given CLI arguments.
    :param arguments: the arguments from the CLI
    :return: parsed configuration
    """
    # XXX: Setting a value other than the display string seems to be non-trivial: https://bugs.python.org/issue23487
    parsed_arguments = _create_parser().parse_args(arguments)
    parsed_arguments = {x.replace("_", "-"): y for x, y in vars(parsed_arguments).items()}
    action_value = parsed_arguments[ACTION_PARAMETER]
    if action_value is None:
        logger.error("No action specified")
        _create_parser().print_help(sys.stderr)
        exit(1)
    parsed_arguments[ACTION_PARAMETER] = ActionValue(action_value)

    cli_configuration_class = {
        ActionValue.BUILD: BuildCliConfiguration,
        ActionValue.PREPARE: PrepareCliConfiguration
    }[parsed_arguments[ACTION_PARAMETER]]

    extra_configuration = {}
    if issubclass(cli_configuration_class, BuildCliConfiguration):
        extra_configuration["image_name"] = parsed_arguments[IMAGE_NAME_PARAMETER]

    cli_configuration = cli_configuration_class(
        log_verbosity=get_verbosity(parsed_arguments),
        dry_run=parsed_arguments[DRY_RUN_LONG_PARAMETER],
        additional_files=parsed_arguments[ADDITIONAL_FILES_LONG_PARAMETER],
        patches=parsed_arguments[PATCHES_LONG_PARAMETER],
        dockerfile_location=parsed_arguments[DOCKERFILE_LOCATION_LONG_PARAMETER],
        build_location=parsed_arguments[BUILD_LOCATION_LONG_PARAMETER],
        import_from=parsed_arguments[IMPORT_REPOSITORY_FROM_PARAMETER],
        base_image=parsed_arguments[BASE_IMAGE_LONG_PARAMETER],
        **extra_configuration
    )

    return cli_configuration


def _set_log_level(level: int):
    """
    Sets the log level to that given.
    :param level: log level to set
    """
    logzero.loglevel(level)
    if level == logging.WARNING:
        logger.warning("There are not likely to be many WARN level logs: consider increasing the verbosity by adding"
                       f"more -{VERBOSITY_SHORT_PARAMETER}")


def print_configuration(configuration: BaseCliConfiguration):
    """
    Prints information about the given configuration to stdout.
    :param configuration: configuration to output
    """
    configuration_as_json = json.dumps(dataclasses.asdict(configuration))
    print(configuration_as_json)


def build(core: PatchworkDocker, configuration: BuildCliConfiguration):
    """
    Builds patchwork Docker image with the given configuration.
    :param core: patchwork Docker core
    :param configuration: build configuration
    :return:
    """
    core.build(configuration.image_name, configuration.build_location)


def prepare(core: PatchworkDocker, configuration: PrepareCliConfiguration):
    """
    Prepares for patchwork Docker build.
    :param core: patchwork Docker core
    :param configuration: build configuration
    :return:
    """
    output = core.prepare(configuration.build_location)
    print(output)


def main(cli_arguments: List[str]):
    """
    Main.
    :param cli_arguments: arguments passed in via the CLI
    :raises SystemExit: always raised
    """
    cli_configuration = parse_cli_configuration(cli_arguments)

    if cli_configuration.log_verbosity:
        _set_log_level(cli_configuration.log_verbosity)

    if cli_configuration.dry_run:
        print_configuration(cli_configuration)
        exit(0)

    # XXX: Ideally, we would use `configuration: Intersect[ContextUsingCliConfiguration, SubcommandCliConfiguration]
    # but multiple bounds are sadly not supported in Python's type hinting: https://github.com/python/typing/issues/213
    def create_core(configuration: Union[PrepareCliConfiguration, BuildCliConfiguration]) -> PatchworkDocker:
        return PatchworkDocker(configuration.import_from, additional_files=configuration.additional_files,
                               patches=configuration.patches, dockerfile_location=configuration.dockerfile_location,
                               base_image=cli_configuration.base_image)

    {
        BuildCliConfiguration: lambda: build(create_core(cli_configuration), cli_configuration),
        PrepareCliConfiguration: lambda: prepare(create_core(cli_configuration), cli_configuration),
    }[type(cli_configuration)]()


def entrypoint():
    """
    Entry-point to be used by CLI.
    """
    logger.setLevel(DEFAULT_VERBOSITY)
    main(sys.argv[1:])


if __name__ == "__main__":
    entrypoint()
<file_sep>FROM alpine

ADD hello-world.txt /test.txt

CMD cat /test.txt<file_sep>import fileinput
import itertools
import os
import shutil
from distutils import dir_util
from tempfile import TemporaryDirectory

from patch import fromfile


def copy_file(file: str, destination: str):
    """
    Copy the given file (which could be a directory) to the given destination.
    :param file: the file (which could be a directory) to copy
    :param destination: where to copy the file to
    """
    if not os.path.exists(file):
        raise FileExistsError(f"Cannot copy file {file} as it does not exist")
    if os.path.isfile(file):
        shutil.copy(file, destination)
    else:
        dir_util.copy_tree(file, destination)


def apply_patch(patch_file: str, target_file: str):
    """
    Applies the given patch file (plain, git, mercurial or svn styles accepted) to the given target file.

    To create a compatible patch between two files, you could use:
    ```
    diff -uNr src_1 src_2
    ```
    :param patch_file: the patch to apply
    :param target_file: the patch target
    """
    patch_set = fromfile(patch_file)
    if not patch_set:
        raise SyntaxError(f"Could not parse contents of patch file: {patch_file}")

    hunks = list(itertools.chain(*[item.hunks for item in patch_set.items]))

    with TemporaryDirectory() as temp_directory:
        temp_file = os.path.join(temp_directory, os.path.basename(target_file))
        patch_set.write_hunks(target_file, os.path.join(temp_file), hunks)
        shutil.move(temp_file, target_file)


def change_base_image(dockerfile_location: str, desired_base: str):
    """
    TODO
    :param dockerfile_location:
    :param desired_base:
    :return:
    """
    changed = False
    for line in fileinput.input(dockerfile_location, inplace=True):
        if line.upper().startswith("FROM"):
            assert not changed
            print(f"FROM {desired_base}")
            changed = True
        else:
            print(line, end="")

    if not changed:
        raise ValueError(f"Dockerfile did not contain FROM line: {dockerfile_location}")
