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

Add docker_context_info module (#1039)

Browse Source 
* Vendor parts of the Docker SDK for Python

This is a combination of the latest git version
(db7f8b8bb6)
with some fixes to make it compatible with Python 2.7
and adjusting some imports.

* Polishing.

* Fix bug that prevents contexts to be found when no Docker config file is present.

Ref: https://github.com/docker/docker-py/issues/3190

* Linting.

* Fix typos.

* Adjust more to behavior of Docker CLI.

* Add first iteration of docker_context_info module.

* Improvements.

* Add basic CI.

* Add caveat on contexts[].config result.
This commit is contained in:
Felix Fontein 2025-02-10 21:59:05 +01:00 committed by GitHub
parent ea3ac5f195
commit 3b6068e44b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 1180 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

242
plugins/module_utils/_api/context/api.py Normal file
View File

@ -0,0 +1,242 @@
# -*- coding: utf-8 -*-
# This code is part of the Ansible collection community.docker, but is an independent component.
# This particular file, and this file only, is based on the Docker SDK for Python (https://github.com/docker/docker-py/)
#
# Copyright (c) 2016-2025 Docker, Inc.
#
# 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)
__metaclass__ = type
import json
import os
from ansible.module_utils.six import raise_from
from .. import errors
from .config import (
METAFILE,
get_current_context_name,
get_meta_dir,
write_context_name_to_docker_config,
)
from .context import Context
def create_default_context():
host = None
if os.environ.get('DOCKER_HOST'):
host = os.environ.get('DOCKER_HOST')
return Context("default", "swarm", host, description="Current DOCKER_HOST based configuration")
class ContextAPI(object):
"""Context API.
Contains methods for context management:
create, list, remove, get, inspect.
"""
DEFAULT_CONTEXT = None
@classmethod
def get_default_context(cls):
context = cls.DEFAULT_CONTEXT
if context is None:
context = create_default_context()
cls.DEFAULT_CONTEXT = context
return context
@classmethod
def create_context(
cls, name, orchestrator=None, host=None, tls_cfg=None,
default_namespace=None, skip_tls_verify=False):
"""Creates a new context.
Returns:
(Context): a Context object.
Raises:
:py:class:`docker.errors.MissingContextParameter`
If a context name is not provided.
:py:class:`docker.errors.ContextAlreadyExists`
If a context with the name already exists.
:py:class:`docker.errors.ContextException`
If name is default.
Example:
>>> from docker.context import ContextAPI
>>> ctx = ContextAPI.create_context(name='test')
>>> print(ctx.Metadata)
{
"Name": "test",
"Metadata": {},
"Endpoints": {
"docker": {
"Host": "unix:///var/run/docker.sock",
"SkipTLSVerify": false
}
}
}
"""
if not name:
raise errors.MissingContextParameter("name")
if name == "default":
raise errors.ContextException(
'"default" is a reserved context name')
ctx = Context.load_context(name)
if ctx:
raise errors.ContextAlreadyExists(name)
endpoint = "docker"
if orchestrator and orchestrator != "swarm":
endpoint = orchestrator
ctx = Context(name, orchestrator)
ctx.set_endpoint(
endpoint, host, tls_cfg,
skip_tls_verify=skip_tls_verify,
def_namespace=default_namespace)
ctx.save()
return ctx
@classmethod
def get_context(cls, name=None):
"""Retrieves a context object.
Args:
name (str): The name of the context
Example:
>>> from docker.context import ContextAPI
>>> ctx = ContextAPI.get_context(name='test')
>>> print(ctx.Metadata)
{
"Name": "test",
"Metadata": {},
"Endpoints": {
"docker": {
"Host": "unix:///var/run/docker.sock",
"SkipTLSVerify": false
}
}
}
"""
if not name:
name = get_current_context_name()
if name == "default":
return cls.get_default_context()
return Context.load_context(name)
@classmethod
def contexts(cls):
"""Context list.
Returns:
(Context): List of context objects.
Raises:
:py:class:`docker.errors.APIError`
If something goes wrong.
"""
names = []
for dirname, dummy, fnames in os.walk(get_meta_dir()):
for filename in fnames:
if filename == METAFILE:
filepath = os.path.join(dirname, filename)
try:
with open(filepath, "r") as f:
data = json.load(f)
name = data["Name"]
if name == "default":
raise ValueError('"default" is a reserved context name')
names.append(name)
except Exception as e:
raise_from(errors.ContextException(
"Failed to load metafile {filepath}: {e}".format(filepath=filepath, e=e),
), e)
contexts = [cls.get_default_context()]
for name in names:
context = Context.load_context(name)
if not context:
raise errors.ContextException("Context {context} cannot be found".format(context=name))
contexts.append(context)
return contexts
@classmethod
def get_current_context(cls):
"""Get current context.
Returns:
(Context): current context object.
"""
return cls.get_context()
@classmethod
def set_current_context(cls, name="default"):
ctx = cls.get_context(name)
if not ctx:
raise errors.ContextNotFound(name)
err = write_context_name_to_docker_config(name)
if err:
raise errors.ContextException(
'Failed to set current context: {err}'.format(err=err))
@classmethod
def remove_context(cls, name):
"""Remove a context. Similar to the ``docker context rm`` command.
Args:
name (str): The name of the context
Raises:
:py:class:`docker.errors.MissingContextParameter`
If a context name is not provided.
:py:class:`docker.errors.ContextNotFound`
If a context with the name does not exist.
:py:class:`docker.errors.ContextException`
If name is default.
Example:
>>> from docker.context import ContextAPI
>>> ContextAPI.remove_context(name='test')
>>>
"""
if not name:
raise errors.MissingContextParameter("name")
if name == "default":
raise errors.ContextException(
'context "default" cannot be removed')
ctx = Context.load_context(name)
if not ctx:
raise errors.ContextNotFound(name)
if name == get_current_context_name():
write_context_name_to_docker_config(None)
ctx.remove()
@classmethod
def inspect_context(cls, name="default"):
"""Inspect a context. Similar to the ``docker context inspect`` command.
Args:
name (str): The name of the context
Raises:
:py:class:`docker.errors.MissingContextParameter`
If a context name is not provided.
:py:class:`docker.errors.ContextNotFound`
If a context with the name does not exist.
Example:
>>> from docker.context import ContextAPI
>>> ContextAPI.remove_context(name='test')
>>>
"""
if not name:
raise errors.MissingContextParameter("name")
if name == "default":
return cls.get_default_context()()
ctx = Context.load_context(name)
if not ctx:
raise errors.ContextNotFound(name)
return ctx()

103
plugins/module_utils/_api/context/config.py Normal file
View File

@ -0,0 +1,103 @@
# -*- coding: utf-8 -*-
# This code is part of the Ansible collection community.docker, but is an independent component.
# This particular file, and this file only, is based on the Docker SDK for Python (https://github.com/docker/docker-py/)
#
# Copyright (c) 2016-2025 Docker, Inc.
#
# 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)
__metaclass__ = type
import hashlib
import json
import os
from ..constants import DEFAULT_UNIX_SOCKET, IS_WINDOWS_PLATFORM
from ..utils.config import find_config_file, get_default_config_file
from ..utils.utils import parse_host
METAFILE = "meta.json"
def get_current_context_name_with_source():
if os.environ.get('DOCKER_HOST'):
return "default", "DOCKER_HOST environment variable set"
if os.environ.get('DOCKER_CONTEXT'):
return os.environ['DOCKER_CONTEXT'], "DOCKER_CONTEXT environment variable set"
docker_cfg_path = find_config_file()
if docker_cfg_path:
try:
with open(docker_cfg_path) as f:
return json.load(f).get("currentContext", "default"), "configuration file {file}".format(file=docker_cfg_path)
except Exception:
pass
return "default", "fallback value"
def get_current_context_name():
return get_current_context_name_with_source()[0]
def write_context_name_to_docker_config(name=None):
if name == 'default':
name = None
docker_cfg_path = find_config_file()
config = {}
if docker_cfg_path:
try:
with open(docker_cfg_path) as f:
config = json.load(f)
except Exception as e:
return e
current_context = config.get("currentContext", None)
if current_context and not name:
del config["currentContext"]
elif name:
config["currentContext"] = name
else:
return
if not docker_cfg_path:
docker_cfg_path = get_default_config_file()
try:
with open(docker_cfg_path, "w") as f:
json.dump(config, f, indent=4)
except Exception as e:
return e
def get_context_id(name):
return hashlib.sha256(name.encode('utf-8')).hexdigest()
def get_context_dir():
docker_cfg_path = find_config_file() or get_default_config_file()
return os.path.join(os.path.dirname(docker_cfg_path), "contexts")
def get_meta_dir(name=None):
meta_dir = os.path.join(get_context_dir(), "meta")
if name:
return os.path.join(meta_dir, get_context_id(name))
return meta_dir
def get_meta_file(name):
return os.path.join(get_meta_dir(name), METAFILE)
def get_tls_dir(name=None, endpoint=""):
context_dir = get_context_dir()
if name:
return os.path.join(context_dir, "tls", get_context_id(name), endpoint)
return os.path.join(context_dir, "tls")
def get_context_host(path=None, tls=False):
host = parse_host(path, IS_WINDOWS_PLATFORM, tls)
if host == DEFAULT_UNIX_SOCKET:
# remove http+ from default docker socket url
if host.startswith("http+"):
host = host[5:]
return host

268
plugins/module_utils/_api/context/context.py Normal file
View File

@ -0,0 +1,268 @@
# -*- coding: utf-8 -*-
# This code is part of the Ansible collection community.docker, but is an independent component.
# This particular file, and this file only, is based on the Docker SDK for Python (https://github.com/docker/docker-py/)
#
# Copyright (c) 2016-2025 Docker, Inc.
#
# 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)
__metaclass__ = type
import json
import os
from shutil import copyfile, rmtree
from ansible.module_utils.six import raise_from
from ..errors import ContextException
from ..tls import TLSConfig
from .config import (
get_context_host,
get_meta_dir,
get_meta_file,
get_tls_dir,
)
IN_MEMORY = "IN MEMORY"
class Context(object):
"""A context."""
def __init__(self, name, orchestrator=None, host=None, endpoints=None,
tls=False, description=None):
if not name:
raise Exception("Name not provided")
self.name = name
self.context_type = None
self.orchestrator = orchestrator
self.endpoints = {}
self.tls_cfg = {}
self.meta_path = IN_MEMORY
self.tls_path = IN_MEMORY
self.description = description
if not endpoints:
# set default docker endpoint if no endpoint is set
default_endpoint = "docker" if (
not orchestrator or orchestrator == "swarm"
) else orchestrator
self.endpoints = {
default_endpoint: {
"Host": get_context_host(host, tls),
"SkipTLSVerify": not tls
}
}
return
# check docker endpoints
for k, v in endpoints.items():
if not isinstance(v, dict):
# unknown format
raise ContextException(
"Unknown endpoint format for context {name}: {v}".format(name=name, v=v),
)
self.endpoints[k] = v
if k != "docker":
continue
self.endpoints[k]["Host"] = v.get("Host", get_context_host(
host, tls))
self.endpoints[k]["SkipTLSVerify"] = bool(v.get(
"SkipTLSVerify", not tls))
def set_endpoint(
self, name="docker", host=None, tls_cfg=None,
skip_tls_verify=False, def_namespace=None):
self.endpoints[name] = {
"Host": get_context_host(host, not skip_tls_verify),
"SkipTLSVerify": skip_tls_verify
}
if def_namespace:
self.endpoints[name]["DefaultNamespace"] = def_namespace
if tls_cfg:
self.tls_cfg[name] = tls_cfg
def inspect(self):
return self.__call__()
@classmethod
def load_context(cls, name):
meta = Context._load_meta(name)
if meta:
instance = cls(
meta["Name"],
orchestrator=meta["Metadata"].get("StackOrchestrator", None),
endpoints=meta.get("Endpoints", None),
description=meta.get('Description'))
instance.context_type = meta["Metadata"].get("Type", None)
instance._load_certs()
instance.meta_path = get_meta_dir(name)
return instance
return None
@classmethod
def _load_meta(cls, name):
meta_file = get_meta_file(name)
if not os.path.isfile(meta_file):
return None
metadata = {}
try:
with open(meta_file) as f:
metadata = json.load(f)
except (OSError, KeyError, ValueError) as e:
# unknown format
raise_from(Exception(
"Detected corrupted meta file for context {name} : {e}".format(name=name, e=e)
), e)
# for docker endpoints, set defaults for
# Host and SkipTLSVerify fields
for k, v in metadata["Endpoints"].items():
if k != "docker":
continue
metadata["Endpoints"][k]["Host"] = v.get(
"Host", get_context_host(None, False))
metadata["Endpoints"][k]["SkipTLSVerify"] = bool(
v.get("SkipTLSVerify", True))
return metadata
def _load_certs(self):
certs = {}
tls_dir = get_tls_dir(self.name)
for endpoint in self.endpoints.keys():
if not os.path.isdir(os.path.join(tls_dir, endpoint)):
continue
ca_cert = None
cert = None
key = None
for filename in os.listdir(os.path.join(tls_dir, endpoint)):
if filename.startswith("ca"):
ca_cert = os.path.join(tls_dir, endpoint, filename)
elif filename.startswith("cert"):
cert = os.path.join(tls_dir, endpoint, filename)
elif filename.startswith("key"):
key = os.path.join(tls_dir, endpoint, filename)
if all([ca_cert, cert, key]):
verify = None
if endpoint == "docker" and not self.endpoints["docker"].get(
"SkipTLSVerify", False):
verify = True
certs[endpoint] = TLSConfig(
client_cert=(cert, key), ca_cert=ca_cert, verify=verify)
self.tls_cfg = certs
self.tls_path = tls_dir
def save(self):
meta_dir = get_meta_dir(self.name)
if not os.path.isdir(meta_dir):
os.makedirs(meta_dir)
with open(get_meta_file(self.name), "w") as f:
f.write(json.dumps(self.Metadata))
tls_dir = get_tls_dir(self.name)
for endpoint, tls in self.tls_cfg.items():
if not os.path.isdir(os.path.join(tls_dir, endpoint)):
os.makedirs(os.path.join(tls_dir, endpoint))
ca_file = tls.ca_cert
if ca_file:
copyfile(ca_file, os.path.join(
tls_dir, endpoint, os.path.basename(ca_file)))
if tls.cert:
cert_file, key_file = tls.cert
copyfile(cert_file, os.path.join(
tls_dir, endpoint, os.path.basename(cert_file)))
copyfile(key_file, os.path.join(
tls_dir, endpoint, os.path.basename(key_file)))
self.meta_path = get_meta_dir(self.name)
self.tls_path = get_tls_dir(self.name)
def remove(self):
if os.path.isdir(self.meta_path):
rmtree(self.meta_path)
if os.path.isdir(self.tls_path):
rmtree(self.tls_path)
def __repr__(self):
return "<{classname}: '{name}'>".format(classname=self.__class__.__name__, name=self.name)
def __str__(self):
return json.dumps(self.__call__(), indent=2)
def __call__(self):
result = self.Metadata
result.update(self.TLSMaterial)
result.update(self.Storage)
return result
def is_docker_host(self):
return self.context_type is None
@property
def Name(self):
return self.name
@property
def Host(self):
if not self.orchestrator or self.orchestrator == "swarm":
endpoint = self.endpoints.get("docker", None)
if endpoint:
return endpoint.get("Host", None)
return None
return self.endpoints[self.orchestrator].get("Host", None)
@property
def Orchestrator(self):
return self.orchestrator
@property
def Metadata(self):
meta = {}
if self.orchestrator:
meta = {"StackOrchestrator": self.orchestrator}
return {
"Name": self.name,
"Metadata": meta,
"Endpoints": self.endpoints
}
@property
def TLSConfig(self):
key = self.orchestrator
if not key or key == "swarm":
key = "docker"
if key in self.tls_cfg.keys():
return self.tls_cfg[key]
return None
@property
def TLSMaterial(self):
certs = {}
for endpoint, tls in self.tls_cfg.items():
cert, key = tls.cert
certs[endpoint] = list(
map(os.path.basename, [tls.ca_cert, cert, key]))
return {
"TLSMaterial": certs
}
@property
def Storage(self):
return {
"Storage": {
"MetadataPath": self.meta_path,
"TLSPath": self.tls_path
}}

9
plugins/module_utils/_api/utils/config.py
View File

@ -22,12 +22,17 @@ LEGACY_DOCKER_CONFIG_FILENAME = '.dockercfg'
log = logging.getLogger(__name__)
def get_default_config_file():
return os.path.join(home_dir(), DOCKER_CONFIG_FILENAME)
def find_config_file(config_path=None):
homedir = home_dir()
paths = list(filter(None, [
config_path, # 1
config_path_from_environment(), # 2
os.path.join(home_dir(), DOCKER_CONFIG_FILENAME), # 3
os.path.join(home_dir(), LEGACY_DOCKER_CONFIG_FILENAME), # 4
os.path.join(homedir, DOCKER_CONFIG_FILENAME), # 3
os.path.join(homedir, LEGACY_DOCKER_CONFIG_FILENAME), # 4
]))
log.debug("Trying paths: %s", repr(paths))

7
plugins/module_utils/_api/utils/utils.py
View File

@ -303,12 +303,7 @@ def parse_host(addr, is_win32=False, tls=False):
if proto in ('tcp', 'ssh'):
port = parsed_url.port or 0
if port <= 0:
if proto != 'ssh':
raise errors.DockerException(
'Invalid bind address format: port is required:'
' {0}'.format(addr)
)
port = 22
port = 22 if proto == 'ssh' else (2375 if tls else 2376)
netloc = '{0}:{1}'.format(parsed_url.netloc, port)
if not parsed_url.hostname:

2
plugins/module_utils/common.py
View File

@ -321,7 +321,7 @@ class AnsibleDockerClientBase(Client):
cert_path=self._get_value('cert_path', params['client_cert'], 'DOCKER_CERT_PATH', None, type='str'),
key_path=self._get_value('key_path', params['client_key'], 'DOCKER_CERT_PATH', None, type='str'),
tls=self._get_value('tls', params['tls'], 'DOCKER_TLS', DEFAULT_TLS, type='bool'),
tls_verify=self._get_value('tls_verfy', params['validate_certs'], 'DOCKER_TLS_VERIFY',
tls_verify=self._get_value('validate_certs', params['validate_certs'], 'DOCKER_TLS_VERIFY',
DEFAULT_TLS_VERIFY, type='bool'),
timeout=self._get_value('timeout', params['timeout'], 'DOCKER_TIMEOUT',
DEFAULT_TIMEOUT_SECONDS, type='int'),

2
plugins/module_utils/common_api.py
View File

@ -204,7 +204,7 @@ class AnsibleDockerClientBase(Client):
cert_path=self._get_value('cert_path', params['client_cert'], 'DOCKER_CERT_PATH', None, type='str'),
key_path=self._get_value('key_path', params['client_key'], 'DOCKER_CERT_PATH', None, type='str'),
tls=self._get_value('tls', params['tls'], 'DOCKER_TLS', DEFAULT_TLS, type='bool'),
tls_verify=self._get_value('tls_verfy', params['validate_certs'], 'DOCKER_TLS_VERIFY',
tls_verify=self._get_value('validate_certs', params['validate_certs'], 'DOCKER_TLS_VERIFY',
DEFAULT_TLS_VERIFY, type='bool'),
timeout=self._get_value('timeout', params['timeout'], 'DOCKER_TIMEOUT',
DEFAULT_TIMEOUT_SECONDS, type='int'),

305
plugins/modules/docker_context_info.py Normal file
View File

@ -0,0 +1,305 @@
#!/usr/bin/python
#
# Copyright 2025 Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = r"""
module: docker_context_info
short_description: Retrieve information on Docker contexts for the current user
description:
- Return information on Docker contexts.
- This includes some generic information, as well as a RV(contexts[].config) dictionary that can be used for module defaults for all community.docker modules
that use the C(community.docker.docker) module defaults group.
extends_documentation_fragment:
- community.docker.attributes
- community.docker.attributes.info_module
- community.docker.attributes.idempotent_not_modify_state
options:
only_current:
description:
- If set to V(true), RV(contexts) will just contain the current context and none else.
- If set to V(false) (default), RV(contexts) will list all contexts, unless O(name) is specified.
- Mutually exclusive to O(name).
type: bool
default: false
name:
description:
- A specific Docker CLI context to query.
- The module will fail if this context does not exist. If you simply want to query whether a context exists,
do not specify this parameter and use Jinja2 to search the resulting list for a context of the given name instead.
- Mutually exclusive with O(only_current).
type: str
cli_context:
description:
- Override for the default context's name.
- This is preferably used for context selection when O(only_current=true),
and it is used to compute the return values RV(contexts[].current) and RV(current_context_name).
type: str
author:
- "Felix Fontein (@felixfontein)"
"""
EXAMPLES = r"""
- name: Get infos on contexts
community.docker.docker_context_info:
register: result
- name: Show all contexts
ansible.builtin.debug:
msg: "{{ result.contexts }}"
- name: Get current context
community.docker.docker_context_info:
only_current: true
register: docker_current_context
- name: Run community.docker modules with current context
module_defaults:
group/community.docker.docker: "{{ docker_current_context.contexts[0].config }}"
block:
- name: Task using the current context
community.docker.docker_container:
image: ubuntu:latest
name: ubuntu
state: started
"""
RETURN = r"""
contexts:
description:
- A list of all contexts (O(only_current=false), O(name) not specified),
only the current context (O(only_current=true)),
or the requested context (O(name) specified).
type: list
elements: dict
returned: success
contains:
current:
description:
- Whether this context is the current one.
type: bool
returned: success
sample: true
name:
description:
- The context's name.
type: bool
returned: success
sample: default
description:
description:
- The context's description, if available.
type: bool
returned: success
sample: My context
meta_path:
description:
- The path to the context's meta directory.
- Not present for RV(contexts[].name=default).
type: str
returned: success
sample: /home/felix/.docker/contexts/meta/0123456789abcdef01234567890abcdef0123456789abcdef0123456789abcde
tls_path:
description:
- The path to the context's TLS config directory.
- Not present for RV(contexts[].name=default).
type: str
returned: success
sample: /home/user/.docker/contexts/tls/0123456789abcdef01234567890abcdef0123456789abcdef0123456789abcde/
config:
description:
- In case the context is for Docker, contains option values to configure the community.docker modules to use this context.
- Note that the exact values returned here and their values might change over time if incompatibilities to existing modules are found.
The goal is that this configuration works fine with all modules in this collection, but we do not have the capabilities to
test all possible configuration options at the moment.
type: dict
returned: success
sample: {}
contains:
docker_host:
description:
- The Docker daemon to connect to.
type: str
returned: success and context is for Docker
sample: unix:///var/run/docker.sock
tls:
description:
- Whether the Docker context should use an unvalidated TLS connection.
type: bool
returned: success and context is for Docker
sample: false
ca_path:
description:
- The CA certificate used to validate the Docker daemon's certificate.
type: bool
returned: success, context is for Docker, TLS config is present, and CA cert is present
sample: /path/to/ca-cert.pem
client_cert:
description:
- The client certificate to authenticate with to the Docker daemon.
type: bool
returned: success, context is for Docker, TLS config is present, and client cert info is present
sample: /path/to/client-cert.pem
client_key:
description:
- The client certificate's key to authenticate with to the Docker daemon.
type: bool
returned: success, context is for Docker, TLS config is present, and client cert info is present
sample: /path/to/client-key.pem
validate_certs:
description:
- Whether the Docker context should use a validated TLS connection.
type: bool
returned: success, context is for Docker, and TLS config is present
sample: true
current_context_name:
description:
- The name of the current Docker context.
type: str
returned: success
sample: default
"""
import traceback
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.text.converters import to_native, to_text
from ansible.module_utils.six import string_types
from ansible_collections.community.docker.plugins.module_utils._api.context.api import (
ContextAPI,
)
from ansible_collections.community.docker.plugins.module_utils._api.context.config import (
get_current_context_name_with_source,
)
from ansible_collections.community.docker.plugins.module_utils._api.context.context import (
IN_MEMORY,
)
from ansible_collections.community.docker.plugins.module_utils._api.errors import (
ContextException,
DockerException,
)
def tls_context_to_json(context):
if context is None:
return None
return {
'client_cert': context.cert[0] if context.cert else None,
'client_key': context.cert[1] if context.cert else None,
'ca_cert': context.ca_cert,
'verify': context.verify,
# 'ssl_version': context.ssl_version, -- this isn't used anymore
}
def to_bool(value):
return True if value else False
def context_to_json(context, current):
module_config = {}
if 'docker' in context.endpoints:
endpoint = context.endpoints['docker']
if isinstance(endpoint.get('Host'), string_types):
host_str = to_text(endpoint['Host'])
# Adjust protocol name so that it works with the Docker CLI tool as well
proto = None
idx = host_str.find('://')
if idx >= 0:
proto = host_str[:idx]
host_str = host_str[idx + 3:]
if proto in ('http', 'https'):
proto = 'tcp'
if proto == 'http+unix':
proto = 'unix'
if proto:
host_str = "{0}://{1}".format(proto, host_str)
# Create config for the modules
module_config['docker_host'] = host_str
module_config['tls'] = not to_bool(endpoint.get('SkipTLSVerify'))
if context.tls_cfg.get('docker'):
tls_cfg = context.tls_cfg['docker']
if tls_cfg.ca_cert:
module_config['ca_path'] = tls_cfg.ca_cert
if tls_cfg.cert:
module_config['client_cert'] = tls_cfg.cert[0]
module_config['client_key'] = tls_cfg.cert[1]
module_config['validate_certs'] = tls_cfg.verify
module_config['tls'] = to_bool(tls_cfg.verify)
return {
'current': current,
'name': context.name,
'description': context.description,
'meta_path': None if context.meta_path is IN_MEMORY else context.meta_path,
'tls_path': None if context.tls_path is IN_MEMORY else context.tls_path,
'config': module_config,
}
def main():
argument_spec = dict(
only_current=dict(type='bool', default=False),
name=dict(type='str'),
cli_context=dict(type='str'),
)
module = AnsibleModule(
argument_spec=argument_spec,
supports_check_mode=True,
mutually_exclusive=[
("only_current", "name"),
],
)
try:
if module.params['cli_context']:
current_context_name, current_context_source = module.params['cli_context'], "cli_context module option"
else:
current_context_name, current_context_source = get_current_context_name_with_source()
if module.params['name']:
contexts = [ContextAPI.get_context(module.params['name'])]
if not contexts[0]:
module.fail_json(msg="There is no context of name {name!r}".format(name=module.params['name']))
elif module.params['only_current']:
contexts = [ContextAPI.get_context(current_context_name)]
if not contexts[0]:
module.fail_json(
msg="There is no context of name {name!r}, which is configured as the default context ({source})".format(
name=current_context_name,
source=current_context_source,
),
)
else:
contexts = ContextAPI.contexts()
json_contexts = [
context_to_json(context, context.name == current_context_name)
for context in contexts
]
module.exit_json(
changed=False,
contexts=json_contexts,
current_context_name=current_context_name,
)
except ContextException as e:
module.fail_json(msg='Error when handling Docker contexts: {0}'.format(to_native(e)), exception=traceback.format_exc())
except DockerException as e:
module.fail_json(msg='An unexpected Docker error occurred: {0}'.format(to_native(e)), exception=traceback.format_exc())
if __name__ == '__main__':
main()

6
tests/integration/targets/docker_context_info/aliases Normal file
View File

@ -0,0 +1,6 @@
# Copyright (c) Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
azp/5
destructive

8
tests/integration/targets/docker_context_info/meta/main.yml Normal file
View File

@ -0,0 +1,8 @@
---
# Copyright (c) Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
dependencies:
- setup_docker
- setup_docker_python_deps

40
tests/integration/targets/docker_context_info/tasks/main.yml Normal file
View File

@ -0,0 +1,40 @@
---
# Copyright (c) Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
####################################################################
# WARNING: These are designed specifically for Ansible tests #
# and should not be used as examples of how to write Ansible roles #
####################################################################
# Create random name prefix (for containers, networks, ...)
- name: Create random container name prefix
set_fact:
cname_prefix: "{{ 'ansible-docker-test-%0x' % ((2**32) | random) }}"
cnames: []
- debug:
msg: "Using container name prefix {{ cname_prefix }}"
# Run the tests
- block:
- include_tasks: run-test.yml
with_fileglob:
- "tests/*.yml"
loop_control:
loop_var: test_name
always:
- name: "Make sure all containers are removed"
docker_container:
name: "{{ item }}"
state: absent
force_kill: true
with_items: "{{ cnames }}"
diff: false
when: docker_api_version is version('1.25', '>=')
- fail: msg="Too old docker / docker-py version to run all docker_container tests!"
when: not(docker_api_version is version('1.25', '>=')) and (ansible_distribution != 'CentOS' or ansible_distribution_major_version|int > 6)

7
tests/integration/targets/docker_context_info/tasks/run-test.yml Normal file
View File

@ -0,0 +1,7 @@
---
# Copyright (c) Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
- name: "Loading tasks from {{ test_name }}"
include_tasks: "{{ test_name }}"

121
tests/integration/targets/docker_context_info/tasks/tests/default-context.yml Normal file
View File

@ -0,0 +1,121 @@
---
# Copyright (c) Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
- name: Registering container name
set_fact:
cname: "{{ cname_prefix ~ '-hi' }}"
- name: Registering container name
set_fact:
cnames: "{{ cnames + [cname] }}"
- name: Get current context
community.docker.docker_context_info:
only_current: true
register: docker_current_context
- assert:
that:
- docker_current_context is not changed
# Some of the following tests will not be true on all machines, but they should be in CI:
- docker_current_context.current_context_name == 'default'
- docker_current_context.contexts | length == 1
- docker_current_context.contexts[0].name == 'default'
- docker_current_context.contexts[0].current == true
- docker_current_context.contexts[0].description == 'Current DOCKER_HOST based configuration'
- docker_current_context.contexts[0].meta_path is none
- docker_current_context.contexts[0].tls_path is none
- docker_current_context.contexts[0].config.docker_host == 'unix:///var/run/docker.sock'
- docker_current_context.contexts[0].config.tls == false
- name: Run community.docker modules with current context
module_defaults:
group/community.docker.docker: "{{ docker_current_context.contexts[0].config }}"
block:
- name: Create container
docker_container:
image: "{{ docker_test_image_alpine }}"
command: '/bin/sh -c "sleep 10m"'
name: "{{ cname }}"
state: present
register: create_1
- name: Create container (idempotent)
docker_container:
image: "{{ docker_test_image_alpine }}"
command: '/bin/sh -c "sleep 10m"'
name: "{{ cname }}"
state: present
register: create_2
- assert:
that:
- create_1 is changed
- create_2 is not changed
- name: Inspect container with CLI tool
ansible.builtin.command:
cmd: docker inspect {{ cname }}
register: result
- assert:
that:
- (result.stdout | from_json) | length == 1
- (result.stdout | from_json)[0].State.Status == "created"
- name: Start container
docker_container:
name: "{{ cname }}"
state: started
register: start_1
- name: Start container (idempotent)
docker_container:
name: "{{ cname }}"
state: started
register: start_2
- assert:
that:
- start_1 is changed
- start_2 is not changed
- name: Inspect container with CLI tool
ansible.builtin.command:
cmd: docker inspect {{ cname }}
register: result
- assert:
that:
- (result.stdout | from_json) | length == 1
- (result.stdout | from_json)[0].State.Status == "running"
- name: Remove container
docker_container:
name: "{{ cname }}"
state: absent
force_kill: true
register: remove_1
- name: Remove container (idempotent)
docker_container:
name: "{{ cname }}"
state: absent
force_kill: true
register: remove_2
- assert:
that:
- remove_1 is changed
- remove_2 is not changed
- name: Inspect container with CLI tool
ansible.builtin.command:
cmd: docker inspect {{ cname }}
register: result
failed_when: result.rc != 1
- assert:
that:
- (result.stdout | from_json) | length == 0

1
tests/sanity/extra/action-group.py
View File

@ -20,6 +20,7 @@ ACTION_GROUPS = {
'pattern': re.compile('^.*$'),
'exclusions': [
'current_container_facts',
'docker_context_info',
],
'doc_fragment': 'community.docker.attributes.actiongroup_docker',
},

68
tests/unit/plugins/module_utils/_api/test_context.py Normal file
View File

@ -0,0 +1,68 @@
# -*- coding: utf-8 -*-
# This code is part of the Ansible collection community.docker, but is an independent component.
# This particular file, and this file only, is based on the Docker SDK for Python (https://github.com/docker/docker-py/)
#
# Copyright (c) 2016-2025 Docker, Inc.
#
# 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)
__metaclass__ = type
import unittest
import pytest
from ansible_collections.community.docker.plugins.module_utils._api import errors
from ansible_collections.community.docker.plugins.module_utils._api.constants import (
DEFAULT_NPIPE,
DEFAULT_UNIX_SOCKET,
IS_WINDOWS_PLATFORM,
)
from ansible_collections.community.docker.plugins.module_utils._api.context.api import ContextAPI
from ansible_collections.community.docker.plugins.module_utils._api.context.context import Context
class BaseContextTest(unittest.TestCase):
@pytest.mark.skipif(
IS_WINDOWS_PLATFORM, reason='Linux specific path check'
)
def test_url_compatibility_on_linux(self):
c = Context("test")
assert c.Host == DEFAULT_UNIX_SOCKET[5:]
@pytest.mark.skipif(
not IS_WINDOWS_PLATFORM, reason='Windows specific path check'
)
def test_url_compatibility_on_windows(self):
c = Context("test")
assert c.Host == DEFAULT_NPIPE
def test_fail_on_default_context_create(self):
with pytest.raises(errors.ContextException):
ContextAPI.create_context("default")
def test_default_in_context_list(self):
found = False
ctx = ContextAPI.contexts()
for c in ctx:
if c.Name == "default":
found = True
assert found is True
def test_get_current_context(self):
assert ContextAPI.get_current_context().Name == "default"
def test_https_host(self):
c = Context("test", host="tcp://testdomain:8080", tls=True)
assert c.Host == "https://testdomain:8080"
def test_context_inspect_without_params(self):
ctx = ContextAPI.inspect_context()
assert ctx["Name"] == "default"
assert ctx["Metadata"]["StackOrchestrator"] == "swarm"
assert ctx["Endpoints"]["docker"]["Host"] in (
DEFAULT_NPIPE,
DEFAULT_UNIX_SOCKET[5:],
)

2
tests/unit/plugins/module_utils/_api/utils/test_utils.py
View File

@ -255,7 +255,7 @@ class ParseEnvFileTest(unittest.TestCase):
class ParseHostTest(unittest.TestCase):
def test_parse_host(self):
invalid_hosts = [
'0.0.0.0',
'foo://0.0.0.0',
'tcp://',
'udp://127.0.0.1',
'udp://127.0.0.1:2375',