ansible-systemd/plugins/modules/socket.py

#!/usr/bin/python3
import pathlib
from typing import List, Optional, Union

try:
    from ansible_module.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, modspec
    from ansible_module.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable
except ImportError:
    from ansible_collections.sebastian.base.plugins.module_utils.generic import SYSTEMD_SERVICE_CONFIG, Types, modspec
    from ansible_collections.sebastian.base.plugins.module_utils.module import SystemdReloadMixin, SystemdUnitModule, installable


@installable
class Module(SystemdUnitModule, SystemdReloadMixin):  # type: ignore
    """Creates socket units."""

    name = "socket"
    module_spec = modspec(
        argument_spec=dict(
            name=Types.str(required=True, help="Name of the socket"),
            stream=Types.list(
                Types.str(
                    help="Name of the stream socket. The name can be a path, an portnumber or an ip with an port. addresses in square brackets are always ipv6 addresses"
                )
            ),
            datagram=Types.list(
                Types.str(
                    help="Name of the datagram socket. The name can be a path, an portnumber or an ip with an port. addresses in square brackets are always ipv6 addresses"
                )
            ),
            sequential=Types.list(
                Types.str(
                    help="Name of the sequential socket. The name can be a path, an portnumber or an ip with an port. addresses in square brackets are always ipv6 addresses"
                )
            ),
            fifo=Types.list(Types.path(help="Name of the fifo. The name must be an absolute path")),
            socketuser=Types.str(help="User that owns the socket/fifo"),
            socketgroup=Types.str(help="Group that owns the socket/fifo"),
            socketmode=Types.str(help="mode of the socket in octal notation", default="0666"),
            service=Types.list(str, help="Name of the services that use this socket"),
        ),
        required_one_of=(("stream", "datagram", "sequential", "fifo"),),
    )
    restartable = False

    def prepare(self):
        self.unitfile = (SYSTEMD_SERVICE_CONFIG / self.get("name")).with_suffix(".socket")
        self.__unit = None

    def socket(self) -> Optional[str]:
        section = self.map_param(
            stream="ListenStream",
            datagram="ListenDatagram",
            sequential="ListenSequential",
            fifo="ListenFIFO",
            socketuser="SocketUser",
            socketgroup="SocketGroup",
            socketmode="SocketMode",
        )
        if len(section) == 0:
            return None
        return "[Socket]\n" + "".join(section)

    def unit(self) -> str:
        if self.__unit is None:
            self.__unit = self._unit(
                self.header(),
                self.socket(),
                self.install(),  # type: ignore[call-arg,misc]
            )
        return self.__unit


DOCUMENTATION = """---
description:
- Creates socket units.
module: socket
options:
  after:
    default: []
    description:
    - list of units that this unit wants to be started after this unit
    elements: str
    required: false
    type: list
  before:
    default: []
    description:
    - list of units that this unit needs to be started before this unit.
    elements: str
    required: false
    type: list
  datagram:
    default: []
    elements: str
    required: false
    type: list
  description:
    description:
    - An description for programs that access systemd
    required: false
    type: str
  documentation:
    default: []
    description:
    - Paths where documentation can be found
    elements: str
    required: false
    type: list
  fifo:
    default: []
    elements: path
    required: false
    type: list
  name:
    description:
    - Name of the socket
    required: true
    type: str
  partof:
    default: []
    description:
    - list of units that this unit is part of.
    - If the restart this unit does it too, but if this restarts it does not affect
      the other units.
    elements: str
    required: false
    type: list
  required_by:
    default: []
    description:
    - systemd units that require this mount
    elements: str
    required: false
    type: list
  requires:
    default: []
    description:
    - list of units that this unit requires. If it fails or can't be started this
      unit fails. without before/after this is started at the same time
    elements: str
    required: false
    type: list
  sequential:
    default: []
    elements: str
    required: false
    type: list
  service:
    default: []
    description:
    - Name of the services that use this socket
    elements: str
    required: false
    type: list
  socketgroup:
    description:
    - Group that owns the socket/fifo
    required: false
    type: str
  socketmode:
    default: '0666'
    description:
    - mode of the socket in octal notation
    required: false
    type: str
  socketuser:
    description:
    - User that owns the socket/fifo
    required: false
    type: str
  stream:
    default: []
    elements: str
    required: false
    type: list
  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
  wants:
    default: []
    description:
    - list of units that this unit wants. If it fails or can't be started it does
      not affect this unit
    elements: str
    required: false
    type: list
short_description: Creates socket units.
"""

if __name__ == "__main__":
    Module()()