Core API

SimNOS Class

SimNOS class is a main entry point to interact with SimNOS servers - start, stop, list.

:param inventory: SimNOS inventory dictionary or OS path to .yaml file with inventory data :param plugins: Plugins to add extra devices/commands currently not supported easily.

Sample usage:

from simnos import SimNOS

net = SimNOS()
net.start()

Source code in simnos/core/simnos.py

class SimNOS:
    """
    SimNOS class is a main entry point to interact
    with SimNOS servers - start, stop, list.

    :param inventory: SimNOS inventory dictionary or
                      OS path to .yaml file with inventory data
    :param plugins: Plugins to add extra devices/commands
                    currently not supported easily.

    Sample usage:

    ```python
    from simnos import SimNOS

    net = SimNOS()
    net.start()
    ```
    """

    def __init__(
        self,
        inventory: dict | str | None = None,
        plugins: list | None = None,
    ) -> None:
        self.inventory: dict | str = inventory or default_inventory
        self.plugins: list = plugins or []

        self.hosts: dict[str, Host] = {}
        self.allocated_ports: set[int] = set()

        self.shell_plugins = shell_plugins
        self.nos_plugins = nos_plugins
        self.servers_plugins = servers_plugins

        self._load_inventory()
        self._init()
        self._register_nos_plugins()

    def __enter__(self):
        """
        Method to start the SimNOS servers when entering the context manager.
        It is meant to be used with the `with` statement.
        """
        self.start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Method to stop the SimNOS servers when exiting the context manager.
        It is meant to be used with the `with` statement.

        If ``stop()`` itself raises, the traceback is logged via
        ``log.exception`` (so it is not lost) and the exception is re-raised
        to preserve normal context-manager semantics. This prevents
        silent failures during shutdown.
        """
        try:
            self.stop()
        except Exception:
            log.exception("stop() failed during SimNOS context manager __exit__")
            raise

    def _is_inventory_in_yaml(self) -> bool:
        """method that checks if the inventory is a yaml file."""
        return isinstance(self.inventory, str) and self.inventory.endswith((".yaml", ".yml"))

    def _load_inventory_yaml(self) -> None:
        """Helper method to load SimNOS inventory if it is yaml."""
        # narrow for ty; _is_inventory_in_yaml() guarantees self.inventory is a str.
        assert isinstance(self.inventory, str)  # noqa: S101 — caller-side invariant
        with open(self.inventory, encoding="utf-8") as f:
            self.inventory = yaml.safe_load(f.read())

    def _load_inventory(self) -> None:
        """Helper method to load SimNOS inventory"""
        if self._is_inventory_in_yaml():
            self._load_inventory_yaml()

        if isinstance(self.inventory, str):
            raise ValueError(f"Inventory file must end with .yaml or .yml, got '{self.inventory}'")
        if not isinstance(self.inventory, dict):
            raise TypeError(f"Inventory must be a dict or a path to a YAML file, got {type(self.inventory).__name__}")

        self.inventory["default"] = {
            **default_inventory["default"],
            **self.inventory.get("default", {}),
        }
        ModelSimnosInventory(**self.inventory)
        log.debug("SimNOS inventory validation succeeded")

    def _init(self) -> None:
        """
        Helper method to initiate host objects
        and store them in self.hosts, this
        method called automatically on SimNOS object instantiation.
        """
        # narrow for ty; _load_inventory() ran first and resolved/validated the
        # inventory to a dict (str paths loaded, non-dict raised), so it is a dict here.
        assert isinstance(self.inventory, dict)  # noqa: S101 — caller-side invariant
        for host_name, host_config in self.inventory["hosts"].items():
            params = {
                **copy.deepcopy(self.inventory["default"]),
                **copy.deepcopy(host_config),
            }
            port: int | list[int] = params.pop("port")
            replicas: int | None = params.pop("replicas", None)
            self._check_ports_and_replicas(port, replicas)
            self._instantiate_host_object(host_name, port, replicas, params)

    def _check_ports_and_replicas(self, port: int | list[int], replicas: int | None) -> None:
        """
        Method to check if the port and replicas are valid

        :param port: integer or list of two integers - port to allocate
        :param replicas: integer - number of hosts to create
        """
        # `is None` (not `not replicas`) so that replicas=0 is treated as "set
        # with invalid value" and reaches the `replicas < 1` check below. T-12
        # PR #211 recorded this as a known quirk; #220 promotes the check.
        # Guard cascade also lets ty narrow `port` to list[int] after the
        # initial isinstance check.
        if replicas is None:
            if isinstance(port, list):
                raise ValueError("If replicas is not set, port must be an integer.")
            return  # port is int, no further check needed
        if replicas < 1:
            raise ValueError("If replicas is set, replicas must be greater than 0.")
        if not isinstance(port, list):
            raise ValueError("If replicas is set, port must be a list of two integers.")
        if len(port) != 2:
            raise ValueError("If replicas is set, port must be a list of two integers.")
        if port[0] >= port[1]:
            raise ValueError("If replicas is set, port[0] must be less than port[1].")
        if port[1] - port[0] + 1 != replicas:
            raise ValueError("If replicas is set, port range must be equal to the number of replicas.")

    def _instantiate_host_object(
        self, host_name: str, port: int | list[int], replicas: int | None, params: dict
    ) -> None:
        """
        Method that instantiate the host objects. It initializes the hosts
        with the corresponding name, port and network operating system

        :param host_name: string - name of the host
        :param port: integer or list of two integers - port to allocate
        :param replicas: integer - number of hosts to create
        :param params: dictionary - parameters to pass to
                                    the host like configurations
        """
        hosts_name, ports = self._get_hosts_and_ports(host_name, port, replicas)
        for h_name, p in zip(hosts_name, ports, strict=True):
            self._instantiate_single_host_object(h_name, p, params)

    def _get_hosts_and_ports(
        self, host_name: str, port: int | list[int], replicas: int | None = None
    ) -> tuple[list[str], list[int]]:
        """
        Method to get hosts and ports correctly
        depending on the number of replicas (if exists).

        Pre-condition: ``_check_ports_and_replicas`` already validated that
        when ``replicas`` is truthy, ``port`` is a list[int] of length 2,
        and otherwise ``port`` is an int. The ``isinstance`` assertions
        below narrow ``port`` for ty without changing runtime behavior.

        :param host_name: string - name of the host
        :param port: integer or list of two integers - port to allocate
        :param replicas: integer - number of hosts to create
        """
        if replicas:
            assert isinstance(port, list)  # noqa: S101 — caller-side invariant from _check_ports_and_replicas
            hosts_name = [f"{host_name}{i}" for i in range(replicas)]
            ports = list(range(port[0], port[1] + 1))
        else:
            assert isinstance(port, int)  # noqa: S101 — caller-side invariant
            hosts_name = [host_name]
            ports = [port]
        return hosts_name, ports

    def _instantiate_single_host_object(self, host: str, port: int, params: dict) -> None:
        """
        Method that instantiates a single host object.

        :param host: name of the host
        :param port: port to allocate
        :param params: parameters to pass to the host like configurations
        """
        self._allocate_port(port)
        self.hosts[host] = Host(name=host, port=port, simnos=self, **params)

    def _allocate_port(self, port: int | list[int]) -> None:
        """
        Method to allocate port for host

        :param port: integer or list of two integers -
                     range to allocate port from
        """
        if isinstance(port, int):
            port: list[int] = [port]

        for p in port:
            self._allocate_port_single(p)

    def _allocate_port_single(self, port: int) -> None:
        """
        Method to allocate single port for host.

        :param port: integer - port to allocate
        """
        if not (0 < port <= 65535):
            raise ValueError(f"Port {port} out of valid range (1-65535)")
        if port in self.allocated_ports:
            raise ValueError(f"Port {port} already in use")
        self.allocated_ports.add(port)

    def _get_hosts_as_list(self, hosts: str | list[str] | None = None) -> list[Host]:
        """
        Helper method to get hosts as list

        :param hosts: string or list of strings
        :return: list of Host objects
        """
        if not hosts:
            hosts = list(self.hosts.keys())
        if isinstance(hosts, str):
            hosts = [hosts]
        return [self.hosts[host] for host in hosts]

    def start(
        self,
        hosts: str | list[str] | None = None,
        parallel: bool = False,
        workers: int | None = None,
    ) -> None:
        """
        Function to start NOS servers instances

        :param hosts: single or list of hosts to start by their name.
        :param parallel: if True, start hosts in parallel using threads.
        :param workers: max number of worker threads (default: min(32, host_count)).
        """
        hosts: list[Host] = self._get_hosts_as_list(hosts)
        self._execute_function_over_hosts(
            hosts,
            "start",
            host_running=False,
            parallel=parallel,
            workers=workers,
        )
        log.info(
            "The following devices have been initiated: %s",
            [host.name for host in hosts],
        )
        for host in hosts:
            log.info("Device %s is running on port %s", host.name, host.port)
            self._warn_security(host)

    def stop(
        self,
        hosts: str | list[str] | None = None,
        parallel: bool = False,
        workers: int | None = None,
    ) -> None:
        """
        Function to stop NOS servers instances and join managed threads.

        Uses a global deadline (SHUTDOWN_GLOBAL_DEADLINE seconds) to bound the
        total wall-clock time.  If the deadline is exceeded, remaining hosts
        may be left running and a warning is logged.

        :param hosts: single or list of hosts to stop by their name.
        :param parallel: if True, stop hosts in parallel using threads.
        :param workers: max number of worker threads (default: min(32, host_count)).
        """
        deadline = time.monotonic() + SHUTDOWN_GLOBAL_DEADLINE
        hosts: list[Host] = self._get_hosts_as_list(hosts)
        # Collect managed threads before stopping (Host.stop sets server to None)
        managed_threads = self._collect_server_threads(hosts)
        self._execute_function_over_hosts(
            hosts,
            "stop",
            host_running=True,
            parallel=parallel,
            workers=workers,
            deadline=deadline,
        )
        if managed_threads:
            remaining = max(0, deadline - time.monotonic())
            self._join_threads(managed_threads, timeout=min(SHUTDOWN_SAFETY_NET_DEADLINE, remaining))

    def _collect_server_threads(self, hosts: list[Host]) -> list[threading.Thread]:
        """Collect all managed threads from host servers before stopping."""
        threads: list[threading.Thread] = []
        for host in hosts:
            if host.server is not None:
                threads.extend(host.server.managed_threads)
        return threads

    def _join_threads(
        self,
        threads: list[threading.Thread],
        timeout: float | None = None,
    ) -> None:
        """
        Join SimNOS-managed threads after all hosts are stopped.
        Server threads are already joined by TCPServerBase.stop();
        this is a safety net for any stragglers.
        """
        total = timeout if timeout is not None else SHUTDOWN_SAFETY_NET_DEADLINE
        alive = join_threads_with_deadline(threads, total, SHUTDOWN_SAFETY_NET_PER_THREAD)
        if alive:
            log.warning("%d SimNOS thread(s) did not exit within timeout", len(alive))

    def _execute_function_over_hosts(
        self,
        hosts: list[Host],
        func: str,
        host_running: bool = True,
        parallel: bool = False,
        workers: int | None = None,
        deadline: float | None = None,
    ) -> None:
        """
        Function that executes a function like start or stop over
        the selected hosts.

        :param hosts: list of Hosts objects in which the function will
        be executed.
        :param parallel: if True, execute in parallel using threads.
        :param workers: max number of worker threads.
        :param deadline: optional monotonic deadline; skip remaining hosts if exceeded.
        """
        for host in hosts:
            if host not in self.hosts.values():
                raise ValueError(f"Host {host} not found")
        targets = [h for h in hosts if h.running == host_running]
        if not parallel or len(targets) <= 1:
            for i, h in enumerate(targets):
                if deadline is not None and time.monotonic() >= deadline:
                    log.warning("Global stop deadline exceeded, %d host(s) not stopped", len(targets) - i)
                    break
                getattr(h, func)()
            return
        if workers is not None and workers < 1:
            raise ValueError(f"workers must be >= 1, got {workers}")
        max_workers = workers or min(32, len(targets))
        remaining = max(0, deadline - time.monotonic()) if deadline is not None else None
        ex = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)
        futures = [ex.submit(getattr(h, func)) for h in targets]
        timed_out = False
        try:
            for f in concurrent.futures.as_completed(futures, timeout=remaining):
                f.result()
        except TimeoutError:
            timed_out = True
            log.warning("Global stop deadline exceeded during parallel %s", func)
        finally:
            if timed_out:
                ex.shutdown(wait=False, cancel_futures=True)
            else:
                ex.shutdown(wait=True)

    @staticmethod
    def _warn_security(host: Host) -> None:
        """Emit warnings for common security misconfigurations."""
        if host.username == "user" and host.password == "user":  # noqa: S105
            log.warning(
                "Device %s uses default credentials (user/user). "
                "Change username/password in the inventory for non-local use.",
                host.name,
            )
        address = host.server_inventory.get("configuration", {}).get("address", "")
        if address == "0.0.0.0":  # noqa: S104
            log.warning(
                "Device %s binds to 0.0.0.0 (all interfaces). Use 127.0.0.1 to restrict access to localhost only.",
                host.name,
            )

    def _register_nos_plugins(self) -> None:
        """
        Method to register NOS plugins with SimNOS object, all plugins
        must be registered before calling start method.
        """
        for plugin in self.plugins:
            if isinstance(plugin, Nos):
                nos_instance = plugin
            else:
                nos_instance = Nos()
                if isinstance(plugin, dict):
                    nos_instance.from_dict(plugin)
                elif isinstance(plugin, str):
                    nos_instance.from_file(plugin)
                else:
                    raise TypeError(f"Unsupported NOS type {type(plugin)}, supported str, dict or Nos")
            self.nos_plugins[nos_instance.name] = nos_instance

__enter__()

Method to start the SimNOS servers when entering the context manager. It is meant to be used with the with statement.

Source code in simnos/core/simnos.py

def __enter__(self):
    """
    Method to start the SimNOS servers when entering the context manager.
    It is meant to be used with the `with` statement.
    """
    self.start()
    return self

__exit__(exc_type, exc_val, exc_tb)

Method to stop the SimNOS servers when exiting the context manager. It is meant to be used with the with statement.

If stop() itself raises, the traceback is logged via log.exception (so it is not lost) and the exception is re-raised to preserve normal context-manager semantics. This prevents silent failures during shutdown.

Source code in simnos/core/simnos.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """
    Method to stop the SimNOS servers when exiting the context manager.
    It is meant to be used with the `with` statement.

    If ``stop()`` itself raises, the traceback is logged via
    ``log.exception`` (so it is not lost) and the exception is re-raised
    to preserve normal context-manager semantics. This prevents
    silent failures during shutdown.
    """
    try:
        self.stop()
    except Exception:
        log.exception("stop() failed during SimNOS context manager __exit__")
        raise

start(hosts=None, parallel=False, workers=None)

Function to start NOS servers instances

:param hosts: single or list of hosts to start by their name. :param parallel: if True, start hosts in parallel using threads. :param workers: max number of worker threads (default: min(32, host_count)).

Source code in simnos/core/simnos.py

def start(
    self,
    hosts: str | list[str] | None = None,
    parallel: bool = False,
    workers: int | None = None,
) -> None:
    """
    Function to start NOS servers instances

    :param hosts: single or list of hosts to start by their name.
    :param parallel: if True, start hosts in parallel using threads.
    :param workers: max number of worker threads (default: min(32, host_count)).
    """
    hosts: list[Host] = self._get_hosts_as_list(hosts)
    self._execute_function_over_hosts(
        hosts,
        "start",
        host_running=False,
        parallel=parallel,
        workers=workers,
    )
    log.info(
        "The following devices have been initiated: %s",
        [host.name for host in hosts],
    )
    for host in hosts:
        log.info("Device %s is running on port %s", host.name, host.port)
        self._warn_security(host)

stop(hosts=None, parallel=False, workers=None)

Function to stop NOS servers instances and join managed threads.

Uses a global deadline (SHUTDOWN_GLOBAL_DEADLINE seconds) to bound the total wall-clock time. If the deadline is exceeded, remaining hosts may be left running and a warning is logged.

:param hosts: single or list of hosts to stop by their name. :param parallel: if True, stop hosts in parallel using threads. :param workers: max number of worker threads (default: min(32, host_count)).

Source code in simnos/core/simnos.py

def stop(
    self,
    hosts: str | list[str] | None = None,
    parallel: bool = False,
    workers: int | None = None,
) -> None:
    """
    Function to stop NOS servers instances and join managed threads.

    Uses a global deadline (SHUTDOWN_GLOBAL_DEADLINE seconds) to bound the
    total wall-clock time.  If the deadline is exceeded, remaining hosts
    may be left running and a warning is logged.

    :param hosts: single or list of hosts to stop by their name.
    :param parallel: if True, stop hosts in parallel using threads.
    :param workers: max number of worker threads (default: min(32, host_count)).
    """
    deadline = time.monotonic() + SHUTDOWN_GLOBAL_DEADLINE
    hosts: list[Host] = self._get_hosts_as_list(hosts)
    # Collect managed threads before stopping (Host.stop sets server to None)
    managed_threads = self._collect_server_threads(hosts)
    self._execute_function_over_hosts(
        hosts,
        "stop",
        host_running=True,
        parallel=parallel,
        workers=workers,
        deadline=deadline,
    )
    if managed_threads:
        remaining = max(0, deadline - time.monotonic())
        self._join_threads(managed_threads, timeout=min(SHUTDOWN_SAFETY_NET_DEADLINE, remaining))

Host Class

Host class to build host instances to use with SIMNOS.

Source code in simnos/core/host.py

class Host:
    """
    Host class to build host instances to use with SIMNOS.
    """

    def __init__(
        self,
        name: str,
        username: str,
        password: str,
        port: int,
        server: dict,
        shell: dict,
        nos: dict,
        simnos: "SimNOS",
        platform: str | None = None,
        configuration_file: str | None = None,
    ) -> None:
        self.name: str = name
        self.server_inventory: dict = server
        self.shell_inventory: dict = shell
        self.nos_inventory: dict = nos
        self.username: str = username
        self.password: str = password
        self.port: int = port
        self.simnos = simnos  # SimNOS object
        self.shell_inventory["configuration"].setdefault("base_prompt", self.name)
        self.running = False
        self.server = None
        self.server_plugin = None
        self.shell_plugin = None
        self.nos_plugin = None
        self.nos = None
        self.platform: str | None = platform
        self.configuration_file: str | None = configuration_file

        if self.platform:
            self.nos_inventory["plugin"] = self.platform

        self._validate()

    def start(self):
        """Method to start server instance for this host.

        No-op if the server is already running (``self.running``),
        symmetric with the double-stop guard in ``stop()``. This protects
        direct ``host.start()`` callers from spawning a duplicate server
        instance (and orphaning the first one); the SimNOS-level
        orchestration already filters on ``host_running=False`` and never
        double-starts.
        """
        if self.running:
            log.debug("Host %s is already running; start() is a no-op", self.name)
            return
        self.server_plugin = self.simnos.servers_plugins[self.server_inventory["plugin"]]
        self.shell_plugin = self.simnos.shell_plugins[self.shell_inventory["plugin"]]
        self.nos_plugin = self.simnos.nos_plugins.get(self.nos_inventory["plugin"], self.nos_inventory["plugin"])
        self.nos = (
            Nos(filename=self.nos_plugin, configuration_file=self.configuration_file)
            if not isinstance(self.nos_plugin, Nos)
            else self.nos_plugin
        )
        self.server = self.server_plugin(
            shell=self.shell_plugin,
            shell_configuration=self.shell_inventory["configuration"],
            nos=self.nos,
            nos_inventory_config=self.nos_inventory.get("configuration", {}),
            port=self.port,
            username=self.username,
            password=self.password,
            **self.server_inventory["configuration"],
        )
        self.server.start()
        self.running = True

    def stop(self):
        """Method to stop server instance for this host.

        No-op if the server was never started or has already been stopped
        (``self.server is None``); this guards against double-stop calls.
        """
        if self.server is None:
            log.debug("Host %s has no running server; stop() is a no-op", self.name)
            return
        self.server.stop()
        self.server = None
        self.running = False

    def _validate(self):
        """Validate that the host has the required attributes using pydantic"""
        if self.platform:
            self._check_if_platform_is_supported(self.platform)
        ModelHost(**self.__dict__)

    def _check_if_platform_is_supported(self, platform: str):
        """Check if the platform is supported.

        Thin wrapper around the registry-level helper; kept as a method
        because tests patch / call it as the Host-level seam (#237).
        """
        assert_platform_supported(platform)

start()

Method to start server instance for this host.

No-op if the server is already running (self.running), symmetric with the double-stop guard in stop(). This protects direct host.start() callers from spawning a duplicate server instance (and orphaning the first one); the SimNOS-level orchestration already filters on host_running=False and never double-starts.

Source code in simnos/core/host.py

def start(self):
    """Method to start server instance for this host.

    No-op if the server is already running (``self.running``),
    symmetric with the double-stop guard in ``stop()``. This protects
    direct ``host.start()`` callers from spawning a duplicate server
    instance (and orphaning the first one); the SimNOS-level
    orchestration already filters on ``host_running=False`` and never
    double-starts.
    """
    if self.running:
        log.debug("Host %s is already running; start() is a no-op", self.name)
        return
    self.server_plugin = self.simnos.servers_plugins[self.server_inventory["plugin"]]
    self.shell_plugin = self.simnos.shell_plugins[self.shell_inventory["plugin"]]
    self.nos_plugin = self.simnos.nos_plugins.get(self.nos_inventory["plugin"], self.nos_inventory["plugin"])
    self.nos = (
        Nos(filename=self.nos_plugin, configuration_file=self.configuration_file)
        if not isinstance(self.nos_plugin, Nos)
        else self.nos_plugin
    )
    self.server = self.server_plugin(
        shell=self.shell_plugin,
        shell_configuration=self.shell_inventory["configuration"],
        nos=self.nos,
        nos_inventory_config=self.nos_inventory.get("configuration", {}),
        port=self.port,
        username=self.username,
        password=self.password,
        **self.server_inventory["configuration"],
    )
    self.server.start()
    self.running = True

stop()

Method to stop server instance for this host.

No-op if the server was never started or has already been stopped (self.server is None); this guards against double-stop calls.

Source code in simnos/core/host.py

def stop(self):
    """Method to stop server instance for this host.

    No-op if the server was never started or has already been stopped
    (``self.server is None``); this guards against double-stop calls.
    """
    if self.server is None:
        log.debug("Host %s has no running server; stop() is a no-op", self.name)
        return
    self.server.stop()
    self.server = None
    self.running = False

Nos Class

Base class to build NOS plugins instances to use with SIMNOS.

Source code in simnos/core/nos.py

class Nos:
    """
    Base class to build NOS plugins instances to use with SIMNOS.
    """

    # Mapping of module-level UPPERCASE constants in a Python plugin file
    # to the corresponding lowercase attribute on the Nos instance.
    # Used by `_from_module` to sync module constants into self.
    _MODULE_ATTR_MAP: dict[str, str] = {
        "NAME": "name",
        "INITIAL_PROMPT": "initial_prompt",
        "AUTH": "auth",
        "ENABLE_PROMPT": "enable_prompt",
        "CONFIG_PROMPT": "config_prompt",
    }

    def __init__(
        self,
        name: str = "SimNOS",
        commands: dict | None = None,
        initial_prompt: str = "SimNOS>",
        filename: str | list[str] | None = None,
        configuration_file: str | None = None,
        dict_args: dict | None = None,
    ) -> None:
        """
        Method to instantiate Nos Instance

        :param name: NOS plugin name
        :param commands: dictionary of NOS commands
        :param initial_prompt: NOS initial prompt
        """
        self.name = name
        # Constructor-passed commands are an inflow like any other: keep
        # the caller's dict untouched and normalize the runtime copy
        # (#244 / D3). A non-dict value is passed through for the trailing
        # `validate()` to reject with the usual ValidationError.
        commands = commands or {}
        if isinstance(commands, dict):
            commands = self.normalize_command_prompts(copy.deepcopy(commands))
        self.commands = commands
        self.initial_prompt = initial_prompt
        self.auth: str | None = None
        self.enable_prompt: str | None = None
        self.config_prompt: str | None = None
        self.device = None
        self.configuration_file = configuration_file
        if isinstance(filename, str):
            self.from_file(filename)
        elif isinstance(filename, list):
            for file in filename:
                self.from_file(file)
        elif dict_args:
            self.from_dict(dict_args)

        self.validate()

    def validate(self) -> None:
        """
        Method to validate NOS attributes: commands, name,
        initial prompt - using Pydantic models,
        raises ValidationError on failure.

        Only the schema fields are passed (explicitly extracted via
        `ModelNosAttributes.model_fields`) — `self.__dict__` also holds
        non-schema runtime state (`device`, `configuration_file`) that
        must never reach the model (#244 / D8).
        """
        ModelNosAttributes(**{field: getattr(self, field) for field in ModelNosAttributes.model_fields})
        log.debug("%s NOS attributes validation succeeded", self.name)

    @staticmethod
    def normalize_command_prompts(commands: dict) -> dict:
        """Normalize each command's `prompt` to `list[str] | None`.

        Called on a deepcopied candidate (never on caller-owned dicts or
        module-level `commands` constants). Authoring accepts both a bare
        str (sugar) and a list; runtime consumers see lists only, so
        read-side isinstance branches are unnecessary (#244 / P-12c).
        """
        for cmd in commands.values():
            if not isinstance(cmd, dict):
                # Malformed value — left for the ModelNosAttributes
                # validation to reject (one error surface, not two).
                continue
            prompt = cmd.get("prompt")
            if isinstance(prompt, str):
                cmd["prompt"] = [prompt]
        return commands

    def from_dict(self, data: dict) -> None:
        """
        Method to build NOS from dictionary data.

        The per-command schema follows :class:`simnos.core.pydantic_models.ModelNosCommand`;
        a live Python plugin example is :mod:`simnos.plugins.nos.platforms_py.cisco_ios`.

        Minimal sample::

            nos_plugin_dict = {
                "name": "MySimNOSPlugin",
                "initial_prompt": "{base_prompt}>",
                "commands": {
                    "show clock": {"output": "12:00:00", "help": "Show clock"},
                },
            }

        :param data: NOS dictionary
        :raises ValueError: if the 'commands' value is not a mapping, or
            if `data` holds a top-level key outside the
            `ModelNosAttributes` schema (a typo like `enable_promt` used
            to be dropped silently, #244 / D8)
        :raises pydantic.ValidationError: if the merged result would not
            satisfy `ModelNosAttributes` — validated before any attribute
            is committed, so malformed data never leaves partial state
            behind (same no-partial-state contract as `_from_module`,
            #232); this also covers the hot-reload path, which calls
            `from_file` directly and never reaches `__init__`'s trailing
            `validate()` (#244 / D8)
        """
        unknown = data.keys() - ModelNosAttributes.model_fields.keys()
        if unknown:
            raise ValueError(f"NOS data has unknown top-level field(s): {sorted(unknown)}")
        commands = data.get("commands", {})
        if not isinstance(commands, dict):
            raise ValueError(f"NOS data 'commands' must be a mapping (got {type(commands).__name__})")
        # Normalize a deepcopied candidate — never the caller's dict, which
        # stays in its original authoring form (#244 / D3).
        candidate = self.normalize_command_prompts(copy.deepcopy(commands))
        # Validate the exact post-commit state (a merged view, not `data`
        # alone): commands merge cumulatively across multi-file loads and
        # scalars keep their current value when absent from `data`.
        merged_name = data.get("name", self.name)
        merged_initial_prompt = data.get("initial_prompt", self.initial_prompt)
        merged_auth = data.get("auth", self.auth)
        merged_enable_prompt = data.get("enable_prompt", self.enable_prompt)
        merged_config_prompt = data.get("config_prompt", self.config_prompt)
        ModelNosAttributes(
            name=merged_name,
            initial_prompt=merged_initial_prompt,
            auth=merged_auth,
            enable_prompt=merged_enable_prompt,
            config_prompt=merged_config_prompt,
            commands={**self.commands, **candidate},
        )
        # Commit phase — mirrors the validated merged view (normalized
        # candidate included), so the validated state and the committed
        # state cannot drift apart.
        self.name = merged_name
        self.commands.update(candidate)
        self.initial_prompt = merged_initial_prompt
        self.auth = merged_auth
        self.enable_prompt = merged_enable_prompt
        self.config_prompt = merged_config_prompt

    def _from_yaml(self, filepath: str) -> None:
        """
        Method to build NOS from YAML file.

        The YAML mirrors the dict schema accepted by :meth:`from_dict`;
        see :class:`simnos.core.pydantic_models.ModelNosCommand` for the
        per-command schema and ``simnos/plugins/nos/platforms_yaml/cisco_ios.yaml``
        for a live example.

        :param filepath: OS path to YAML file with NOS data
        :raises ValueError: if the file holds no YAML mapping (empty file
            or a non-dict top level), e.g. a half-written file caught
            mid-save — `yaml.safe_load` returns None for it and
            :meth:`from_dict` would crash on `data.get`
        """
        with open(filepath, encoding="utf-8") as f:
            data = yaml.safe_load(f)
        if not isinstance(data, dict):
            raise ValueError(f"NOS YAML file '{filepath}' does not contain a mapping (got {type(data).__name__})")
        self.from_dict(data)

    def _from_module(self, filename: str) -> None:
        """
        Method to import NOS data from python file or python module.

        Loads from the .py file using the recipe:
        https://docs.python.org/3/library/importlib.html#importing-a-source-file-directly

        The module is expected to define module-level constants (``NAME``,
        ``INITIAL_PROMPT``, optional ``ENABLE_PROMPT`` / ``CONFIG_PROMPT`` /
        ``DEFAULT_CONFIGURATION``) and a ``commands`` dict; the device class
        is auto-detected as the module's single locally-defined `BaseDevice`
        subclass, if any (see `_find_device_classes` — the legacy
        ``DEVICE_NAME`` constant is ignored with a warning since #241).
        See :mod:`simnos.plugins.nos.platforms_py.cisco_ios` for a live example.

        :param filename: OS path string to Python .py file
        """
        spec = importlib.util.spec_from_file_location("module.name", filename)
        if spec is None or spec.loader is None:
            raise ValueError(f"Cannot load NOS module from '{filename}' (spec_from_file_location returned None)")
        module = importlib.util.module_from_spec(spec)
        try:
            spec.loader.exec_module(module)
        except FileNotFoundError:
            # Preserve FileNotFoundError as-is so callers can distinguish
            # "path does not exist" from "plugin code itself failed".
            raise
        except Exception as e:
            raise RuntimeError(f"Failed to load NOS plugin '{filename}': {e}") from e
        # Build/validate everything that can still fail BEFORE mutating self,
        # so a broken plugin raises without leaving partial state behind
        # (#232 cross-review: attrs/commands used to be committed before the
        # device-class validation, so the hot-reload per-file guard skipped
        # the file but a later `commands.update(nos.commands)` leaked the
        # broken plugin's commands into the running shell).
        module_commands = getattr(module, "commands", {})
        if not isinstance(module_commands, dict):
            raise ValueError(f"Module '{filename}' 'commands' must be a mapping (got {type(module_commands).__name__})")
        # Normalize a deepcopied candidate — never the module-level
        # `commands` constant, which stays in its authoring form (#244 /
        # D3; deepcopy treats callables as atomic, so handler identity is
        # preserved).
        candidate_commands = self.normalize_command_prompts(copy.deepcopy(module_commands))
        # Validate the exact post-commit state before committing, mirroring
        # `from_dict`'s merged view (#244 / D8) — this also covers hot
        # reload, which calls `from_file` directly. No top-level key check
        # here: unrelated module-level names are legitimate in a py plugin
        # (only `_MODULE_ATTR_MAP` constants are mapped).
        ModelNosAttributes(
            **{
                self_attr: getattr(module, module_attr, getattr(self, self_attr))
                for module_attr, self_attr in self._MODULE_ATTR_MAP.items()
            },
            commands={**self.commands, **candidate_commands},
        )
        device_classes = _find_device_classes(module)
        if len(device_classes) > 1:
            raise ValueError(
                f"Module '{filename}' defines multiple BaseDevice subclasses "
                f"({', '.join(sorted(c.__name__ for c in device_classes))}); expected exactly one"
            )
        if hasattr(module, "DEVICE_NAME"):
            # G4 (#241) removed the DEVICE_NAME indirection; the device class
            # is auto-detected now. Nudge plugin authors to drop the leftover.
            log.warning(
                "Module '%s' still defines DEVICE_NAME; it is deprecated and ignored (auto-detection is used)",
                filename,
            )
        device = None
        if device_classes:
            configuration_file = self.configuration_file or getattr(module, "DEFAULT_CONFIGURATION", None)
            device = device_classes[0](configuration_file=configuration_file)
        else:
            log.warning("Module '%s' defines no BaseDevice subclass; no device will be set from this module", filename)
        # Commit phase — nothing below is expected to raise.
        for module_attr, self_attr in self._MODULE_ATTR_MAP.items():
            setattr(self, self_attr, getattr(module, module_attr, getattr(self, self_attr)))
        # P-7 (#241): a py module replaces same-named already-loaded
        # commands wholesale (typically yaml-defined ones, but multi-file
        # py loads count too; per-command full replacement, no deep
        # merge) — make the implicit precedence observable for authors.
        overridden = self.commands.keys() & candidate_commands.keys()
        if overridden:
            log.debug(
                "module '%s' overrides %d already-loaded command(s): %s",
                filename,
                len(overridden),
                sorted(overridden),
            )
        self.commands.update(candidate_commands)
        if self.name == "SimNOS":
            log.warning(
                "Module '%s' does not define NAME; falling back to default 'SimNOS' "
                "(plugin will be registered under that key)",
                filename,
            )
        if device_classes:
            # Only a detected class updates the device — a no-device module
            # keeps whatever a previously loaded file set (same "existing
            # device survives" semantics as the pre-#241
            # `if classname is not None` commit).
            self.device = device

    def from_file(self, filename: str) -> None:
        """
        Method to load NOS from YAML or Python file

        :param filename: OS path string to `.yaml/.yml` or `.py` file with NOS data
        """
        if not self._is_file_ending_correct(filename):
            raise ValueError(f'Unsupported "{filename}" file extension. Supported: .py, .yml, .yaml')
        if not os.path.isfile(filename):
            raise FileNotFoundError(filename)
        if filename.endswith((".yaml", ".yml")):
            self._from_yaml(filename)
        elif filename.endswith(".py"):
            self._from_module(filename)

    def _is_file_ending_correct(self, filename: str) -> bool:
        """
        Method to check if file extension is supported.
        Supported types are: .yaml, .yml and .py
        """
        return filename.endswith((".yaml", ".yml", ".py"))

__init__(name='SimNOS', commands=None, initial_prompt='SimNOS>', filename=None, configuration_file=None, dict_args=None)

Method to instantiate Nos Instance

:param name: NOS plugin name :param commands: dictionary of NOS commands :param initial_prompt: NOS initial prompt

Source code in simnos/core/nos.py

def __init__(
    self,
    name: str = "SimNOS",
    commands: dict | None = None,
    initial_prompt: str = "SimNOS>",
    filename: str | list[str] | None = None,
    configuration_file: str | None = None,
    dict_args: dict | None = None,
) -> None:
    """
    Method to instantiate Nos Instance

    :param name: NOS plugin name
    :param commands: dictionary of NOS commands
    :param initial_prompt: NOS initial prompt
    """
    self.name = name
    # Constructor-passed commands are an inflow like any other: keep
    # the caller's dict untouched and normalize the runtime copy
    # (#244 / D3). A non-dict value is passed through for the trailing
    # `validate()` to reject with the usual ValidationError.
    commands = commands or {}
    if isinstance(commands, dict):
        commands = self.normalize_command_prompts(copy.deepcopy(commands))
    self.commands = commands
    self.initial_prompt = initial_prompt
    self.auth: str | None = None
    self.enable_prompt: str | None = None
    self.config_prompt: str | None = None
    self.device = None
    self.configuration_file = configuration_file
    if isinstance(filename, str):
        self.from_file(filename)
    elif isinstance(filename, list):
        for file in filename:
            self.from_file(file)
    elif dict_args:
        self.from_dict(dict_args)

    self.validate()

from_dict(data)

Method to build NOS from dictionary data.

The per-command schema follows :class:simnos.core.pydantic_models.ModelNosCommand; a live Python plugin example is :mod:simnos.plugins.nos.platforms_py.cisco_ios.

Minimal sample::

nos_plugin_dict = {
    "name": "MySimNOSPlugin",
    "initial_prompt": "{base_prompt}>",
    "commands": {
        "show clock": {"output": "12:00:00", "help": "Show clock"},
    },
}

:param data: NOS dictionary :raises ValueError: if the 'commands' value is not a mapping, or if data holds a top-level key outside the ModelNosAttributes schema (a typo like enable_promt used to be dropped silently, #244 / D8) :raises pydantic.ValidationError: if the merged result would not satisfy ModelNosAttributes — validated before any attribute is committed, so malformed data never leaves partial state behind (same no-partial-state contract as _from_module, #232); this also covers the hot-reload path, which calls from_file directly and never reaches __init__'s trailing validate() (#244 / D8)

Source code in simnos/core/nos.py

def from_dict(self, data: dict) -> None:
    """
    Method to build NOS from dictionary data.

    The per-command schema follows :class:`simnos.core.pydantic_models.ModelNosCommand`;
    a live Python plugin example is :mod:`simnos.plugins.nos.platforms_py.cisco_ios`.

    Minimal sample::

        nos_plugin_dict = {
            "name": "MySimNOSPlugin",
            "initial_prompt": "{base_prompt}>",
            "commands": {
                "show clock": {"output": "12:00:00", "help": "Show clock"},
            },
        }

    :param data: NOS dictionary
    :raises ValueError: if the 'commands' value is not a mapping, or
        if `data` holds a top-level key outside the
        `ModelNosAttributes` schema (a typo like `enable_promt` used
        to be dropped silently, #244 / D8)
    :raises pydantic.ValidationError: if the merged result would not
        satisfy `ModelNosAttributes` — validated before any attribute
        is committed, so malformed data never leaves partial state
        behind (same no-partial-state contract as `_from_module`,
        #232); this also covers the hot-reload path, which calls
        `from_file` directly and never reaches `__init__`'s trailing
        `validate()` (#244 / D8)
    """
    unknown = data.keys() - ModelNosAttributes.model_fields.keys()
    if unknown:
        raise ValueError(f"NOS data has unknown top-level field(s): {sorted(unknown)}")
    commands = data.get("commands", {})
    if not isinstance(commands, dict):
        raise ValueError(f"NOS data 'commands' must be a mapping (got {type(commands).__name__})")
    # Normalize a deepcopied candidate — never the caller's dict, which
    # stays in its original authoring form (#244 / D3).
    candidate = self.normalize_command_prompts(copy.deepcopy(commands))
    # Validate the exact post-commit state (a merged view, not `data`
    # alone): commands merge cumulatively across multi-file loads and
    # scalars keep their current value when absent from `data`.
    merged_name = data.get("name", self.name)
    merged_initial_prompt = data.get("initial_prompt", self.initial_prompt)
    merged_auth = data.get("auth", self.auth)
    merged_enable_prompt = data.get("enable_prompt", self.enable_prompt)
    merged_config_prompt = data.get("config_prompt", self.config_prompt)
    ModelNosAttributes(
        name=merged_name,
        initial_prompt=merged_initial_prompt,
        auth=merged_auth,
        enable_prompt=merged_enable_prompt,
        config_prompt=merged_config_prompt,
        commands={**self.commands, **candidate},
    )
    # Commit phase — mirrors the validated merged view (normalized
    # candidate included), so the validated state and the committed
    # state cannot drift apart.
    self.name = merged_name
    self.commands.update(candidate)
    self.initial_prompt = merged_initial_prompt
    self.auth = merged_auth
    self.enable_prompt = merged_enable_prompt
    self.config_prompt = merged_config_prompt

from_file(filename)

Method to load NOS from YAML or Python file

:param filename: OS path string to .yaml/.yml or .py file with NOS data

Source code in simnos/core/nos.py

def from_file(self, filename: str) -> None:
    """
    Method to load NOS from YAML or Python file

    :param filename: OS path string to `.yaml/.yml` or `.py` file with NOS data
    """
    if not self._is_file_ending_correct(filename):
        raise ValueError(f'Unsupported "{filename}" file extension. Supported: .py, .yml, .yaml')
    if not os.path.isfile(filename):
        raise FileNotFoundError(filename)
    if filename.endswith((".yaml", ".yml")):
        self._from_yaml(filename)
    elif filename.endswith(".py"):
        self._from_module(filename)

normalize_command_prompts(commands) staticmethod

Normalize each command's prompt to list[str] | None.

Called on a deepcopied candidate (never on caller-owned dicts or module-level commands constants). Authoring accepts both a bare str (sugar) and a list; runtime consumers see lists only, so read-side isinstance branches are unnecessary (#244 / P-12c).

Source code in simnos/core/nos.py

@staticmethod
def normalize_command_prompts(commands: dict) -> dict:
    """Normalize each command's `prompt` to `list[str] | None`.

    Called on a deepcopied candidate (never on caller-owned dicts or
    module-level `commands` constants). Authoring accepts both a bare
    str (sugar) and a list; runtime consumers see lists only, so
    read-side isinstance branches are unnecessary (#244 / P-12c).
    """
    for cmd in commands.values():
        if not isinstance(cmd, dict):
            # Malformed value — left for the ModelNosAttributes
            # validation to reject (one error surface, not two).
            continue
        prompt = cmd.get("prompt")
        if isinstance(prompt, str):
            cmd["prompt"] = [prompt]
    return commands

validate()

Method to validate NOS attributes: commands, name, initial prompt - using Pydantic models, raises ValidationError on failure.

Only the schema fields are passed (explicitly extracted via ModelNosAttributes.model_fields) — self.__dict__ also holds non-schema runtime state (device, configuration_file) that must never reach the model (#244 / D8).

Source code in simnos/core/nos.py

def validate(self) -> None:
    """
    Method to validate NOS attributes: commands, name,
    initial prompt - using Pydantic models,
    raises ValidationError on failure.

    Only the schema fields are passed (explicitly extracted via
    `ModelNosAttributes.model_fields`) — `self.__dict__` also holds
    non-schema runtime state (`device`, `configuration_file`) that
    must never reach the model (#244 / D8).
    """
    ModelNosAttributes(**{field: getattr(self, field) for field in ModelNosAttributes.model_fields})
    log.debug("%s NOS attributes validation succeeded", self.name)

TCPServerBase Class

Bases: ABC

Base class for a TCP Server. It provides the methods to start and stop the server.

Source code in simnos/core/servers.py

class TCPServerBase(ABC):
    """
    Base class for a TCP Server.
    It provides the methods to start and stop the server.
    """

    def __init__(self, address="localhost", port=6000, timeout=1):
        """
        Initialize the server with the address and port
        and the safety-net timeout for select() (seconds).
        """
        self.address = address
        self.port = port
        self.timeout = timeout
        self._is_running = threading.Event()
        self._socket = None
        self.client_shell = None
        self._listen_thread = None
        self._connection_threads = []
        self._wakeup_r: socket.socket | None = None
        self._wakeup_w: socket.socket | None = None
        self._selector: selectors.BaseSelector | None = None

    def start(self):
        """
        Start Server which distributes the connections.
        It handles the creation of the socket, binding to the address and port,
        and starting the listening thread.
        """
        if self._is_running.is_set():
            return

        self._is_running.set()
        try:
            self._bind_sockets()
            # _bind_sockets() assigns self._socket; narrow for ty.
            assert self._socket is not None  # noqa: S101 — post-condition of _bind_sockets
            self._socket.listen()

            self._wakeup_r, self._wakeup_w = socket.socketpair()
            self._wakeup_r.setblocking(False)

            self._selector = selectors.DefaultSelector()
            self._selector.register(self._socket, selectors.EVENT_READ, data="listen")
            self._selector.register(self._wakeup_r, selectors.EVENT_READ, data="wakeup")

            self._listen_thread = threading.Thread(target=self._listen)
            self._listen_thread.start()
        except Exception:
            self._cleanup_resources()
            self._is_running.clear()
            raise

    def _bind_sockets(self):
        """
        It binds the sockets to the corresponding IPs and Ports.
        In Linux and OSX it reuses the port if needed but
        not in Windows
        """
        self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)

        if sys.platform in ["linux"]:
            self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, True)

        self._socket.setblocking(False)
        self._socket.bind((self.address, self.port))

    @property
    def managed_threads(self) -> list[threading.Thread]:
        """Return all threads managed by this server (listen + connections)."""
        threads = list(self._connection_threads)
        if self._listen_thread is not None:
            threads.append(self._listen_thread)
        return threads

    def stop(self):
        """
        It stops the server joining the threads
        and closing the corresponding sockets.
        """
        if not self._is_running.is_set():
            return

        self._is_running.clear()

        if self._wakeup_w is not None:
            with contextlib.suppress(OSError):
                self._wakeup_w.send(b"\x00")

        # fail-safe join — wakeup socket may have failed; bound by SHUTDOWN_IO_TIMEOUT.
        # _is_running.is_set() guard at the top of stop() returns early if start()
        # hasn't been called, so self._listen_thread is set here; narrow for ty.
        assert self._listen_thread is not None  # noqa: S101 — post-condition of start()
        self._listen_thread.join(timeout=SHUTDOWN_IO_TIMEOUT)

        try:
            alive = join_threads_with_deadline(
                self._connection_threads,
                SHUTDOWN_SERVER_STOP_DEADLINE,
                SHUTDOWN_SERVER_PER_THREAD_JOIN,
            )
            if alive:
                log.warning(
                    "%d connection thread(s) did not exit within %ds",
                    len(alive),
                    SHUTDOWN_SERVER_STOP_DEADLINE,
                )
        finally:
            self._cleanup_resources()

    def _cleanup_resources(self):
        """Safely close selector, wakeup socketpair, and listen socket."""
        if self._selector is not None:
            with contextlib.suppress(Exception):
                self._selector.close()
            self._selector = None
        if self._socket is not None:
            with contextlib.suppress(OSError):
                self._socket.close()
            self._socket = None
        for sock in (self._wakeup_r, self._wakeup_w):
            if sock is not None:
                with contextlib.suppress(OSError):
                    sock.close()
        self._wakeup_r = self._wakeup_w = None

    def _listen(self):
        """
        Wait for connections using selectors.
        A wakeup socketpair allows stop() to unblock select() instantly
        instead of waiting for the timeout to expire.
        """
        # _listen only runs from the thread started by start(), where both
        # self._socket and self._selector are non-None; narrow for ty.
        assert self._selector is not None  # noqa: S101 — post-condition of start()
        assert self._socket is not None  # noqa: S101 — post-condition of start()
        while self._is_running.is_set():
            try:
                # select(timeout) is a safety net. Normal shutdown is
                # signalled via the wakeup socket and returns immediately.
                events = self._selector.select(timeout=self.timeout)
            except (OSError, ValueError):
                break  # selector was closed

            # Check wakeup first: if shutdown and accept fire simultaneously,
            # skip accept and exit immediately.
            for key, _ in events:
                if key.data == "wakeup":
                    return

            # No wakeup — process accept events.
            for key, _ in events:
                if key.data == "listen":
                    try:
                        client, _ = self._socket.accept()
                    except BlockingIOError:
                        continue  # spurious wakeup
                    except OSError:
                        break  # socket was closed (shutdown path)

                    # listen socket is non-blocking, so accepted client
                    # inherits that mode (OS-dependent). Restore blocking
                    # before handing to connection_function (e.g. paramiko).
                    client.setblocking(True)

                    connection_thread = threading.Thread(
                        target=self.connection_function,
                        args=(client, self._is_running),
                    )
                    connection_thread.start()
                    self._connection_threads.append(connection_thread)

            # Prune finished threads to prevent unbounded growth
            self._connection_threads = [t for t in self._connection_threads if t.is_alive()]

    @abstractmethod
    def connection_function(self, client, is_running):
        """
        This abstract method is called when a new connection
        is made. The implementation should handle the
        connection afterwards.
        """

managed_threads property

Return all threads managed by this server (listen + connections).

__init__(address='localhost', port=6000, timeout=1)

Initialize the server with the address and port and the safety-net timeout for select() (seconds).

Source code in simnos/core/servers.py

def __init__(self, address="localhost", port=6000, timeout=1):
    """
    Initialize the server with the address and port
    and the safety-net timeout for select() (seconds).
    """
    self.address = address
    self.port = port
    self.timeout = timeout
    self._is_running = threading.Event()
    self._socket = None
    self.client_shell = None
    self._listen_thread = None
    self._connection_threads = []
    self._wakeup_r: socket.socket | None = None
    self._wakeup_w: socket.socket | None = None
    self._selector: selectors.BaseSelector | None = None

connection_function(client, is_running) abstractmethod

This abstract method is called when a new connection is made. The implementation should handle the connection afterwards.

Source code in simnos/core/servers.py

@abstractmethod
def connection_function(self, client, is_running):
    """
    This abstract method is called when a new connection
    is made. The implementation should handle the
    connection afterwards.
    """

start()

Start Server which distributes the connections. It handles the creation of the socket, binding to the address and port, and starting the listening thread.

Source code in simnos/core/servers.py

def start(self):
    """
    Start Server which distributes the connections.
    It handles the creation of the socket, binding to the address and port,
    and starting the listening thread.
    """
    if self._is_running.is_set():
        return

    self._is_running.set()
    try:
        self._bind_sockets()
        # _bind_sockets() assigns self._socket; narrow for ty.
        assert self._socket is not None  # noqa: S101 — post-condition of _bind_sockets
        self._socket.listen()

        self._wakeup_r, self._wakeup_w = socket.socketpair()
        self._wakeup_r.setblocking(False)

        self._selector = selectors.DefaultSelector()
        self._selector.register(self._socket, selectors.EVENT_READ, data="listen")
        self._selector.register(self._wakeup_r, selectors.EVENT_READ, data="wakeup")

        self._listen_thread = threading.Thread(target=self._listen)
        self._listen_thread.start()
    except Exception:
        self._cleanup_resources()
        self._is_running.clear()
        raise

stop()

It stops the server joining the threads and closing the corresponding sockets.

Source code in simnos/core/servers.py

def stop(self):
    """
    It stops the server joining the threads
    and closing the corresponding sockets.
    """
    if not self._is_running.is_set():
        return

    self._is_running.clear()

    if self._wakeup_w is not None:
        with contextlib.suppress(OSError):
            self._wakeup_w.send(b"\x00")

    # fail-safe join — wakeup socket may have failed; bound by SHUTDOWN_IO_TIMEOUT.
    # _is_running.is_set() guard at the top of stop() returns early if start()
    # hasn't been called, so self._listen_thread is set here; narrow for ty.
    assert self._listen_thread is not None  # noqa: S101 — post-condition of start()
    self._listen_thread.join(timeout=SHUTDOWN_IO_TIMEOUT)

    try:
        alive = join_threads_with_deadline(
            self._connection_threads,
            SHUTDOWN_SERVER_STOP_DEADLINE,
            SHUTDOWN_SERVER_PER_THREAD_JOIN,
        )
        if alive:
            log.warning(
                "%d connection thread(s) did not exit within %ds",
                len(alive),
                SHUTDOWN_SERVER_STOP_DEADLINE,
            )
    finally:
        self._cleanup_resources()