commit 1b4783dcd8e28899b860963cacef1b1ec25f4ed5 Author: Peter V. Saveliev peter@svinota.eu Date: Thu Nov 20 16:04:43 2014 +0100
NetUtils: fix PEP8 issues
Having automated flake8 testing it is easier to catch some issues prior they will cause real bugs; but to be used with flake8, the code must be PEP8-compliant.
Signed-off-by: Peter V. Saveliev peter@svinota.eu Signed-off-by: Jiri Pirko jiri@resnulli.us
lnst/Common/NetUtils.py | 14 +++++++++++--- 1 files changed, 11 insertions(+), 3 deletions(-) --- diff --git a/lnst/Common/NetUtils.py b/lnst/Common/NetUtils.py index a578fb7..d9440ed 100644 --- a/lnst/Common/NetUtils.py +++ b/lnst/Common/NetUtils.py @@ -10,7 +10,6 @@ __author__ = """ rpazdera@redhat.com (Radek Pazdera) """
-import logging import os import re import socket @@ -28,12 +27,14 @@ except ImportError: from pyroute2.iproute import RTM_NEWLINK from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg
+ def normalize_hwaddr(hwaddr): try: return hwaddr.upper().rstrip("\n") except: return ""
+ def scan_netdevs(): scan = [] nl_socket = IPRSocket() @@ -45,7 +46,7 @@ def scan_netdevs(): msg["header"]["sequence_number"] = 1 msg.encode()
- nl_socket.sendto(msg.buf.getvalue(), (0,0)) + nl_socket.sendto(msg.buf.getvalue(), (0, 0))
finished = False while not finished: @@ -71,6 +72,7 @@ def scan_netdevs(): nl_socket.close() return scan
+ def test_tcp_connection(host, port): try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -80,6 +82,7 @@ def test_tcp_connection(host, port): except: return False
+ def verify_ip_address(addr): if len(addr.split('.')) != 4: return False @@ -89,12 +92,14 @@ def verify_ip_address(addr): except: return False
+ def verify_mac_address(addr): if re.match("^[0-9a-f]{2}([:][0-9a-f]{2}){5}$", addr, re.I): return True else: return False
+ def get_corespond_local_ip(query_ip): """ Get ip address in local system which can communicate with query_ip. @@ -107,11 +112,12 @@ def get_corespond_local_ip(query_ip): shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) ip = ip.communicate()[0] - ip = re.search(r"src ([0-9.]*)",ip) + ip = re.search(r"src ([0-9.]*)", ip) if ip is None: return ip return ip.group(1)
+ class AddressPool: def __init__(self, start, end): self._next = self._addr_to_byte_string(start) @@ -141,6 +147,7 @@ class AddressPool:
return addr_str
+ class MacPool(AddressPool): def _addr_to_byte_string(self, addr): if not verify_mac_address(addr): @@ -153,6 +160,7 @@ class MacPool(AddressPool): def _byte_string_to_addr(self, byte_string): return ':'.join(map(lambda x: "%02x" % x, byte_string))
+ class IpPool(AddressPool): def _addr_to_byte_string(self, addr): if not verify_ip_address(addr):
lnst-developers@lists.fedorahosted.org