diff --git a/.gitignore b/.gitignore index 3040828..89381dd 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # ---> Ansible *.retry +tests/output # ---> Python # Byte-compiled / optimized / DLL files diff --git a/README.md b/README.md index 77ab911..cf7c0e0 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,3 @@ -# ansible-systemd +# Ansible Collection - stop50.systemd -an collection for my own collection for different things \ No newline at end of file +Documentation for the collection. diff --git a/galaxy.yml b/galaxy.yml new file mode 100644 index 0000000..d799914 --- /dev/null +++ b/galaxy.yml @@ -0,0 +1,24 @@ +--- +namespace: sebastian +name: systemd +version: 0.1.0 + +# The path to the Markdown (.md) readme file. This path is relative to the root of the collection +readme: README.md + +# A list of the collection's content authors. Can be just the name or in the format 'Full Name (url) +# @nicks:irc/im.site#channel' +authors: + - Sebastian Tobie +description: An simple for generating systemd units with ansible +license_file: 'LICENSE' +tags: + - systemd + - linux +dependencies: {} +repository: https://gitea.sebastian-tobie.de/sebastian/ansible-systemd +# documentation: http://docs.example.com +# homepage: +issues: https://gitea.sebastian-tobie.de/sebastian/ansible-systemd/issues +build_ignore: [] +# manifest: null diff --git a/meta/runtime.yml b/meta/runtime.yml new file mode 100644 index 0000000..20f709e --- /dev/null +++ b/meta/runtime.yml @@ -0,0 +1,52 @@ +--- +# Collections must specify a minimum required ansible version to upload +# to galaxy +# requires_ansible: '>=2.9.10' + +# Content that Ansible needs to load from another location or that has +# been deprecated/removed +# plugin_routing: +# action: +# redirected_plugin_name: +# redirect: ns.col.new_location +# deprecated_plugin_name: +# deprecation: +# removal_version: "4.0.0" +# warning_text: | +# See the porting guide on how to update your playbook to +# use ns.col.another_plugin instead. +# removed_plugin_name: +# tombstone: +# removal_version: "2.0.0" +# warning_text: | +# See the porting guide on how to update your playbook to +# use ns.col.another_plugin instead. +# become: +# cache: +# callback: +# cliconf: +# connection: +# doc_fragments: +# filter: +# httpapi: +# inventory: +# lookup: +# module_utils: +# modules: +# netconf: +# shell: +# strategy: +# terminal: +# test: +# vars: + +# Python import statements that Ansible needs to load from another location +# import_redirection: +# ansible_collections.ns.col.plugins.module_utils.old_location: +# redirect: ansible_collections.ns.col.plugins.module_utils.new_location + +# Groups of actions/modules that take a common set of options +# action_groups: +# group_name: +# - module1 +# - module2 diff --git a/plugins/README.md b/plugins/README.md new file mode 100644 index 0000000..34cd30a --- /dev/null +++ b/plugins/README.md @@ -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.14/plugins/plugins.html). diff --git a/plugins/__init__.py b/plugins/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/plugins/module_utils/__init__.py b/plugins/module_utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/plugins/module_utils/generic.py b/plugins/module_utils/generic.py new file mode 100644 index 0000000..8b43f71 --- /dev/null +++ b/plugins/module_utils/generic.py @@ -0,0 +1,121 @@ +import pathlib +from functools import partial +from typing import Any, Callable, Dict, Optional, Sequence, Tuple, Type, Union + +__all__ = ( + "Types", + "SYSTEMD_SERVICE_CONFIG", + "SYSTEMD_NETWORK_CONFIG", + "SYSTEMD_CONFIG_ROOT", +) + + +SYSTEMD_CONFIG_ROOT = pathlib.Path("/etc/systemd") +SYSTEMD_NETWORK_CONFIG = SYSTEMD_CONFIG_ROOT / "network" +SYSTEMD_SERVICE_CONFIG = SYSTEMD_CONFIG_ROOT / "system" + + +class _sdict(dict): + _help: Optional[str] + __name__: str + + +class _Type(type): + def __new__(metacls, cls, bases, classdict, **kwds): + individual = dict() + virtfunc = None + virtual = () + special = dict() + for key, value in classdict.items(): + if key.startswith("_"): + if key == "__getattr__": + virtfunc = value + elif key == "__dir__": + virtual = tuple(value(None)) + elif key in ("__doc__",): + special[key] = value + else: + individual[key] = value + if len(virtual) != 0 and virtfunc is None: # pragma: nocover + raise TypeError( + "Virtual funcs defined, but no func to generate them defined" + ) + special["_attr"] = tuple(virtual + tuple(individual.keys())) + special["_vfunc"] = virtfunc + special["_virtual"] = virtual + special["_individual"] = individual + annotations = dict() + if len(virtual) != 0 and virtfunc is not None: # pragma: nocover + anno = virtfunc(None, virtual[0]).__annotations__ + for virtualkey in virtual: + annotations[virtualkey] = Callable[[*anno.values()], Dict[str, Any]] + annotations["__dir__"] = Callable[[], Tuple[str]] + special["__annotations__"] = annotations + inst = super().__new__(metacls, cls, bases, special, **kwds) + return inst + + def __getattribute__(self, __name: str) -> Any: + if __name in ( + "__dict__", + "__doc__", + "_attr", + "__annotations__", + "_virtual", + "_vfunc", + "_individual", + ): + return super().__getattribute__(__name) + if __name in self._virtual: + return self._vfunc(self, __name) + if __name in self._individual: + return partial(self._individual[__name], self) + raise AttributeError(f"Attribut {__name} not found.") + + +class Types(metaclass=_Type): + """Provides helpers for the ansible types""" + + def list( + self, + elements: Union[Type[object], str], + required: bool = False, + help: Optional[str] = None, + ) -> dict: + if not isinstance(elements, str): + elements = elements.__name__ + option = _sdict(type="list", elements=elements, required=required) + option._help = help + return option + + def __dir__(self) -> tuple: + return ( + "str", + "dict", + "bool", + "int", + "float", + "path", + "raw", + "jsonarg", + "json", + "bytes", + "bits", + ) + + def __getattr__(self, name: str): + def argument( + required: bool = False, + help: Optional[str] = None, + choices: Optional[Sequence] = None, + default: Optional[Any] = None, + ): + output = _sdict(type=name, required=required) + if choices is not None: + output["choices"] = choices + if default is not None: + output["default"] = default + output._help = help + return output + + argument.__name__ = name + return argument diff --git a/plugins/module_utils/module.py b/plugins/module_utils/module.py new file mode 100644 index 0000000..37912ca --- /dev/null +++ b/plugins/module_utils/module.py @@ -0,0 +1,221 @@ +import pathlib +from typing import Any, Callable, ClassVar, Dict, Optional, TypeVar + +import ansible.module_utils.basic as basic +from ansible.module_utils.generic import _sdict + +__all__ = ( + "AnsibleModule", + "SystemdUnitModule", +) + + +T = TypeVar("T") + + +class AnsibleModule(object): + """Simple wrapper for the mo""" + + name: ClassVar[str] + module: basic.AnsibleModule + msg: str + result: dict + module_spec: ClassVar[dict] + _common_args = dict() + + @property + def params(self) -> Dict[str, Any]: + return self.module.params # type: ignore + + def __init__(self): + self.result = dict(changed=False) + specs = dict() + specs.update(self._common_args) + specs.update(self.module_spec) + self.module = basic.AnsibleModule(**specs) + self.msg = "" + self.tmpdir = pathlib.Path(self.module.tmpdir) + + def set(self, key: str, value): + self.result[key] = value + + def diff( + self, + before, + after, + before_header: Optional[str] = None, + after_header: Optional[str] = None, + ): + if "diff" not in self.result: + self.result["diff"] = list() + 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 + self.result["diff"].append(diff) + + def get(self, key: str, default: T = None) -> T: + if key not in self.params.keys(): + return default + if self.params[key] is None and default is not None: + return default + if self.params[key] is None or key not in self.params: + raise KeyError() + return self.params[key] + + @property + def changed(self): + return self.result["changed"] + + @changed.setter + def changed_set(self, value): + self.result["changed"] = not not value + + def prepare(self): + raise NotImplementedError() + + def check(self): + raise NotImplementedError() + + def run(self): + raise NotImplementedError() + + def __call__(self): + 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) + + @classmethod + def doc(cls) -> str: + try: + import yaml + except ImportError: + return "---\n" + doc = cls.__doc__ + if doc is None: + doc = "" + options = dict() + help: _sdict + for option, help in cls.module_spec["argument_spec"].items(): + options[option] = dict( + type=help["type"], + ) + if hasattr(help, "_help") and help._help is not None: + options[option]["description"] = help._help.split("\n") + if "required" in help and help["required"]: + options[option]["required"] = True + else: + options[option]["required"] = False + if help["type"] == "list": + options[option]["elements"] = help["elements"] + if not options[option]["required"]: + options[option]["default"] = [] + if "default" in help: + options[option]["default"] = help["default"] + if "choices" in help: + options[option]["choices"] = tuple(help["choices"]) + 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="---", + ) + ) + + +class SystemdUnitModule(AnsibleModule): + unitfile: pathlib.Path + _common_args = dict( + supports_check_mode=True, + add_file_common_args=True, + ) + post: Optional[Callable[[], None]] + + def unit(self) -> str: + raise NotImplementedError() + + def unitfile_gen(self): + 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(): + if "diff" not in self.result: + self.result["diff"] = list() + diff = dict() + self.result["changed"] = self.module.set_owner_if_different( + self.unitfile.as_posix(), + "root", + self.result["changed"], + diff, + ) + self.result["changed"] = self.module.set_group_if_different( + self.unitfile.as_posix(), + "root", + self.result["changed"], + diff, + ) + self.result["changed"] = self.module.set_mode_if_different( + self.unitfile.as_posix(), + "0644", + self.result["changed"], + diff, + ) + self.result["diff"].append(diff) + + def check(self): + if "changed" in self.result: + changed = self.result["changed"] + else: + changed = False + self.unitfile_gen() + if not self.unitfile.exists(): + self.diff("", self.unit(), self.unitfile.as_posix()) + changed = True + else: + if self.module.sha256(self.unitfile.as_posix()) != self.module.sha256( + (self.tmpdir / "newunit").as_posix() + ): + changed = True + self.diff( + before=self.unitfile.read_text(), + after=self.unit(), + before_header=self.unitfile.as_posix(), + ) + self.set("changed", changed) + if hasattr(self, "post") and self.post is not None: + self.post() + return changed + + def run(self): + if not self.check(): + 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() diff --git a/plugins/modules/__init__.py b/plugins/modules/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/plugins/modules/systemd_link.py b/plugins/modules/systemd_link.py new file mode 100644 index 0000000..def54f1 --- /dev/null +++ b/plugins/modules/systemd_link.py @@ -0,0 +1,157 @@ +#!/usr/bin/python3 +import pathlib +from typing import List + +from ansible.module_utils.generic import SYSTEMD_NETWORK_CONFIG as SYSTEMD_PATH +from ansible.module_utils.generic import Types +from ansible.module_utils.module import SystemdUnitModule + + +class Module(SystemdUnitModule): + """generates an systemd-networkd link""" + + name = "systemd_link" + module_spec = dict( + argument_spec=dict( + mac=Types.str(help="The Mac address of the device"), + permanentmac=Types.str( + help="The Permanent Mac address advertised by the device" + ), + path=Types.str( + help="A shell-style glob matching the persistent path, as exposed by the udev property ID_PATH." + ), + driver=Types.str( + help="A glob matching the driver currently bound to the device" + ), + type=Types.str( + help="A glob matching the device type, as exposed by networkctl list" + ), + kind=Types.str( + help="a glob matching the device kind, as exposed by networkctl status INTERFACE or ip -d link show INTERFACE." + ), + description=Types.str("The description for the link"), + name=Types.str(required=True, help="The new name of the device"), + mtu=Types.int(help="The maximum Transmission unit for the link"), + ), + required_one_of=( + ("mac", "permanentmac", "path", "driver", "type", "kind"), + ("name", "mac", "permanentmac"), + ), + ) + + def prepare(self): + self.__unit = None + newname = ( + self.get("name", "") or self.get("mac", "") or self.get("permanentmac", "") + ) + newname = newname.replace(":", "").replace("/", "-").lower() + self.unitfile = SYSTEMD_PATH.joinpath("50-" + newname).with_suffix(".link") + + def unit(self) -> str: + if self.__unit is None: + self.__unit = "\n".join((self.match(), self.link())) + return self.__unit + + def match(self) -> str: + options = [] + if self.get("mac", False): + options.append("MACAddress={}\n".format(self.get("mac", False))) + if self.get("permanentmac", False): + options.append( + "PermanentAddress={}\n".format(self.get("permanentmac", False)) + ) + if self.get("path", False): + options.append("Path={}\n".format(self.get("path", False))) + if self.get("driver", False): + options.append("Driver={}\n".format(self.get("driver", False))) + if self.get("type", False): + options.append("Type={}\n".format(self.get("type", False))) + if self.get("kind", False): + options.append("Kind={}\n".format(self.get("kind", False))) + return "[Match]\n" + "".join(options) + + def link(self) -> str: + options = [] + if self.get("description", False): + options.append("Description={}\n".format(self.get("description", False))) + if self.get("name", False): + options.append("Name={}\n".format(self.get("name", False))) + if self.get("mtu", False): + options.append("MTUBytes={}\n".format(self.get("mtu", False))) + return "[Link]\n" + "".join(options) + + def post(self): + if not self.changed: + return + args = [ + "/usr/bin/udevadm", + "trigger", + "-c", + "add", + ] + if self.module.check_mode: + args.append("-n") + if self.get("mac", False): + args.append("--attr-match=address={}".format(self.get("mac"))) + if self.get("path", False): + args.append(self.get("path")) + self.module.run_command(args, check_rc=True) + + +DOCUMENTATION = """--- +description: +- generates an systemd-networkd link +module: systemd_link +options: + description: + description: + - The description for the link + required: false + type: str + driver: + description: + - A glob matching the driver currently bound to the device + required: false + type: str + kind: + description: + - a glob matching the device kind, as exposed by networkctl status INTERFACE or + ip -d link show INTERFACE. + required: false + type: str + mac: + description: + - The Mac address of the device + required: false + type: str + mtu: + description: + - The maximum Transmission unit for the link + required: false + type: int + name: + description: + - The new name of the device + required: false + type: str + path: + description: + - A shell-style glob matching the persistent path, as exposed by the udev property + ID_PATH. + required: false + type: str + permanentmac: + description: + - The Permanent Mac address advertised by the device + required: false + type: str + type: + description: + - A glob matching the device type, as exposed by networkctl list + required: false + type: str +short_description: generates an systemd-networkd link +""" + +if __name__ == "__main__": + Module()() diff --git a/plugins/modules/systemd_mount.py b/plugins/modules/systemd_mount.py new file mode 100644 index 0000000..bf5a248 --- /dev/null +++ b/plugins/modules/systemd_mount.py @@ -0,0 +1,165 @@ +#!/usr/bin/python3 +import pathlib +from typing import List, Optional + +from ansible.module_utils.generic import SYSTEMD_SERVICE_CONFIG as SYSTEMD_PATH +from ansible.module_utils.generic import Types +from ansible.module_utils.module import SystemdUnitModule + +SYSTEMD_PATH = pathlib.Path("/etc/systemd/system") + +OPTION_MAPPING = dict( + required_by="RequiredBy", + wanted_by="WantedBy", +) + + +class Module(SystemdUnitModule): + """Creates an systemd mount""" + + name = "systemd_mount" + module_spec = dict( + argument_spec=dict( + fs=Types.str( + required=True, help="The filesystem that is used for the mount" + ), + where=Types.path( + required=True, help="The Path where the filesystem is mounted to" + ), + what=Types.str( + required=True, help="The device or an string that will be mounted" + ), + state=Types.str( + choices=("present", "absent"), + default="present", + help="the state the mount is", + ), + options=Types.list(elements=str, help="The options for the mount"), + description=Types.str( + help="An description for programs that access systemd" + ), + 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.", + ), + ), + ) + + def prepare(self): + self.mountdir = pathlib.Path(self.params["where"]) + self.unitfile = SYSTEMD_PATH.joinpath( + self.mountdir.relative_to("/").as_posix().replace("/", "-") + ).with_suffix(".mount") + self.__unit = None + + def unit(self) -> str: + if self.__unit is None: + self.__unit = "\n".join( + ( + self.header(), + self.mount(), + self.install(), + ) + ) + return self.__unit + + def header(self) -> str: + return "[Unit]\nDescription={}\n".format( + self.get("description", "Mount for {}".format(self.get("where"))) + ) + + def mount(self) -> str: + output = "[Mount]\n" + output += "Where={}\n".format(self.get("where")) + output += "What={}\n".format(self.get("what")) + output += "Type={}\n".format(self.get("fs")) + if self.get("options", False): + output += "Options={}\n".format(",".join(self.get("options"))) + return output + + def install(self) -> str: + output = "[Install]\n" + for argument, key in OPTION_MAPPING.items(): + if self.get(argument, False): + for unit in self.get(argument): + output += "{}={}\n".format(key, unit) + return output + + def post(self): + if not self.changed: + return + systemctl = self.module.get_bin_path("systemctl", required=True) + self.module.run_command([systemctl, "daemon-reload"], check_rc=True) + (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 + ) + + +DOCUMENTATION = """--- +description: +- Creates an systemd mount +module: systemd_mount +options: + description: + description: + - An description for programs that access systemd + required: false + type: str + fs: + description: + - The filesystem that is used for the mount + required: true + type: str + options: + default: [] + description: + - The options for the mount + elements: str + required: false + type: list + required_by: + default: [] + description: + - systemd units that require this mount + elements: str + required: false + type: list + state: + choices: + - present + - absent + default: present + description: + - the state the mount is + required: false + type: str + wanted_by: + default: [] + description: + - systemd units that want the mount, but not explicitly require it. Commonly used + for target if not service explicitly require it. + elements: str + required: false + type: list + what: + description: + - The device or an string that will be mounted + required: true + type: str + where: + description: + - The Path where the filesystem is mounted to + required: true + type: path +short_description: Creates an systemd mount +""" + +if __name__ == "__main__": + Module()() diff --git a/plugins/modules/systemd_network.py b/plugins/modules/systemd_network.py new file mode 100644 index 0000000..e8bb2b2 --- /dev/null +++ b/plugins/modules/systemd_network.py @@ -0,0 +1,157 @@ +#!/usr/bin/python3 +import pathlib +from typing import List, Union + +from ansible.module_utils.generic import SYSTEMD_NETWORK_CONFIG as SYSTEMD_PATH +from ansible.module_utils.generic import Types +from ansible.module_utils.module import SystemdUnitModule + + +def boolconvert(b: Union[bool, str]) -> str: + if b is True: + return "yes" + elif b is False: + return "no" + return b + + +class Module(SystemdUnitModule): + """Sets up the systemd network unit""" + + name = "systemd_network" + module_spec = dict( + argument_spec=dict( + mac=Types.str(), + device=Types.str(), + name=Types.str(required=True), + description=Types.str(), + dot=Types.bool(), + dnssec=Types.bool(), + dns=Types.list(elements=str), + domain=Types.list(elements=str), + defaultdns=Types.bool(), + address=Types.list(elements=str, required=True), + route=Types.list(elements=str), + ), + required_if=(("defaultdns", True, ("dns",), False),), + required_one_of=(("mac", "device"),), + ) + + def prepare(self): + self.unitfile = SYSTEMD_PATH.joinpath(self.get("name")).with_suffix(".network") + self.__unit = None + + def unit(self) -> str: + if self.__unit is None: + self.__unit = "\n".join( + ( + self.match(), + self.network(), + self.addresses(), + self.routes(), + ) + ) + return self.__unit + + def match(self) -> str: + matches = [] + if self.get("mac", False): + matches.append("MACAddress={}\n".format(self.get("mac"))) + if self.get("device", False): + matches.append("Name={}\n".format(self.get("device"))) + return "[Match]\n" + "".join(matches) + + def network(self) -> str: + output = "[Network]\n" + options = [] + try: + options.append("Description={}".format(self.get("description"))) + except KeyError: + pass + try: + for server in self.get("dns", []): + options.append(f"DNS={server}") + options.append("DNSDefaultRoute={}".format(self.get("defaultdns", False))) + except KeyError: + pass + try: + domain = self.get("domain") + self.set("domainlog", str(domain)) + options.append("Domains={}".format(" ".join(domain))) + options.append( + "DNSOverTLS={}".format(boolconvert(self.get("dot", "opportunistic"))) + ) + options.append( + "DNSSEC={}".format(boolconvert(self.get("dnssec", "allow-downgrade"))) + ) + except KeyError: + pass + output += "\n".join(options) + return output + + def addresses(self) -> str: + output = [] + for address in self.get("address"): + output.append(f"[Address]\nAddress={address}\n") + return "\n".join(output) + + def routes(self) -> str: + output = [] + routes = self.get("route", []) + self.set("routes", routes) + for gw in routes: + output.append(f"[Route]\nGateway={gw}\nGatewayOnLink=yes\nQuickAck=yes\n") + self.set("routes", output) + return "\n".join(output) + + +DOCUMENTATION = """--- +description: +- Sets up the systemd network unit +module: systemd_network +options: + address: + elements: str + required: true + type: list + defaultdns: + required: false + type: bool + description: + required: false + type: str + dns: + default: [] + elements: str + required: false + type: list + dnssec: + required: false + type: bool + domain: + default: [] + elements: str + required: false + type: list + dot: + required: false + type: bool + name: + required: true + type: str + mac: + elements: str + required: true + type: list + route: + default: [] + elements: str + required: false + type: list +short_description: Sets up the systemd network unit + +""" + + +if __name__ == "__main__": + Module()() diff --git a/tests/unit/plugins/module_utils/generic/__init__.py b/tests/unit/plugins/module_utils/generic/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/plugins/module_utils/generic/test_generic.py b/tests/unit/plugins/module_utils/generic/test_generic.py new file mode 100644 index 0000000..6a1c699 --- /dev/null +++ b/tests/unit/plugins/module_utils/generic/test_generic.py @@ -0,0 +1,40 @@ +import os +import unittest + +try: # pragma: nocover + from ansible_collections.sebastian.systemd.plugins.module_utils.generic import \ + Types +except ImportError: # pragma: nocover + import sys + + sys.path.append("plugins/module_utils") + from generic import Types + + +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.assertEquals(output["type"], "str") + self.assertEquals(Types.str.__name__, "str") + self.assertEquals(output["required"], True) + self.assertEquals(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.assertEquals(output["type"], "list") + self.assertEquals(output["required"], False) + self.assertEquals(output["elements"], "str")