提交 7ed07840 编写于 作者: A air9

clean codesmell

上级 3f8d148e
......@@ -16,7 +16,6 @@ import os
import argparse
from hwcompatible.test import Test
from hwcompatible.command import Command
from hwcompatible.env import CertEnv
from hwcompatible.document import CertDocument
from rdma import RDMATest
......@@ -25,6 +24,9 @@ from rdma import RDMATest
class EthernetTest(RDMATest):
def __init__(self):
RDMATest.__init__(self)
self.args = None
self.cert = None
self.device = None
self.subtests = [self.test_ip_info, self.test_eth_link, self.test_icmp,
self.test_udp_tcp, self.test_http]
self.target_bandwidth_percent = 0.75
......
......@@ -26,7 +26,7 @@ except ImportError:
from urllib2 import urlopen, Request, HTTPError
from hwcompatible.test import Test
from hwcompatible.command import Command
from hwcompatible.command import Command, CertCommandError
from hwcompatible.document import CertDocument
from hwcompatible.env import CertEnv
......@@ -34,6 +34,9 @@ from hwcompatible.env import CertEnv
class NetworkTest(Test):
def __init__(self):
Test.__init__(self)
self.args = None
self.cert = None
self.device = None
self.requirements = ['ethtool', 'iproute', 'psmisc', 'qperf']
self.subtests = [self.test_ip_info, self.test_eth_link, self.test_icmp,
self.test_udp_tcp, self.test_http]
......@@ -79,7 +82,7 @@ class NetworkTest(Test):
try:
c.run()
return c.output
except Exception as e:
except CertCommandError as e:
print(e)
return []
......@@ -98,28 +101,28 @@ class NetworkTest(Test):
def get_speed(self):
c = Command("ethtool %s" % self.interface)
pattern = ".*Speed:\s+(?P<speed>\d+)Mb/s"
pattern = r".*Speed:\s+(?P<speed>\d+)Mb/s"
try:
speed = c.get_str(pattern, 'speed', False)
return int(speed)
except Exception as e:
except CertCommandError as e:
print("[X] No speed found on the interface.")
return None
def get_interface_ip(self):
c = Command("ip addr show %s" % self.interface)
pattern = ".*inet.? (?P<ip>.+)/.*"
pattern = r".*inet.? (?P<ip>.+)/.*"
try:
ip = c.get_str(pattern, 'ip', False)
return ip
except Exception as e:
except CertCommandError as e:
print("[X] No available ip on the interface.")
return None
def test_icmp(self):
count = 500
c = Command("ping -q -c %d -i 0 %s" % (count, self.server_ip))
pattern =".*, (?P<loss>\d+\.{0,1}\d*)% packet loss.*"
pattern = r".*, (?P<loss>\d+\.{0,1}\d*)% packet loss.*"
for _ in range(self.retries):
try:
......@@ -128,7 +131,7 @@ class NetworkTest(Test):
c.print_output()
if float(loss) == 0:
return True
except Exception as e:
except CertCommandError as e:
print(e)
return False
......@@ -145,7 +148,7 @@ class NetworkTest(Test):
try:
request = Request(url, data=data, headers=headers)
response = urlopen(request)
except Exception as e:
except HTTPError as e:
print(e)
return False
print("Status: %u %s" % (response.code, response.msg))
......@@ -171,7 +174,7 @@ class NetworkTest(Test):
cmd = "qperf %s tcp_bw" % self.server_ip
print(cmd)
c = Command(cmd)
pattern = "\s+bw\s+=\s+(?P<bandwidth>[\.0-9]+ [MG]B/sec)"
pattern = r"\s+bw\s+=\s+(?P<bandwidth>[\.0-9]+ [MG]B/sec)"
for _ in range(self.retries):
try:
bandwidth = c.get_str(pattern, 'bandwidth', False)
......@@ -186,7 +189,7 @@ class NetworkTest(Test):
(bandwidth, target_bandwidth))
if bandwidth > target_bandwidth:
return True
except Exception as e:
except CertCommandError as e:
print(e)
return False
......@@ -204,7 +207,7 @@ class NetworkTest(Test):
try:
with open(self.testfile, 'rb') as f:
filetext = base64.b64encode(f.read())
except Exception as e:
except FileNotFoundError as e:
print(e)
return False
......@@ -221,7 +224,7 @@ class NetworkTest(Test):
try:
request = Request(url, data=data, headers=headers)
response = urlopen(request)
except Exception as e:
except HTTPError as e:
print(e)
return False
time_stop = time.time()
......@@ -240,7 +243,7 @@ class NetworkTest(Test):
time_start = time.time()
try:
response = urlopen(url)
except Exception as e:
except HTTPError as e:
print(e)
return False
time_stop = time.time()
......@@ -252,7 +255,7 @@ class NetworkTest(Test):
try:
with open(self.testfile, 'wb') as f:
f.write(filetext)
except Exception as e:
except FileNotFoundError as e:
print(e)
return False
......
......@@ -17,7 +17,7 @@ import re
import argparse
from hwcompatible.test import Test
from hwcompatible.command import Command
from hwcompatible.command import Command, CertCommandError
from hwcompatible.document import CertDocument
from hwcompatible.env import CertEnv
from network import NetworkTest
......@@ -26,11 +26,13 @@ from network import NetworkTest
class RDMATest(NetworkTest):
def __init__(self):
NetworkTest.__init__(self)
self.args = None
self.cert = None
self.device = None
self.requirements = ['ethtool', 'iproute', 'psmisc', 'qperf',
'perftest', 'opensm', 'infiniband-diags',
'librdmacm-utils', 'libibverbs-utils']
self.subtests = [self.test_ibstatus, self.test_icmp, self.test_rdma]
self.device = None
self.interface = None
self.ib_device = None
self.ib_port = None
......@@ -53,7 +55,7 @@ class RDMATest(NetworkTest):
c = Command(cmd)
try:
self.ib_device = c.read()
except Exception as e:
except CertCommandError as e:
print(e)
return False
......@@ -62,7 +64,7 @@ class RDMATest(NetworkTest):
c = Command(cmd)
try:
self.ib_port = int(c.read(), 16) + 1
except Exception as e:
except CertCommandError as e:
print(e)
return False
......@@ -78,14 +80,14 @@ class RDMATest(NetworkTest):
if ib_str not in info:
continue
print(info)
self.gid = re.search("default gid:\s+(.*)", info).group(1)
self.base_lid = re.search("base lid:\s+(.*)", info).group(1)
self.sm_lid = re.search("sm lid:\s+(.*)", info).group(1)
self.state = re.search("state:\s+(.*)", info).group(1)
self.phys_state = re.search("phys state:\s+(.*)", info).group(1)
self.link_layer = re.search("link_layer:\s+(.*)", info).group(1)
self.speed = int(re.search("rate:\s+(\d*)", info).group(1)) * 1024
except Exception as e:
self.gid = re.search(r"default gid:\s+(.*)", info).group(1)
self.base_lid = re.search(r"base lid:\s+(.*)", info).group(1)
self.sm_lid = re.search(r"sm lid:\s+(.*)", info).group(1)
self.state = re.search(r"state:\s+(.*)", info).group(1)
self.phys_state = re.search(r"phys state:\s+(.*)", info).group(1)
self.link_layer = re.search(r"link_layer:\s+(.*)", info).group(1)
self.speed = int(re.search(r"rate:\s+(\d*)", info).group(1)) * 1024
except CertCommandError as e:
print(e)
return False
......@@ -126,7 +128,7 @@ class RDMATest(NetworkTest):
cmd = "%s %s -d %s -i %s" % (cmd, self.server_ip, self.ib_device, self.ib_port)
print(cmd)
c = Command(cmd)
pattern = "\s+(\d+)\s+(\d+)\s+([\.\d]+)\s+(?P<avg_bw>[\.\d]+)\s+([\.\d]+)"
pattern = r"\s+(\d+)\s+(\d+)\s+([\.\d]+)\s+(?P<avg_bw>[\.\d]+)\s+([\.\d]+)"
try:
avg_bw = c.get_str(pattern, 'avg_bw', False) ## MB/sec
avg_bw = float(avg_bw) * 8
......@@ -135,7 +137,7 @@ class RDMATest(NetworkTest):
print("Current bandwidth is %.2fMb/s, target is %.2fMb/s" %
(avg_bw, tgt_bw))
return avg_bw > tgt_bw
except Exception as e:
except CertCommandError as e:
print(e)
self.call_remote_server(cmd, 'stop')
return False
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册