ansible-collections/community.docker
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.docker.git synced 2025-12-17 04:18:42 +00:00
Code Packages Projects Releases Wiki Activity

Address many redefined-outer-name.

Browse Source
This commit is contained in:
Felix Fontein 2025-10-07 21:32:49 +02:00
parent 38795830b7
commit e07db8a832
13 changed files with 69 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.pylintrc
View File

@ -407,7 +407,7 @@ disable=raw-checker-failed,
possibly-used-before-assignment,
protected-access,
raise-missing-from,
redefined-outer-name,
redefined-outer-name, # needed for test fixtures
simplifiable-if-expression,
subprocess-popen-preexec-fn,
super-init-not-called,

26
plugins/module_utils/_api/api/client.py
View File

@ -289,14 +289,14 @@ class APIClient(_Session, DaemonApiMixin):
except _HTTPError as e:
create_api_error_from_http_exception(e)
def _result(self, response, json=False, binary=False):
if json and binary:
def _result(self, response, get_json=False, get_binary=False):
if get_json and get_binary:
raise AssertionError("json and binary must not be both True")
self._raise_for_status(response)
if json:
if get_json:
return response.json()
if binary:
if get_binary:
return response.content
return response.text
@ -360,12 +360,12 @@ class APIClient(_Session, DaemonApiMixin):
else:
# Response is not chunked, meaning we probably
# encountered an error immediately
yield self._result(response, json=decode)
yield self._result(response, get_json=decode)
def _multiplexed_buffer_helper(self, response):
"""A generator of multiplexed data blocks read from a buffered
response."""
buf = self._result(response, binary=True)
buf = self._result(response, get_binary=True)
buf_length = len(buf)
walker = 0
while True:
@ -478,7 +478,7 @@ class APIClient(_Session, DaemonApiMixin):
return (
self._stream_raw_result(res)
if stream
else self._result(res, binary=True)
else self._result(res, get_binary=True)
)
self._raise_for_status(res)
@ -551,13 +551,13 @@ class APIClient(_Session, DaemonApiMixin):
def get_binary(self, pathfmt, *args, **kwargs):
return self._result(
self._get(self._url(pathfmt, *args, versioned_api=True), **kwargs),
binary=True,
get_binary=True,
)
def get_json(self, pathfmt, *args, **kwargs):
return self._result(
self._get(self._url(pathfmt, *args, versioned_api=True), **kwargs),
json=True,
get_json=True,
)
def get_text(self, pathfmt, *args, **kwargs):
@ -581,7 +581,7 @@ class APIClient(_Session, DaemonApiMixin):
def delete_json(self, pathfmt, *args, **kwargs):
return self._result(
self._delete(self._url(pathfmt, *args, versioned_api=True), **kwargs),
json=True,
get_json=True,
)
def post_call(self, pathfmt, *args, **kwargs):
@ -603,7 +603,7 @@ class APIClient(_Session, DaemonApiMixin):
self._post_json(
self._url(pathfmt, *args, versioned_api=True), data, **kwargs
),
binary=True,
get_binary=True,
)
def post_json_to_json(self, pathfmt, *args, **kwargs):
@ -612,7 +612,7 @@ class APIClient(_Session, DaemonApiMixin):
self._post_json(
self._url(pathfmt, *args, versioned_api=True), data, **kwargs
),
json=True,
get_json=True,
)
def post_json_to_text(self, pathfmt, *args, **kwargs):
@ -670,5 +670,5 @@ class APIClient(_Session, DaemonApiMixin):
def post_to_json(self, pathfmt, *args, **kwargs):
return self._result(
self._post(self._url(pathfmt, *args, versioned_api=True), **kwargs),
json=True,
get_json=True,
)

8
plugins/module_utils/_api/api/daemon.py
View File

