fixed for python3.7; replaced := and format {var=} syntax

Author: zack-vii

dhcppython/client.py

@@ -29,11 +29,8 @@ def format_dhcp_packet(pkt: packet.DHCPPacket) -> str:
     broadcast = "BROADCAST" if pkt.flags else "UNICAST"
     client_info_padding = 18
     client_info = f"{pkt.htype} - {pkt.chaddr} ({utils.mac2vendor(pkt.chaddr)})"
-    if (
-        visual_diff := (
-            utils.visual_length(client_info) - (COL_LEN - client_info_padding)
-        )
-    ) > 0:
+    visual_diff = (utils.visual_length(client_info) - (COL_LEN - client_info_padding))
+    if visual_diff > 0:
         client_info = client_info[:-visual_diff]
 
     output = (
@@ -172,15 +169,16 @@ def get_lease(
         if verbose > 1:
             print(format_dhcp_packet(discover))
         start = default_timer()
-        logging.debug(f"Sending discover packet to {server} with {tx_id=}")
+        logging.debug(f"Sending discover packet to {server} with {tx_id}")
         self.send_discover(server, discover, verbose)
         # O
         tries = 0
-        while not (offer := self.receive_offer(tx_id, verbose)):
+        offer = self.receive_offer(tx_id, verbose)
+        while not offer:
             logging.debug(f"Sleeping {self.retry_interval} ms then retrying discover")
             sleep(self.retry_interval / 1000)
             logging.debug(
-                f"Attempt {tries} - Sending discover packet to {server} with {tx_id=}"
+                f"Attempt {tries} - Sending discover packet to {server} with {tx_id}"
             )
             if verbose > 1:
                 print("Resending DISCOVER packet")
@@ -190,6 +188,7 @@ def get_lease(
                     "Unable to obtain offer run client with -d for debug info"
                 )
             tries += 1
+            offer = self.receive_offer(tx_id, verbose)
         # R
         request = packet.DHCPPacket.Request(
             mac_addr,
@@ -200,19 +199,21 @@ def get_lease(
             client_ip=offer.yiaddr,
             relay=relay,
         )
+        print(request)
         if verbose > 1:
             print("REQUEST Packet")
             print(format_dhcp_packet(request))
         logging.debug(f"Constructed request packet: {request}")
-        logging.debug(f"Sending request packet to {server} with {tx_id=}")
+        logging.debug(f"Sending request packet to {server} with {tx_id}")
         self.send_request(server, request, verbose)
         # A
         tries = 0
-        while not (ack := self.receive_ack(tx_id, verbose)):
+        ack = self.receive_ack(tx_id, verbose)
+        while not ack:
             logging.debug(f"Sleeping {self.retry_interval} ms then retrying request")
             sleep(self.retry_interval / 1000)
             logging.debug(
-                f"Attempt {tries} - Sending request packet to {server} with {tx_id=}"
+                f"Attempt {tries} - Sending request packet to {server} with {tx_id}"
             )
             if verbose > 1:
                 print("Resending REQUEST packet")
@@ -222,7 +223,7 @@ def get_lease(
                     "Unable to obtain ack run client with -d for debug info"
                 )
             tries += 1
-
+            ack = self.receive_ack(tx_id, verbose)
         lease_time = default_timer() - start
         lease = Lease(discover, offer, request, ack, lease_time, self.ack_server)
 
@@ -256,19 +257,19 @@ def listen(
             logging.debug(
                 f"Select: {select.select(self.listening_sockets, self.writing_sockets, self.except_sockets, 0)}"
             )
-            if len(
-                socks := select.select(
+            socks = select.select(
                     self.listening_sockets,
                     self.writing_sockets,
                     self.except_sockets,
                     self.select_timout,
                 )[0]
-            ):
+            if len(socks):
                 for sock in socks:
                     data, addr = sock.recvfrom(self.max_pkt_size)
                     logging.debug(f"Received data from {addr}: {data}")
+                    dhcp_packet = self.get_valid_pkt(data)
                     if (
-                        (dhcp_packet := self.get_valid_pkt(data)) is not None
+                        dhcp_packet is not None
                         and dhcp_packet.xid == tx_id
                         and dhcp_packet.msg_type == msg_type
                     ):
@@ -283,12 +284,14 @@ def listen(
                             logging.debug(
                                 f"TX ID does not match expected ID {dhcp_packet.xid} != {tx_id}"
                             )
-                        elif (msg_type_actual := dhcp_packet.msg_type) != msg_type:
-                            logging.debug(
-                                f"DHCP message type does not match expected: {msg_type_actual} != {msg_type}"
-                            )
                         else:
-                            logging.debug("Something is wrong with this packet")
+                            msg_type_actual = dhcp_packet.msg_type
+                            if msg_type_actual != msg_type:
+                                logging.debug(
+                                    f"DHCP message type does not match expected: {msg_type_actual} != {msg_type}"
+                                )
+                            else:
+                                logging.debug("Something is wrong with this packet")
                         logging.debug(dhcp_packet)
                         dhcp_packet = None
                         tries += 1
@@ -346,14 +349,13 @@ def send(self, remote_addr: str, remote_port: int, data: bytes, verbosity: int):
         tries = 0
         while tries < self.max_tries:
             logging.debug(f"Select: {select.select(self.listening_sockets, self.writing_sockets, self.except_sockets, self.select_timout,)}")
-            if len(
-                socks := select.select(
+            socks = select.select(
                     self.listening_sockets,
                     self.writing_sockets,
                     self.except_sockets,
                     self.select_timout,
                 )[1]
-            ):
+            if socks:
                 sock = socks[0]
                 logging.debug(f"Connecting to {remote_addr}:{remote_port}")
                 logging.debug(f"Sending data {data!r}")

dhcppython/options.py

@@ -54,7 +54,7 @@
 import csv
 from abc import ABC, abstractmethod
 import collections.abc
-from typing import Dict, Union, List, Tuple, Optional, TypedDict
+from typing import Dict, Union, List, Tuple, Optional
 import ipaddress
 import struct
 import json
@@ -74,9 +74,7 @@
 }
 
 
-class CodeDataMapping(TypedDict):
-    obj: Option
-    index: int
+CodeDataMapping = dict
 
 
 class OptionList(collections.abc.MutableSequence):

dhcppython/packet.py

@@ -111,7 +111,8 @@ def asbytes(self):
 
     @property
     def msg_type(self) -> Optional[str]:
-        if msg_type_option := self.options.by_code(53):
+        msg_type_option = self.options.by_code(53)
+        if msg_type_option:
             return list(msg_type_option.value.values())[0]
         else:
             return None