AS1024/napalm-danos
1
0
Fork 0
mirror of https://github.com/napalm-automation-community/napalm-vyos.git synced 2026-09-23 12:41:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix python code according to pylama errors

Browse Source
This commit is contained in:
Pieprzycki Piotr
2016-12-16 15:47:51 +01:00
parent 9b4d3aec64
commit 5afafa5522
4 changed files with 762 additions and 798 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
.travis.yml
View File

@@ -13,19 +13,8 @@ deploy:
tags: true tags: true
branch: master branch: master
script: script:
- cd test/unit - py.test --cov-report= --cov=napalm_vyos test/
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_arp_table - pylama .
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_bgp_neighbors after_success:
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_environment - coveralls
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_facts - if [ $TRAVIS_TAG ]; then curl -X POST https://readthedocs.org/build/napalm; fi
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_interfaces
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_interfaces_counters
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_interfaces_ip
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_lldp_neighbors
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_lldp_neighbors_detail
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_mac_address_table
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_ntp_stats
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_get_snmp_information
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_ios_only_bgp_time_conversion
- nosetests -v TestIOSDriver:TestGetterIOSDriver.test_ping
- cd ../..

71
napalm_vyos/vyos.py
View File

@@ -31,9 +31,8 @@ from netmiko import SCPConn
# NAPALM base # NAPALM base
from napalm_base.base import NetworkDriver from napalm_base.base import NetworkDriver
from napalm_base.exceptions import ConnectionException, SessionLockedException, \ from napalm_base.exceptions import ConnectionException, \
MergeConfigException, ReplaceConfigException,\ MergeConfigException, ReplaceConfigException
CommandErrorException
class VyOSDriver(NetworkDriver): class VyOSDriver(NetworkDriver):
@@ -58,7 +57,6 @@ class VyOSDriver(NetworkDriver):
self._old_config = None self._old_config = None
self._ssh_usekeys = False self._ssh_usekeys = False
# Netmiko possible arguments # Netmiko possible arguments
netmiko_argument_map = { netmiko_argument_map = {
'port': None, 'port': None,
@@ -92,16 +90,17 @@ class VyOSDriver(NetworkDriver):
self.global_delay_factor = optional_args.get('global_delay_factor', 1) self.global_delay_factor = optional_args.get('global_delay_factor', 1)
self.port = optional_args.get('port', 22) self.port = optional_args.get('port', 22)
def open(self): def open(self):
self._device = ConnectHandler(device_type='vyos', self._device = ConnectHandler(device_type='vyos',
host=self._hostname, host=self._hostname,
username=self._username, username=self._username,
password=self._password, password=self._password,
**self.netmiko_optional_args) **self.netmiko_optional_args)
self._scp_client = SCPConn(self._device)
try:
self._scp_client = SCPConn(self._device)
except:
raise ConnectionException("Failed to open connection ")
def close(self): def close(self):
self._device.disconnect() self._device.disconnect()
@@ -114,7 +113,7 @@ class VyOSDriver(NetworkDriver):
support a replace using a configuration string. support a replace using a configuration string.
""" """
if filename is not None: if filename is not None:
if os.path.exists(filename) == True: if os.path.exists(filename) is True:
self._scp_client.scp_transfer_file(filename, self._DEST_FILENAME) self._scp_client.scp_transfer_file(filename, self._DEST_FILENAME)
print self._device.send_command("cp "+self._BOOT_FILENAME+" "+self._BACKUP_FILENAME) print self._device.send_command("cp "+self._BOOT_FILENAME+" "+self._BACKUP_FILENAME)
output_loadcmd = self._device.send_config_set(['load '+self._DEST_FILENAME]) output_loadcmd = self._device.send_config_set(['load '+self._DEST_FILENAME])
@@ -141,9 +140,10 @@ class VyOSDriver(NetworkDriver):
Only configuration in set-format is supported with load_merge_candidate. Only configuration in set-format is supported with load_merge_candidate.
""" """
if filename is not None: if filename is not None:
if os.path.exists(filename) == True: if os.path.exists(filename) is True:
with open(filename) as f: with open(filename) as f:
print self._device.send_command("cp "+self._BOOT_FILENAME+" "+self._BACKUP_FILENAME) print self._device.send_command("cp "+self._BOOT_FILENAME+" "
+ self._BACKUP_FILENAME)
self._new_config = f.read() self._new_config = f.read()
cfg = [x for x in self._new_config.split("\n") if x is not ""] cfg = [x for x in self._new_config.split("\n") if x is not ""]
output_loadcmd = self._device.send_config_set(cfg) output_loadcmd = self._device.send_config_set(cfg)
@@ -153,7 +153,6 @@ class VyOSDriver(NetworkDriver):
if match_setfailed or match_delfailed: if match_setfailed or match_delfailed:
raise MergeConfigException("Failed merge config: " raise MergeConfigException("Failed merge config: "
+ output_loadcmd) + output_loadcmd)
else: else:
raise MergeConfigException("config file is not found") raise MergeConfigException("config file is not found")
elif config is not None: elif config is not None:
@@ -161,7 +160,6 @@ class VyOSDriver(NetworkDriver):
else: else:
raise MergeConfigException("no configuration found") raise MergeConfigException("no configuration found")
def discard_config(self): def discard_config(self):
self._device.exit_config_mode() self._device.exit_config_mode()
@@ -193,10 +191,7 @@ class VyOSDriver(NetworkDriver):
else: else:
self._device.send_config_set(['commit', 'save']) self._device.send_config_set(['commit', 'save'])
def get_environment(self): def get_environment(self):
""" """
'vmstat' output: 'vmstat' output:
procs -----------memory---------- ---swap-- -----io---- -system-- ----cpu---- procs -----------memory---------- ---swap-- -----io---- -system-- ----cpu----
@@ -249,7 +244,6 @@ class VyOSDriver(NetworkDriver):
return environment return environment
def get_interfaces(self): def get_interfaces(self):
""" """
"show interfaces" output example: "show interfaces" output example:
@@ -270,7 +264,8 @@ class VyOSDriver(NetworkDriver):
# 'match' example: # 'match' example:
# [("br0", "u", "D"), ("eth0", "u", "u"), ("eth1", "u", "u")...] # [("br0", "u", "D"), ("eth0", "u", "u"), ("eth1", "u", "u")...]
iface_state = {iface_name:{"State": state, "Link": link} for iface_name, state, link in match} iface_state = {iface_name: {"State": state, "Link": link} for iface_name,
state, link in match}
output_conf = self._device.send_command("show configuration") output_conf = self._device.send_command("show configuration")
@@ -284,7 +279,6 @@ class VyOSDriver(NetworkDriver):
ifaces_detail = config["interfaces"][iface_type] ifaces_detail = config["interfaces"][iface_type]
for iface_name in ifaces_detail: for iface_name in ifaces_detail:
description = self._get_value("description", ifaces_detail[iface_name]) description = self._get_value("description", ifaces_detail[iface_name])
if description is None: if description is None:
description = "" description = ""
@@ -313,8 +307,6 @@ class VyOSDriver(NetworkDriver):
return iface_dict return iface_dict
# for avoiding KeyError
@staticmethod @staticmethod
def _get_value(key, target_dict): def _get_value(key, target_dict):
if key in target_dict: if key in target_dict:
@@ -322,7 +314,6 @@ class VyOSDriver(NetworkDriver):
else: else:
return None return None
def get_arp_table(self): def get_arp_table(self):
# 'age' is not implemented yet # 'age' is not implemented yet
@@ -364,7 +355,6 @@ class VyOSDriver(NetworkDriver):
return arp_table return arp_table
def get_ntp_stats(self): def get_ntp_stats(self):
""" """
'ntpq -np' output example 'ntpq -np' output example
@@ -380,7 +370,8 @@ class VyOSDriver(NetworkDriver):
for ntp_info in output: for ntp_info in output:
remote, refid, st, t, when, hostpoll, reachability, delay, offset, jitter = ntp_info.split() remote, refid, st, t, when, hostpoll, reachability, delay, offset, \
jitter = ntp_info.split()
# 'remote' contains '*' if the machine synchronized with NTP server # 'remote' contains '*' if the machine synchronized with NTP server
synchronized = "*" in remote synchronized = "*" in remote
@@ -406,7 +397,6 @@ class VyOSDriver(NetworkDriver):
return ntp_stats return ntp_stats
def get_ntp_peers(self): def get_ntp_peers(self):
output = self._device.send_command("ntpq -np").split("\n")[2:] output = self._device.send_command("ntpq -np").split("\n")[2:]
@@ -436,10 +426,10 @@ class VyOSDriver(NetworkDriver):
192.168.1.4 4 64522 0 0 0 0 0 never Active 192.168.1.4 4 64522 0 0 0 0 0 never Active
""" """
output = self._device.send_command("show ip bgp summary").split("\n") output = self._device.send_command("show ip bgp summary").split("\n")
match = re.search(".* router identifier (\d+\.\d+\.\d+\.\d+), local AS number (\d+)", output[0]) match = re.search(".* router identifier (\d+\.\d+\.\d+\.\d+), local AS number (\d+)",
output[0])
if not match: if not match:
return {} return {}
router_id = unicode(match.group(1)) router_id = unicode(match.group(1))
@@ -516,9 +506,8 @@ class VyOSDriver(NetworkDriver):
return bgp_neighbor_data return bgp_neighbor_data
def _bgp_time_conversion(self, bgp_uptime): def _bgp_time_conversion(self, bgp_uptime):
uptime_letters = set(["y", "w", "h", "d"]) # uptime_letters = set(["y", "w", "h", "d"])
if "never" in bgp_uptime: if "never" in bgp_uptime:
return -1 return -1
@@ -547,14 +536,14 @@ class VyOSDriver(NetworkDriver):
(minutes * self._MINUTE_SECONDS) + seconds) (minutes * self._MINUTE_SECONDS) + seconds)
return uptime return uptime
def get_interfaces_counters(self): def get_interfaces_counters(self):
# 'rx_unicast_packet', 'rx_broadcast_packets', 'tx_unicast_packets', # 'rx_unicast_packet', 'rx_broadcast_packets', 'tx_unicast_packets',
# 'tx_multicast_packets' and 'tx_broadcast_packets' are not implemented yet # 'tx_multicast_packets' and 'tx_broadcast_packets' are not implemented yet
""" """
'show interfaces detail' output example: 'show interfaces detail' output example:
eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc pfifo_fast state UP group default qlen 1000 eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc pfifo_fast state
UP group default qlen 1000
link/ether 00:50:56:86:8c:26 brd ff:ff:ff:ff:ff:ff link/ether 00:50:56:86:8c:26 brd ff:ff:ff:ff:ff:ff
~~~ ~~~
RX: bytes packets errors dropped overrun mcast RX: bytes packets errors dropped overrun mcast
@@ -563,12 +552,9 @@ class VyOSDriver(NetworkDriver):
32776498 279273 0 0 0 0 32776498 279273 0 0 0 0
""" """
output = self._device.send_command("show interfaces detail") output = self._device.send_command("show interfaces detail")
interfaces = re.findall("(\S+): <.*", output) interfaces = re.findall("(\S+): <.*", output)
count = re.findall("(\d+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+)\s+", output) count = re.findall("(\d+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+)\s+", output)
counters = dict() counters = dict()
j = 0 j = 0
for i in count: for i in count:
@@ -600,7 +586,6 @@ class VyOSDriver(NetworkDriver):
return counters return counters
def get_snmp_information(self): def get_snmp_information(self):
# 'acl' is not implemented yet # 'acl' is not implemented yet
@@ -611,7 +596,6 @@ class VyOSDriver(NetworkDriver):
snmp = dict() snmp = dict()
snmp["community"] = dict() snmp["community"] = dict()
try: try:
for i in config["service"]["snmp"]["community"]: for i in config["service"]["snmp"]["community"]:
snmp["community"].update({ snmp["community"].update({
@@ -634,10 +618,8 @@ class VyOSDriver(NetworkDriver):
def get_facts(self): def get_facts(self):
output_uptime = self._device.send_command("cat /proc/uptime | awk '{print $1}'") output_uptime = self._device.send_command("cat /proc/uptime | awk '{print $1}'")
uptime = int(float(output_uptime)) uptime = int(float(output_uptime))
output = self._device.send_command("show version").split("\n") output = self._device.send_command("show version").split("\n")
ver_str = [line for line in output if "Version" in line][0] ver_str = [line for line in output if "Version" in line][0]
version = self.parse_version(ver_str) version = self.parse_version(ver_str)
@@ -679,13 +661,11 @@ class VyOSDriver(NetworkDriver):
return facts return facts
@staticmethod @staticmethod
def parse_version(ver_str): def parse_version(ver_str):
version = ver_str.split()[-1] version = ver_str.split()[-1]
return version return version
@staticmethod @staticmethod
def parse_snumber(sn_str): def parse_snumber(sn_str):
sn = sn_str.split(":") sn = sn_str.split(":")
@@ -696,7 +676,6 @@ class VyOSDriver(NetworkDriver):
model = model_str.split(":") model = model_str.split(":")
return model[1].strip() return model[1].strip()
def get_interfaces_ip(self): def get_interfaces_ip(self):
output = self._device.send_command("show interfaces") output = self._device.send_command("show interfaces")
output = output.split("\n") output = output.split("\n")
@@ -729,7 +708,6 @@ class VyOSDriver(NetworkDriver):
return ifaces_ip return ifaces_ip
@staticmethod @staticmethod
def _get_ip_version(ip_address): def _get_ip_version(ip_address):
if ":" in ip_address: if ":" in ip_address:
@@ -737,7 +715,6 @@ class VyOSDriver(NetworkDriver):
elif "." in ip_address: elif "." in ip_address:
return "ipv4" return "ipv4"
def get_users(self): def get_users(self):
output = self._device.send_command("show configuration commands").split("\n") output = self._device.send_command("show configuration commands").split("\n")
@@ -749,7 +726,6 @@ class VyOSDriver(NetworkDriver):
user_auth = dict() user_auth = dict()
for user in user_name: for user in user_name:
sshkeys = list() sshkeys = list()
# extract the configuration which relates to 'user' # extract the configuration which relates to 'user'
@@ -766,7 +742,8 @@ class VyOSDriver(NetworkDriver):
else: else:
level = 0 level = 0
# "set system login user alice authentication public-keys alice@example.com key 'ABC'" # "set system login user alice authentication public-keys
# alice@example.com key 'ABC'"
elif len(line) == 10 and line[8] == "key": elif len(line) == 10 and line[8] == "key":
sshkeys.append(line[9].strip("'")) sshkeys.append(line[9].strip("'"))
@@ -780,7 +757,6 @@ class VyOSDriver(NetworkDriver):
return user_auth return user_auth
def ping(self, destination, source="", ttl=255, timeout=5, size=100, count=5): def ping(self, destination, source="", ttl=255, timeout=5, size=100, count=5):
# does not support multiple destination yet # does not support multiple destination yet
@@ -793,7 +769,6 @@ class VyOSDriver(NetworkDriver):
command += "interface %s " % source command += "interface %s " % source
ping_result = dict() ping_result = dict()
output_ping = self._device.send_command(command) output_ping = self._device.send_command(command)
if "Unknown host" in output_ping: if "Unknown host" in output_ping:
@@ -805,12 +780,12 @@ class VyOSDriver(NetworkDriver):
ping_result["error"] = err ping_result["error"] = err
else: else:
# 'packet_info' example: # 'packet_info' example:
# ['5', 'packets', 'transmitted,' '5', 'received,' '0%', 'packet', 'loss,', 'time', '3997ms'] # ['5', 'packets', 'transmitted,' '5', 'received,' '0%', 'packet',
# 'loss,', 'time', '3997ms']
packet_info = output_ping.split("\n")[-2] packet_info = output_ping.split("\n")[-2]
packet_info = [x.strip() for x in packet_info.split()] packet_info = [x.strip() for x in packet_info.split()]
sent = int(packet_info[0]) sent = int(packet_info[0])
received = int(packet_info[3]) received = int(packet_info[3])
lost = sent - received lost = sent - received

2
setup.cfg
View File

@@ -5,7 +5,7 @@ ignore = D203,C901
[pylama:pep8] [pylama:pep8]
max_line_length = 100 max_line_length = 100
[pytest] [tools:pytest]
addopts = --cov=./ -vs addopts = --cov=./ -vs
json_report = report.json json_report = report.json
jsonapi = true jsonapi = true