Commits vergleichen

..

Keine gemeinsamen Commits. "stable" und "v0.1.0" haben vollständig unterschiedliche Historien.

33 geänderte Dateien mit 471 neuen und 542 gelöschten Zeilen

Datei anzeigen

@ -1,57 +0,0 @@
============================
Sebastian.Base Release Notes
============================
.. contents:: Topics
v0.4.2
======
Minor Changes
-------------
- removed the empty options dict
v0.4.1
======
Minor Changes
-------------
- added an default Display to the module
- fixed the docification of dictionaries
v0.4.0
======
Release Summary
---------------
to prevent empty sections, the install and header methods return None if the method would just the scetion
header
v0.3.1
======
Release Summary
---------------
removed forgotten print calls
v0.3.0
======
Release Summary
---------------
rewrote the Types helper
v0.2.0
======
Release Summary
---------------
change the module to an ansible module

Datei anzeigen

@ -1,26 +0,0 @@
VERSION := $(shell hatch version)
format:
black .
isort .
version:
@yq --yaml-output-grammar-version 1.2 -i -y -s --arg 'version' "${VERSION}" '.[0] | (.version = $$version)' galaxy.yml
changelog: version
antsibull-changelog generate
docs: format
update-doc
clean-dist:
rm -rf dist
hatch-release: clean-dist version
hatch build
galaxy-release: clean-dist changelog version
ansible-galaxy collection build --output-path dist/galaxy
upload: galaxy-release hatch-release
./upload.sh

Datei anzeigen

