commit 1b4783dcd8e28899b860963cacef1b1ec25f4ed5
Author: Peter V. Saveliev <peter(a)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(a)svinota.eu>
Signed-off-by: Jiri Pirko <jiri(a)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(a)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):
Show replies by date