Skip to content

Commit

Permalink
fixed for python3.7; replaced := and format {var=} syntax
Browse files Browse the repository at this point in the history
  • Loading branch information
zack-vii committed Nov 5, 2020
1 parent b44466d commit 5bb8d8e
Show file tree
Hide file tree
Showing 3 changed files with 29 additions and 29 deletions.
49 changes: 25 additions & 24 deletions dhcppython/client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 = (
Expand Down Expand Up @@ -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")
Expand All @@ -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,
Expand All @@ -204,15 +203,16 @@ def get_lease(
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")
Expand All @@ -222,7 +222,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)

Expand Down Expand Up @@ -256,19 +256,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
):
Expand All @@ -283,12 +283,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
Expand Down Expand Up @@ -346,14 +348,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}")
Expand Down
6 changes: 2 additions & 4 deletions dhcppython/options.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -74,9 +74,7 @@
}


class CodeDataMapping(TypedDict):
obj: Option
index: int
CodeDataMapping = dict


class OptionList(collections.abc.MutableSequence):
Expand Down
3 changes: 2 additions & 1 deletion dhcppython/packet.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down

0 comments on commit 5bb8d8e

Please sign in to comment.