@ -1,3 +1,21 @@
# Ansible Collection - sebastian.base
# Ansible module
Documentation for the collection.
[![PyPI - Version](https://img.shields.io/pypi/v/ansible-module.svg)](https://pypi.org/project/ansible-module)
[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/ansible-module.svg)](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.

Datei anzeigen

@ -1,19 +0,0 @@
objects:
role: {}
plugins:
become: {}
cache: {}
callback: {}
cliconf: {}
connection: {}
filter: {}
httpapi: {}
inventory: {}
lookup: {}
module: {}
netconf: {}
shell: {}
strategy: {}
test: {}
vars: {}
version: 0.4.2

Datei anzeigen

@ -1,47 +0,0 @@
ancestor: null
releases:
0.2.0:
changes:
release_summary: change the module to an ansible module
fragments:
- base_release.yml
release_date: '2024-02-11'
0.3.0:
changes:
release_summary: rewrote the Types helper
fragments:
- types.yml
release_date: '2024-02-24'
0.3.1:
changes:
release_summary: removed forgotten print calls
fragments:
- print_calls.yml
release_date: '2024-02-24'
0.4.0:
changes:
release_summary: 'to prevent empty sections, the install and header methods
return None if the method would just the scetion
header
'
fragments:
- sectioning.yml
release_date: '2024-03-08'
0.4.1:
changes:
minor_changes:
- added an default Display to the module
- fixed the docification of dictionaries
fragments:
- display.yml
- options_fix.yml
release_date: '2024-03-13'
0.4.2:
changes:
minor_changes:
- removed the empty options dict
fragments:
- empty_options.yml
release_date: '2024-03-13'

Datei anzeigen

@ -1,32 +0,0 @@
changelog_filename_template: ../CHANGELOG.rst
changelog_filename_version_depth: 0
changes_file: changelog.yaml
changes_format: combined
ignore_other_fragment_extensions: true
keep_fragments: true
mention_ancestor: true
new_plugins_after_name: removed_features
notesdir: fragments
prelude_section_name: release_summary
prelude_section_title: Release Summary
sanitize_changelog: true
sections:
- - major_changes
- Major Changes
- - minor_changes
- Minor Changes
- - breaking_changes
- Breaking Changes / Porting Guide
- - deprecated_features
- Deprecated Features
- - removed_features
- Removed Features (previously deprecated)
- - security_fixes
- Security Fixes
- - bugfixes
- Bugfixes
- - known_issues
- Known Issues
title: Sebastian.Base
trivial_section_name: trivial
use_fqcn: true

Datei anzeigen

@ -1,6 +0,0 @@
---
minor_changes:
- Added an type hint to the update_doc function
- Added Type hint to the dictionary argument of the diff method
- added an wrapper for fail_json and exit_json
- improved the version update, the version is now only in one place the source of truth

Datei anzeigen

@ -1,2 +0,0 @@
---
release_summary: change the module to an ansible module

Datei anzeigen

@ -1,3 +0,0 @@
---
minor_changes:
- modspec now supports supports_check_mode and add_file_common_args

Datei anzeigen

@ -1,3 +0,0 @@
---
minor_changes:
- added an default Display to the module

Datei anzeigen

@ -1,3 +0,0 @@
---
minor_changes:
- removed the empty options dict

Datei anzeigen

@ -1,3 +0,0 @@
---
minor_changes:
- fixed the docification of dictionaries

Datei anzeigen

@ -1,2 +0,0 @@
---
release_summary: removed forgotten print calls

Datei anzeigen

@ -1,4 +0,0 @@
---
release_summary: |
to prevent empty sections, the install and header methods return None if the method would just the scetion
header

Datei anzeigen

@ -1,2 +0,0 @@
---
release_summary: rewrote the Types helper

Datei anzeigen

@ -1,28 +0,0 @@
namespace: sebastian
name: base
version: 0.4.3
readme: README.md
authors:
- Sebastian Tobie
description: The base of my ansible collections. It provides the nessesary tools for
my modules
license_file: LICENSE.txt
tags:
- linux
- systemd
dependencies: {}
repository: https://gitea.sebastian-tobie.de/ansible/ansible-module.git
homepage: https://gitea.sebastian-tobie.de/ansible/ansible-module
issues: https://gitea.sebastian-tobie.de/ansible/ansible-module/issues
build_ignore:
- '*.gz'
- .*
- Makefile
- pyproject.toml
- upload.sh
- htmlcov
- changelogs
- docs
- src
- coverage.*
- dist

Datei anzeigen

@ -1,4 +0,0 @@
---
# Collections must specify a minimum required ansible version to upload
# to galaxy
requires_ansible: '>=2.9.10'

Datei anzeigen

@ -1,31 +0,0 @@
# Collections Plugins Directory
This directory can be used to ship various plugins inside an Ansible collection. Each plugin is placed in a folder that
is named after the type of plugin it is in. It can also include the `module_utils` and `modules` directory that
would contain module utils and modules respectively.
Here is an example directory of the majority of plugins currently supported by Ansible:
```
└── plugins
├── action
├── become
├── cache
├── callback
├── cliconf
├── connection
├── filter
├── httpapi
├── inventory
├── lookup
├── module_utils
├── modules
├── netconf
├── shell
├── strategy
├── terminal
├── test
└── vars
```
A full list of plugin types can be found at [Working With Plugins](https://docs.ansible.com/ansible-core/2.16/plugins/plugins.html).

Datei anzeigen

@ -1,175 +0,0 @@
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]
def wrap_func(func, **updates):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
attrs = frozenset(
('__module__', '__name__', '__qualname__', '__doc__', '__annotations__', '__type_params__')
) - frozenset(updates.keys())
for attr in attrs:
try:
value = getattr(func, attr)
except AttributeError:
pass
else:
setattr(wrapper, attr, value)
for attr, value in updates.items():
setattr(wrapper, attr, value)
wrapper.__dict__.update(func.__dict__)
setattr(wrapper, "__wrapped__", func)
return wrapper
GENERIC_DOC = """Returns an dictionary for the Ansible {type} type."""
def default(name: str):
def wrapped(
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
return wrapped
class meta(type):
def __new__(cls, clsname, bases, attrs):
types = frozenset(
(
"str",
"bool",
"int",
"float",
"path",
"raw",
"jsonarg",
"json",
"bytes",
"dict",
"list",
"bits",
)
)
for attr in types - set(attrs.keys()):
attrs[attr] = wrap_func(
default(attr), __doc__=GENERIC_DOC.format(type=attr), __name__=attr, __qualname__=f"{clsname}.{attr}"
)
attrs["__slots__"] = ()
return super().__new__(cls, clsname, bases, attrs)
class Types(metaclass=meta):
def list( # type: ignore[misc]
elements: Union[Type[object], str, AnsibleParameter],
required: bool = False,
help: Optional[str] = None,
) -> AnsibleParameter:
"""Wrapper for the Ansible list type
Args:
elements: The type of the elements
required: if the item is absolutly required
help: an helptext for the ansible-doc
"""
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(required: bool = False, help: Optional[str] = None, **options: AnsibleParameter) -> AnsibleParameter: # type: ignore[misc]
"""Wrapper for the Ansible dict type
Args:
required: if the item is absolutly required
help: an helptext for the ansible-doc
options: The individual options that this parameter has
"""
option: AnsibleParameter = dict(type="dict", required=required)
option["options"] = options
if help is not None:
option["description"] = help.split("\n")
return option
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, ...]]] = {},
supports_check_mode: bool = False,
add_file_common_args: bool = False,
) -> 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,
add_file_common_args=add_file_common_args,
supports_check_mode=supports_check_mode,
)
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