@ -33,7 +33,7 @@ class DaemonApiMixin(object):
If the server returns an error.
"""
url = self._url("/system/df")
return self._result(self._get(url), True)
return self._result(self._get(url), get_json=True)
def info(self):
"""
@ -47,7 +47,7 @@ class DaemonApiMixin(object):
:py:class:`docker.errors.APIError`
If the server returns an error.
"""
return self._result(self._get(self._url("/info")), True)
return self._result(self._get(self._url("/info")), get_json=True)
def login(
self,
@ -108,7 +108,7 @@ class DaemonApiMixin(object):
response = self._post_json(self._url("/auth"), data=req_data)
if response.status_code == 200:
self._auth_configs.add_auth(registry or auth.INDEX_NAME, req_data)
return self._result(response, json=True)
return self._result(response, get_json=True)
def ping(self):
"""
@ -137,4 +137,4 @@ class DaemonApiMixin(object):
If the server returns an error.
"""
url = self._url("/version", versioned_api=api_version)
return self._result(self._get(url), json=True)
return self._result(self._get(url), get_json=True)

50
plugins/module_utils/_common.py
View File

@ -133,43 +133,45 @@ def _get_tls_config(fail_function, **kwargs):
fail_function(f"TLS config error: {exc}")
def is_using_tls(auth):
return auth["tls_verify"] or auth["tls"]
def is_using_tls(auth_data):
return auth_data["tls_verify"] or auth_data["tls"]
def get_connect_params(auth, fail_function):
if is_using_tls(auth):
auth["docker_host"] = auth["docker_host"].replace("tcp://", "https://")
result = dict(
base_url=auth["docker_host"],
version=auth["api_version"],
timeout=auth["timeout"],
def get_connect_params(auth_data, fail_function):
if is_using_tls(auth_data):
auth_data["docker_host"] = auth_data["docker_host"].replace(
"tcp://", "https://"
)
if auth["tls_verify"]:
result = dict(
base_url=auth_data["docker_host"],
version=auth_data["api_version"],
timeout=auth_data["timeout"],
)
if auth_data["tls_verify"]:
# TLS with verification
tls_config = dict(
verify=True,
assert_hostname=auth["tls_hostname"],
assert_hostname=auth_data["tls_hostname"],
fail_function=fail_function,
)
if auth["cert_path"] and auth["key_path"]:
tls_config["client_cert"] = (auth["cert_path"], auth["key_path"])
if auth["cacert_path"]:
tls_config["ca_cert"] = auth["cacert_path"]
if auth_data["cert_path"] and auth_data["key_path"]:
tls_config["client_cert"] = (auth_data["cert_path"], auth_data["key_path"])
if auth_data["cacert_path"]:
tls_config["ca_cert"] = auth_data["cacert_path"]
result["tls"] = _get_tls_config(**tls_config)
elif auth["tls"]:
elif auth_data["tls"]:
# TLS without verification
tls_config = dict(
verify=False,
fail_function=fail_function,
)
if auth["cert_path"] and auth["key_path"]:
tls_config["client_cert"] = (auth["cert_path"], auth["key_path"])
if auth_data["cert_path"] and auth_data["key_path"]:
tls_config["client_cert"] = (auth_data["cert_path"], auth_data["key_path"])
result["tls"] = _get_tls_config(**tls_config)
if auth.get("use_ssh_client"):
if auth_data.get("use_ssh_client"):
if LooseVersion(docker_version) < LooseVersion("4.4.0"):
fail_function(
"use_ssh_client=True requires Docker SDK for Python 4.4.0 or newer"
@ -579,7 +581,7 @@ class AnsibleDockerClientBase(Client):
break
return images
def pull_image(self, name, tag="latest", platform=None):
def pull_image(self, name, tag="latest", image_platform=None):
"""
Pull an image
"""
@ -588,8 +590,8 @@ class AnsibleDockerClientBase(Client):
stream=True,
decode=True,
)
if platform is not None:
kwargs["platform"] = platform
if image_platform is not None:
kwargs["platform"] = image_platform
self.log(f"Pulling image {name}:{tag}")
old_tag = self.find_image(name, tag)
try:
@ -624,7 +626,7 @@ class AnsibleDockerClientBase(Client):
self._url("/distribution/{0}/json", image),
headers={"X-Registry-Auth": header},
),
json=True,
get_json=True,
)
return super(AnsibleDockerClientBase, self).inspect_distribution(
image, **kwargs

6
plugins/module_utils/_common_api.py
View File

@ -495,7 +495,7 @@ class AnsibleDockerClientBase(Client):
except Exception as exc:
self.fail(f"Error inspecting image ID {image_id} - {exc}")
def pull_image(self, name, tag="latest", platform=None):
def pull_image(self, name, tag="latest", image_platform=None):
"""
Pull an image
"""
@ -508,8 +508,8 @@ class AnsibleDockerClientBase(Client):
"tag": tag or image_tag or "latest",
"fromImage": repository,
}
if platform is not None:
params["platform"] = platform
if image_platform is not None:
params["platform"] = image_platform
headers = {}
header = auth.get_config_header(self, registry)

2
plugins/module_utils/_module_container/base.py
View File

@ -330,7 +330,7 @@ class EngineDriver(object):
pass
@abc.abstractmethod
def pull_image(self, client, repository, tag, platform=None):
def pull_image(self, client, repository, tag, image_platform=None):
pass
@abc.abstractmethod

4
plugins/module_utils/_module_container/docker_api.py
View File

@ -232,8 +232,8 @@ class DockerAPIEngineDriver(EngineDriver):
def inspect_image_by_name(self, client, repository, tag):
return client.find_image(repository, tag)
def pull_image(self, client, repository, tag, platform=None):
return client.pull_image(repository, tag, platform=platform)
def pull_image(self, client, repository, tag, image_platform=None):
return client.pull_image(repository, tag, image_platform=image_platform)
def pause_container(self, client, container_id):
client.post_call("/containers/{0}/pause", container_id)

2
plugins/module_utils/_module_container/module.py
View File

@ -569,7 +569,7 @@ class ContainerManager(DockerBaseClass):
self.client,
repository,
tag,
platform=self.module.params["platform"],
image_platform=self.module.params["platform"],
)
if alreadyToLatest:
self.results["changed"] = False

2
plugins/modules/docker_image.py
View File

@ -570,7 +570,7 @@ class ImageManager(DockerBaseClass):
self.results["changed"] = True
if not self.check_mode:
self.results["image"], dummy = self.client.pull_image(
self.name, tag=self.tag, platform=self.pull_platform
self.name, tag=self.tag, image_platform=self.pull_platform
)
elif self.source == "local":
if image is None:

2
plugins/modules/docker_image_pull.py
View File

@ -181,7 +181,7 @@ class ImagePuller(DockerBaseClass):
results["diff"]["after"] = image_info(dict(Id="unknown"))
else:
results["image"], not_changed = self.client.pull_image(
self.name, tag=self.tag, platform=self.platform
self.name, tag=self.tag, image_platform=self.platform
)
results["changed"] = not not_changed
results["diff"]["after"] = image_info(results["image"])

6
plugins/modules/docker_login.py
View File

@ -214,13 +214,13 @@ class DockerFileStore(object):
"""
b64auth = base64.b64encode(to_bytes(username) + b":" + to_bytes(password))
auth = to_text(b64auth)
tauth = to_text(b64auth)
# build up the auth structure
if "auths" not in self._config:
self._config["auths"] = dict()
self._config["auths"][server] = dict(auth=auth)
self._config["auths"][server] = dict(auth=tauth)
self._write()
@ -294,7 +294,7 @@ class LoginManager(DockerBaseClass):
self.client._auth_configs.add_auth(
self.registry_url or auth.INDEX_NAME, req_data
)
return self.client._result(response, json=True)
return self.client._result(response, get_json=True)
def login(self):
"""

26
tests/unit/plugins/module_utils/_api/api/test_client.py
View File

@ -619,29 +619,29 @@ class DisableSocketTest(unittest.TestCase):
def test_disable_socket_timeout(self):
"""Test that the timeout is disabled on a generic socket object."""
socket = self.DummySocket()
the_socket = self.DummySocket()
self.client._disable_socket_timeout(socket)
self.client._disable_socket_timeout(the_socket)
assert socket.timeout is None
assert the_socket.timeout is None
def test_disable_socket_timeout2(self):
"""Test that the timeouts are disabled on a generic socket object
and it's _sock object if present."""
socket = self.DummySocket()
socket._sock = self.DummySocket()
the_socket = self.DummySocket()
the_socket._sock = self.DummySocket()
self.client._disable_socket_timeout(socket)
self.client._disable_socket_timeout(the_socket)
assert socket.timeout is None
assert socket._sock.timeout is None
assert the_socket.timeout is None
assert the_socket._sock.timeout is None
def test_disable_socket_timout_non_blocking(self):
"""Test that a non-blocking socket does not get set to blocking."""
socket = self.DummySocket()
socket._sock = self.DummySocket(0.0)
the_socket = self.DummySocket()
the_socket._sock = self.DummySocket(0.0)
self.client._disable_socket_timeout(socket)
self.client._disable_socket_timeout(the_socket)
assert socket.timeout is None
assert socket._sock.timeout == 0.0
assert the_socket.timeout is None
assert the_socket._sock.timeout == 0.0

4
tests/unit/plugins/module_utils/_api/test_errors.py
View File

@ -123,8 +123,8 @@ class APIErrorTest(unittest.TestCase):
except requests.exceptions.HTTPError as e:
try:
create_api_error_from_http_exception(e)
except APIError as e:
err = e
except APIError as e2:
err = e2
assert err.is_server_error() is True