Working on improving examples code.

Author: ccie18643

examples/icmp_echo_client.py

@@ -39,26 +39,19 @@
 import os
 import struct
 import time
-from typing import override
+from typing import Any, override
 
 import click
+from net_addr import Ip4Address
+from net_addr import Ip6Address
+from net_addr import IpVersion
 
 from examples.lib.client import Client
 from net_addr import (
-    ClickTypeIp4Address,
-    ClickTypeIp4Host,
-    ClickTypeIp6Address,
-    ClickTypeIp6Host,
     ClickTypeIpAddress,
-    ClickTypeMacAddress,
-    Ip4Address,
-    Ip4Host,
-    Ip6Address,
-    Ip6Host,
-    IpAddress,
-    MacAddress,
 )
-from pytcp import socket, stack
+from pytcp import socket
+from run_stack import cli as stack_cli
 
 ICMP4__ECHO_REQUEST__TYPE = 8
 ICMP4__ECHO_REQUEST__CODE = 0
@@ -78,8 +71,7 @@ class IcmpEchoClient(Client):
     def __init__(
         self,
         *,
-        local_ip_address: IpAddress,
-        remote_ip_address: IpAddress,
+        remote_ip_address: Ip6Address | Ip4Address,
         message_count: int = -1,
         message_delay: int = 1,
         message_size: int = 5,
@@ -88,7 +80,6 @@ def __init__(
         Class constructor.
         """
 
-        self._local_ip_address = local_ip_address
         self._remote_ip_address = remote_ip_address
         self._message_count = message_count
         self._message_delay = message_delay
@@ -147,22 +138,17 @@ def _thread__client__sender(self) -> None:
             message_count = self._message_count
 
             while self._run_thread and message_count:
-                match (
-                    self._local_ip_address.version,
-                    self._remote_ip_address.version,
-                ):
-                    case 6, 6:
+                match self._remote_ip_address.version:
+                    case IpVersion.IP6:
                         icmp_message = self._create_icmp6_message(
                             identifier=identifier,
                             sequence=self._message_count - message_count + 1,
                         )
-                    case 4, 4:
+                    case IpVersion.IP4:
                         icmp_message = self._create_icmp4_message(
                             identifier=identifier,
                             sequence=self._message_count - message_count + 1,
                         )
-                    case _:
-                        raise ValueError("Invalid IP address versions.")
 
                 try:
                     self._client_socket.send(icmp_message)
@@ -212,110 +198,31 @@ def _thread__client__receiver(self) -> None:
 
 
 @click.command()
-@click.option(
-    "--stack-interface",
-    "stack__interface",
-    default="tap7",
-    help="Name of the interface to be used by the stack.",
-)
-@click.option(
-    "--stack-mac-address",
-    "stack__mac_address",
-    type=ClickTypeMacAddress(),
-    default=None,
-    help="MAC address to be assigned to the stack interface.",
-)
-@click.option(
-    "--stack-ip6-address",
-    "stack__ip6_host",
-    type=ClickTypeIp6Host(),
-    default=None,
-    help="IPv6 address/mask to be assigned to the stack interface.",
-)
-@click.option(
-    "--stack-ip6-gateway",
-    "stack__ip6_gateway",
-    type=ClickTypeIp6Address(),
-    default=None,
-    help="IPv6 gateway address to be assigned to the stack interface.",
-)
-@click.option(
-    "--stack-ip4-address",
-    "stack__ip4_host",
-    type=ClickTypeIp4Host(),
-    default=None,
-    help="IPv4 address/mask to be assigned to the stack interface.",
-)
-@click.option(
-    "--stack-ip4-gateway",
-    "stack__ip4_gateway",
-    type=ClickTypeIp4Address(),
-    default=None,
-    help="IPv4 gateway address to be assigned to the stack interface.",
-)
 @click.argument(
     "remote_ip_address",
     type=ClickTypeIpAddress(),
     required=True,
 )
+@click.pass_context
 def cli(
+    ctx: click.Context,
     *,
-    stack__interface: str,
-    stack__mac_address: MacAddress | None,
-    stack__ip6_host: Ip6Host | None,
-    stack__ip6_gateway: Ip6Address | None,
-    stack__ip4_host: Ip4Host | None,
-    stack__ip4_gateway: Ip4Address | None,
-    remote_ip_address: IpAddress,
+    remote_ip_address: Ip6Address | Ip4Address,
+    **kwargs: dict[str, Any],
 ) -> None:
     """
     Start PyTCP stack and stop it when user presses Ctrl-C.
     Start ICMP Echo client.
     """
 
-    if stack__ip6_host:
-        stack__ip6_host.gateway = stack__ip6_gateway
-
-    if stack__ip4_host:
-        stack__ip4_host.gateway = stack__ip4_gateway
-
-    stack.init(
-        *stack.initialize_interface(stack__interface),
-        mac_address=stack__mac_address,
-        ip6_host=stack__ip6_host,
-        ip4_host=stack__ip4_host,
-    )
-
     match remote_ip_address.version:
-        case 6:
-            client = IcmpEchoClient(
-                local_ip_address=(
-                    stack__ip6_host.address if stack__ip6_host else Ip6Address()
-                ),
-                remote_ip_address=remote_ip_address,
-            )
-        case 4:
-            client = IcmpEchoClient(
-                local_ip_address=(
-                    stack__ip4_host.address if stack__ip4_host else Ip4Address()
-                ),
-                remote_ip_address=remote_ip_address,
-            )
-        case _:
-            raise ValueError(
-                f"Invalid remote IP address version: {remote_ip_address.version}"
-            )
-
-    try:
-        stack.start()
-        client.start()
-        while True:
-            time.sleep(1)
+        case IpVersion.IP6:
+            client = IcmpEchoClient(remote_ip_address=remote_ip_address)
+        case IpVersion.IP4:
+            client = IcmpEchoClient(remote_ip_address=remote_ip_address)
 
-    except KeyboardInterrupt:
-        client.stop()
-        stack.stop()
+    ctx.invoke(stack_cli, subsystem=client, **kwargs)
 
 
 if __name__ == "__main__":
-    cli()  # pylint: disable = missing-kwoa
+    cli.main()

examples/lib/client.py

@@ -41,7 +41,9 @@
 from typing import TYPE_CHECKING, override
 
 import click
-from net_addr.ip_address import IpAddress
+from net_addr.ip4_address import Ip4Address
+from net_addr.ip6_address import Ip6Address
+from net_addr.ip_address import IpVersion
 
 from examples.lib.subsystem import Subsystem
 from pytcp.socket import (
@@ -66,9 +68,9 @@ class Client(Subsystem):
 
     _protocol_name: str
     _client_name: str
-    _local_ip_address: IpAddress
+    _local_ip_address: Ip4Address | Ip6Address
     _local_port: int
-    _remote_ip_address: IpAddress
+    _remote_ip_address: Ip4Address | Ip6Address
     _remote_port: int
     _run_thread: bool
     _client_socket: Socket | None
@@ -83,6 +85,12 @@ def start(self) -> None:
             f"Starting the {self._protocol_name} {self._client_name} service."
         )
 
+        if isinstance(self._remote_ip_address, Ip4Address):
+            self._local_ip_address = self.stack_ip4_address
+
+        if isinstance(self._remote_ip_address, Ip6Address):
+            self._local_ip_address = self.stack_ip6_address
+
         self._client_socket = self._get_client_socket()
 
         self._run_thread = True
@@ -108,25 +116,24 @@ def _get_client_socket(self) -> Socket | None:
         """
 
         match (
-            self._local_ip_address.version,
             self._remote_ip_address.version,
             self._protocol_name,
         ):
-            case 6, 6, "TCP":
+            case IpVersion.IP6, "TCP":
                 client_socket = socket(family=AF_INET6, type=SOCK_STREAM)
-            case 4, 4, "TCP":
+            case IpVersion.IP4, "TCP":
                 client_socket = socket(family=AF_INET4, type=SOCK_STREAM)
-            case 6, 6, "UDP":
+            case IpVersion.IP6, "UDP":
                 client_socket = socket(family=AF_INET6, type=SOCK_DGRAM)
-            case 4, 4, "UDP":
+            case IpVersion.IP4, "UDP":
                 client_socket = socket(family=AF_INET4, type=SOCK_DGRAM)
-            case 6, 6, "ICMP":
+            case IpVersion.IP6, "ICMP":
                 client_socket = socket(
                     family=AF_INET6, type=SOCK_RAW, protocol=IPPROTO_ICMP6
                 )
                 self._local_port = int(IPPROTO_ICMP6)
                 self._remote_port = 0
-            case 4, 4, "ICMP":
+            case IpVersion.IP4, "ICMP":
                 client_socket = socket(
                     family=AF_INET4, type=SOCK_RAW, protocol=IPPROTO_ICMP4
                 )

examples/lib/subsystem.py

@@ -36,13 +36,21 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from net_addr.ip4_address import Ip4Address
+    from net_addr.ip6_address import Ip6Address
 
 
 class Subsystem(ABC):
     """
     Base class for 'user space' services like clients and servers.
     """
 
+    stack_ip4_address: Ip4Address
+    stack_ip6_address: Ip6Address
+
     @abstractmethod
     def start(self) -> None:
         """

examples/run_stack.py

@@ -126,6 +126,12 @@ def cli(
     try:
         stack.start()
         if subsystem is not None:
+            subsystem.stack_ip4_address = (
+                stack__ip4_host.address if stack__ip4_host else Ip4Address()
+            )
+            subsystem.stack_ip6_address = (
+                stack__ip6_host.address if stack__ip6_host else Ip6Address()
+            )
             subsystem.start()
 
         while True:
@@ -138,4 +144,4 @@ def cli(
 
 
 if __name__ == "__main__":
-    cli()  # pylint: disable = missing-kwoa
+    cli.main()

net_addr/__init__.py

@@ -74,5 +74,5 @@
 from .ip6_host import Ip6Host, Ip6HostOrigin
 from .ip6_mask import Ip6Mask
 from .ip6_network import Ip6Network
-from .ip_address import IpAddress
+from .ip_address import IpAddress, IpVersion
 from .mac_address import MacAddress

net_addr/ip4_address.py

@@ -40,7 +40,7 @@
 from typing import override
 
 from .errors import Ip4AddressFormatError
-from .ip_address import IpAddress
+from .ip_address import IpAddress, IpVersion
 from .mac_address import MacAddress
 
 IP4__ADDRESS_LEN = 4
@@ -56,7 +56,7 @@ class Ip4Address(IpAddress):
     IPv4 address support class.
     """
 
-    _version: int = 4
+    _version: IpVersion = IpVersion.IP4
 
     def __init__(
         self,

net_addr/ip4_host.py

@@ -39,6 +39,8 @@
 from enum import auto
 from typing import override
 
+from net_addr.ip_address import IpVersion
+
 from .errors import (
     Ip4AddressFormatError,
     Ip4HostFormatError,
@@ -67,7 +69,7 @@ class Ip4Host(IpHost[Ip4Address, Ip4Network, Ip4HostOrigin]):
     IPv4 host support class.
     """
 
-    _version: int = 4
+    _version: IpVersion = IpVersion.IP4
     _primary: bool
     _gateway: Ip4Address | None
     _origin: Ip4HostOrigin

net_addr/ip4_mask.py

@@ -39,6 +39,8 @@
 import socket
 from typing import override
 
+from net_addr.ip_address import IpVersion
+
 from .errors import Ip4MaskFormatError
 from .ip4_address import IP4__ADDRESS_LEN, IP4__REGEX
 from .ip_mask import IpMask
@@ -49,7 +51,7 @@ class Ip4Mask(IpMask):
     IPv4 mask support class.
     """
 
-    _version: int = 4
+    _version: IpVersion = IpVersion.IP4
 
     def __init__(
         self,

net_addr/ip4_network.py

@@ -37,6 +37,8 @@
 
 from typing import override
 
+from net_addr.ip_address import IpVersion
+
 from .errors import (
     Ip4AddressFormatError,
     Ip4MaskFormatError,
@@ -52,7 +54,7 @@ class Ip4Network(IpNetwork[Ip4Address, Ip4Mask]):
     IPv4 network support class.
     """
 
-    _version = 4
+    _version = IpVersion.IP4
 
     def __init__(
         self,