#2464 - Linting changes to see if the python 3.12 pipeline will pass
This commit is contained in:
@@ -231,7 +231,7 @@ class WebpageUnavailablePenalty(AbstractReward):
|
|||||||
|
|
||||||
# If the last request did actually go through, then check if the webpage also loaded
|
# If the last request did actually go through, then check if the webpage also loaded
|
||||||
web_browser_state = access_from_nested_dict(state, self.location_in_state)
|
web_browser_state = access_from_nested_dict(state, self.location_in_state)
|
||||||
if web_browser_state is NOT_PRESENT_IN_STATE or "history" not in web_browser_state:
|
if web_browser_state is NOT_PRESENT_IN_STATE or "history" not in web_browser_state: # noqa
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"Web browser reward could not be calculated because the web browser history on node",
|
"Web browser reward could not be calculated because the web browser history on node",
|
||||||
f"{self._node} was not reported in the simulation state. Returning 0.0",
|
f"{self._node} was not reported in the simulation state. Returning 0.0",
|
||||||
|
|||||||
@@ -248,7 +248,7 @@ class Network(SimComponent):
|
|||||||
hostname_b = node_b.hostname if node_b else None
|
hostname_b = node_b.hostname if node_b else None
|
||||||
port_a = link.endpoint_a.port_num
|
port_a = link.endpoint_a.port_num
|
||||||
port_b = link.endpoint_b.port_num
|
port_b = link.endpoint_b.port_num
|
||||||
link_key = f"{hostname_a}:eth-{port_a}<->{hostname_b}:eth-{port_b}"
|
link_key = f"{hostname_a}: eth-{port_a}<->{hostname_b}: eth-{port_b}"
|
||||||
state["links"][link_key] = link.describe_state()
|
state["links"][link_key] = link.describe_state()
|
||||||
state["links"][link_key]["hostname_a"] = hostname_a
|
state["links"][link_key]["hostname_a"] = hostname_a
|
||||||
state["links"][link_key]["hostname_b"] = hostname_b
|
state["links"][link_key]["hostname_b"] = hostname_b
|
||||||
|
|||||||
@@ -56,7 +56,7 @@ def generate_mac_address(oui: Optional[str] = None) -> str:
|
|||||||
if oui:
|
if oui:
|
||||||
oui_pattern = re.compile(r"^([0-9A-Fa-f]{2}[:-]){2}[0-9A-Fa-f]{2}$")
|
oui_pattern = re.compile(r"^([0-9A-Fa-f]{2}[:-]){2}[0-9A-Fa-f]{2}$")
|
||||||
if not oui_pattern.match(oui):
|
if not oui_pattern.match(oui):
|
||||||
msg = f"Invalid oui. The oui should be in the format xx:xx:xx, where x is a hexadecimal digit, got '{oui}'"
|
msg = f"Invalid oui. The oui should be in the format xx:xx:xx, where x is a hexadecimal digit, got '{oui}'" # noqa
|
||||||
_LOGGER.error(msg)
|
_LOGGER.error(msg)
|
||||||
raise ValueError(msg)
|
raise ValueError(msg)
|
||||||
oui_bytes = [int(chunk, 16) for chunk in oui.split(":")]
|
oui_bytes = [int(chunk, 16) for chunk in oui.split(":")]
|
||||||
@@ -64,7 +64,7 @@ def generate_mac_address(oui: Optional[str] = None) -> str:
|
|||||||
else:
|
else:
|
||||||
mac = random_bytes
|
mac = random_bytes
|
||||||
|
|
||||||
return ":".join(f"{b:02x}" for b in mac)
|
return ":".join(f"{b:02x}" for b in mac) # noqa
|
||||||
|
|
||||||
|
|
||||||
class NetworkInterface(SimComponent, ABC):
|
class NetworkInterface(SimComponent, ABC):
|
||||||
@@ -599,7 +599,7 @@ class Link(SimComponent):
|
|||||||
@property
|
@property
|
||||||
def current_load_percent(self) -> str:
|
def current_load_percent(self) -> str:
|
||||||
"""Get the current load formatted as a percentage string."""
|
"""Get the current load formatted as a percentage string."""
|
||||||
return f"{self.current_load / self.bandwidth:.5f}%"
|
return f"{self.current_load / self.bandwidth:.5f}%" # noqa
|
||||||
|
|
||||||
def endpoint_up(self):
|
def endpoint_up(self):
|
||||||
"""Let the Link know and endpoint has been brought up."""
|
"""Let the Link know and endpoint has been brought up."""
|
||||||
|
|||||||
@@ -100,7 +100,7 @@ class ICMPPacket(BaseModel):
|
|||||||
icmp_type = info.data["icmp_type"]
|
icmp_type = info.data["icmp_type"]
|
||||||
if get_icmp_type_code_description(icmp_type, v):
|
if get_icmp_type_code_description(icmp_type, v):
|
||||||
return v
|
return v
|
||||||
msg = f"No Matching ICMP code for type:{icmp_type.name}, code:{v}"
|
msg = f"No Matching ICMP code for type: {icmp_type.name}, code: {v}"
|
||||||
_LOGGER.error(msg)
|
_LOGGER.error(msg)
|
||||||
raise ValueError(msg)
|
raise ValueError(msg)
|
||||||
|
|
||||||
@@ -109,6 +109,6 @@ class ICMPPacket(BaseModel):
|
|||||||
description = get_icmp_type_code_description(self.icmp_type, self.icmp_code)
|
description = get_icmp_type_code_description(self.icmp_type, self.icmp_code)
|
||||||
if description:
|
if description:
|
||||||
return description
|
return description
|
||||||
msg = f"No Matching ICMP code for type:{self.icmp_type.name}, code:{self.icmp_code}"
|
msg = f"No Matching ICMP code for type: {self.icmp_type.name}, code: {self.icmp_code}"
|
||||||
_LOGGER.error(msg)
|
_LOGGER.error(msg)
|
||||||
raise ValueError(msg)
|
raise ValueError(msg)
|
||||||
|
|||||||
@@ -81,7 +81,7 @@ class DNSClient(Service):
|
|||||||
# check if the domain is already in the DNS cache
|
# check if the domain is already in the DNS cache
|
||||||
if target_domain in self.dns_cache:
|
if target_domain in self.dns_cache:
|
||||||
self.sys_log.info(
|
self.sys_log.info(
|
||||||
f"{self.name}: Domain lookup for {target_domain} successful,"
|
f"{self.name}: Domain lookup for {target_domain} successful, "
|
||||||
f"resolves to {self.dns_cache[target_domain]}"
|
f"resolves to {self.dns_cache[target_domain]}"
|
||||||
)
|
)
|
||||||
return True
|
return True
|
||||||
|
|||||||
@@ -147,7 +147,7 @@ class FTPServiceABC(Service, ABC):
|
|||||||
retrieved_file: File = self.file_system.get_file(folder_name=folder_name, file_name=file_name)
|
retrieved_file: File = self.file_system.get_file(folder_name=folder_name, file_name=file_name)
|
||||||
|
|
||||||
# if file does not exist, return an error
|
# if file does not exist, return an error
|
||||||
if not retrieved_file:
|
if not retrieved_file: # noqa
|
||||||
self.sys_log.error(
|
self.sys_log.error(
|
||||||
f"File {payload.ftp_command_args['dest_folder_name']}/"
|
f"File {payload.ftp_command_args['dest_folder_name']}/"
|
||||||
f"{payload.ftp_command_args['dest_file_name']} does not exist in {self.sys_log.hostname}"
|
f"{payload.ftp_command_args['dest_file_name']} does not exist in {self.sys_log.hostname}"
|
||||||
|
|||||||
@@ -60,7 +60,7 @@ class ICMP(Service):
|
|||||||
if target_ip_address.is_loopback:
|
if target_ip_address.is_loopback:
|
||||||
self.sys_log.info("Pinging loopback address")
|
self.sys_log.info("Pinging loopback address")
|
||||||
return any(network_interface.enabled for network_interface in self.network_interfaces.values())
|
return any(network_interface.enabled for network_interface in self.network_interfaces.values())
|
||||||
self.sys_log.info(f"Pinging {target_ip_address}:", to_terminal=True)
|
self.sys_log.info(f"Pinging {target_ip_address}: ", to_terminal=True)
|
||||||
sequence, identifier = 0, None
|
sequence, identifier = 0, None
|
||||||
while sequence < pings:
|
while sequence < pings:
|
||||||
sequence, identifier = self._send_icmp_echo_request(target_ip_address, sequence, identifier, pings)
|
sequence, identifier = self._send_icmp_echo_request(target_ip_address, sequence, identifier, pings)
|
||||||
|
|||||||
Reference in New Issue
Block a user