Skip to content

Core API

FakeNOS Class#

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

:param inventory: FakeNOS 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 fakenos import FakeNOS

net = FakeNOS()
net.start()
Source code in fakenos/core/fakenos.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
class FakeNOS:
    """
    FakeNOS class is a main entry point to interact
    with fake NOS servers - start, stop, list.

    :param inventory: FakeNOS 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 fakenos import FakeNOS

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

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

        self.hosts: Dict[str, Host] = {}
        self.allocated_ports: Set[str] = 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 FakeNOS servers when entering the context manager.
        It is meant to be used with the `with` statement.
        """
        self.start()
        return self

    def __exit__(self, *args):
        """
        Method to stop the FakeNOS servers when exiting the context manager.
        It is meant to be used with the `with` statement.
        """
        self.stop()

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

    def _load_inventory_yaml(self) -> None:
        """Helper method to load FakeNOS inventory if it is yaml."""
        with open(self.inventory, "r", encoding="utf-8") as f:
            self.inventory = yaml.safe_load(f.read())

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

        self.inventory["default"] = {
            **default_inventory["default"],
            **self.inventory.get("default", {}),
        }

        ModelFakenosInventory(**self.inventory)
        log.debug("FakeNOS inventory validation succeeded")

    def _init(self) -> None:
        """
        Helper method to initiate host objects
        and store them in self.hosts, this
        method called automatically on FakeNOS object instantiation.
        """
        for host_name, host_config in self.inventory["hosts"].items():
            params = {
                **copy.deepcopy(self.inventory["default"]),
                **copy.deepcopy(host_config),
            }
            port: Union[int, list] = params.pop("port")
            replicas: int = params.pop("replicas", None)
            self._check_ports_and_replicas_are_okey(port, replicas)
            self._instantiate_host_object(host_name, port, replicas, params)

    def _check_ports_and_replicas_are_okey(self, port, replicas):
        """
        Method to check if the port and replicas are okey

        :param port: integer or list of two integers - port to allocate
        :param replicas: integer - number of hosts to create
        """
        if not replicas and isinstance(port, list):
            raise ValueError("If replicas is not set, port must be an integer.")
        if replicas and not isinstance(port, list):
            raise ValueError("If replicas is set, port must be a list of two integers.")
        if replicas and len(port) != 2:
            raise ValueError("If replicas is set, port must be a list of two integers.")
        if replicas and port[0] >= port[1]:
            raise ValueError("If replicas is set, port[0] must be less than port[1].")
        if replicas and replicas < 1:
            raise ValueError("If replicas is set, replicas must be greater than 0.")
        if replicas and 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: Union[int, List[int]], replicas: int, params: dict):
        """
        Method that instantiate the host objects. It initializes the hosts
        with the corresponding name, port and network operating system

        :param host: string - name of the host
        :param port: integer or list of two integers - port to allocate
        :param count: 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):
            self._instantiate_single_host_object(h_name, p, params)

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

        :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
        """
        hosts_name: Set[str] = {}
        ports: Set[int] = {}

        if replicas:
            hosts_name = {f"{host_name}{i}" for i in range(replicas)}
            ports = set(range(port[0], port[1] + 1))
        else:
            hosts_name = {host_name}
            ports = {port}
        return hosts_name, ports

    def _instantiate_single_host_object(self, host, port, params):
        """
        Method that instantiate the host objects. It initializes the hosts

        :param host: string - name of the host
        :param port: integer or list of two integers - port to allocate
        :param params: dictionary - parameters to pass to
                                    the host like configurations
        """
        self._allocate_port(port)
        self.hosts[host] = Host(name=host, port=port, fakenos=self, **params)

    def _allocate_port(self, port: Union[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:
            allocated_port = self._allocate_port_single(p)
            self.allocated_ports.add(allocated_port)

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

        :param port: integer - port to allocate
        """
        if port in self.allocated_ports:
            raise ValueError(f"Port {port} already in use")
        self.allocated_ports.add(port)
        return port

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

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

    def start(self, hosts: Union[str, list] = None) -> None:  # type: ignore
        """
        Function to start NOS servers instances

        :param hosts: single or list of hosts to start by their name.
        """
        hosts: List[str] = self._get_hosts_as_list(hosts)
        self._execute_function_over_hosts(hosts, "start", host_running=False)
        log.info("The following devices has 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)

    def stop(self, hosts: Union[str, List[str]] = None) -> None:
        """
        Function to stop NOS servers instances. It waits 2 seconds
        just in case that there is any thread doing something.

        :param hosts: single or list of hosts to stop by their name.
        """
        hosts: List[str] = self._get_hosts_as_list(hosts)
        self._execute_function_over_hosts(hosts, "stop", host_running=True)
        if hosts == list(self.hosts.values()):
            self._join_threads()

    def _join_threads(self) -> None:
        """
        Method to join threads in case that all hosts are stopped.
        """
        all_threads = threading.enumerate()
        for thread in all_threads:
            if thread is not threading.main_thread() and "pytest_timeout" not in thread.name:
                thread.join()
        n_threads: int = 2 if detect.windows else 1
        while threading.active_count() > n_threads:
            time.sleep(0.01)

    def _execute_function_over_hosts(self, hosts: List[Host], func: str, host_running: bool = True):
        """
        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.
        """
        for host in hosts:
            if host not in self.hosts.values():
                raise ValueError(f"Host {host} not found")
            if host.running == host_running:
                getattr(host, func)()

    def _register_nos_plugins(self) -> None:
        """
        Method to register NOS plugin with FakeNOS object, all plugins
        must be registered before calling start method.

        :param plugin: OS path string to NOS plugin `.yaml/.yml` or `.py` file,
          dictionary or instance if Nos class
        """
        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 FakeNOS servers when entering the context manager. It is meant to be used with the with statement.

Source code in fakenos/core/fakenos.py 
 96
 97
 98
 99
100
101
102
def __enter__(self):
    """
    Method to start the FakeNOS servers when entering the context manager.
    It is meant to be used with the `with` statement.
    """
    self.start()
    return self

__exit__(*args) #

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

Source code in fakenos/core/fakenos.py 
104
105
106
107
108
109
def __exit__(self, *args):
    """
    Method to stop the FakeNOS servers when exiting the context manager.
    It is meant to be used with the `with` statement.
    """
    self.stop()

start(hosts=None) #

Function to start NOS servers instances

:param hosts: single or list of hosts to start by their name.

Source code in fakenos/core/fakenos.py 
259
260
261
262
263
264
265
266
267
268
269
def start(self, hosts: Union[str, list] = None) -> None:  # type: ignore
    """
    Function to start NOS servers instances

    :param hosts: single or list of hosts to start by their name.
    """
    hosts: List[str] = self._get_hosts_as_list(hosts)
    self._execute_function_over_hosts(hosts, "start", host_running=False)
    log.info("The following devices has 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)

stop(hosts=None) #

Function to stop NOS servers instances. It waits 2 seconds just in case that there is any thread doing something.

:param hosts: single or list of hosts to stop by their name.

Source code in fakenos/core/fakenos.py 
271
272
273
274
275
276
277
278
279
280
281
def stop(self, hosts: Union[str, List[str]] = None) -> None:
    """
    Function to stop NOS servers instances. It waits 2 seconds
    just in case that there is any thread doing something.

    :param hosts: single or list of hosts to stop by their name.
    """
    hosts: List[str] = self._get_hosts_as_list(hosts)
    self._execute_function_over_hosts(hosts, "stop", host_running=True)
    if hosts == list(self.hosts.values()):
        self._join_threads()

Host Class#

Host class to build host instances to use with FakeNOS.

Source code in fakenos/core/host.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class Host:
    """
    Host class to build host instances to use with FakeNOS.
    """

    # pylint: disable=too-many-arguments
    # pylint: disable=too-many-instance-attributes
    def __init__(
        self,
        name: str,
        username: str,
        password: str,
        port: int,
        server: dict,
        shell: dict,
        nos: dict,
        fakenos,
        platform: str = None,
        configuration_file: str = 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.fakenos = fakenos  # FakeNOS 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 = platform
        self.configuration_file: str = configuration_file

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

        self._validate()

    def start(self):
        """Method to start server instance for this hosts"""
        self.server_plugin = self.fakenos.servers_plugins[self.server_inventory["plugin"]]
        self.shell_plugin = self.fakenos.shell_plugins[self.shell_inventory["plugin"]]
        if self.platform:
            self.nos_inventory["plugin"] = self.platform
        self.nos_plugin = self.fakenos.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 of this host"""
        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"""
        if platform not in available_platforms:
            raise ValueError(
                f"Platform {platform} is not supported by FakeNOS. \
                    Supported platforms are: {available_platforms}"
            )

start() #

Method to start server instance for this hosts

Source code in fakenos/core/host.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def start(self):
    """Method to start server instance for this hosts"""
    self.server_plugin = self.fakenos.servers_plugins[self.server_inventory["plugin"]]
    self.shell_plugin = self.fakenos.shell_plugins[self.shell_inventory["plugin"]]
    if self.platform:
        self.nos_inventory["plugin"] = self.platform
    self.nos_plugin = self.fakenos.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 of this host

Source code in fakenos/core/host.py 
83
84
85
86
87
def stop(self):
    """Method to stop server instance of this host"""
    self.server.stop()
    self.server = None
    self.running = False

Nos Class#

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

Source code in fakenos/core/nos.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
class Nos:
    """
    Base class to build NOS plugins instances to use with FakeNOS.
    """

    # pylint: disable=too-many-arguments
    def __init__(
        self,
        name: str = "FakeNOS",
        commands: dict = None,
        initial_prompt: str = "FakeNOS>",
        filename: Optional[Union[str, List[str]]] = None,
        configuration_file: Optional[str] = None,
        dict_args: Optional[dict] = 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
        self.commands = commands or {}
        self.initial_prompt = initial_prompt
        self.enable_prompt = None
        self.config_prompt = 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.
        """
        ModelNosAttributes(**self.__dict__)
        log.debug("%s NOS attributes validation succeeded", self.name)

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

        Sample NOS dictionary::

            nos_plugin_dict = {
                "name": "MyFakeNOSPlugin",
                "initial_prompt": "{base_prompt}>",
                "commands": {
                    "terminal width 511": {
                        "output": "",
                        "help": "Set terminal width to 511",
                        "prompt": "{base_prompt}>",
                    },
                    "terminal length 0": {
                        "output": "",
                        "help": "Set terminal length to 0",
                        "prompt": "{base_prompt}>",
                    },
                    "show clock": {
                        "output": "MyFakeNOSPlugin system time is 00:00:00",
                        "help": "Show system time",
                        "prompt": "{base_prompt}>",
                    },
                },
            }

        :param data: NOS dictionary
        """
        self.name = data.get("name", self.name)
        self.commands.update(data.get("commands", self.commands))
        self.initial_prompt = data.get("initial_prompt", self.initial_prompt)

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

        Sample NOS YAML file content::

            name: "MyFakeNOSPlugin"
            initial_prompt: "{base_prompt}>"
            commands:
                terminal width 511: {
                    "output": "",
                    "help": "Set terminal width to 511",
                    "prompt": "{base_prompt}>",
                }
                terminal length 0: {
                    "output": "",
                    "help": "Set terminal length to 0",
                    "prompt": "{base_prompt}>",
                }
                show clock: {
                    "output": "MyFakeNOSPlugin system time is 00:00:00",
                    "help": "Show system time",
                    "prompt": "{base_prompt}>",
                }

        :param data: YAML structured text
        """
        with open(data, "r", encoding="utf-8") as f:
            self.from_dict(yaml.safe_load(f))

    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

        Sample Python NOS plugin file::

            name = "MyFakeNOSPlugin"

            INITIAL_PROMPT = "{base_prompt}>"

            commands = {
                "terminal width 511": {
                    "output": "",
                    "help": "Set terminal width to 511",
                    "prompt": "{base_prompt}>",
                },
                "terminal length 0": {
                    "output": "",
                    "help": "Set terminal length to 0",
                    "prompt": "{base_prompt}>",
                },
                "show clock": {
                    "output": "MyFakeNOSPlugin system time is 00:00:00",
                    "help": "Show system time",
                    "prompt": "{base_prompt}>",
                },
            }

        :param data: OS path string to Python .py file
        """
        spec = importlib.util.spec_from_file_location("module.name", filename)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        self.name = getattr(module, "NAME", self.name)
        self.commands.update(getattr(module, "commands", self.commands))
        self.initial_prompt = getattr(module, "INITIAL_PROMPT", self.initial_prompt)
        self.enable_prompt = getattr(module, "ENABLE_PROMPT", None)
        self.config_prompt = getattr(module, "CONFIG_PROMPT", None)
        classname = getattr(module, "DEVICE_NAME", None)
        configuration_file = self.configuration_file
        if not self.configuration_file:
            configuration_file = getattr(module, "DEFAULT_CONFIGURATION", None)
        self.device = getattr(module, classname)(configuration_file=configuration_file)

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

        :param data: 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) -> None:
        """
        Method to check if file extension is correct and load NOS data.
        Correct types are: .yaml, .yml and .py
        """
        return filename.endswith((".yaml", ".yml", ".py"))

__init__(name='FakeNOS', commands=None, initial_prompt='FakeNOS>', 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 fakenos/core/nos.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def __init__(
    self,
    name: str = "FakeNOS",
    commands: dict = None,
    initial_prompt: str = "FakeNOS>",
    filename: Optional[Union[str, List[str]]] = None,
    configuration_file: Optional[str] = None,
    dict_args: Optional[dict] = 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
    self.commands = commands or {}
    self.initial_prompt = initial_prompt
    self.enable_prompt = None
    self.config_prompt = 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.

Sample NOS dictionary::

nos_plugin_dict = {
    "name": "MyFakeNOSPlugin",
    "initial_prompt": "{base_prompt}>",
    "commands": {
        "terminal width 511": {
            "output": "",
            "help": "Set terminal width to 511",
            "prompt": "{base_prompt}>",
        },
        "terminal length 0": {
            "output": "",
            "help": "Set terminal length to 0",
            "prompt": "{base_prompt}>",
        },
        "show clock": {
            "output": "MyFakeNOSPlugin system time is 00:00:00",
            "help": "Show system time",
            "prompt": "{base_prompt}>",
        },
    },
}

:param data: NOS dictionary

Source code in fakenos/core/nos.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def from_dict(self, data: dict) -> None:
    """
    Method to build NOS from dictionary data.

    Sample NOS dictionary::

        nos_plugin_dict = {
            "name": "MyFakeNOSPlugin",
            "initial_prompt": "{base_prompt}>",
            "commands": {
                "terminal width 511": {
                    "output": "",
                    "help": "Set terminal width to 511",
                    "prompt": "{base_prompt}>",
                },
                "terminal length 0": {
                    "output": "",
                    "help": "Set terminal length to 0",
                    "prompt": "{base_prompt}>",
                },
                "show clock": {
                    "output": "MyFakeNOSPlugin system time is 00:00:00",
                    "help": "Show system time",
                    "prompt": "{base_prompt}>",
                },
            },
        }

    :param data: NOS dictionary
    """
    self.name = data.get("name", self.name)
    self.commands.update(data.get("commands", self.commands))
    self.initial_prompt = data.get("initial_prompt", self.initial_prompt)

from_file(filename) #

Method to load NOS from YAML or Python file

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

Source code in fakenos/core/nos.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def from_file(self, filename: str) -> None:
    """
    Method to load NOS from YAML or Python file

    :param data: 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)

is_file_ending_correct(filename) #

Method to check if file extension is correct and load NOS data. Correct types are: .yaml, .yml and .py

Source code in fakenos/core/nos.py 
237
238
239
240
241
242
def is_file_ending_correct(self, filename: str) -> None:
    """
    Method to check if file extension is correct and load NOS data.
    Correct types are: .yaml, .yml and .py
    """
    return filename.endswith((".yaml", ".yml", ".py"))

validate() #

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

Source code in fakenos/core/nos.py 
 99
100
101
102
103
104
105
106
def validate(self) -> None:
    """
    Method to validate NOS attributes: commands, name,
    initial prompt - using Pydantic models,
    raises ValidationError on failure.
    """
    ModelNosAttributes(**self.__dict__)
    log.debug("%s NOS attributes validation succeeded", self.name)

TCPServerBase Class#

Bases: ABC

This module provides the base class for a TCP Server. It provides the methods to start and stop the server.

Note: We are looking to switch to socketserver as it is the standard library in python.

Source code in fakenos/core/servers.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class TCPServerBase(ABC):
    """
    This module provides the base class for a TCP Server.
    It provides the methods to start and stop the server.

    Note: We are looking to switch to socketserver as it is
    the standard library in python.
    """

    def __init__(self, address="localhost", port=6000, timeout=1):
        """
        Initialize the server with the address and port
        and the timeout for the socket.
        """
        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 = []

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

        self._bind_sockets()

        self._listen_thread = threading.Thread(target=self._listen)
        self._listen_thread.start()

    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.settimeout(self.timeout)
        self._socket.bind((self.address, self.port))

    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()
        self._listen_thread.join()
        self._socket.close()

        for connection_thread in self._connection_threads:
            connection_thread.join()

    def _listen(self):
        """
        This function is constantly running if the server is running.
        It waits for a connection, and if a connection is made, it will
        call the connection function.
        """
        while self._is_running.is_set():
            try:
                self._socket.listen()
                client, _ = self._socket.accept()
                connection_thread = threading.Thread(
                    target=self.connection_function,
                    args=(
                        client,
                        self._is_running,
                    ),
                )
                connection_thread.start()
                self._connection_threads.append(connection_thread)
            except socket.timeout:
                pass

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

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

Initialize the server with the address and port and the timeout for the socket.

Source code in fakenos/core/servers.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(self, address="localhost", port=6000, timeout=1):
    """
    Initialize the server with the address and port
    and the timeout for the socket.
    """
    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 = []

connection_function(client, is_running) abstractmethod #

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

Source code in fakenos/core/servers.py 
109
110
111
112
113
114
115
@abstractmethod
def connection_function(self, client, is_running):
    """
    This abstract method it is called when a new connection
    is made. The implementation should handle all the
    latter connection.
    """

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 fakenos/core/servers.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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()

    self._bind_sockets()

    self._listen_thread = threading.Thread(target=self._listen)
    self._listen_thread.start()

stop() #

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

Source code in fakenos/core/servers.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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()
    self._listen_thread.join()
    self._socket.close()

    for connection_thread in self._connection_threads:
        connection_thread.join()