Datei anzeigen

@ -22,9 +22,8 @@ classifiers = [
"Programming Language :: Python :: Implementation :: CPython",
"Programming Language :: Python :: Implementation :: PyPy",
]
dependencies = []
[tool.hatch.build]
directory = "dist/python"
dependencies = ["ansible-core>=2.9.10"]
[project.urls]
Documentation = "https://github.com/unknown/ansible-module#readme"
Issues = "https://github.com/unknown/ansible-module/issues"
@ -33,9 +32,6 @@ Source = "https://github.com/unknown/ansible-module"
[project.scripts]
update-doc = "ansible_module.update_doc:main"
[tool.hatch.build.targets.wheel.force-include]
"plugins/module_utils" = "ansible_module/module_utils"
[tool.hatch.version]
path = "src/ansible_module/__init__.py"
@ -59,9 +55,6 @@ style = ["ruff {args:.}", "black --check --diff {args:.}"]
fmt = ["black {args:.}", "ruff --fix {args:.}", "style"]
all = ["style", "typing"]
[tool.hatch.publish.index.repos.private]
url = "https://gitea.sebastian-tobie.de/api/packages/sebastian/pypi"
[tool.black]
target-version = ["py37"]
line-length = 120
@ -139,30 +132,26 @@ 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",
]
exclude_lines = ["no cov", "if __name__ == .__main__.:", "if TYPE_CHECKING:", "pragma: nocover"]
[tool.mypy]
disable_error_code = "import-untyped"
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-report", "json",
"--cov-report", "term-missing:skip-covered",
"--cov",
]
pythonpath = ["src"]
required_plugins = ["pytest-cov", "pytest-isort", "pytest-mypy"]
required_plugins = [
"pytest-cov",
"pytest-isort",
"pytest-mypy"
]

Datei anzeigen

@ -1 +1 @@
__version__ = "0.4.3"
__version__ = "0.1.0"

Datei anzeigen

@ -1,4 +1,6 @@
from . import update_doc
if __name__ == "__main__":
update_doc.main()

135
src/ansible_module/generic.py Normale Datei
Datei anzeigen

@ -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

Datei anzeigen

@ -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: ...

Datei anzeigen

