Commits vergleichen

..

22 Commits

Autor SHA1 Nachricht Datum
Sebastian Tobie c5e6129682 modspec now supports supports_check_mode and add_file_common_args 2024-03-16 14:15:50 +01:00
Sebastian Tobie 1ea397da14 improved the build process 2024-03-16 10:39:59 +01:00
Sebastian Tobie 84eb37ffa7 added wrapper for fail_json and exit_json 2024-03-16 10:04:38 +01:00
Sebastian Tobie 85880d2867 added type hint for the diff. 2024-03-16 09:56:39 +01:00
Sebastian Tobie 380bc582e3 added returntype 2024-03-16 09:28:26 +01:00
Sebastian Tobie 8190ee94d3 fixed the empty options 2024-03-13 20:17:36 +01:00
Sebastian Tobie 2f36e69a73 fixed the documentation for dictionaries 2024-03-13 20:13:35 +01:00
Sebastian Tobie 1da65334e2 added an default display 2024-03-12 21:18:09 +01:00
Sebastian Tobie 47c8d3319e added fragments again 2024-03-12 21:10:44 +01:00
Sebastian Tobie cd1d5aa84e improved the build and upload 2024-03-08 22:44:08 +01:00
Sebastian Tobie 0ba608406e to prevent empty sections, the install and header methods return None if the method would just the scetion header 2024-03-08 21:58:32 +01:00
Sebastian Tobie d03b45d678 updated changelog 2024-03-04 23:06:53 +01:00
Sebastian Tobie e583ad0f96 small remove print calls 2024-02-24 10:45:49 +01:00
Sebastian Tobie 8dbb67faf8 the upload script now upload python and ansible packages correctly 2024-02-24 10:21:48 +01:00
Sebastian Tobie e089dbc82a rewrote the Types class 2024-02-24 10:20:47 +01:00
Sebastian Tobie 768713aadc created an ansible module with module utils 2024-02-11 20:32:38 +01:00
Sebastian Tobie 5b9dd8df0b removed dep 2023-12-09 13:04:00 +01:00
Sebastian Tobie 1a8278f763 fixed the generator expection 2023-11-26 10:29:12 +01:00
Sebastian Tobie fbe73d2d83 fixed an exception 2023-11-26 10:11:43 +01:00
Sebastian Tobie 67fcb665f3 fixed version 2023-11-25 13:35:20 +01:00
Sebastian Tobie d390eb1370 added some upload info 2023-11-25 13:34:37 +01:00
Sebastian Tobie a53a14adf6 removed the badges 2023-11-25 13:34:27 +01:00
33 geänderte Dateien mit 542 neuen und 471 gelöschten Zeilen

57
CHANGELOG.rst Normale Datei
Datei anzeigen

@ -0,0 +1,57 @@
============================
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

26
Makefile Normale Datei
Datei anzeigen

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

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

47
changelogs/changelog.yaml Normale Datei
Datei anzeigen

@ -0,0 +1,47 @@
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'

32
changelogs/config.yaml Normale Datei
Datei anzeigen

@ -0,0 +1,32 @@
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

@ -0,0 +1,6 @@
---
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

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

28
galaxy.yml Normale Datei
Datei anzeigen

@ -0,0 +1,28 @@
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

4
meta/runtime.yml Normale Datei
Datei anzeigen

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

31
plugins/README.md Normale Datei
Datei anzeigen

@ -0,0 +1,31 @@
# 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

@ -0,0 +1,175 @@
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

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

Datei anzeigen

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

@ -1,135 +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]
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

@ -1,76 +0,0 @@
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,60 +0,0 @@
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
src/ansible_module/py.typed Normale Datei
Datei anzeigen

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

@ -1,57 +0,0 @@
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))

Datei anzeigen

@ -1,71 +0,0 @@
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])

7
upload.sh Ausführbare Datei
Datei anzeigen

@ -0,0 +1,7 @@
#!/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/*