ansible-collections/community.docker
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.docker.git synced 2026-07-29 11:55:04 +00:00
Code Packages Projects Releases Wiki Activity

Reformat code with black and isort.

Browse Source
This commit is contained in:
Felix Fontein
2025-10-06 18:34:59 +02:00
parent f45232635c
commit d65d37e9e9
132 changed files with 17581 additions and 14729 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/plugins/module_utils/_api/utils/test_build.py
+254 -239
View File
@@ -7,7 +7,9 @@
# It is licensed under the Apache 2.0 license (see LICENSES/Apache-2.0.txt in this collection)
# SPDX-License-Identifier: Apache-2.0
from __future__ import (absolute_import, division, print_function)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import os
@@ -19,9 +21,13 @@ import tempfile
import unittest
import pytest
from ansible_collections.community.docker.plugins.module_utils._api.constants import IS_WINDOWS_PLATFORM
from ansible_collections.community.docker.plugins.module_utils._api.utils.build import exclude_paths, tar
from ansible_collections.community.docker.plugins.module_utils._api.constants import (
IS_WINDOWS_PLATFORM,
)
from ansible_collections.community.docker.plugins.module_utils._api.utils.build import (
exclude_paths,
tar,
)
def make_tree(dirs, files):
@@ -31,7 +37,7 @@ def make_tree(dirs, files):
os.makedirs(os.path.join(base, path))
for path in files:
with open(os.path.join(base, path), 'w') as f:
with open(os.path.join(base, path), "w") as f:
f.write("content")
return base
@@ -42,45 +48,45 @@ def convert_paths(collection):
def convert_path(path):
return path.replace('/', os.path.sep)
return path.replace("/", os.path.sep)
class ExcludePathsTest(unittest.TestCase):
dirs = [
'foo',
'foo/bar',
'bar',
'target',
'target/subdir',
'subdir',
'subdir/target',
'subdir/target/subdir',
'subdir/subdir2',
'subdir/subdir2/target',
'subdir/subdir2/target/subdir'
"foo",
"foo/bar",
"bar",
"target",
"target/subdir",
"subdir",
"subdir/target",
"subdir/target/subdir",
"subdir/subdir2",
"subdir/subdir2/target",
"subdir/subdir2/target/subdir",
]
files = [
'Dockerfile',
'Dockerfile.alt',
'.dockerignore',
'a.py',
'a.go',
'b.py',
'cde.py',
'foo/a.py',
'foo/b.py',
'foo/bar/a.py',
'bar/a.py',
'foo/Dockerfile3',
'target/file.txt',
'target/subdir/file.txt',
'subdir/file.txt',
'subdir/target/file.txt',
'subdir/target/subdir/file.txt',
'subdir/subdir2/file.txt',
'subdir/subdir2/target/file.txt',
'subdir/subdir2/target/subdir/file.txt',
"Dockerfile",
"Dockerfile.alt",
".dockerignore",
"a.py",
"a.go",
"b.py",
"cde.py",
"foo/a.py",
"foo/b.py",
"foo/bar/a.py",
"bar/a.py",
"foo/Dockerfile3",
"target/file.txt",
"target/subdir/file.txt",
"subdir/file.txt",
"subdir/target/file.txt",
"subdir/target/subdir/file.txt",
"subdir/subdir2/file.txt",
"subdir/subdir2/target/file.txt",
"subdir/subdir2/target/subdir/file.txt",
]
all_paths = set(dirs + files)
@@ -95,14 +101,14 @@ class ExcludePathsTest(unittest.TestCase):
return set(exclude_paths(self.base, patterns, dockerfile=dockerfile))
def test_no_excludes(self):
assert self.exclude(['']) == convert_paths(self.all_paths)
assert self.exclude([""]) == convert_paths(self.all_paths)
def test_no_dupes(self):
paths = exclude_paths(self.base, ['!a.py'])
paths = exclude_paths(self.base, ["!a.py"])
assert sorted(paths) == sorted(set(paths))
def test_wildcard_exclude(self):
assert self.exclude(['*']) == set(['Dockerfile', '.dockerignore'])
assert self.exclude(["*"]) == set(["Dockerfile", ".dockerignore"])
def test_exclude_dockerfile_dockerignore(self):
"""
@@ -110,7 +116,7 @@ class ExcludePathsTest(unittest.TestCase):
Dockerfile and/or .dockerignore, don't exclude them from
the actual tar file.
"""
assert self.exclude(['Dockerfile', '.dockerignore']) == convert_paths(
assert self.exclude(["Dockerfile", ".dockerignore"]) == convert_paths(
self.all_paths
)
@@ -119,287 +125,296 @@ class ExcludePathsTest(unittest.TestCase):
If we're using a custom Dockerfile, make sure that's not
excluded.
"""
assert self.exclude(['*'], dockerfile='Dockerfile.alt') == set(['Dockerfile.alt', '.dockerignore'])
assert self.exclude(["*"], dockerfile="Dockerfile.alt") == set(
["Dockerfile.alt", ".dockerignore"]
)
assert self.exclude(
['*'], dockerfile='foo/Dockerfile3'
) == convert_paths(set(['foo/Dockerfile3', '.dockerignore']))
assert self.exclude(["*"], dockerfile="foo/Dockerfile3") == convert_paths(
set(["foo/Dockerfile3", ".dockerignore"])
)
# https://github.com/docker/docker-py/issues/1956
assert self.exclude(
['*'], dockerfile='./foo/Dockerfile3'
) == convert_paths(set(['foo/Dockerfile3', '.dockerignore']))
assert self.exclude(["*"], dockerfile="./foo/Dockerfile3") == convert_paths(
set(["foo/Dockerfile3", ".dockerignore"])
)
def test_exclude_dockerfile_child(self):
includes = self.exclude(['foo/'], dockerfile='foo/Dockerfile3')
assert convert_path('foo/Dockerfile3') in includes
assert convert_path('foo/a.py') not in includes
includes = self.exclude(["foo/"], dockerfile="foo/Dockerfile3")
assert convert_path("foo/Dockerfile3") in includes
assert convert_path("foo/a.py") not in includes
def test_single_filename(self):
assert self.exclude(['a.py']) == convert_paths(
self.all_paths - set(['a.py'])
)
assert self.exclude(["a.py"]) == convert_paths(self.all_paths - set(["a.py"]))
def test_single_filename_leading_dot_slash(self):
assert self.exclude(['./a.py']) == convert_paths(
self.all_paths - set(['a.py'])
)
assert self.exclude(["./a.py"]) == convert_paths(self.all_paths - set(["a.py"]))
# As odd as it sounds, a filename pattern with a trailing slash on the
# end *will* result in that file being excluded.
def test_single_filename_trailing_slash(self):
assert self.exclude(['a.py/']) == convert_paths(
self.all_paths - set(['a.py'])
)
assert self.exclude(["a.py/"]) == convert_paths(self.all_paths - set(["a.py"]))
def test_wildcard_filename_start(self):
assert self.exclude(['*.py']) == convert_paths(
self.all_paths - set(['a.py', 'b.py', 'cde.py'])
assert self.exclude(["*.py"]) == convert_paths(
self.all_paths - set(["a.py", "b.py", "cde.py"])
)
def test_wildcard_with_exception(self):
assert self.exclude(['*.py', '!b.py']) == convert_paths(
self.all_paths - set(['a.py', 'cde.py'])
assert self.exclude(["*.py", "!b.py"]) == convert_paths(
self.all_paths - set(["a.py", "cde.py"])
)
def test_wildcard_with_wildcard_exception(self):
assert self.exclude(['*.*', '!*.go']) == convert_paths(
self.all_paths - set([
'a.py', 'b.py', 'cde.py', 'Dockerfile.alt',
])
assert self.exclude(["*.*", "!*.go"]) == convert_paths(
self.all_paths
- set(
[
"a.py",
"b.py",
"cde.py",
"Dockerfile.alt",
]
)
)
def test_wildcard_filename_end(self):
assert self.exclude(['a.*']) == convert_paths(
self.all_paths - set(['a.py', 'a.go'])
assert self.exclude(["a.*"]) == convert_paths(
self.all_paths - set(["a.py", "a.go"])
)
def test_question_mark(self):
assert self.exclude(['?.py']) == convert_paths(
self.all_paths - set(['a.py', 'b.py'])
assert self.exclude(["?.py"]) == convert_paths(
self.all_paths - set(["a.py", "b.py"])
)
def test_single_subdir_single_filename(self):
assert self.exclude(['foo/a.py']) == convert_paths(
self.all_paths - set(['foo/a.py'])
assert self.exclude(["foo/a.py"]) == convert_paths(
self.all_paths - set(["foo/a.py"])
)
def test_single_subdir_single_filename_leading_slash(self):
assert self.exclude(['/foo/a.py']) == convert_paths(
self.all_paths - set(['foo/a.py'])
assert self.exclude(["/foo/a.py"]) == convert_paths(
self.all_paths - set(["foo/a.py"])
)
def test_exclude_include_absolute_path(self):
base = make_tree([], ['a.py', 'b.py'])
assert exclude_paths(
base,
['/*', '!/*.py']
) == set(['a.py', 'b.py'])
base = make_tree([], ["a.py", "b.py"])
assert exclude_paths(base, ["/*", "!/*.py"]) == set(["a.py", "b.py"])
def test_single_subdir_with_path_traversal(self):
assert self.exclude(['foo/whoops/../a.py']) == convert_paths(
self.all_paths - set(['foo/a.py'])
assert self.exclude(["foo/whoops/../a.py"]) == convert_paths(
self.all_paths - set(["foo/a.py"])
)
def test_single_subdir_wildcard_filename(self):
assert self.exclude(['foo/*.py']) == convert_paths(
self.all_paths - set(['foo/a.py', 'foo/b.py'])
assert self.exclude(["foo/*.py"]) == convert_paths(
self.all_paths - set(["foo/a.py", "foo/b.py"])
)
def test_wildcard_subdir_single_filename(self):
assert self.exclude(['*/a.py']) == convert_paths(
self.all_paths - set(['foo/a.py', 'bar/a.py'])
assert self.exclude(["*/a.py"]) == convert_paths(
self.all_paths - set(["foo/a.py", "bar/a.py"])
)
def test_wildcard_subdir_wildcard_filename(self):
assert self.exclude(['*/*.py']) == convert_paths(
self.all_paths - set(['foo/a.py', 'foo/b.py', 'bar/a.py'])
assert self.exclude(["*/*.py"]) == convert_paths(
self.all_paths - set(["foo/a.py", "foo/b.py", "bar/a.py"])
)
def test_directory(self):
assert self.exclude(['foo']) == convert_paths(
self.all_paths - set([
'foo', 'foo/a.py', 'foo/b.py', 'foo/bar', 'foo/bar/a.py',
'foo/Dockerfile3'
])
assert self.exclude(["foo"]) == convert_paths(
self.all_paths
- set(
[
"foo",
"foo/a.py",
"foo/b.py",
"foo/bar",
"foo/bar/a.py",
"foo/Dockerfile3",
]
)
)
def test_directory_with_trailing_slash(self):
assert self.exclude(['foo']) == convert_paths(
self.all_paths - set([
'foo', 'foo/a.py', 'foo/b.py',
'foo/bar', 'foo/bar/a.py', 'foo/Dockerfile3'
])
assert self.exclude(["foo"]) == convert_paths(
self.all_paths
- set(
[
"foo",
"foo/a.py",
"foo/b.py",
"foo/bar",
"foo/bar/a.py",
"foo/Dockerfile3",
]
)
)
def test_directory_with_single_exception(self):
assert self.exclude(['foo', '!foo/bar/a.py']) == convert_paths(
self.all_paths - set([
'foo/a.py', 'foo/b.py', 'foo', 'foo/bar',
'foo/Dockerfile3'
])
assert self.exclude(["foo", "!foo/bar/a.py"]) == convert_paths(
self.all_paths
- set(["foo/a.py", "foo/b.py", "foo", "foo/bar", "foo/Dockerfile3"])
)
def test_directory_with_subdir_exception(self):
assert self.exclude(['foo', '!foo/bar']) == convert_paths(
self.all_paths - set([
'foo/a.py', 'foo/b.py', 'foo', 'foo/Dockerfile3'
])
assert self.exclude(["foo", "!foo/bar"]) == convert_paths(
self.all_paths - set(["foo/a.py", "foo/b.py", "foo", "foo/Dockerfile3"])
)
@pytest.mark.skipif(
not IS_WINDOWS_PLATFORM, reason='Backslash patterns only on Windows'
not IS_WINDOWS_PLATFORM, reason="Backslash patterns only on Windows"
)
def test_directory_with_subdir_exception_win32_pathsep(self):
assert self.exclude(['foo', '!foo\\bar']) == convert_paths(
self.all_paths - set([
'foo/a.py', 'foo/b.py', 'foo', 'foo/Dockerfile3'
])
assert self.exclude(["foo", "!foo\\bar"]) == convert_paths(
self.all_paths - set(["foo/a.py", "foo/b.py", "foo", "foo/Dockerfile3"])
)
def test_directory_with_wildcard_exception(self):
assert self.exclude(['foo', '!foo/*.py']) == convert_paths(
self.all_paths - set([
'foo/bar', 'foo/bar/a.py', 'foo', 'foo/Dockerfile3'
])
assert self.exclude(["foo", "!foo/*.py"]) == convert_paths(
self.all_paths - set(["foo/bar", "foo/bar/a.py", "foo", "foo/Dockerfile3"])
)
def test_subdirectory(self):
assert self.exclude(['foo/bar']) == convert_paths(
self.all_paths - set(['foo/bar', 'foo/bar/a.py'])
assert self.exclude(["foo/bar"]) == convert_paths(
self.all_paths - set(["foo/bar", "foo/bar/a.py"])
)
@pytest.mark.skipif(
not IS_WINDOWS_PLATFORM, reason='Backslash patterns only on Windows'
not IS_WINDOWS_PLATFORM, reason="Backslash patterns only on Windows"
)
def test_subdirectory_win32_pathsep(self):
assert self.exclude(['foo\\bar']) == convert_paths(
self.all_paths - set(['foo/bar', 'foo/bar/a.py'])
assert self.exclude(["foo\\bar"]) == convert_paths(
self.all_paths - set(["foo/bar", "foo/bar/a.py"])
)
def test_double_wildcard(self):
assert self.exclude(['**/a.py']) == convert_paths(
self.all_paths - set([
'a.py', 'foo/a.py', 'foo/bar/a.py', 'bar/a.py'
])
assert self.exclude(["**/a.py"]) == convert_paths(
self.all_paths - set(["a.py", "foo/a.py", "foo/bar/a.py", "bar/a.py"])
)
assert self.exclude(['foo/**/bar']) == convert_paths(
self.all_paths - set(['foo/bar', 'foo/bar/a.py'])
assert self.exclude(["foo/**/bar"]) == convert_paths(
self.all_paths - set(["foo/bar", "foo/bar/a.py"])
)
def test_single_and_double_wildcard(self):
assert self.exclude(['**/target/*/*']) == convert_paths(
self.all_paths - set([
'target/subdir/file.txt',
'subdir/target/subdir/file.txt',
'subdir/subdir2/target/subdir/file.txt',
])
assert self.exclude(["**/target/*/*"]) == convert_paths(
self.all_paths
- set(
[
"target/subdir/file.txt",
"subdir/target/subdir/file.txt",
"subdir/subdir2/target/subdir/file.txt",
]
)
)
def test_trailing_double_wildcard(self):
assert self.exclude(['subdir/**']) == convert_paths(
self.all_paths - set([
'subdir/file.txt',
'subdir/target/file.txt',
'subdir/target/subdir/file.txt',
'subdir/subdir2/file.txt',
'subdir/subdir2/target/file.txt',
'subdir/subdir2/target/subdir/file.txt',
'subdir/target',
'subdir/target/subdir',
'subdir/subdir2',
'subdir/subdir2/target',
'subdir/subdir2/target/subdir',
])
assert self.exclude(["subdir/**"]) == convert_paths(
self.all_paths
- set(
[
"subdir/file.txt",
"subdir/target/file.txt",
"subdir/target/subdir/file.txt",
"subdir/subdir2/file.txt",
"subdir/subdir2/target/file.txt",
"subdir/subdir2/target/subdir/file.txt",
"subdir/target",
"subdir/target/subdir",
"subdir/subdir2",
"subdir/subdir2/target",
"subdir/subdir2/target/subdir",
]
)
)
def test_double_wildcard_with_exception(self):
assert self.exclude(['**', '!bar', '!foo/bar']) == convert_paths(
set([
'foo/bar', 'foo/bar/a.py', 'bar', 'bar/a.py', 'Dockerfile',
'.dockerignore',
])
assert self.exclude(["**", "!bar", "!foo/bar"]) == convert_paths(
set(
[
"foo/bar",
"foo/bar/a.py",
"bar",
"bar/a.py",
"Dockerfile",
".dockerignore",
]
)
)
def test_include_wildcard(self):
# This may be surprising but it matches the CLI's behavior
# (tested with 18.05.0-ce on linux)
base = make_tree(['a'], ['a/b.py'])
assert exclude_paths(
base,
['*', '!*/b.py']
) == set()
base = make_tree(["a"], ["a/b.py"])
assert exclude_paths(base, ["*", "!*/b.py"]) == set()
def test_last_line_precedence(self):
base = make_tree(
[],
['garbage.md',
'trash.md',
'README.md',
'README-bis.md',
'README-secret.md'])
assert exclude_paths(
base,
['*.md', '!README*.md', 'README-secret.md']
) == set(['README.md', 'README-bis.md'])
[
"garbage.md",
"trash.md",
"README.md",
"README-bis.md",
"README-secret.md",
],
)
assert exclude_paths(base, ["*.md", "!README*.md", "README-secret.md"]) == set(
["README.md", "README-bis.md"]
)
def test_parent_directory(self):
base = make_tree(
[],
['a.py',
'b.py',
'c.py'])
base = make_tree([], ["a.py", "b.py", "c.py"])
# Dockerignore reference stipulates that absolute paths are
# equivalent to relative paths, hence /../foo should be
# equivalent to ../foo. It also stipulates that paths are run
# through Go's filepath.Clean, which explicitly "replace
# "/.." by "/" at the beginning of a path".
assert exclude_paths(
base,
['../a.py', '/../b.py']
) == set(['c.py'])
assert exclude_paths(base, ["../a.py", "/../b.py"]) == set(["c.py"])
class TarTest(unittest.TestCase):
def test_tar_with_excludes(self):
dirs = [
'foo',
'foo/bar',
'bar',
"foo",
"foo/bar",
"bar",
]
files = [
'Dockerfile',
'Dockerfile.alt',
'.dockerignore',
'a.py',
'a.go',
'b.py',
'cde.py',
'foo/a.py',
'foo/b.py',
'foo/bar/a.py',
'bar/a.py',
"Dockerfile",
"Dockerfile.alt",
".dockerignore",
"a.py",
"a.go",
"b.py",
"cde.py",
"foo/a.py",
"foo/b.py",
"foo/bar/a.py",
"bar/a.py",
]
exclude = [
'*.py',
'!b.py',
'!a.go',
'foo',
'Dockerfile*',
'.dockerignore',
"*.py",
"!b.py",
"!a.go",
"foo",
"Dockerfile*",
".dockerignore",
]
expected_names = set([
'Dockerfile',
'.dockerignore',
'a.go',
'b.py',
'bar',
'bar/a.py',
])
expected_names = set(
[
"Dockerfile",
".dockerignore",
"a.go",
"b.py",
"bar",
"bar/a.py",
]
)
base = make_tree(dirs, files)
self.addCleanup(shutil.rmtree, base)
@@ -411,99 +426,99 @@ class TarTest(unittest.TestCase):
def test_tar_with_empty_directory(self):
base = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, base)
for d in ['foo', 'bar']:
for d in ["foo", "bar"]:
os.makedirs(os.path.join(base, d))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
assert sorted(tar_data.getnames()) == ['bar', 'foo']
assert sorted(tar_data.getnames()) == ["bar", "foo"]
@pytest.mark.skipif(
IS_WINDOWS_PLATFORM or os.geteuid() == 0,
reason='root user always has access ; no chmod on Windows'
reason="root user always has access ; no chmod on Windows",
)
def test_tar_with_inaccessible_file(self):
base = tempfile.mkdtemp()
full_path = os.path.join(base, 'foo')
full_path = os.path.join(base, "foo")
self.addCleanup(shutil.rmtree, base)
with open(full_path, 'w') as f:
f.write('content')
with open(full_path, "w") as f:
f.write("content")
os.chmod(full_path, 0o222)
with pytest.raises(IOError) as ei:
tar(base)
assert f'Can not read file in context: {full_path}' in ei.exconly()
assert f"Can not read file in context: {full_path}" in ei.exconly()
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason='No symlinks on Windows')
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason="No symlinks on Windows")
def test_tar_with_file_symlinks(self):
base = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, base)
with open(os.path.join(base, 'foo'), 'w') as f:
with open(os.path.join(base, "foo"), "w") as f:
f.write("content")
os.makedirs(os.path.join(base, 'bar'))
os.symlink('../foo', os.path.join(base, 'bar/foo'))
os.makedirs(os.path.join(base, "bar"))
os.symlink("../foo", os.path.join(base, "bar/foo"))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
assert sorted(tar_data.getnames()) == ['bar', 'bar/foo', 'foo']
assert sorted(tar_data.getnames()) == ["bar", "bar/foo", "foo"]
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason='No symlinks on Windows')
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason="No symlinks on Windows")
def test_tar_with_directory_symlinks(self):
base = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, base)
for d in ['foo', 'bar']:
for d in ["foo", "bar"]:
os.makedirs(os.path.join(base, d))
os.symlink('../foo', os.path.join(base, 'bar/foo'))
os.symlink("../foo", os.path.join(base, "bar/foo"))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
assert sorted(tar_data.getnames()) == ['bar', 'bar/foo', 'foo']
assert sorted(tar_data.getnames()) == ["bar", "bar/foo", "foo"]
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason='No symlinks on Windows')
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason="No symlinks on Windows")
def test_tar_with_broken_symlinks(self):
base = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, base)
for d in ['foo', 'bar']:
for d in ["foo", "bar"]:
os.makedirs(os.path.join(base, d))
os.symlink('../baz', os.path.join(base, 'bar/foo'))
os.symlink("../baz", os.path.join(base, "bar/foo"))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
assert sorted(tar_data.getnames()) == ['bar', 'bar/foo', 'foo']
assert sorted(tar_data.getnames()) == ["bar", "bar/foo", "foo"]
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason='No UNIX sockets on Win32')
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason="No UNIX sockets on Win32")
def test_tar_socket_file(self):
base = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, base)
for d in ['foo', 'bar']:
for d in ["foo", "bar"]:
os.makedirs(os.path.join(base, d))
sock = socket.socket(socket.AF_UNIX)
self.addCleanup(sock.close)
sock.bind(os.path.join(base, 'test.sock'))
sock.bind(os.path.join(base, "test.sock"))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
assert sorted(tar_data.getnames()) == ['bar', 'foo']
assert sorted(tar_data.getnames()) == ["bar", "foo"]
def tar_test_negative_mtime_bug(self):
base = tempfile.mkdtemp()
filename = os.path.join(base, 'th.txt')
filename = os.path.join(base, "th.txt")
self.addCleanup(shutil.rmtree, base)
with open(filename, 'w') as f:
f.write('Invisible Full Moon')
with open(filename, "w") as f:
f.write("Invisible Full Moon")
os.utime(filename, (12345, -3600.0))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
assert tar_data.getnames() == ['th.txt']
assert tar_data.getmember('th.txt').mtime == -3600
assert tar_data.getnames() == ["th.txt"]
assert tar_data.getmember("th.txt").mtime == -3600
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason='No symlinks on Windows')
@pytest.mark.skipif(IS_WINDOWS_PLATFORM, reason="No symlinks on Windows")
def test_tar_directory_link(self):
dirs = ['a', 'b', 'a/c']
files = ['a/hello.py', 'b/utils.py', 'a/c/descend.py']
dirs = ["a", "b", "a/c"]
files = ["a/hello.py", "b/utils.py", "a/c/descend.py"]
base = make_tree(dirs, files)
self.addCleanup(shutil.rmtree, base)
os.symlink(os.path.join(base, 'b'), os.path.join(base, 'a/c/b'))
os.symlink(os.path.join(base, "b"), os.path.join(base, "a/c/b"))
with tar(base) as archive:
tar_data = tarfile.open(fileobj=archive)
names = tar_data.getnames()
for member in dirs + files:
assert member in names
assert 'a/c/b' in names
assert 'a/c/b/utils.py' not in names
assert "a/c/b" in names
assert "a/c/b/utils.py" not in names
tests/unit/plugins/module_utils/_api/utils/test_config.py
+35 -43
View File
@@ -7,18 +7,20 @@
# It is licensed under the Apache 2.0 license (see LICENSES/Apache-2.0.txt in this collection)
# SPDX-License-Identifier: Apache-2.0
from __future__ import (absolute_import, division, print_function)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import json
import os
import unittest
import shutil
import tempfile
import json
from pytest import mark, fixture
import unittest
from ansible_collections.community.docker.plugins.module_utils._api.utils import config
from pytest import fixture, mark
try:
from unittest import mock
@@ -33,46 +35,46 @@ class FindConfigFileTest(unittest.TestCase):
self.mkdir = tmpdir.mkdir
def test_find_config_fallback(self):
tmpdir = self.mkdir('test_find_config_fallback')
tmpdir = self.mkdir("test_find_config_fallback")
with mock.patch.dict(os.environ, {'HOME': str(tmpdir)}):
with mock.patch.dict(os.environ, {"HOME": str(tmpdir)}):
assert config.find_config_file() is None
def test_find_config_from_explicit_path(self):
tmpdir = self.mkdir('test_find_config_from_explicit_path')
config_path = tmpdir.ensure('my-config-file.json')
tmpdir = self.mkdir("test_find_config_from_explicit_path")
config_path = tmpdir.ensure("my-config-file.json")
assert config.find_config_file(str(config_path)) == str(config_path)
def test_find_config_from_environment(self):
tmpdir = self.mkdir('test_find_config_from_environment')
config_path = tmpdir.ensure('config.json')
tmpdir = self.mkdir("test_find_config_from_environment")
config_path = tmpdir.ensure("config.json")
with mock.patch.dict(os.environ, {'DOCKER_CONFIG': str(tmpdir)}):
with mock.patch.dict(os.environ, {"DOCKER_CONFIG": str(tmpdir)}):
assert config.find_config_file() == str(config_path)
@mark.skipif("sys.platform == 'win32'")
def test_find_config_from_home_posix(self):
tmpdir = self.mkdir('test_find_config_from_home_posix')
config_path = tmpdir.ensure('.docker', 'config.json')
tmpdir = self.mkdir("test_find_config_from_home_posix")
config_path = tmpdir.ensure(".docker", "config.json")
with mock.patch.dict(os.environ, {'HOME': str(tmpdir)}):
with mock.patch.dict(os.environ, {"HOME": str(tmpdir)}):
assert config.find_config_file() == str(config_path)
@mark.skipif("sys.platform == 'win32'")
def test_find_config_from_home_legacy_name(self):
tmpdir = self.mkdir('test_find_config_from_home_legacy_name')
config_path = tmpdir.ensure('.dockercfg')
tmpdir = self.mkdir("test_find_config_from_home_legacy_name")
config_path = tmpdir.ensure(".dockercfg")
with mock.patch.dict(os.environ, {'HOME': str(tmpdir)}):
with mock.patch.dict(os.environ, {"HOME": str(tmpdir)}):
assert config.find_config_file() == str(config_path)
@mark.skipif("sys.platform != 'win32'")
def test_find_config_from_home_windows(self):
tmpdir = self.mkdir('test_find_config_from_home_windows')
config_path = tmpdir.ensure('.docker', 'config.json')
tmpdir = self.mkdir("test_find_config_from_home_windows")
config_path = tmpdir.ensure(".docker", "config.json")
with mock.patch.dict(os.environ, {'USERPROFILE': str(tmpdir)}):
with mock.patch.dict(os.environ, {"USERPROFILE": str(tmpdir)}):
assert config.find_config_file() == str(config_path)
@@ -89,32 +91,24 @@ class LoadConfigTest(unittest.TestCase):
folder = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, folder)
dockercfg_path = os.path.join(folder, 'config.json')
dockercfg_path = os.path.join(folder, "config.json")
config_data = {
'HttpHeaders': {
'Name': 'Spike',
'Surname': 'Spiegel'
},
"HttpHeaders": {"Name": "Spike", "Surname": "Spiegel"},
}
with open(dockercfg_path, 'w') as f:
with open(dockercfg_path, "w") as f:
json.dump(config_data, f)
cfg = config.load_general_config(dockercfg_path)
assert 'HttpHeaders' in cfg
assert cfg['HttpHeaders'] == {
'Name': 'Spike',
'Surname': 'Spiegel'
}
assert "HttpHeaders" in cfg
assert cfg["HttpHeaders"] == {"Name": "Spike", "Surname": "Spiegel"}
def test_load_config_detach_keys(self):
folder = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, folder)
dockercfg_path = os.path.join(folder, 'config.json')
config_data = {
'detachKeys': 'ctrl-q, ctrl-u, ctrl-i'
}
with open(dockercfg_path, 'w') as f:
dockercfg_path = os.path.join(folder, "config.json")
config_data = {"detachKeys": "ctrl-q, ctrl-u, ctrl-i"}
with open(dockercfg_path, "w") as f:
json.dump(config_data, f)
cfg = config.load_general_config(dockercfg_path)
@@ -123,13 +117,11 @@ class LoadConfigTest(unittest.TestCase):
def test_load_config_from_env(self):
folder = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, folder)
dockercfg_path = os.path.join(folder, 'config.json')
config_data = {
'detachKeys': 'ctrl-q, ctrl-u, ctrl-i'
}
with open(dockercfg_path, 'w') as f:
dockercfg_path = os.path.join(folder, "config.json")
config_data = {"detachKeys": "ctrl-q, ctrl-u, ctrl-i"}
with open(dockercfg_path, "w") as f:
json.dump(config_data, f)
with mock.patch.dict(os.environ, {'DOCKER_CONFIG': folder}):
with mock.patch.dict(os.environ, {"DOCKER_CONFIG": folder}):
cfg = config.load_general_config(None)
assert cfg == config_data
tests/unit/plugins/module_utils/_api/utils/test_decorators.py
+19 -13
View File
@@ -7,20 +7,28 @@
# It is licensed under the Apache 2.0 license (see LICENSES/Apache-2.0.txt in this collection)
# SPDX-License-Identifier: Apache-2.0
from __future__ import (absolute_import, division, print_function)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import unittest
from ansible_collections.community.docker.plugins.module_utils._api.api.client import APIClient
from ansible_collections.community.docker.plugins.module_utils._api.utils.decorators import update_headers
from ansible_collections.community.docker.tests.unit.plugins.module_utils._api.constants import DEFAULT_DOCKER_API_VERSION
from ansible_collections.community.docker.plugins.module_utils._api.api.client import (
APIClient,
)
from ansible_collections.community.docker.plugins.module_utils._api.utils.decorators import (
update_headers,
)
from ansible_collections.community.docker.tests.unit.plugins.module_utils._api.constants import (
DEFAULT_DOCKER_API_VERSION,
)
class DecoratorsTest(unittest.TestCase):
def test_update_headers(self):
sample_headers = {
'X-Docker-Locale': 'en-US',
"X-Docker-Locale": "en-US",
}
def f(self, headers=None):
@@ -32,17 +40,15 @@ class DecoratorsTest(unittest.TestCase):
g = update_headers(f)
assert g(client, headers=None) is None
assert g(client, headers={}) == {}
assert g(client, headers={'Content-type': 'application/json'}) == {
'Content-type': 'application/json',
assert g(client, headers={"Content-type": "application/json"}) == {
"Content-type": "application/json",
}
client._general_configs = {
'HttpHeaders': sample_headers
}
client._general_configs = {"HttpHeaders": sample_headers}
assert g(client, headers=None) == sample_headers
assert g(client, headers={}) == sample_headers
assert g(client, headers={'Content-type': 'application/json'}) == {
'Content-type': 'application/json',
'X-Docker-Locale': 'en-US',
assert g(client, headers={"Content-type": "application/json"}) == {
"Content-type": "application/json",
"X-Docker-Locale": "en-US",
}
tests/unit/plugins/module_utils/_api/utils/test_json_stream.py
+19 -21
View File
@@ -7,10 +7,16 @@
# It is licensed under the Apache 2.0 license (see LICENSES/Apache-2.0.txt in this collection)
# SPDX-License-Identifier: Apache-2.0
from __future__ import (absolute_import, division, print_function)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from ansible_collections.community.docker.plugins.module_utils._api.utils.json_stream import json_splitter, stream_as_text, json_stream
from ansible_collections.community.docker.plugins.module_utils._api.utils.json_stream import (
json_splitter,
json_stream,
stream_as_text,
)
class TestJsonSplitter:
@@ -21,24 +27,24 @@ class TestJsonSplitter:
def test_json_splitter_with_object(self):
data = '{"foo": "bar"}\n \n{"next": "obj"}'
assert json_splitter(data) == ({'foo': 'bar'}, '{"next": "obj"}')
assert json_splitter(data) == ({"foo": "bar"}, '{"next": "obj"}')
def test_json_splitter_leading_whitespace(self):
data = '\n \r{"foo": "bar"}\n\n {"next": "obj"}'
assert json_splitter(data) == ({'foo': 'bar'}, '{"next": "obj"}')
assert json_splitter(data) == ({"foo": "bar"}, '{"next": "obj"}')
class TestStreamAsText:
def test_stream_with_non_utf_unicode_character(self):
stream = [b'\xed\xf3\xf3']
output, = stream_as_text(stream)
assert output == '���'
stream = [b"\xed\xf3\xf3"]
(output,) = stream_as_text(stream)
assert output == "���"
def test_stream_with_utf_character(self):
stream = ['ěĝ'.encode('utf-8')]
output, = stream_as_text(stream)
assert output == 'ěĝ'
stream = ["ěĝ".encode("utf-8")]
(output,) = stream_as_text(stream)
assert output == "ěĝ"
class TestJsonStream:
@@ -50,21 +56,13 @@ class TestJsonStream:
]
output = list(json_stream(stream))
assert output == [
{'one': 'two'},
{"one": "two"},
{},
[1, 2, 3],
[],
]
def test_with_leading_whitespace(self):
stream = [
'\n \r\n {"one": "two"}{"x": 1}',
' {"three": "four"}\t\t{"x": 2}'
]
stream = ['\n \r\n {"one": "two"}{"x": 1}', ' {"three": "four"}\t\t{"x": 2}']
output = list(json_stream(stream))
assert output == [
{'one': 'two'},
{'x': 1},
{'three': 'four'},
{'x': 2}
]
assert output == [{"one": "two"}, {"x": 1}, {"three": "four"}, {"x": 2}]
tests/unit/plugins/module_utils/_api/utils/test_ports.py
+24 -27
View File
@@ -7,14 +7,18 @@
# It is licensed under the Apache 2.0 license (see LICENSES/Apache-2.0.txt in this collection)
# SPDX-License-Identifier: Apache-2.0
from __future__ import (absolute_import, division, print_function)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import unittest
import pytest
from ansible_collections.community.docker.plugins.module_utils._api.utils.ports import build_port_bindings, split_port
from ansible_collections.community.docker.plugins.module_utils._api.utils.ports import (
build_port_bindings,
split_port,
)
class PortsTest(unittest.TestCase):
@@ -24,10 +28,8 @@ class PortsTest(unittest.TestCase):
assert external_port == [("127.0.0.1", "1000")]
def test_split_port_with_protocol(self):
for protocol in ['tcp', 'udp', 'sctp']:
internal_port, external_port = split_port(
"127.0.0.1:1000:2000/" + protocol
)
for protocol in ["tcp", "udp", "sctp"]:
internal_port, external_port = split_port("127.0.0.1:1000:2000/" + protocol)
assert internal_port == ["2000/" + protocol]
assert external_port == [("127.0.0.1", "1000")]
@@ -67,20 +69,17 @@ class PortsTest(unittest.TestCase):
assert external_port is None
def test_split_port_range_with_protocol(self):
internal_port, external_port = split_port(
"127.0.0.1:1000-1001:2000-2001/udp")
internal_port, external_port = split_port("127.0.0.1:1000-1001:2000-2001/udp")
assert internal_port == ["2000/udp", "2001/udp"]
assert external_port == [("127.0.0.1", "1000"), ("127.0.0.1", "1001")]
def test_split_port_with_ipv6_address(self):
internal_port, external_port = split_port(
"2001:abcd:ef00::2:1000:2000")
internal_port, external_port = split_port("2001:abcd:ef00::2:1000:2000")
assert internal_port == ["2000"]
assert external_port == [("2001:abcd:ef00::2", "1000")]
def test_split_port_with_ipv6_square_brackets_address(self):
internal_port, external_port = split_port(
"[2001:abcd:ef00::2]:1000:2000")
internal_port, external_port = split_port("[2001:abcd:ef00::2]:1000:2000")
assert internal_port == ["2000"]
assert external_port == [("2001:abcd:ef00::2", "1000")]
@@ -117,7 +116,7 @@ class PortsTest(unittest.TestCase):
split_port("")
def test_split_port_non_string(self):
assert split_port(1243) == (['1243'], None)
assert split_port(1243) == (["1243"], None)
def test_build_port_bindings_with_one_port(self):
port_bindings = build_port_bindings(["127.0.0.1:1000:1000"])
@@ -125,14 +124,14 @@ class PortsTest(unittest.TestCase):
def test_build_port_bindings_with_matching_internal_ports(self):
port_bindings = build_port_bindings(
["127.0.0.1:1000:1000", "127.0.0.1:2000:1000"])
assert port_bindings["1000"] == [
("127.0.0.1", "1000"), ("127.0.0.1", "2000")
]
["127.0.0.1:1000:1000", "127.0.0.1:2000:1000"]
)
assert port_bindings["1000"] == [("127.0.0.1", "1000"), ("127.0.0.1", "2000")]
def test_build_port_bindings_with_nonmatching_internal_ports(self):
port_bindings = build_port_bindings(
["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"])
["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"]
)
assert port_bindings["1000"] == [("127.0.0.1", "1000")]
assert port_bindings["2000"] == [("127.0.0.1", "2000")]
@@ -143,16 +142,14 @@ class PortsTest(unittest.TestCase):
def test_build_port_bindings_with_matching_internal_port_ranges(self):
port_bindings = build_port_bindings(
["127.0.0.1:1000-1001:1000-1001", "127.0.0.1:2000-2001:1000-1001"])
assert port_bindings["1000"] == [
("127.0.0.1", "1000"), ("127.0.0.1", "2000")
]
assert port_bindings["1001"] == [
("127.0.0.1", "1001"), ("127.0.0.1", "2001")
]
["127.0.0.1:1000-1001:1000-1001", "127.0.0.1:2000-2001:1000-1001"]
)
assert port_bindings["1000"] == [("127.0.0.1", "1000"), ("127.0.0.1", "2000")]
assert port_bindings["1001"] == [("127.0.0.1", "1001"), ("127.0.0.1", "2001")]
def test_build_port_bindings_with_nonmatching_internal_port_ranges(self):
port_bindings = build_port_bindings(
["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"])
["127.0.0.1:1000:1000", "127.0.0.1:2000:2000"]
)
assert port_bindings["1000"] == [("127.0.0.1", "1000")]
assert port_bindings["2000"] == [("127.0.0.1", "2000")]
tests/unit/plugins/module_utils/_api/utils/test_proxy.py
+39 -32
View File
@@ -7,40 +7,46 @@
# It is licensed under the Apache 2.0 license (see LICENSES/Apache-2.0.txt in this collection)
# SPDX-License-Identifier: Apache-2.0
from __future__ import (absolute_import, division, print_function)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import unittest
from ansible_collections.community.docker.plugins.module_utils._api.utils.proxy import ProxyConfig
from ansible_collections.community.docker.plugins.module_utils._api.utils.proxy import (
ProxyConfig,
)
HTTP = 'http://test:80'
HTTPS = 'https://test:443'
FTP = 'ftp://user:password@host:23'
NO_PROXY = 'localhost,.localdomain'
HTTP = "http://test:80"
HTTPS = "https://test:443"
FTP = "ftp://user:password@host:23"
NO_PROXY = "localhost,.localdomain"
CONFIG = ProxyConfig(http=HTTP, https=HTTPS, ftp=FTP, no_proxy=NO_PROXY)
ENV = {
'http_proxy': HTTP,
'HTTP_PROXY': HTTP,
'https_proxy': HTTPS,
'HTTPS_PROXY': HTTPS,
'ftp_proxy': FTP,
'FTP_PROXY': FTP,
'no_proxy': NO_PROXY,
'NO_PROXY': NO_PROXY,
"http_proxy": HTTP,
"HTTP_PROXY": HTTP,
"https_proxy": HTTPS,
"HTTPS_PROXY": HTTPS,
"ftp_proxy": FTP,
"FTP_PROXY": FTP,
"no_proxy": NO_PROXY,
"NO_PROXY": NO_PROXY,
}
class ProxyConfigTest(unittest.TestCase):
def test_from_dict(self):
config = ProxyConfig.from_dict({
'httpProxy': HTTP,
'httpsProxy': HTTPS,
'ftpProxy': FTP,
'noProxy': NO_PROXY
})
config = ProxyConfig.from_dict(
{
"httpProxy": HTTP,
"httpsProxy": HTTPS,
"ftpProxy": FTP,
"noProxy": NO_PROXY,
}
)
self.assertEqual(CONFIG.http, config.http)
self.assertEqual(CONFIG.https, config.https)
self.assertEqual(CONFIG.ftp, config.ftp)
@@ -53,18 +59,18 @@ class ProxyConfigTest(unittest.TestCase):
self.assertIsNone(config.ftp)
self.assertIsNone(config.no_proxy)
config = ProxyConfig(http='a', https='b', ftp='c', no_proxy='d')
self.assertEqual(config.http, 'a')
self.assertEqual(config.https, 'b')
self.assertEqual(config.ftp, 'c')
self.assertEqual(config.no_proxy, 'd')
config = ProxyConfig(http="a", https="b", ftp="c", no_proxy="d")
self.assertEqual(config.http, "a")
self.assertEqual(config.https, "b")
self.assertEqual(config.ftp, "c")
self.assertEqual(config.no_proxy, "d")
def test_truthiness(self):
assert not ProxyConfig()
assert ProxyConfig(http='non-zero')
assert ProxyConfig(https='non-zero')
assert ProxyConfig(ftp='non-zero')
assert ProxyConfig(no_proxy='non-zero')
assert ProxyConfig(http="non-zero")
assert ProxyConfig(https="non-zero")
assert ProxyConfig(ftp="non-zero")
assert ProxyConfig(no_proxy="non-zero")
def test_environment(self):
self.assertDictEqual(CONFIG.get_environment(), ENV)
@@ -75,16 +81,17 @@ class ProxyConfigTest(unittest.TestCase):
# Proxy config is non null, env is None.
self.assertSetEqual(
set(CONFIG.inject_proxy_environment(None)),
set(f'{k}={v}' for k, v in ENV.items()))
set(f"{k}={v}" for k, v in ENV.items()),
)
# Proxy config is null, env is None.
self.assertIsNone(ProxyConfig().inject_proxy_environment(None), None)
env = ['FOO=BAR', 'BAR=BAZ']
env = ["FOO=BAR", "BAR=BAZ"]
# Proxy config is non null, env is non null
actual = CONFIG.inject_proxy_environment(env)
expected = [f'{k}={v}' for k, v in ENV.items()] + env
expected = [f"{k}={v}" for k, v in ENV.items()] + env
# It's important that the first 8 variables are the ones from the proxy
# config, and the last 2 are the ones from the input environment
self.assertSetEqual(set(actual[:8]), set(expected[:8]))
tests/unit/plugins/module_utils/_api/utils/test_utils.py
+206 -206
View File
@@ -7,7 +7,9 @@
# It is licensed under the Apache 2.0 license (see LICENSES/Apache-2.0.txt in this collection)
# SPDX-License-Identifier: Apache-2.0
from __future__ import (absolute_import, division, print_function)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
import base64
@@ -19,22 +21,36 @@ import tempfile
import unittest
import pytest
from ansible_collections.community.docker.plugins.module_utils._api.api.client import APIClient
from ansible_collections.community.docker.plugins.module_utils._api.constants import IS_WINDOWS_PLATFORM
from ansible_collections.community.docker.plugins.module_utils._api.errors import DockerException
from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
convert_filters, convert_volume_binds,
decode_json_header, kwargs_from_env, parse_bytes,
parse_devices, parse_env_file, parse_host,
parse_repository_tag, split_command, format_environment,
from ansible_collections.community.docker.plugins.module_utils._api.api.client import (
APIClient,
)
from ansible_collections.community.docker.plugins.module_utils._api.constants import (
IS_WINDOWS_PLATFORM,
)
from ansible_collections.community.docker.plugins.module_utils._api.errors import (
DockerException,
)
from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
convert_filters,
convert_volume_binds,
decode_json_header,
format_environment,
kwargs_from_env,
parse_bytes,
parse_devices,
parse_env_file,
parse_host,
parse_repository_tag,
split_command,
)
from ansible_collections.community.docker.tests.unit.plugins.module_utils._api.constants import (
DEFAULT_DOCKER_API_VERSION,
)
from ansible_collections.community.docker.tests.unit.plugins.module_utils._api.constants import DEFAULT_DOCKER_API_VERSION
TEST_CERT_DIR = os.path.join(
os.path.dirname(__file__),
'testdata/certs',
"testdata/certs",
)
@@ -46,84 +62,85 @@ class KwargsFromEnvTest(unittest.TestCase):
os.environ = self.os_environ
def test_kwargs_from_env_empty(self):
os.environ.update(DOCKER_HOST='',
DOCKER_CERT_PATH='')
os.environ.pop('DOCKER_TLS_VERIFY', None)
os.environ.update(DOCKER_HOST="", DOCKER_CERT_PATH="")
os.environ.pop("DOCKER_TLS_VERIFY", None)
kwargs = kwargs_from_env()
assert kwargs.get('base_url') is None
assert kwargs.get('tls') is None
assert kwargs.get("base_url") is None
assert kwargs.get("tls") is None
def test_kwargs_from_env_tls(self):
os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376',
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY='1')
os.environ.update(
DOCKER_HOST="tcp://192.168.59.103:2376",
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY="1",
)
kwargs = kwargs_from_env(assert_hostname=False)
assert 'tcp://192.168.59.103:2376' == kwargs['base_url']
assert 'ca.pem' in kwargs['tls'].ca_cert
assert 'cert.pem' in kwargs['tls'].cert[0]
assert 'key.pem' in kwargs['tls'].cert[1]
assert kwargs['tls'].assert_hostname is False
assert kwargs['tls'].verify
assert "tcp://192.168.59.103:2376" == kwargs["base_url"]
assert "ca.pem" in kwargs["tls"].ca_cert
assert "cert.pem" in kwargs["tls"].cert[0]
assert "key.pem" in kwargs["tls"].cert[1]
assert kwargs["tls"].assert_hostname is False
assert kwargs["tls"].verify
parsed_host = parse_host(kwargs['base_url'], IS_WINDOWS_PLATFORM, True)
kwargs['version'] = DEFAULT_DOCKER_API_VERSION
parsed_host = parse_host(kwargs["base_url"], IS_WINDOWS_PLATFORM, True)
kwargs["version"] = DEFAULT_DOCKER_API_VERSION
try:
client = APIClient(**kwargs)
assert parsed_host == client.base_url
assert kwargs['tls'].ca_cert == client.verify
assert kwargs['tls'].cert == client.cert
assert kwargs["tls"].ca_cert == client.verify
assert kwargs["tls"].cert == client.cert
except TypeError as e:
self.fail(e)
def test_kwargs_from_env_tls_verify_false(self):
os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376',
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY='')
os.environ.update(
DOCKER_HOST="tcp://192.168.59.103:2376",
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY="",
)
kwargs = kwargs_from_env(assert_hostname=True)
assert 'tcp://192.168.59.103:2376' == kwargs['base_url']
assert 'ca.pem' in kwargs['tls'].ca_cert
assert 'cert.pem' in kwargs['tls'].cert[0]
assert 'key.pem' in kwargs['tls'].cert[1]
assert kwargs['tls'].assert_hostname is True
assert kwargs['tls'].verify is False
parsed_host = parse_host(kwargs['base_url'], IS_WINDOWS_PLATFORM, True)
kwargs['version'] = DEFAULT_DOCKER_API_VERSION
assert "tcp://192.168.59.103:2376" == kwargs["base_url"]
assert "ca.pem" in kwargs["tls"].ca_cert
assert "cert.pem" in kwargs["tls"].cert[0]
assert "key.pem" in kwargs["tls"].cert[1]
assert kwargs["tls"].assert_hostname is True
assert kwargs["tls"].verify is False
parsed_host = parse_host(kwargs["base_url"], IS_WINDOWS_PLATFORM, True)
kwargs["version"] = DEFAULT_DOCKER_API_VERSION
try:
client = APIClient(**kwargs)
assert parsed_host == client.base_url
assert kwargs['tls'].cert == client.cert
assert not kwargs['tls'].verify
assert kwargs["tls"].cert == client.cert
assert not kwargs["tls"].verify
except TypeError as e:
self.fail(e)
def test_kwargs_from_env_tls_verify_false_no_cert(self):
temp_dir = tempfile.mkdtemp()
cert_dir = os.path.join(temp_dir, '.docker')
cert_dir = os.path.join(temp_dir, ".docker")
shutil.copytree(TEST_CERT_DIR, cert_dir)
os.environ.update(DOCKER_HOST='tcp://192.168.59.103:2376',
HOME=temp_dir,
DOCKER_TLS_VERIFY='')
os.environ.pop('DOCKER_CERT_PATH', None)
os.environ.update(
DOCKER_HOST="tcp://192.168.59.103:2376", HOME=temp_dir, DOCKER_TLS_VERIFY=""
)
os.environ.pop("DOCKER_CERT_PATH", None)
kwargs = kwargs_from_env(assert_hostname=True)
assert 'tcp://192.168.59.103:2376' == kwargs['base_url']
assert "tcp://192.168.59.103:2376" == kwargs["base_url"]
def test_kwargs_from_env_no_cert_path(self):
try:
temp_dir = tempfile.mkdtemp()
cert_dir = os.path.join(temp_dir, '.docker')
cert_dir = os.path.join(temp_dir, ".docker")
shutil.copytree(TEST_CERT_DIR, cert_dir)
os.environ.update(HOME=temp_dir,
DOCKER_CERT_PATH='',
DOCKER_TLS_VERIFY='1')
os.environ.update(HOME=temp_dir, DOCKER_CERT_PATH="", DOCKER_TLS_VERIFY="1")
kwargs = kwargs_from_env()
assert kwargs['tls'].verify
assert cert_dir in kwargs['tls'].ca_cert
assert cert_dir in kwargs['tls'].cert[0]
assert cert_dir in kwargs['tls'].cert[1]
assert kwargs["tls"].verify
assert cert_dir in kwargs["tls"].ca_cert
assert cert_dir in kwargs["tls"].cert[0]
assert cert_dir in kwargs["tls"].cert[1]
finally:
if temp_dir:
shutil.rmtree(temp_dir)
@@ -132,15 +149,17 @@ class KwargsFromEnvTest(unittest.TestCase):
# Values in os.environ are entirely ignored if an alternate is
# provided
os.environ.update(
DOCKER_HOST='tcp://192.168.59.103:2376',
DOCKER_HOST="tcp://192.168.59.103:2376",
DOCKER_CERT_PATH=TEST_CERT_DIR,
DOCKER_TLS_VERIFY=''
DOCKER_TLS_VERIFY="",
)
kwargs = kwargs_from_env(environment={
'DOCKER_HOST': 'http://docker.gensokyo.jp:2581',
})
assert 'http://docker.gensokyo.jp:2581' == kwargs['base_url']
assert 'tls' not in kwargs
kwargs = kwargs_from_env(
environment={
"DOCKER_HOST": "http://docker.gensokyo.jp:2581",
}
)
assert "http://docker.gensokyo.jp:2581" == kwargs["base_url"]
assert "tls" not in kwargs
class ConverVolumeBindsTest(unittest.TestCase):
@@ -149,52 +168,36 @@ class ConverVolumeBindsTest(unittest.TestCase):
assert convert_volume_binds([]) == []
def test_convert_volume_binds_list(self):
data = ['/a:/a:ro', '/b:/c:z']
data = ["/a:/a:ro", "/b:/c:z"]
assert convert_volume_binds(data) == data
def test_convert_volume_binds_complete(self):
data = {
'/mnt/vol1': {
'bind': '/data',
'mode': 'ro'
}
}
assert convert_volume_binds(data) == ['/mnt/vol1:/data:ro']
data = {"/mnt/vol1": {"bind": "/data", "mode": "ro"}}
assert convert_volume_binds(data) == ["/mnt/vol1:/data:ro"]
def test_convert_volume_binds_compact(self):
data = {
'/mnt/vol1': '/data'
}
assert convert_volume_binds(data) == ['/mnt/vol1:/data:rw']
data = {"/mnt/vol1": "/data"}
assert convert_volume_binds(data) == ["/mnt/vol1:/data:rw"]
def test_convert_volume_binds_no_mode(self):
data = {
'/mnt/vol1': {
'bind': '/data'
}
}
assert convert_volume_binds(data) == ['/mnt/vol1:/data:rw']
data = {"/mnt/vol1": {"bind": "/data"}}
assert convert_volume_binds(data) == ["/mnt/vol1:/data:rw"]
def test_convert_volume_binds_unicode_bytes_input(self):
expected = ['/mnt/지연:/unicode/박:rw']
expected = ["/mnt/지연:/unicode/박:rw"]
data = {
'/mnt/지연'.encode('utf-8'): {
'bind': '/unicode/박'.encode('utf-8'),
'mode': 'rw'
"/mnt/지연".encode("utf-8"): {
"bind": "/unicode/박".encode("utf-8"),
"mode": "rw",
}
}
assert convert_volume_binds(data) == expected
def test_convert_volume_binds_unicode_unicode_input(self):
expected = ['/mnt/지연:/unicode/박:rw']
expected = ["/mnt/지연:/unicode/박:rw"]
data = {
'/mnt/지연': {
'bind': '/unicode/박',
'mode': 'rw'
}
}
data = {"/mnt/지연": {"bind": "/unicode/박", "mode": "rw"}}
assert convert_volume_binds(data) == expected
@@ -206,41 +209,36 @@ class ParseEnvFileTest(unittest.TestCase):
Don't forget to unlink the file with os.unlink() after.
"""
local_tempfile = tempfile.NamedTemporaryFile(delete=False)
local_tempfile.write(file_content.encode('UTF-8'))
local_tempfile.write(file_content.encode("UTF-8"))
local_tempfile.close()
return local_tempfile.name
def test_parse_env_file_proper(self):
env_file = self.generate_tempfile(
file_content='USER=jdoe\nPASS=secret')
env_file = self.generate_tempfile(file_content="USER=jdoe\nPASS=secret")
get_parse_env_file = parse_env_file(env_file)
assert get_parse_env_file == {'USER': 'jdoe', 'PASS': 'secret'}
assert get_parse_env_file == {"USER": "jdoe", "PASS": "secret"}
os.unlink(env_file)
def test_parse_env_file_with_equals_character(self):
env_file = self.generate_tempfile(
file_content='USER=jdoe\nPASS=sec==ret')
env_file = self.generate_tempfile(file_content="USER=jdoe\nPASS=sec==ret")
get_parse_env_file = parse_env_file(env_file)
assert get_parse_env_file == {'USER': 'jdoe', 'PASS': 'sec==ret'}
assert get_parse_env_file == {"USER": "jdoe", "PASS": "sec==ret"}
os.unlink(env_file)
def test_parse_env_file_commented_line(self):
env_file = self.generate_tempfile(
file_content='USER=jdoe\n#PASS=secret')
env_file = self.generate_tempfile(file_content="USER=jdoe\n#PASS=secret")
get_parse_env_file = parse_env_file(env_file)
assert get_parse_env_file == {'USER': 'jdoe'}
assert get_parse_env_file == {"USER": "jdoe"}
os.unlink(env_file)
def test_parse_env_file_newline(self):
env_file = self.generate_tempfile(
file_content='\nUSER=jdoe\n\n\nPASS=secret')
env_file = self.generate_tempfile(file_content="\nUSER=jdoe\n\n\nPASS=secret")
get_parse_env_file = parse_env_file(env_file)
assert get_parse_env_file == {'USER': 'jdoe', 'PASS': 'secret'}
assert get_parse_env_file == {"USER": "jdoe", "PASS": "secret"}
os.unlink(env_file)
def test_parse_env_file_invalid_line(self):
env_file = self.generate_tempfile(
file_content='USER jdoe')
env_file = self.generate_tempfile(file_content="USER jdoe")
with pytest.raises(DockerException):
parse_env_file(env_file)
os.unlink(env_file)
@@ -249,46 +247,44 @@ class ParseEnvFileTest(unittest.TestCase):
class ParseHostTest(unittest.TestCase):
def test_parse_host(self):
invalid_hosts = [
'foo://0.0.0.0',
'tcp://',
'udp://127.0.0.1',
'udp://127.0.0.1:2375',
'ssh://:22/path',
'tcp://netloc:3333/path?q=1',
'unix:///sock/path#fragment',
'https://netloc:3333/path;params',
'ssh://:clearpassword@host:22',
"foo://0.0.0.0",
"tcp://",
"udp://127.0.0.1",
"udp://127.0.0.1:2375",
"ssh://:22/path",
"tcp://netloc:3333/path?q=1",
"unix:///sock/path#fragment",
"https://netloc:3333/path;params",
"ssh://:clearpassword@host:22",
]
valid_hosts = {
'0.0.0.1:5555': 'http://0.0.0.1:5555',
':6666': 'http://127.0.0.1:6666',
'tcp://:7777': 'http://127.0.0.1:7777',
'http://:7777': 'http://127.0.0.1:7777',
'https://kokia.jp:2375': 'https://kokia.jp:2375',
'unix:///var/run/docker.sock': 'http+unix:///var/run/docker.sock',
'unix://': 'http+unix:///var/run/docker.sock',
'12.234.45.127:2375/docker/engine': (
'http://12.234.45.127:2375/docker/engine'
"0.0.0.1:5555": "http://0.0.0.1:5555",
":6666": "http://127.0.0.1:6666",
"tcp://:7777": "http://127.0.0.1:7777",
"http://:7777": "http://127.0.0.1:7777",
"https://kokia.jp:2375": "https://kokia.jp:2375",
"unix:///var/run/docker.sock": "http+unix:///var/run/docker.sock",
"unix://": "http+unix:///var/run/docker.sock",
"12.234.45.127:2375/docker/engine": (
"http://12.234.45.127:2375/docker/engine"
),
'somehost.net:80/service/swarm': (
'http://somehost.net:80/service/swarm'
"somehost.net:80/service/swarm": ("http://somehost.net:80/service/swarm"),
"npipe:////./pipe/docker_engine": "npipe:////./pipe/docker_engine",
"[fd12::82d1]:2375": "http://[fd12::82d1]:2375",
"https://[fd12:5672::12aa]:1090": "https://[fd12:5672::12aa]:1090",
"[fd12::82d1]:2375/docker/engine": (
"http://[fd12::82d1]:2375/docker/engine"
),
'npipe:////./pipe/docker_engine': 'npipe:////./pipe/docker_engine',
'[fd12::82d1]:2375': 'http://[fd12::82d1]:2375',
'https://[fd12:5672::12aa]:1090': 'https://[fd12:5672::12aa]:1090',
'[fd12::82d1]:2375/docker/engine': (
'http://[fd12::82d1]:2375/docker/engine'
),
'ssh://[fd12::82d1]': 'ssh://[fd12::82d1]:22',
'ssh://user@[fd12::82d1]:8765': 'ssh://user@[fd12::82d1]:8765',
'ssh://': 'ssh://127.0.0.1:22',
'ssh://user@localhost:22': 'ssh://user@localhost:22',
'ssh://user@remote': 'ssh://user@remote:22',
"ssh://[fd12::82d1]": "ssh://[fd12::82d1]:22",
"ssh://user@[fd12::82d1]:8765": "ssh://user@[fd12::82d1]:8765",
"ssh://": "ssh://127.0.0.1:22",
"ssh://user@localhost:22": "ssh://user@localhost:22",
"ssh://user@remote": "ssh://user@remote:22",
}
for host in invalid_hosts:
msg = f'Should have failed to parse invalid host: {host}'
msg = f"Should have failed to parse invalid host: {host}"
with self.assertRaises(DockerException, msg=msg):
parse_host(host, None)
@@ -296,35 +292,35 @@ class ParseHostTest(unittest.TestCase):
self.assertEqual(
parse_host(host, None),
expected,
msg=f'Failed to parse valid host: {host}',
msg=f"Failed to parse valid host: {host}",
)
def test_parse_host_empty_value(self):
unix_socket = 'http+unix:///var/run/docker.sock'
npipe = 'npipe:////./pipe/docker_engine'
unix_socket = "http+unix:///var/run/docker.sock"
npipe = "npipe:////./pipe/docker_engine"
for val in [None, '']:
for val in [None, ""]:
assert parse_host(val, is_win32=False) == unix_socket
assert parse_host(val, is_win32=True) == npipe
def test_parse_host_tls(self):
host_value = 'myhost.docker.net:3348'
expected_result = 'https://myhost.docker.net:3348'
host_value = "myhost.docker.net:3348"
expected_result = "https://myhost.docker.net:3348"
assert parse_host(host_value, tls=True) == expected_result
def test_parse_host_tls_tcp_proto(self):
host_value = 'tcp://myhost.docker.net:3348'
expected_result = 'https://myhost.docker.net:3348'
host_value = "tcp://myhost.docker.net:3348"
expected_result = "https://myhost.docker.net:3348"
assert parse_host(host_value, tls=True) == expected_result
def test_parse_host_trailing_slash(self):
host_value = 'tcp://myhost.docker.net:2376/'
expected_result = 'http://myhost.docker.net:2376'
host_value = "tcp://myhost.docker.net:2376/"
expected_result = "http://myhost.docker.net:2376"
assert parse_host(host_value) == expected_result
class ParseRepositoryTagTest(unittest.TestCase):
sha = 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'
sha = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
def test_index_image_no_tag(self):
assert parse_repository_tag("root") == ("root", None)
@@ -342,77 +338,83 @@ class ParseRepositoryTagTest(unittest.TestCase):
assert parse_repository_tag("url:5000/repo") == ("url:5000/repo", None)
def test_private_reg_image_tag(self):
assert parse_repository_tag("url:5000/repo:tag") == (
"url:5000/repo", "tag"
)
assert parse_repository_tag("url:5000/repo:tag") == ("url:5000/repo", "tag")
def test_index_image_sha(self):
assert parse_repository_tag(f"root@sha256:{self.sha}") == (
"root", f"sha256:{self.sha}"
"root",
f"sha256:{self.sha}",
)
def test_private_reg_image_sha(self):
assert parse_repository_tag(
f"url:5000/repo@sha256:{self.sha}"
) == ("url:5000/repo", f"sha256:{self.sha}")
assert parse_repository_tag(f"url:5000/repo@sha256:{self.sha}") == (
"url:5000/repo",
f"sha256:{self.sha}",
)
class ParseDeviceTest(unittest.TestCase):
def test_dict(self):
devices = parse_devices([{
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/mnt1',
'CgroupPermissions': 'r'
}])
devices = parse_devices(
[
{
"PathOnHost": "/dev/sda1",
"PathInContainer": "/dev/mnt1",
"CgroupPermissions": "r",
}
]
)
assert devices[0] == {
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/mnt1',
'CgroupPermissions': 'r'
"PathOnHost": "/dev/sda1",
"PathInContainer": "/dev/mnt1",
"CgroupPermissions": "r",
}
def test_partial_string_definition(self):
devices = parse_devices(['/dev/sda1'])
devices = parse_devices(["/dev/sda1"])
assert devices[0] == {
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/sda1',
'CgroupPermissions': 'rwm'
"PathOnHost": "/dev/sda1",
"PathInContainer": "/dev/sda1",
"CgroupPermissions": "rwm",
}
def test_permissionless_string_definition(self):
devices = parse_devices(['/dev/sda1:/dev/mnt1'])
devices = parse_devices(["/dev/sda1:/dev/mnt1"])
assert devices[0] == {
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/mnt1',
'CgroupPermissions': 'rwm'
"PathOnHost": "/dev/sda1",
"PathInContainer": "/dev/mnt1",
"CgroupPermissions": "rwm",
}
def test_full_string_definition(self):
devices = parse_devices(['/dev/sda1:/dev/mnt1:r'])
devices = parse_devices(["/dev/sda1:/dev/mnt1:r"])
assert devices[0] == {
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/mnt1',
'CgroupPermissions': 'r'
"PathOnHost": "/dev/sda1",
"PathInContainer": "/dev/mnt1",
"CgroupPermissions": "r",
}
def test_hybrid_list(self):
devices = parse_devices([
'/dev/sda1:/dev/mnt1:rw',
{
'PathOnHost': '/dev/sda2',
'PathInContainer': '/dev/mnt2',
'CgroupPermissions': 'r'
}
])
devices = parse_devices(
[
"/dev/sda1:/dev/mnt1:rw",
{
"PathOnHost": "/dev/sda2",
"PathInContainer": "/dev/mnt2",
"CgroupPermissions": "r",
},
]
)
assert devices[0] == {
'PathOnHost': '/dev/sda1',
'PathInContainer': '/dev/mnt1',
'CgroupPermissions': 'rw'
"PathOnHost": "/dev/sda1",
"PathInContainer": "/dev/mnt1",
"CgroupPermissions": "rw",
}
assert devices[1] == {
'PathOnHost': '/dev/sda2',
'PathInContainer': '/dev/mnt2',
'CgroupPermissions': 'r'
"PathOnHost": "/dev/sda2",
"PathInContainer": "/dev/mnt2",
"CgroupPermissions": "r",
}
@@ -439,37 +441,35 @@ class UtilsTest(unittest.TestCase):
def test_convert_filters(self):
tests = [
({'dangling': True}, '{"dangling": ["true"]}'),
({'dangling': "true"}, '{"dangling": ["true"]}'),
({'exited': 0}, '{"exited": ["0"]}'),
({'exited': [0, 1]}, '{"exited": ["0", "1"]}'),
({"dangling": True}, '{"dangling": ["true"]}'),
({"dangling": "true"}, '{"dangling": ["true"]}'),
({"exited": 0}, '{"exited": ["0"]}'),
({"exited": [0, 1]}, '{"exited": ["0", "1"]}'),
]
for filters, expected in tests:
assert convert_filters(filters) == expected
def test_decode_json_header(self):
obj = {'a': 'b', 'c': 1}
data = base64.urlsafe_b64encode(bytes(json.dumps(obj), 'utf-8'))
obj = {"a": "b", "c": 1}
data = base64.urlsafe_b64encode(bytes(json.dumps(obj), "utf-8"))
decoded_data = decode_json_header(data)
assert obj == decoded_data
class SplitCommandTest(unittest.TestCase):
def test_split_command_with_unicode(self):
assert split_command('echo μμ') == ['echo', 'μμ']
assert split_command("echo μμ") == ["echo", "μμ"]
class FormatEnvironmentTest(unittest.TestCase):
def test_format_env_binary_unicode_value(self):
env_dict = {
'ARTIST_NAME': b'\xec\x86\xa1\xec\xa7\x80\xec\x9d\x80'
}
assert format_environment(env_dict) == ['ARTIST_NAME=송지은']
env_dict = {"ARTIST_NAME": b"\xec\x86\xa1\xec\xa7\x80\xec\x9d\x80"}
assert format_environment(env_dict) == ["ARTIST_NAME=송지은"]
def test_format_env_no_value(self):
env_dict = {
'FOO': None,
'BAR': '',
"FOO": None,
"BAR": "",
}
assert sorted(format_environment(env_dict)) == ['BAR=', 'FOO']
assert sorted(format_environment(env_dict)) == ["BAR=", "FOO"]