first version of my ansible-modle module
this helps with the development of new modules by abstracting some things away
Dieser Commit ist enthalten in:
Commit
511f4e615f
|
|
@ -0,0 +1,18 @@
|
|||
# EditorConfig is awesome: https://EditorConfig.org
|
||||
|
||||
# top-most EditorConfig file
|
||||
root = true
|
||||
|
||||
[*]
|
||||
indent_style = space
|
||||
indent_size = 4
|
||||
tab_width = 4
|
||||
end_of_line = lf
|
||||
charset = utf-8
|
||||
trim_trailing_whitespace = true
|
||||
insert_final_newline = true
|
||||
|
||||
[*.{yaml,yml}]
|
||||
indent_size = 2
|
||||
[Makefile]
|
||||
indent_style = tab
|
||||
|
|
@ -0,0 +1,166 @@
|
|||
# ---> Python
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
|
||||
# C extensions
|
||||
*.so
|
||||
|
||||
# Distribution / packaging
|
||||
.Python
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
downloads/
|
||||
eggs/
|
||||
.eggs/
|
||||
lib/
|
||||
lib64/
|
||||
parts/
|
||||
sdist/
|
||||
var/
|
||||
wheels/
|
||||
share/python-wheels/
|
||||
*.egg-info/
|
||||
.installed.cfg
|
||||
*.egg
|
||||
MANIFEST
|
||||
|
||||
# PyInstaller
|
||||
# Usually these files are written by a python script from a template
|
||||
# before PyInstaller builds the exe, so as to inject date/other infos into it.
|
||||
*.manifest
|
||||
*.spec
|
||||
|
||||
# Installer logs
|
||||
pip-log.txt
|
||||
pip-delete-this-directory.txt
|
||||
|
||||
# Unit test / coverage reports
|
||||
htmlcov/
|
||||
.tox/
|
||||
.nox/
|
||||
.coverage
|
||||
.coverage.*
|
||||
coverage.*
|
||||
.cache
|
||||
nosetests.xml
|
||||
coverage.xml
|
||||
*.cover
|
||||
*.py,cover
|
||||
.hypothesis/
|
||||
.pytest_cache/
|
||||
cover/
|
||||
|
||||
# Translations
|
||||
*.mo
|
||||
*.pot
|
||||
|
||||
# Django stuff:
|
||||
*.log
|
||||
local_settings.py
|
||||
db.sqlite3
|
||||
db.sqlite3-journal
|
||||
|
||||
# Flask stuff:
|
||||
instance/
|
||||
.webassets-cache
|
||||
|
||||
# Scrapy stuff:
|
||||
.scrapy
|
||||
|
||||
# Sphinx documentation
|
||||
docs/_build/
|
||||
|
||||
# PyBuilder
|
||||
.pybuilder/
|
||||
target/
|
||||
|
||||
# Jupyter Notebook
|
||||
.ipynb_checkpoints
|
||||
|
||||
# IPython
|
||||
profile_default/
|
||||
ipython_config.py
|
||||
|
||||
# pyenv
|
||||
# For a library or package, you might want to ignore these files since the code is
|
||||
# intended to run in multiple environments; otherwise, check them in:
|
||||
# .python-version
|
||||
|
||||
# pipenv
|
||||
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
|
||||
# However, in case of collaboration, if having platform-specific dependencies or dependencies
|
||||
# having no cross-platform support, pipenv may install dependencies that don't work, or not
|
||||
# install all needed dependencies.
|
||||
#Pipfile.lock
|
||||
|
||||
# poetry
|
||||
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
|
||||
# This is especially recommended for binary packages to ensure reproducibility, and is more
|
||||
# commonly ignored for libraries.
|
||||
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
|
||||
#poetry.lock
|
||||
|
||||
# pdm
|
||||
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
|
||||
#pdm.lock
|
||||
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
|
||||
# in version control.
|
||||
# https://pdm.fming.dev/#use-with-ide
|
||||
.pdm.toml
|
||||
|
||||
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
|
||||
__pypackages__/
|
||||
|
||||
# Celery stuff
|
||||
celerybeat-schedule
|
||||
celerybeat.pid
|
||||
|
||||
# SageMath parsed files
|
||||
*.sage.py
|
||||
|
||||
# Environments
|
||||
.env
|
||||
.venv
|
||||
env/
|
||||
venv/
|
||||
ENV/
|
||||
env.bak/
|
||||
venv.bak/
|
||||
|
||||
# Spyder project settings
|
||||
.spyderproject
|
||||
.spyproject
|
||||
|
||||
# Rope project settings
|
||||
.ropeproject
|
||||
|
||||
# mkdocs documentation
|
||||
/site
|
||||
|
||||
# mypy
|
||||
.mypy_cache/
|
||||
.dmypy.json
|
||||
dmypy.json
|
||||
|
||||
# Pyre type checker
|
||||
.pyre/
|
||||
|
||||
# pytype static type analyzer
|
||||
.pytype/
|
||||
|
||||
# Cython debug symbols
|
||||
cython_debug/
|
||||
|
||||
# PyCharm
|
||||
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
|
||||
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
|
||||
# and can be added to the global gitignore or merged into this file. For a more nuclear
|
||||
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
|
||||
#.idea/
|
||||
|
||||
# ---> Ansible
|
||||
*.retry
|
||||
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
MIT License
|
||||
|
||||
Copyright (c) 2023-present Sebastian Tobie <sebastian@sebastian-tobie.de>
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
# Ansible module
|
||||
|
||||
[](https://pypi.org/project/ansible-module)
|
||||
[](https://pypi.org/project/ansible-module)
|
||||
|
||||
-----
|
||||
|
||||
**Table of Contents**
|
||||
|
||||
- [Installation](#installation)
|
||||
- [License](#license)
|
||||
|
||||
## Installation
|
||||
|
||||
```console
|
||||
pip install ansible-module
|
||||
```
|
||||
|
||||
## License
|
||||
|
||||
`ansible-module` is distributed under the terms of the [MIT](https://spdx.org/licenses/MIT.html) license.
|
||||
|
|
@ -0,0 +1,154 @@
|
|||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "ansible-module"
|
||||
dynamic = ["version"]
|
||||
description = 'Helps with developing modules for ansible in an easier manner'
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.7"
|
||||
license = "MIT"
|
||||
keywords = []
|
||||
authors = [{ name = "Sebastian Tobie", email = "sebastian@sebastian-tobie.de" }]
|
||||
classifiers = [
|
||||
"Development Status :: 4 - Beta",
|
||||
"Programming Language :: Python",
|
||||
"Programming Language :: Python :: 3.7",
|
||||
"Programming Language :: Python :: 3.8",
|
||||
"Programming Language :: Python :: 3.9",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: Implementation :: CPython",
|
||||
"Programming Language :: Python :: Implementation :: PyPy",
|
||||
]
|
||||
dependencies = ["ansible-core>=2.9.10"]
|
||||
|
||||
[project.urls]
|
||||
Documentation = "https://github.com/unknown/ansible-module#readme"
|
||||
Issues = "https://github.com/unknown/ansible-module/issues"
|
||||
Source = "https://github.com/unknown/ansible-module"
|
||||
|
||||
[tool.hatch.version]
|
||||
path = "src/ansible_module/__init__.py"
|
||||
|
||||
[tool.hatch.envs.default]
|
||||
dependencies = ["coverage[toml]>=6.5", "pytest"]
|
||||
[tool.hatch.envs.default.scripts]
|
||||
test = "pytest {args:tests}"
|
||||
test-cov = "coverage run -m pytest {args:tests}"
|
||||
cov-report = ["- coverage combine", "coverage report"]
|
||||
cov = ["test-cov", "cov-report"]
|
||||
|
||||
[[tool.hatch.envs.all.matrix]]
|
||||
python = ["3.7", "3.8", "3.9", "3.10", "3.11"]
|
||||
|
||||
[tool.hatch.envs.lint]
|
||||
detached = true
|
||||
dependencies = ["black>=23.1.0", "mypy>=1.0.0", "ruff>=0.0.243"]
|
||||
[tool.hatch.envs.lint.scripts]
|
||||
typing = "mypy --install-types --non-interactive {args:src/ansible_module tests}"
|
||||
style = ["ruff {args:.}", "black --check --diff {args:.}"]
|
||||
fmt = ["black {args:.}", "ruff --fix {args:.}", "style"]
|
||||
all = ["style", "typing"]
|
||||
|
||||
[tool.black]
|
||||
target-version = ["py37"]
|
||||
line-length = 120
|
||||
skip-string-normalization = true
|
||||
|
||||
[tool.ruff]
|
||||
target-version = "py37"
|
||||
line-length = 120
|
||||
select = [
|
||||
"A",
|
||||
"ARG",
|
||||
"B",
|
||||
"C",
|
||||
"DTZ",
|
||||
"E",
|
||||
"EM",
|
||||
"F",
|
||||
"FBT",
|
||||
"I",
|
||||
"ICN",
|
||||
"ISC",
|
||||
"N",
|
||||
"PLC",
|
||||
"PLE",
|
||||
"PLR",
|
||||
"PLW",
|
||||
"Q",
|
||||
"RUF",
|
||||
"S",
|
||||
"T",
|
||||
"TID",
|
||||
"UP",
|
||||
"W",
|
||||
"YTT",
|
||||
]
|
||||
ignore = [
|
||||
# Allow non-abstract empty methods in abstract base classes
|
||||
"B027",
|
||||
# Allow boolean positional values in function calls, like `dict.get(... True)`
|
||||
"FBT003",
|
||||
# Ignore checks for possible passwords
|
||||
"S105",
|
||||
"S106",
|
||||
"S107",
|
||||
# Ignore complexity
|
||||
"C901",
|
||||
"PLR0911",
|
||||
"PLR0912",
|
||||
"PLR0913",
|
||||
"PLR0915",
|
||||
]
|
||||
unfixable = [
|
||||
# Don't touch unused imports
|
||||
"F401",
|
||||
]
|
||||
|
||||
[tool.ruff.isort]
|
||||
known-first-party = ["ansible_module"]
|
||||
|
||||
[tool.ruff.flake8-tidy-imports]
|
||||
ban-relative-imports = "all"
|
||||
|
||||
[tool.ruff.per-file-ignores]
|
||||
# Tests can use magic values, assertions, and relative imports
|
||||
"tests/**/*" = ["PLR2004", "S101", "TID252"]
|
||||
|
||||
[tool.coverage.run]
|
||||
source_pkgs = ["ansible_module", "tests"]
|
||||
branch = true
|
||||
parallel = true
|
||||
omit = ["src/ansible_module/__about__.py"]
|
||||
|
||||
[tool.coverage.paths]
|
||||
ansible_module = ["src/ansible_module", "*/ansible-module/src/ansible_module"]
|
||||
tests = ["tests", "*/ansible-module/tests"]
|
||||
|
||||
[tool.coverage.report]
|
||||
exclude_lines = ["no cov", "if __name__ == .__main__.:", "if TYPE_CHECKING:", "pragma: nocover"]
|
||||
|
||||
[tool.mypy]
|
||||
python_version = "3.11"
|
||||
warn_return_any = true
|
||||
warn_unused_configs = true
|
||||
|
||||
[[tool.mypy.overrides]]
|
||||
module = ["ansible", "ansible.module_utils", "ansible.module_utils.basic"]
|
||||
ignore_missing_imports = true
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
addopts = [
|
||||
"--cov-report", "json",
|
||||
"--cov-report", "term-missing:skip-covered",
|
||||
"--cov",
|
||||
]
|
||||
pythonpath = ["src"]
|
||||
required_plugins = [
|
||||
"pytest-cov",
|
||||
"pytest-isort",
|
||||
"pytest-mypy"
|
||||
]
|
||||
|
|
@ -0,0 +1 @@
|
|||
__version__ = "0.1.0"
|
||||
|
|
@ -0,0 +1,135 @@
|
|||
import pathlib
|
||||
import warnings
|
||||
from typing import Any, Callable, Dict, Optional, Sequence, Tuple, Type, Union
|
||||
|
||||
__all__ = (
|
||||
"Types",
|
||||
"SYSTEMD_SERVICE_CONFIG",
|
||||
"SYSTEMD_NETWORK_CONFIG",
|
||||
"SYSTEMD_CONFIG_ROOT",
|
||||
"systemdbool",
|
||||
"AnsibleParameter",
|
||||
)
|
||||
|
||||
|
||||
SYSTEMD_CONFIG_ROOT = pathlib.Path("/etc/systemd")
|
||||
SYSTEMD_NETWORK_CONFIG = SYSTEMD_CONFIG_ROOT / "network"
|
||||
SYSTEMD_SERVICE_CONFIG = SYSTEMD_CONFIG_ROOT / "system"
|
||||
|
||||
AnsibleParameter = Dict[str, Any]
|
||||
|
||||
|
||||
class _Type:
|
||||
def __dir__(self) -> tuple: # pragma: nocover
|
||||
return (
|
||||
"str",
|
||||
"bool",
|
||||
"int",
|
||||
"float",
|
||||
"path",
|
||||
"raw",
|
||||
"jsonarg",
|
||||
"json",
|
||||
"bytes",
|
||||
"dict",
|
||||
"list",
|
||||
"bits",
|
||||
"__doc__",
|
||||
)
|
||||
|
||||
def __repr__(self) -> str: # pragma: nocover
|
||||
return "Types()"
|
||||
|
||||
def __call__(self) -> "_Type": # pragma: nocover
|
||||
return self
|
||||
|
||||
def list(
|
||||
self,
|
||||
elements: Union[Type[object], str, AnsibleParameter],
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
) -> AnsibleParameter:
|
||||
option: AnsibleParameter = dict(type="list", required=required)
|
||||
if not isinstance(elements, (str, dict)):
|
||||
option["elements"] = elements.__name__
|
||||
elif isinstance(elements, dict):
|
||||
option["elements"] = elements["type"]
|
||||
if elements["type"] == "dict":
|
||||
option["options"] = dict()
|
||||
for name, value in elements["option"].items():
|
||||
option["options"][name] = value
|
||||
if "description" not in option["options"][name]:
|
||||
warnings.warn( # pragma: nocover
|
||||
f"helptext of option {name} is unset."
|
||||
" Ansible requires suboptions to have an documentation"
|
||||
)
|
||||
if help is not None:
|
||||
option["description"] = help.split("\n")
|
||||
return option
|
||||
|
||||
def dict(self, required: bool = False, help: Optional[str] = None, **options: dict) -> AnsibleParameter:
|
||||
option: AnsibleParameter = dict(type="dict", required=required)
|
||||
option["option"] = options
|
||||
if help is not None:
|
||||
option["description"] = help.split("\n")
|
||||
return option
|
||||
|
||||
def __getattr__(self, name: str):
|
||||
def argument(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter:
|
||||
option: AnsibleParameter = dict(type=name, required=required)
|
||||
if choices is not None:
|
||||
option["choices"] = choices
|
||||
if default is not None:
|
||||
option["default"] = default
|
||||
if help is not None:
|
||||
option["description"] = help.split("\n")
|
||||
return option
|
||||
|
||||
argument.__name__ = name
|
||||
argument.__qualname__ = f"Types.{name}"
|
||||
argument.__doc__ = f"Simple wrapper for Ansible {name} argument dict"
|
||||
return argument
|
||||
|
||||
|
||||
Types = _Type()
|
||||
|
||||
|
||||
def systemdbool(b: Union[bool, str]) -> str:
|
||||
"""Converts values into things systemd can parse"""
|
||||
if b is True:
|
||||
return "yes"
|
||||
elif b is False:
|
||||
return "no"
|
||||
return b
|
||||
|
||||
|
||||
def modspec(
|
||||
argument_spec: Dict[str, Dict[str, Any]],
|
||||
mutually_exclusive: Sequence[Tuple[str, ...]] = (),
|
||||
required_together: Sequence[Tuple[str, ...]] = (),
|
||||
required_one_of: Sequence[Tuple[str, ...]] = (),
|
||||
required_if: Sequence[Union[Tuple[str, Any, Tuple[str, ...]], Tuple[str, Any, Tuple[str, ...], bool]]] = (),
|
||||
required_by: Dict[str, Union[str, Tuple[str, ...]]] = {},
|
||||
) -> Dict[str, Any]: # pragma: nocover
|
||||
return dict(
|
||||
argument_spec=argument_spec,
|
||||
mutually_exclusive=mutually_exclusive,
|
||||
required_together=required_together,
|
||||
required_one_of=required_one_of,
|
||||
required_if=required_if,
|
||||
required_by=required_by,
|
||||
)
|
||||
|
||||
|
||||
def joindict(*items: dict) -> dict:
|
||||
"""merges one or more dictionaries into one"""
|
||||
odict = dict()
|
||||
for item in items:
|
||||
for key, value in item.items():
|
||||
odict[key] = value
|
||||
return odict
|
||||
|
|
@ -0,0 +1,76 @@
|
|||
from typing import Any, Dict, Optional, Type, Union
|
||||
|
||||
from _typeshed import Incomplete
|
||||
|
||||
SYSTEMD_CONFIG_ROOT: Incomplete
|
||||
SYSTEMD_NETWORK_CONFIG: Incomplete
|
||||
SYSTEMD_SERVICE_CONFIG: Incomplete
|
||||
AnsibleParameter = Dict[str, Any]
|
||||
|
||||
class Types:
|
||||
def list(
|
||||
elements: Union[Type[object], str, AnsibleParameter], required: bool = False, help: Optional[str] = None
|
||||
) -> AnsibleParameter: ...
|
||||
def dict(required: bool = False, help: Optional[str] = None, **options: dict) -> AnsibleParameter: ...
|
||||
def str(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter: ...
|
||||
def bool(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter: ...
|
||||
def int(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter: ...
|
||||
def float(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter: ...
|
||||
def path(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter: ...
|
||||
def raw(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter: ...
|
||||
def jsonarg(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter: ...
|
||||
def json(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter: ...
|
||||
def bytes(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter: ...
|
||||
def bits(
|
||||
required: bool = False,
|
||||
help: Optional[str] = None,
|
||||
choices: Optional[Sequence] = None,
|
||||
default: Optional[Any] = None,
|
||||
) -> AnsibleParameter: ...
|
||||
|
||||
def systemdbool(b: Union[bool, str]) -> str: ...
|
||||
|
|
@ -0,0 +1,374 @@
|
|||
import pathlib
|
||||
from copy import deepcopy
|
||||
from typing import (Any, Callable, ClassVar, Dict, NoReturn, Optional, Type,
|
||||
TypeVar, Union, overload)
|
||||
|
||||
import ansible.module_utils.basic as basic
|
||||
|
||||
from ansible_module.generic import AnsibleParameter, Types, systemdbool
|
||||
|
||||
__all__ = (
|
||||
"AnsibleModule",
|
||||
"SystemdUnitModule",
|
||||
"installable",
|
||||
"SystemdReloadMixin",
|
||||
)
|
||||
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
def docify(input: Union[dict, AnsibleParameter]) -> dict:
|
||||
options = dict()
|
||||
for name, help in input.items():
|
||||
options[name] = dict(type=help["type"])
|
||||
if "description" in help:
|
||||
if isinstance(help["description"], str):
|
||||
help["description"] = help["description"].split("\n")
|
||||
options[name]["description"] = help["description"]
|
||||
if "required" in help and help["required"]:
|
||||
options[name]["required"] = True
|
||||
else:
|
||||
options[name]["required"] = False
|
||||
if help["type"] == "list":
|
||||
options[name]["elements"] = help["elements"]
|
||||
if not options[name]["required"]:
|
||||
options[name]["default"] = []
|
||||
if "default" in help:
|
||||
options[name]["default"] = help["default"]
|
||||
if "options" in help:
|
||||
options[name]["options"] = docify(help["options"])
|
||||
if "choices" in help:
|
||||
options[name]["choices"] = tuple(help["choices"])
|
||||
return options
|
||||
|
||||
|
||||
class AnsibleModule(object):
|
||||
"""Simple wrapper for the basic.AnsibleModule"""
|
||||
|
||||
#: name of the module. This is required for the generation of the Ansible documentation
|
||||
name: ClassVar[str]
|
||||
#: The AnsibleModule for this Module
|
||||
module: basic.AnsibleModule
|
||||
#: The result of this Module call. It always contains the changed key, so in any case an Module can report if it changed anything
|
||||
result: dict
|
||||
#: the specification of the arguments. Subclasses that are usable Modules must set this value.
|
||||
module_spec: ClassVar[dict]
|
||||
#: This is set by classes that define common things for their subclasses, like behaviour of the run and check methods. This is used by `SystemdUnitModule`
|
||||
_common_args: ClassVar[dict[str, Any]] = dict()
|
||||
|
||||
@property
|
||||
def params(self) -> Dict[str, Any]:
|
||||
"""params is an wrapper for the module.params"""
|
||||
return self.module.params # type: ignore
|
||||
|
||||
def __init__(self):
|
||||
self.result = dict(changed=False)
|
||||
specs = dict()
|
||||
specs.update(deepcopy(self._common_args))
|
||||
modspec = self.module_spec
|
||||
if "argument_spec" in modspec and "argument_spec" in self._common_args:
|
||||
specs["argument_spec"].update(modspec["argument_spec"])
|
||||
del modspec["argument_spec"]
|
||||
specs.update(modspec)
|
||||
self.module = basic.AnsibleModule(**specs)
|
||||
self.tmpdir = pathlib.Path(self.module.tmpdir)
|
||||
|
||||
def set(self, key: str, value):
|
||||
"""sets an value for the result"""
|
||||
self.result[key] = value
|
||||
|
||||
@overload
|
||||
def diff(self, diff: Dict[str, str]): # pragma: nocover
|
||||
pass
|
||||
|
||||
@overload
|
||||
def diff(
|
||||
self,
|
||||
before: Optional[str] = None,
|
||||
after: Optional[str] = None,
|
||||
before_header: Optional[str] = None,
|
||||
after_header: Optional[str] = None,
|
||||
): # pragma: nocover
|
||||
pass
|
||||
|
||||
def diff( # type: ignore
|
||||
self,
|
||||
diff=None,
|
||||
*,
|
||||
before=None,
|
||||
after=None,
|
||||
before_header=None,
|
||||
after_header=None,
|
||||
):
|
||||
"""adds the special return value "diff". This allows Modules to present the changes of files to the caller. it takes care of the special semantics of the return value"""
|
||||
if "diff" not in self.result:
|
||||
self.result["diff"] = list()
|
||||
if diff is not None and not any((before is not None, after is not None)):
|
||||
pass
|
||||
elif all((before is not None, after is not None, diff is None)):
|
||||
diff = dict(
|
||||
before=before,
|
||||
after=after,
|
||||
)
|
||||
if before_header is not None:
|
||||
diff["before_header"] = before_header
|
||||
if after_header is not None:
|
||||
diff["after_header"] = after_header
|
||||
else:
|
||||
raise TypeError("only diff or before and after can be set, not both of them")
|
||||
self.result["diff"].append(diff)
|
||||
|
||||
def get(self, key: str, default: Optional[T] = None) -> T: # type: ignore
|
||||
"""returns an Parameter of the Module."""
|
||||
if self.params[key] is None and default is not None:
|
||||
return default
|
||||
elif self.params[key] is None and default is None:
|
||||
raise KeyError()
|
||||
return self.params[key] # type:ignore
|
||||
|
||||
def changed_get(self):
|
||||
"""value that shows if changes were detected/made"""
|
||||
return self.result["changed"]
|
||||
|
||||
def changed_set(self, value):
|
||||
self.result["changed"] = not not value
|
||||
|
||||
changed = property(changed_get, changed_set, None, changed_get.__doc__)
|
||||
|
||||
def prepare(self): # pragma: nocover
|
||||
raise NotImplementedError()
|
||||
|
||||
def check(self): # pragma: nocover
|
||||
raise NotImplementedError()
|
||||
|
||||
def run(self): # pragma: nocover
|
||||
raise NotImplementedError()
|
||||
|
||||
def __call__(self) -> NoReturn: # pragma: nocover
|
||||
"""This calls the module. first prepare is called and then check or run, depending on the check mode.
|
||||
If an exception is raised this is catched and the module automatically fails with an traceback
|
||||
"""
|
||||
self.prepare()
|
||||
try:
|
||||
if self.module.check_mode:
|
||||
self.check()
|
||||
else:
|
||||
self.run()
|
||||
except Exception as exc:
|
||||
import traceback
|
||||
|
||||
self.module.fail_json(
|
||||
"".join(traceback.format_exception(type(exc), exc, exc.__traceback__)),
|
||||
**self.result,
|
||||
)
|
||||
self.module.exit_json(**self.result)
|
||||
raise Exception("exit_json failed")
|
||||
|
||||
@classmethod
|
||||
def doc(cls) -> str:
|
||||
"""this returns the documentation string of the module. If the help arguments of an Types method was given, it adds this as an helptext of this parameter"""
|
||||
try:
|
||||
import yaml
|
||||
except ImportError: # pragma: nocover
|
||||
return "---\n"
|
||||
doc = cls.__doc__
|
||||
if doc is None:
|
||||
doc = ""
|
||||
specs = dict()
|
||||
if "argument_spec" in cls._common_args: # pragma: nocover
|
||||
specs.update(cls._common_args["argument_spec"])
|
||||
if "argument_spec" in cls.module_spec: # pragma: nocover
|
||||
specs.update(cls.module_spec["argument_spec"])
|
||||
options = docify(specs)
|
||||
docu = doc.split("\n")
|
||||
return str(
|
||||
yaml.safe_dump(
|
||||
dict(
|
||||
module=cls.name,
|
||||
short_description=docu[0],
|
||||
description=docu,
|
||||
options=options,
|
||||
),
|
||||
stream=None,
|
||||
explicit_start=True,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class SystemdUnitModule(AnsibleModule):
|
||||
#: path of the unitfile managed by this module
|
||||
unitfile: pathlib.Path
|
||||
#: subclasses of this always support the file common args and the check mode
|
||||
_common_args = dict(
|
||||
supports_check_mode=True,
|
||||
add_file_common_args=True,
|
||||
argument_spec=dict(
|
||||
description=Types.str(help="An description for programs that access systemd"),
|
||||
documentation=Types.list(str, help="Paths where documentation can be found"),
|
||||
requires=Types.list(
|
||||
str,
|
||||
help="list of units that this unit requires. If it fails or can't be started this unit fails. without before/after this is started at the same time",
|
||||
),
|
||||
wants=Types.list(
|
||||
str,
|
||||
help="list of units that this unit wants. If it fails or can't be started it does not affect this unit",
|
||||
),
|
||||
partof=Types.list(
|
||||
str,
|
||||
help="list of units that this unit is part of.\nIf the restart this unit does it too, but if this restarts it does not affect the other units.",
|
||||
),
|
||||
before=Types.list(
|
||||
str,
|
||||
help="list of units that this unit needs to be started before this unit.",
|
||||
),
|
||||
after=Types.list(
|
||||
str,
|
||||
help="list of units that this unit wants to be started after this unit",
|
||||
),
|
||||
),
|
||||
)
|
||||
#: if defined it will be called after run has changed the unitfile
|
||||
post: Optional[Callable[[], None]]
|
||||
|
||||
#: generates the install section if the unit is installable
|
||||
install: ClassVar[Optional[Callable[["SystemdUnitModule"], str]]]
|
||||
|
||||
def unit(self) -> str: # pragma: nocover
|
||||
raise NotImplementedError()
|
||||
|
||||
def header(self) -> str:
|
||||
header = "[Unit]\n"
|
||||
header += "".join(
|
||||
self.map_param(
|
||||
description="Description",
|
||||
documentation="Documentation",
|
||||
requires="Requires",
|
||||
wants="Wants",
|
||||
partof="PartOf",
|
||||
before="Before",
|
||||
after="After",
|
||||
)
|
||||
)
|
||||
return header
|
||||
|
||||
def map_param(self, **parammap: str):
|
||||
"""maps an dict with keys for an section with given params. The key of the dictionary is the parameter and the value is the key in the unitfile. If an parameter has multiple values it adds multiple entries"""
|
||||
output: list[str] = []
|
||||
for param, key in parammap.items():
|
||||
if self.params[param] is not None:
|
||||
params = self.params[param]
|
||||
if isinstance(params, (list, tuple, set)):
|
||||
output.extend((f"{key}={systemdbool(p)}\n" for p in params))
|
||||
else:
|
||||
output.append(f"{key}={systemdbool(params)}\n")
|
||||
return output
|
||||
|
||||
def unitfile_gen(self): # pragma: nocover
|
||||
path = self.tmpdir / "newunit"
|
||||
with open(path, "w") as unit:
|
||||
unit.write(self.unit())
|
||||
self.module.set_owner_if_different(path.as_posix(), "root", False)
|
||||
self.module.set_group_if_different(path.as_posix(), "root", False)
|
||||
self.module.set_mode_if_different(path.as_posix(), "0644", False)
|
||||
if self.unitfile.exists():
|
||||
diff = dict()
|
||||
self.changed = self.changed | self.module.set_owner_if_different(
|
||||
self.unitfile.as_posix(),
|
||||
"root",
|
||||
self.result["changed"],
|
||||
diff,
|
||||
)
|
||||
self.diff(diff)
|
||||
diff = dict()
|
||||
self.changed = self.changed | self.module.set_group_if_different(
|
||||
self.unitfile.as_posix(),
|
||||
"root",
|
||||
self.result["changed"],
|
||||
diff,
|
||||
)
|
||||
self.diff(diff)
|
||||
diff = dict()
|
||||
self.changed = self.changed | self.module.set_mode_if_different(
|
||||
self.unitfile.as_posix(),
|
||||
"0644",
|
||||
self.result["changed"],
|
||||
diff,
|
||||
)
|
||||
self.diff(diff)
|
||||
|
||||
def check(self): # pragma: nocover
|
||||
self.set("unitfile", self.unitfile.as_posix())
|
||||
self.unitfile_gen()
|
||||
if not self.unitfile.exists():
|
||||
self.diff(before="", after=self.unit(), before_header=self.unitfile.as_posix())
|
||||
self.changed = True
|
||||
else:
|
||||
if self.module.sha256(self.unitfile.as_posix()) != self.module.sha256((self.tmpdir / "newunit").as_posix()):
|
||||
self.changed = True
|
||||
self.diff(
|
||||
before=self.unitfile.read_text(),
|
||||
after=self.unit(),
|
||||
before_header=self.unitfile.as_posix(),
|
||||
)
|
||||
|
||||
def run(self): # pragma: nocover
|
||||
self.check()
|
||||
if not self.changed:
|
||||
return
|
||||
self.module.atomic_move(
|
||||
src=(self.tmpdir / "newunit").as_posix(),
|
||||
dest=self.unitfile.as_posix(),
|
||||
)
|
||||
if hasattr(self, "post") and self.post is not None:
|
||||
self.post()
|
||||
|
||||
|
||||
_INSTALL_MAPPING = dict(
|
||||
required_by="RequiredBy",
|
||||
wanted_by="WantedBy",
|
||||
)
|
||||
|
||||
|
||||
def installable(_class: Type[SystemdUnitModule]): # pragma: nocover
|
||||
"""adds the required arguments to the spec and adds the install method for the unit method"""
|
||||
specs = _class.module_spec
|
||||
|
||||
arguments = dict(
|
||||
required_by=Types.list(elements=str, help="systemd units that require this mount"),
|
||||
wanted_by=Types.list(
|
||||
elements=str,
|
||||
help="systemd units that want the mount, but not explicitly require it. Commonly used for target if not service explicitly require it.",
|
||||
),
|
||||
)
|
||||
specs["argument_spec"].update(arguments)
|
||||
|
||||
def install(self: SystemdUnitModule) -> str:
|
||||
output = "[Install]\n"
|
||||
for argument, key in _INSTALL_MAPPING.items():
|
||||
if self.get(argument, False):
|
||||
for unit in self.get(argument): # type: ignore
|
||||
output += "{}={}\n".format(key, unit)
|
||||
return output
|
||||
|
||||
_class.install = install
|
||||
_class.module_spec = specs
|
||||
return _class
|
||||
|
||||
|
||||
class SystemdReloadMixin:
|
||||
module: basic.AnsibleModule
|
||||
unitfile: pathlib.Path
|
||||
restartable: bool = True
|
||||
changed: bool
|
||||
|
||||
def post(self): # pragma: nocover
|
||||
if not self.changed:
|
||||
return
|
||||
systemctl = self.module.get_bin_path("systemctl", required=True)
|
||||
self.module.run_command([systemctl, "daemon-reload"], check_rc=True)
|
||||
if self.unitfile.stem.endswith("@") or not self.restartable:
|
||||
return
|
||||
(rc, _, _) = self.module.run_command([systemctl, "is-enabled", self.unitfile.name], check_rc=False)
|
||||
if rc == 0:
|
||||
self.module.run_command([systemctl, "restart", self.unitfile.name], check_rc=True)
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
import pathlib
|
||||
from typing import (Any, Callable, ClassVar, Dict, NoReturn, Optional, Type,
|
||||
TypeVar, overload)
|
||||
|
||||
import ansible.module_utils.basic as basic
|
||||
from _typeshed import Incomplete
|
||||
|
||||
T = TypeVar('T')
|
||||
|
||||
class AnsibleModule:
|
||||
name: ClassVar[str]
|
||||
module: basic.AnsibleModule
|
||||
result: dict
|
||||
module_spec: ClassVar[dict]
|
||||
@property
|
||||
def params(self) -> Dict[str, Any]: ...
|
||||
tmpdir: Incomplete
|
||||
def __init__(self) -> None: ...
|
||||
def set(self, key: str, value): ...
|
||||
@overload
|
||||
def diff(self, diff: Dict[str, str]): ...
|
||||
@overload
|
||||
def diff(
|
||||
self,
|
||||
before: Optional[str] = ...,
|
||||
after: Optional[str] = ...,
|
||||
before_header: Optional[str] = ...,
|
||||
after_header: Optional[str] = ...,
|
||||
): ...
|
||||
def get(self, key: str, default: Optional[T] = ...) -> T: ...
|
||||
def changed_get(self): ...
|
||||
def changed_set(self, value) -> None: ...
|
||||
changed: Incomplete
|
||||
def prepare(self) -> None: ...
|
||||
def check(self) -> None: ...
|
||||
def run(self) -> None: ...
|
||||
def __call__(self) -> NoReturn: ...
|
||||
@classmethod
|
||||
def doc(cls) -> str: ...
|
||||
|
||||
class SystemdUnitModule(AnsibleModule):
|
||||
unitfile: pathlib.Path
|
||||
post: Optional[Callable[[], None]]
|
||||
install: ClassVar[Optional[Callable[[SystemdUnitModule], str]]]
|
||||
def unit(self) -> str: ...
|
||||
def header(self) -> str: ...
|
||||
def map_param(self, **parammap: str): ...
|
||||
changed: Incomplete
|
||||
def unitfile_gen(self) -> None: ...
|
||||
def check(self) -> None: ...
|
||||
def run(self) -> None: ...
|
||||
|
||||
def installable(_class: Type[SystemdUnitModule]): ...
|
||||
|
||||
class SystemdReloadMixin:
|
||||
module: basic.AnsibleModule
|
||||
unitfile: pathlib.Path
|
||||
restartable: bool
|
||||
changed: bool
|
||||
def post(self) -> None: ...
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
# SPDX-FileCopyrightText: 2023-present Sebastian Tobie <sebastian@sebastian-tobie.de>
|
||||
#
|
||||
# SPDX-License-Identifier: MIT
|
||||
|
|
@ -0,0 +1,57 @@
|
|||
import os
|
||||
import unittest
|
||||
|
||||
from ansible_module.generic import Types, joindict, systemdbool
|
||||
|
||||
|
||||
class TestTypes(unittest.TestCase):
|
||||
"""tests the Types class"""
|
||||
|
||||
def testsimpletype(self):
|
||||
"""this tests if an simple type is correctly build"""
|
||||
output = Types.str(required=True, help="test", choices=("a", "1"), default="1")
|
||||
self.assertIn("type", output)
|
||||
self.assertIn("required", output)
|
||||
self.assertIn("default", output)
|
||||
self.assertIn("choices", output)
|
||||
self.assertEqual(output["type"], "str")
|
||||
self.assertEqual(Types.str.__name__, "str")
|
||||
self.assertEqual(output["required"], True)
|
||||
self.assertEqual(output["default"], "1")
|
||||
self.assertTupleEqual(output["choices"], ("a", "1"))
|
||||
Types.str()
|
||||
|
||||
def testlisttype(self):
|
||||
"""this tests if the special type list is correctly build"""
|
||||
output = Types.list(str)
|
||||
Types.list("str")
|
||||
self.assertIn("type", output)
|
||||
self.assertIn("elements", output)
|
||||
self.assertIn("required", output)
|
||||
self.assertEqual(output["type"], "list")
|
||||
self.assertEqual(output["required"], False)
|
||||
self.assertEqual(output["elements"], "str")
|
||||
Types.list(Types.dict(a=Types.str(help="")), help="")
|
||||
Types.list("str")
|
||||
|
||||
def testdicttype(self):
|
||||
output = Types.dict(help="HILFE")
|
||||
self.assertIn("type", output)
|
||||
self.assertIn("required", output)
|
||||
self.assertIn("description", output)
|
||||
self.assertIn("option", output)
|
||||
|
||||
|
||||
class TestFuncs(unittest.TestCase):
|
||||
def testsystemdbool(self):
|
||||
self.assertEqual("Text", systemdbool("Text"))
|
||||
self.assertEqual("no", systemdbool(False))
|
||||
self.assertEqual("yes", systemdbool(True))
|
||||
|
||||
def testjoindict(self):
|
||||
dicts = (
|
||||
dict(a=1, b=2),
|
||||
dict(b=3, c=4),
|
||||
)
|
||||
output = dict(a=1, b=3, c=4)
|
||||
self.assertDictEqual(output, joindict(*dicts))
|
||||
|
|
@ -0,0 +1,71 @@
|
|||
import sys
|
||||
import unittest
|
||||
|
||||
from ansible_module.generic import modspec
|
||||
from ansible_module.module import SystemdUnitModule, docify
|
||||
|
||||
|
||||
class ModuleMock(SystemdUnitModule):
|
||||
name = "mock"
|
||||
module_spec = modspec({})
|
||||
|
||||
|
||||
class TestSystemdUnitModule(unittest.TestCase):
|
||||
def setUp(self):
|
||||
sys.argv = [sys.argv[0], '{"ANSIBLE_MODULE_ARGS":{}}']
|
||||
self.mod = ModuleMock()
|
||||
|
||||
def test_header(self):
|
||||
self.assertEqual("[Unit]\n", self.mod.header())
|
||||
|
||||
def test_doc(self):
|
||||
ModuleMock.doc()
|
||||
|
||||
def test_params(self):
|
||||
self.mod.module.params = dict(ok="ok", none=None, true=True, false=False, list=("item1", "item2"))
|
||||
self.assertEqual("ok", self.mod.get("ok"))
|
||||
self.assertEqual("default", self.mod.get("none", "default"))
|
||||
self.assertRaises(KeyError, self.mod.get, "none", None)
|
||||
self.assertListEqual(
|
||||
['TRUE=yes\n', "FALSE=no\n", "OK=ok\n", "LIST=item1\n", "LIST=item2\n"],
|
||||
self.mod.map_param(true="TRUE", false="FALSE", ok="OK", list="LIST"),
|
||||
)
|
||||
self.assertRaises(TypeError, self.mod.diff, diff=dict(), before=True, after=False)
|
||||
self.mod.diff(before="a", after="b", before_header="a", after_header="b")
|
||||
self.mod.diff(dict(before="c", after="d"))
|
||||
|
||||
def test_result(self):
|
||||
self.assertFalse(self.mod.changed)
|
||||
self.mod.changed = 5
|
||||
self.assertTrue(self.mod.changed)
|
||||
self.mod.set("key", "value")
|
||||
self.assertDictEqual(
|
||||
dict(
|
||||
changed=True,
|
||||
key="value",
|
||||
),
|
||||
self.mod.result,
|
||||
)
|
||||
|
||||
|
||||
class TestFunctions(unittest.TestCase):
|
||||
def test_docify(self):
|
||||
input = dict(
|
||||
item1=dict(type="str", description=""),
|
||||
item2=dict(required=True, type="int", default=5),
|
||||
item3=dict(type="list", required=False, elements="int"),
|
||||
item4=dict(type="list", required=True, elements="int"),
|
||||
item5=dict(type="str", choices=["a", "b", "c"]),
|
||||
item6=dict(type="dict", options=dict(a=dict(type="str"))),
|
||||
)
|
||||
expected = dict(
|
||||
item1=dict(required=False, type="str", description=[""]),
|
||||
item2=dict(required=True, type="int", default=5),
|
||||
item3=dict(type="list", required=False, elements="int", default=[]),
|
||||
item4=dict(type="list", required=True, elements="int"),
|
||||
item5=dict(type="str", required=False, choices=("a", "b", "c")),
|
||||
item6=dict(type="dict", required=False, options=dict(a=dict(type="str", required=False))),
|
||||
)
|
||||
output = docify(input)
|
||||
for key, value in input.items():
|
||||
self.assertDictEqual(expected[key], output[key])
|
||||
Laden…
In neuem Issue referenzieren