@ -1,11 +1,11 @@
import pathlib
from copy import deepcopy
from typing import Any, Callable, ClassVar, Dict, NoReturn, Optional, Type, TypeVar, Union, overload, TypedDict
from ansible.utils.display import Display
from typing import (Any, Callable, ClassVar, Dict, NoReturn, Optional, Type,
TypeVar, Union, overload)
import ansible.module_utils.basic as basic
from .generic import AnsibleParameter, Types, systemdbool
from ansible_module.generic import AnsibleParameter, Types, systemdbool
__all__ = (
"AnsibleModule",
@ -18,13 +18,6 @@ __all__ = (
T = TypeVar("T")
class TypedDiff(TypedDict):
before: str
after: str
before_header: Optional[str] = None
after_header: Optional[str] = None
def docify(input: Union[dict, AnsibleParameter]) -> dict:
options = dict()
for name, help in input.items():
@ -43,9 +36,9 @@ def docify(input: Union[dict, AnsibleParameter]) -> dict:
options[name]["default"] = []
if "default" in help:
options[name]["default"] = help["default"]
if "options" in help and help["options"] != {}:
if "options" in help:
options[name]["options"] = docify(help["options"])
if "choices" in help and len(help["choices"]) > 0:
if "choices" in help:
options[name]["choices"] = tuple(help["choices"])
return options
@ -63,8 +56,6 @@ class AnsibleModule(object):
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()
#: For log messages
display: Display
@property
def params(self) -> Dict[str, Any]:
@ -80,7 +71,6 @@ class AnsibleModule(object):
specs["argument_spec"].update(modspec["argument_spec"])
del modspec["argument_spec"]
specs.update(modspec)
self.display = Display()
self.module = basic.AnsibleModule(**specs)
self.tmpdir = pathlib.Path(self.module.tmpdir)
@ -89,7 +79,7 @@ class AnsibleModule(object):
self.result[key] = value
@overload
def diff(self, diff: TypedDiff): # pragma: nocover
def diff(self, diff: Dict[str, str]): # pragma: nocover
pass
@overload
@ -168,8 +158,12 @@ class AnsibleModule(object):
except Exception as exc:
import traceback
self.fail("".join(traceback.format_exception(type(exc), exc, exc.__traceback__)))
self.exit()
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:
@ -201,14 +195,6 @@ class AnsibleModule(object):
)
)
def fail(self, message: str) -> NoReturn:
"""Wrapper for AnsibleModule.fail_json"""
self.module.fail_json(message, **self.result)
def exit(self) -> NoReturn:
"""Wrapper for AnsibleModule.exit_json"""
self.module.exit_json(**self.result)
class SystemdUnitModule(AnsibleModule):
#: path of the unitfile managed by this module
@ -243,16 +229,18 @@ class SystemdUnitModule(AnsibleModule):
),
)
#: if defined it will be called after run has changed the unitfile
post: Optional[Callable[[], None]] = None
post: Optional[Callable[[], None]]
#: generates the install section if the unit is installable
install: ClassVar[Optional[Callable[["SystemdUnitModule"], str | None]]] = None
install: ClassVar[Optional[Callable[["SystemdUnitModule"], str]]]
def unit(self) -> str: # pragma: nocover
raise NotImplementedError()
def header(self) -> Optional[str]:
parts = self.map_param(
def header(self) -> str:
header = "[Unit]\n"
header += "".join(
self.map_param(
description="Description",
documentation="Documentation",
requires="Requires",
@ -261,19 +249,10 @@ class SystemdUnitModule(AnsibleModule):
before="Before",
after="After",
)
if len(parts) == 0:
return None
header = "[Unit]\n" + "".join(parts)
)
return header
def _unit(self, *parts: Optional[str]) -> str:
opart = []
for part in parts:
if part is not None:
opart.append(part)
return "\n".join(opart)
def map_param(self, **parammap: str) -> list[str]:
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():
@ -285,7 +264,7 @@ class SystemdUnitModule(AnsibleModule):
output.append(f"{key}={systemdbool(params)}\n")
return output
def unitfile_gen(self) -> None: # pragma: nocover
def unitfile_gen(self): # pragma: nocover
path = self.tmpdir / "newunit"
with open(path, "w") as unit:
unit.write(self.unit())
@ -293,7 +272,7 @@ class SystemdUnitModule(AnsibleModule):
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[str, str] = dict()
diff = dict()
self.changed = self.changed | self.module.set_owner_if_different(
self.unitfile.as_posix(),
"root",
@ -364,15 +343,13 @@ def installable(_class: Type[SystemdUnitModule]): # pragma: nocover
)
specs["argument_spec"].update(arguments)
def install(self: SystemdUnitModule) -> Optional[str]:
parts = []
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
parts.append("{}={}\n".format(key, unit))
if len(parts) == 0:
return None
return "[Install]\n" + "".join(parts)
output += "{}={}\n".format(key, unit)
return output
_class.install = install
_class.module_spec = specs

Datei anzeigen

@ -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: ...

Datei anzeigen

@ -10,13 +10,8 @@ moduledir = pathlib.Path("plugins/modules")
regex = re.compile("DOCUMENTATION *= *r?(?P<quote>\"{3}|'{3})(---)?.*?(?P=quote)", re.MULTILINE | re.DOTALL)
def main() -> None:
try:
modules = list(moduledir.iterdir())
except:
print("failed to find modules")
return
for modfile in modules:
def main():
for modfile in moduledir.iterdir():
if modfile.name in ("__init__.py", "__pycache__", "unit.py.example"):
continue
mod = importlib.import_module(".".join((modfile.parts[:-1]) + (modfile.stem,)))

3
tests/__init__.py Normale Datei
Datei anzeigen

@ -0,0 +1,3 @@
# SPDX-FileCopyrightText: 2023-present Sebastian Tobie <sebastian@sebastian-tobie.de>
#
# SPDX-License-Identifier: MIT

57
tests/test_generic.py Normale Datei
Datei anzeigen

@ -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))

71
tests/test_modules.py Normale Datei
Datei anzeigen

@ -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])

Datei anzeigen

@ -1,7 +0,0 @@
#!/bin/bash
user=$(yq -r .namespace galaxy.yml)
package=$(yq -r .name galaxy.yml)
version=$(yq -r .version galaxy.yml)
printf "Namespace: %s\nPackage: %s\nVersion: %s\n" $user $package $version
ansible-galaxy collection publish dist/galaxy/*
hatch publish -r ansible dist/python/*