yellowbox/tests/test_container_utils.py at 2420dedd516f834d124e92b122fe174b0395e0de · biocatchltd/yellowbox · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
import io
import tempfile
from time import sleep
from typing import IO

import pytest
from pytest import mark

from yellowbox.containers import download_file, is_removed, removing, upload_file
from yellowbox.image_build import build_image


def test_upload_file(docker_client, create_and_pull):
    container = create_and_pull(docker_client, "alpine:latest", ["cat", "/tmp/test"], remove=True)
    container.start()
    assert container.wait()["StatusCode"] != 0

    container = create_and_pull(docker_client, "alpine:latest", ["cat", "/tmp/test"])
    upload_file(container, "/tmp/test", b"testfile")
    container.start()
    assert container.wait()["StatusCode"] == 0
    assert download_file(container, "/tmp/test").read() == b"testfile"


def _create_temp_file(data: bytes) -> IO[bytes]:
    f = tempfile.TemporaryFile()
    f.write(data)
    f.seek(0)
    return f


@pytest.mark.parametrize("fileobj_creation", [io.BytesIO, _create_temp_file])
def test_upload_fileobj(docker_client, fileobj_creation, create_and_pull):
    container = create_and_pull(docker_client, "alpine:latest", ["cat", "/tmp/test"])
    with fileobj_creation(b"testfile") as file:
        upload_file(container, "/tmp/test", fileobj=file)
    container.start()
    assert container.wait()["StatusCode"] == 0
    assert download_file(container, "/tmp/test").read() == b"testfile"


def test_download_file(docker_client, create_and_pull):
    container = create_and_pull(docker_client, "alpine:latest")
    upload_file(container, "/tmp/test", b"abcd")
    with download_file(container, "/tmp/test") as file:
        assert file.read() == b"abcd"


@pytest.mark.parametrize("expected_exit_code", [0, None, (1, 0)])
def test_removing(docker_client, create_and_pull, expected_exit_code):
    container = create_and_pull(docker_client, "alpine:latest", "sh -c exit 0")
    with removing(container, expected_exit_code=expected_exit_code):
        container.start()
        assert container.wait()["StatusCode"] == 0
    assert is_removed(container)


@pytest.mark.parametrize("expected_exit_code", [1, (1, 12)])
def test_removing_fails(docker_client, create_and_pull, expected_exit_code):
    container = create_and_pull(docker_client, "alpine:latest", "sh -c exit 0")
    with pytest.raises(RuntimeError), removing(container, expected_exit_code=expected_exit_code):
        container.start()
    assert container.wait()["StatusCode"] == 0
    assert not is_removed(container)


def test_create_and_pull(docker_client, create_and_pull):
    container = create_and_pull(docker_client, "alpine:latest", "sh -c exit 0")
    assert "alpine:latest" in container.image.tags


@mark.xfail(reason="I really don't know why it succeeds locally for me but fails on Github.")
@mark.parametrize("image_name", ["yellowbox", "yellowbox:test", None])
def test_build_create_and_pull(docker_client, create_and_pull, image_name):
    # we create an anonymous image to test this
    with build_image(
        docker_client, image_name, path=".", dockerfile="tests/resources/valid_dockerfile/Dockerfile"
    ) as image:
        # sometimes we need to wait for the image to be acknowledged by docker
        sleep(1)
        container = create_and_pull(docker_client, image, "sh -c exit 0")
        expected_tags = [] if image_name is None else ["yellowbox:test"]
        assert container.image.tags == expected_tags
        with removing(container):
            container.start()
            assert container.wait()["StatusCode"] == 0