fix: correct exception on test Signed-off-by: Daniel Lombardi <lombardi.daniel.o@gmail.com>

Signed-off-by: Daniel Lombardi <lombardi.daniel.o@gmail.com>
This commit is contained in:
Daniel Lombardi 2023-12-04 14:31:38 -03:00
parent b8670edcf8
commit 24907a524e
1 changed files with 141 additions and 120 deletions

View File

@ -2,187 +2,206 @@ import gzip
import io import io
import shutil import shutil
import docker
from docker import auth
from docker.api.build import process_dockerfile
import pytest import pytest
import docker
from docker import auth, errors
from docker.api.build import process_dockerfile
from ..helpers import make_tree from ..helpers import make_tree
from .api_test import BaseAPIClientTest, fake_request, url_prefix from .api_test import BaseAPIClientTest, fake_request, url_prefix
class BuildTest(BaseAPIClientTest): class BuildTest(BaseAPIClientTest):
def test_build_container(self): def test_build_container(self):
script = io.BytesIO('\n'.join([ script = io.BytesIO(
'FROM busybox', "\n".join(
'RUN mkdir -p /tmp/test', [
'EXPOSE 8080', "FROM busybox",
'ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz' "RUN mkdir -p /tmp/test",
' /tmp/silence.tar.gz' "EXPOSE 8080",
]).encode('ascii')) "ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz"
" /tmp/silence.tar.gz",
]
).encode("ascii")
)
self.client.build(fileobj=script) self.client.build(fileobj=script)
def test_build_container_pull(self): def test_build_container_pull(self):
script = io.BytesIO('\n'.join([ script = io.BytesIO(
'FROM busybox', "\n".join(
'RUN mkdir -p /tmp/test', [
'EXPOSE 8080', "FROM busybox",
'ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz' "RUN mkdir -p /tmp/test",
' /tmp/silence.tar.gz' "EXPOSE 8080",
]).encode('ascii')) "ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz"
" /tmp/silence.tar.gz",
]
).encode("ascii")
)
self.client.build(fileobj=script, pull=True) self.client.build(fileobj=script, pull=True)
def test_build_container_custom_context(self): def test_build_container_custom_context(self):
script = io.BytesIO('\n'.join([ script = io.BytesIO(
'FROM busybox', "\n".join(
'RUN mkdir -p /tmp/test', [
'EXPOSE 8080', "FROM busybox",
'ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz' "RUN mkdir -p /tmp/test",
' /tmp/silence.tar.gz' "EXPOSE 8080",
]).encode('ascii')) "ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz"
" /tmp/silence.tar.gz",
]
).encode("ascii")
)
context = docker.utils.mkbuildcontext(script) context = docker.utils.mkbuildcontext(script)
self.client.build(fileobj=context, custom_context=True) self.client.build(fileobj=context, custom_context=True)
def test_build_container_custom_context_gzip(self): def test_build_container_custom_context_gzip(self):
script = io.BytesIO('\n'.join([ script = io.BytesIO(
'FROM busybox', "\n".join(
'RUN mkdir -p /tmp/test', [
'EXPOSE 8080', "FROM busybox",
'ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz' "RUN mkdir -p /tmp/test",
' /tmp/silence.tar.gz' "EXPOSE 8080",
]).encode('ascii')) "ADD https://dl.dropboxusercontent.com/u/20637798/silence.tar.gz"
" /tmp/silence.tar.gz",
]
).encode("ascii")
)
context = docker.utils.mkbuildcontext(script) context = docker.utils.mkbuildcontext(script)
gz_context = gzip.GzipFile(fileobj=context) gz_context = gzip.GzipFile(fileobj=context)
self.client.build( self.client.build(fileobj=gz_context, custom_context=True, encoding="gzip")
fileobj=gz_context,
custom_context=True,
encoding="gzip"
)
def test_build_remote_with_registry_auth(self): def test_build_remote_with_registry_auth(self):
self.client._auth_configs = auth.AuthConfig({ self.client._auth_configs = auth.AuthConfig(
'auths': { {
'https://example.com': { "auths": {
'user': 'example', "https://example.com": {
'password': 'example', "user": "example",
'email': 'example@example.com' "password": "example",
"email": "example@example.com",
}
} }
} }
})
expected_params = {'t': None, 'q': False, 'dockerfile': None,
'rm': False, 'nocache': False, 'pull': False,
'forcerm': False,
'remote': 'https://github.com/docker-library/mongo'}
expected_headers = {
'X-Registry-Config': auth.encode_header(
self.client._auth_configs.auths
) )
expected_params = {
"t": None,
"q": False,
"dockerfile": None,
"rm": False,
"nocache": False,
"pull": False,
"forcerm": False,
"remote": "https://github.com/docker-library/mongo",
}
expected_headers = {
"X-Registry-Config": auth.encode_header(self.client._auth_configs.auths)
} }
self.client.build(path='https://github.com/docker-library/mongo') self.client.build(path="https://github.com/docker-library/mongo")
fake_request.assert_called_with( fake_request.assert_called_with(
'POST', "POST",
f"{url_prefix}build", f"{url_prefix}build",
stream=True, stream=True,
data=None, data=None,
headers=expected_headers, headers=expected_headers,
params=expected_params, params=expected_params,
timeout=None timeout=None,
) )
def test_build_container_with_named_dockerfile(self): def test_build_container_with_named_dockerfile(self):
self.client.build('.', dockerfile='nameddockerfile') self.client.build(".", dockerfile="nameddockerfile")
def test_build_with_invalid_tag(self): def test_build_with_invalid_tag(self):
with pytest.raises(TypeError): with pytest.raises(errors.DockerException):
self.client.build( self.client.build(".", tag="https://example.com")
".", tag="https://example.com"
)
def test_build_container_with_container_limits(self): def test_build_container_with_container_limits(self):
self.client.build('.', container_limits={ self.client.build(
'memory': 1024 * 1024, ".",
'cpusetcpus': 1, container_limits={
'cpushares': 1000, "memory": 1024 * 1024,
'memswap': 1024 * 1024 * 8 "cpusetcpus": 1,
}) "cpushares": 1000,
"memswap": 1024 * 1024 * 8,
},
)
def test_build_container_invalid_container_limits(self): def test_build_container_invalid_container_limits(self):
with pytest.raises(docker.errors.DockerException): with pytest.raises(docker.errors.DockerException):
self.client.build('.', container_limits={ self.client.build(".", container_limits={"foo": "bar"})
'foo': 'bar'
})
def test_set_auth_headers_with_empty_dict_and_auth_configs(self): def test_set_auth_headers_with_empty_dict_and_auth_configs(self):
self.client._auth_configs = auth.AuthConfig({ self.client._auth_configs = auth.AuthConfig(
'auths': { {
'https://example.com': { "auths": {
'user': 'example', "https://example.com": {
'password': 'example', "user": "example",
'email': 'example@example.com' "password": "example",
"email": "example@example.com",
} }
} }
}) }
)
headers = {} headers = {}
expected_headers = { expected_headers = {
'X-Registry-Config': auth.encode_header( "X-Registry-Config": auth.encode_header(self.client._auth_configs.auths)
self.client._auth_configs.auths
)
} }
self.client._set_auth_headers(headers) self.client._set_auth_headers(headers)
assert headers == expected_headers assert headers == expected_headers
def test_set_auth_headers_with_dict_and_auth_configs(self): def test_set_auth_headers_with_dict_and_auth_configs(self):
self.client._auth_configs = auth.AuthConfig({ self.client._auth_configs = auth.AuthConfig(
'auths': { {
'https://example.com': { "auths": {
'user': 'example', "https://example.com": {
'password': 'example', "user": "example",
'email': 'example@example.com' "password": "example",
"email": "example@example.com",
} }
} }
}) }
)
headers = {'foo': 'bar'} headers = {"foo": "bar"}
expected_headers = { expected_headers = {
'X-Registry-Config': auth.encode_header( "X-Registry-Config": auth.encode_header(self.client._auth_configs.auths),
self.client._auth_configs.auths "foo": "bar",
),
'foo': 'bar'
} }
self.client._set_auth_headers(headers) self.client._set_auth_headers(headers)
assert headers == expected_headers assert headers == expected_headers
def test_set_auth_headers_with_dict_and_no_auth_configs(self): def test_set_auth_headers_with_dict_and_no_auth_configs(self):
headers = {'foo': 'bar'} headers = {"foo": "bar"}
expected_headers = { expected_headers = {"foo": "bar"}
'foo': 'bar'
}
self.client._set_auth_headers(headers) self.client._set_auth_headers(headers)
assert headers == expected_headers assert headers == expected_headers
@pytest.mark.skipif( @pytest.mark.skipif(
not docker.constants.IS_WINDOWS_PLATFORM, not docker.constants.IS_WINDOWS_PLATFORM, reason="Windows-specific syntax"
reason='Windows-specific syntax') )
def test_process_dockerfile_win_longpath_prefix(self): def test_process_dockerfile_win_longpath_prefix(self):
dirs = [ dirs = [
'foo', 'foo/bar', 'baz', "foo",
"foo/bar",
"baz",
] ]
files = [ files = [
'Dockerfile', 'foo/Dockerfile.foo', 'foo/bar/Dockerfile.bar', "Dockerfile",
'baz/Dockerfile.baz', "foo/Dockerfile.foo",
"foo/bar/Dockerfile.bar",
"baz/Dockerfile.baz",
] ]
base = make_tree(dirs, files) base = make_tree(dirs, files)
@ -192,40 +211,42 @@ class BuildTest(BaseAPIClientTest):
return docker.constants.WINDOWS_LONGPATH_PREFIX + path return docker.constants.WINDOWS_LONGPATH_PREFIX + path
assert process_dockerfile(None, pre(base)) == (None, None) assert process_dockerfile(None, pre(base)) == (None, None)
assert process_dockerfile('Dockerfile', pre(base)) == ( assert process_dockerfile("Dockerfile", pre(base)) == ("Dockerfile", None)
'Dockerfile', None assert process_dockerfile("foo/Dockerfile.foo", pre(base)) == (
"foo/Dockerfile.foo",
None,
) )
assert process_dockerfile('foo/Dockerfile.foo', pre(base)) == ( assert process_dockerfile("../Dockerfile", pre(f"{base}\\foo"))[1] is not None
'foo/Dockerfile.foo', None assert process_dockerfile("../baz/Dockerfile.baz", pre(f"{base}/baz")) == (
"../baz/Dockerfile.baz",
None,
) )
assert process_dockerfile(
'../Dockerfile', pre(f"{base}\\foo")
)[1] is not None
assert process_dockerfile(
'../baz/Dockerfile.baz', pre(f"{base}/baz")
) == ('../baz/Dockerfile.baz', None)
def test_process_dockerfile(self): def test_process_dockerfile(self):
dirs = [ dirs = [
'foo', 'foo/bar', 'baz', "foo",
"foo/bar",
"baz",
] ]
files = [ files = [
'Dockerfile', 'foo/Dockerfile.foo', 'foo/bar/Dockerfile.bar', "Dockerfile",
'baz/Dockerfile.baz', "foo/Dockerfile.foo",
"foo/bar/Dockerfile.bar",
"baz/Dockerfile.baz",
] ]
base = make_tree(dirs, files) base = make_tree(dirs, files)
self.addCleanup(shutil.rmtree, base) self.addCleanup(shutil.rmtree, base)
assert process_dockerfile(None, base) == (None, None) assert process_dockerfile(None, base) == (None, None)
assert process_dockerfile('Dockerfile', base) == ('Dockerfile', None) assert process_dockerfile("Dockerfile", base) == ("Dockerfile", None)
assert process_dockerfile('foo/Dockerfile.foo', base) == ( assert process_dockerfile("foo/Dockerfile.foo", base) == (
'foo/Dockerfile.foo', None "foo/Dockerfile.foo",
None,
) )
assert process_dockerfile( assert process_dockerfile("../Dockerfile", f"{base}/foo")[1] is not None
'../Dockerfile', f"{base}/foo" assert process_dockerfile("../baz/Dockerfile.baz", f"{base}/baz") == (
)[1] is not None "../baz/Dockerfile.baz",
assert process_dockerfile('../baz/Dockerfile.baz', f"{base}/baz") == ( None,
'../baz/Dockerfile.baz', None
) )