tests/integration/models_images_test.py

import io
import tempfile

import pytest

import docker

from ..helpers import random_name
from .base import TEST_API_VERSION, TEST_IMG, BaseIntegrationTest


class ImageCollectionTest(BaseIntegrationTest):

    def test_build(self):
        client = docker.from_env(version=TEST_API_VERSION)
        image, _ = client.images.build(fileobj=io.BytesIO(
            b"FROM alpine\n"
            b"CMD echo hello world"
        ))
        self.tmp_imgs.append(image.id)
        assert client.containers.run(image) == b"hello world\n"

    # @pytest.mark.xfail(reason='Engine 1.13 responds with status 500')
    def test_build_with_error(self):
        client = docker.from_env(version=TEST_API_VERSION)
        with pytest.raises(docker.errors.BuildError) as cm:
            client.images.build(fileobj=io.BytesIO(
                b"FROM alpine\n"
                b"RUN exit 1"
            ))
        assert (
            "The command '/bin/sh -c exit 1' returned a non-zero code: 1"
        ) in cm.exconly()
        assert cm.value.build_log

    def test_build_with_multiple_success(self):
        client = docker.from_env(version=TEST_API_VERSION)
        image, _ = client.images.build(
            tag='some-tag', fileobj=io.BytesIO(
                b"FROM alpine\n"
                b"CMD echo hello world"
            )
        )
        self.tmp_imgs.append(image.id)
        assert client.containers.run(image) == b"hello world\n"

    def test_build_with_success_build_output(self):
        client = docker.from_env(version=TEST_API_VERSION)
        image, _ = client.images.build(
            tag='dup-txt-tag', fileobj=io.BytesIO(
                b"FROM alpine\n"
                b"CMD echo Successfully built abcd1234"
            )
        )
        self.tmp_imgs.append(image.id)
        assert client.containers.run(image) == b"Successfully built abcd1234\n"

    def test_list(self):
        client = docker.from_env(version=TEST_API_VERSION)
        image = client.images.pull('alpine:latest')
        assert image.id in get_ids(client.images.list())

    def test_list_with_repository(self):
        client = docker.from_env(version=TEST_API_VERSION)
        image = client.images.pull('alpine:latest')
        assert image.id in get_ids(client.images.list('alpine'))
        assert image.id in get_ids(client.images.list('alpine:latest'))

    def test_pull(self):
        client = docker.from_env(version=TEST_API_VERSION)
        image = client.images.pull('alpine:latest')
        assert 'alpine:latest' in image.attrs['RepoTags']

    def test_pull_with_tag(self):
        client = docker.from_env(version=TEST_API_VERSION)
        image = client.images.pull('alpine', tag='3.10')
        assert 'alpine:3.10' in image.attrs['RepoTags']

    def test_pull_with_sha(self):
        image_ref = (
            'hello-world@sha256:083de497cff944f969d8499ab94f07134c50bcf5e6b95'
            '59b27182d3fa80ce3f7'
        )
        client = docker.from_env(version=TEST_API_VERSION)
        image = client.images.pull(image_ref)
        assert image_ref in image.attrs['RepoDigests']

    def test_pull_multiple(self):
        client = docker.from_env(version=TEST_API_VERSION)
        images = client.images.pull('hello-world', all_tags=True)
        assert len(images) >= 1
        assert any('hello-world:latest' in img.attrs['RepoTags'] for img in images)

    def test_load_error(self):
        client = docker.from_env(version=TEST_API_VERSION)
        with pytest.raises(docker.errors.ImageLoadError):
            client.images.load('abc')

    def test_save_and_load(self):
        client = docker.from_env(version=TEST_API_VERSION)
        image = client.images.get(TEST_IMG)
        with tempfile.TemporaryFile() as f:
            stream = image.save()
            for chunk in stream:
                f.write(chunk)

            f.seek(0)
            result = client.images.load(f.read())

        assert len(result) == 1
        assert result[0].id == image.id

    def test_save_and_load_repo_name(self):
        client = docker.from_env(version=TEST_API_VERSION)
        image = client.images.get(TEST_IMG)
        additional_tag = random_name()
        image.tag(additional_tag)
        self.tmp_imgs.append(additional_tag)
        image.reload()
        with tempfile.TemporaryFile() as f:
            stream = image.save(named=f'{additional_tag}:latest')
            for chunk in stream:
                f.write(chunk)

            f.seek(0)
            client.images.remove(additional_tag, force=True)
            result = client.images.load(f.read())

        assert len(result) == 1
        assert result[0].id == image.id
        assert f'{additional_tag}:latest' in result[0].tags

    def test_save_name_error(self):
        client = docker.from_env(version=TEST_API_VERSION)
        image = client.images.get(TEST_IMG)
        with pytest.raises(docker.errors.InvalidArgument):
            image.save(named='sakuya/izayoi')


class ImageTest(BaseIntegrationTest):

    def test_tag_and_remove(self):
        repo = 'dockersdk.tests.images.test_tag'
        tag = 'some-tag'
        identifier = f'{repo}:{tag}'

        client = docker.from_env(version=TEST_API_VERSION)
        image = client.images.pull('alpine:latest')

        result = image.tag(repo, tag)
        assert result is True
        self.tmp_imgs.append(identifier)
        assert image.id in get_ids(client.images.list(repo))
        assert image.id in get_ids(client.images.list(identifier))

        client.images.remove(identifier)
        assert image.id not in get_ids(client.images.list(repo))
        assert image.id not in get_ids(client.images.list(identifier))

        assert image.id in get_ids(client.images.list('alpine:latest'))


def get_ids(images):
    return [i.id for i in images]