aioarp

python_net/aioarp/aioarp/_backends/_base.py

@@ -0,0 +1,24 @@
+import typing
+from typing_extensions import Protocol
+from typing_extensions import TypeAlias
+
+
+_Address: TypeAlias = typing.Tuple[typing.Any, ...]
+
+
+class SocketInterface(Protocol):
+
+    def send(self, data: bytes) -> int: ...
+
+    def sendall(self, data: bytes) -> None: ...
+    
+    def recv(self, bufsize: int) -> bytes: ...
+
+    def bind(self, address: typing.Union[_Address, bytes]) -> None: ...
+
+    def fileno(self) -> int: ...
+
+    def settimeout(self, value: typing.Union[float, None]) -> None: ...
+
+    def close(self) -> None: ...
+    

python_net/aioarp/aioarp/_backends/_mock.py

@@ -0,0 +1,42 @@
+import tempfile
+import typing
+
+from typing_extensions import TypeAlias
+
+
+_Address: TypeAlias = typing.Tuple[typing.Any, ...]
+
+
+class MockSocket:
+
+    def __init__(self, to_receive: bytes, delay: typing.Optional[float] = None):
+        self._file = tempfile.TemporaryFile('wb+')
+        self._to_receive = to_receive
+
+    def send(self, data: bytes) -> int:
+        self._file.write(data)
+        return len(data)
+
+    def sendall(self, data: bytes) -> None:
+        self._file.write(data)
+
+    def recv(self, max_bytes: typing.Optional[int] = None) -> bytes:
+        return self._to_receive
+    
+    def bind(self, address: typing.Union[_Address, bytes]) -> None: ...
+
+    def settimeout(self, value: typing.Union[float, None]) -> None: ...
+
+    def fileno(self) -> int:
+        return self._file.fileno()
+    
+    def close(self) -> None:
+        self._file.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+
+    

python_net/aioarp/aioarp/_backends/_utils.py → python_net/aioarp/aioarp/_utils.py

@@ -7,8 +7,13 @@ import getmac
 
 __all__ = (
     'LocalNetwork',
+    'is_valid_ipv4',
+    'enforce_mac',
+    'enforce_mac',
+    'parse_ip',
 )
 
+
 OUR_MAC = None
 OUR_IP = None
 OUR_INTERFACE = None
@@ -53,9 +58,49 @@ class LocalNetwork:
         raise RuntimeError('Could not find default interface')
 
 
+def is_valid_ipv4(ip: str) -> bool:
+    try:
+        ipaddress.IPv4Address(ip)
+        return True
+    except ipaddress.AddressValueError:
+        return False
+
+
+def enforce_mac(mac: str) -> bytes:
+    mac_bytes = []
+    for b in mac.split(':'):
+        mac_bytes.append(int(b, 16))
+    return bytes(mac_bytes)
+
+
+def enforce_ip(ip: str) -> bytes:
+    ip_bytes = []
+    for b in ip.split('.'):
+        ip_bytes.append(int(b))
+    return bytes(ip_bytes)
+
+
+def parse_mac(mac: bytes) -> str:
+    mac_parts = []
+    for b in mac:
+        hex = f'{b:x}'
+        if len(hex) == 1:
+            hex = '0' + hex
+        mac_parts.append(hex)
+    return ':'.join(mac_parts)
+
+
+def parse_ip(ip: bytes) -> str:
+    ip_parts = []
+    for b in ip:
+        ip_parts.append(str(b))
+    return '.'.join(ip_parts)
+
+
+
 if __name__ == '__main__':
     foo = LocalNetwork()
-    print(foo.get_default_interface())
-    print(foo.get_mac())
-    print(foo.get_ip())
-# print(getmac.get_mac_address('Ethernet 3'))
+    # print(foo.get_default_interface())
+    # print(foo.get_mac())
+    # print(foo.get_ip())
+    # print(is_valid_ipv4('1.1.1.1'))

python_net/aioarp/aioarp/defaults.py

@@ -0,0 +1,11 @@
+
+__all__ = (
+    'DEFAULT_READ_TIMEOUT',
+    'DEFAULT_WRITE_TIMEOUT',
+    'DEFAULT_REPLAY_MISSING_TIME'
+)
+
+
+DEFAULT_READ_TIMEOUT = 5
+DEFAULT_WRITE_TIMEOUT = 5
+DEFAULT_REPLAY_MISSING_TIME = 5