Skip to content
Snippets Groups Projects
Commit 3c474fa9 authored by Erik Reid's avatar Erik Reid
Browse files

simplified pyEZ and initial mocking attempt

parent 074d797d
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
resource_management/hardware/router.py
+ 86
− 123
View file @ 3c474fa9
import logging
from pickle import NONE
from jnpr.junos import Device
from jnpr.junos.exception import ConnectError, ConnectTimeoutError
from jnpr.junos.op.fpc import FpcHwTable
from jnpr.junos.op.phyport import PhyPortTable
import environment
environment.setup_logging()
logger = logging.getLogger(__name__)
class junos_router(object):
"""Gets line card and ports information from a Junos router
params:
fqdn(str): The fqdn of the router
creds(dict): Authorization data for the router
"""
def __init__(self, fqdn: str, creds: dict):
self.fqdn = fqdn
self.creds = creds
self.dev = NONE
self._line_card_json = NONE
self._ports_json = NONE
self._open_device()
if self.dev:
self._get_hardware_info()
def _open_device(self):
# Establish connection to router
creds = self.creds
dev = NONE
try:
if "password" in creds and "ssh-config" in creds:
dev = Device(host=self.fqdn,
passwd=creds["password"],
ssh_config=creds["ssh-config"])
elif "password" in creds:
dev = Device(host=self.fqdn,
user=creds["username"],
passwd=creds["password"])
elif "private-key" in creds and "ssh-config" in creds:
dev = Device(host=self.fqdn,
ssh_private_key_file=creds["private-key"],
ssh_config=creds["ssh-config"])
elif "private-key" in creds:
dev = Device(host=self.fqdn,
ssh_private_key_file=creds["private-key"])
except ConnectError as err:
logger.error("Cannot connect to device: {0}".format(err))
except ConnectTimeoutError as err:
logger.error("Timeout connecting to device: {0}".format(err))
except Exception as err:
logger.error("Error reading hardware information for {1}: {0}"
.format(err, self.fqdn))
if dev:
dev.open()
self.dev = dev
else:
raise Exception("Error during connection to router")
def _get_hardware_info(self):
# For line_cards table
self._line_cards = FpcHwTable(self.dev)
self._line_cards.get()
self._line_card_json = self._line_cards.to_json()
# For ports table
self._ports = PhyPortTable(self.dev)
self._ports.get()
self._ports_json = self._ports.to_json()
def close_device(self):
self.dev.close
def load_router(self) -> dict:
"""Returns hardware info of a junos router
by line cards and ports.
"""
line_cards = []
ports = []
router = {}
try:
for key, value in self._line_cards.items():
position = int(key.replace("FPC ", ""))
model = value[4][1] or "Unknown"
line_cards.append({"position": position,
"model": model})
for key, value in self._ports.items():
name = key.split("/", 1)[1]
speed = key[:2]
if speed == "ge":
speed = 1000
elif speed == "xe":
speed = 10000
elif speed == "et":
speed = 40000
else:
speed = -1
line_card_position = int(key[3])
ports.append({"name": name,
"speed": speed,
"_line_card_position": line_card_position})
router = {"node": self.fqdn, "line_cards": line_cards, "ports": ports}
except Exception as err:
logger.error("Error reading hardware information for {1}: {0}"
.format(err, self.fqdn))
return router
def get_hardware_info_as_json(self):
"""
Returns the line cards and ports of the router
in the JSON format.
The returned dict can be accessed as follows:
-------------------------------
line_card_json = router['line_cards']
ports_json = router['ports']
"""
return {'line_cards': self._line_card_json, 'ports': self._ports_json}
import contextlib
@contextlib.contextmanager
def device(hostname, username, key_filename, port=830):
dev = Device(
host=hostname,
port=port,
user=username,
ssh_private_key_file=key_filename)
try:
dev.open()
yield dev
finally:
dev.close()
def load_line_cards(hostname, username, key_filename, port=830):
with device(
hostname=hostname,
port=port,
username=username,
key_filename=key_filename) as router:
phy_port_table = PhyPortTable(router)
phy_port_table.get()
def _make_port_element(p):
return {
'name': p.name,
'line-card-position': int(p.name[3])
}
ports = list(map(_make_port_element, phy_port_table))
fpc_hw_table = FpcHwTable(router)
fpc_hw_table.get()
for idx, fpc in enumerate(fpc_hw_table):
yield {
'name': fpc.name,
'ports': [p for p in ports if p['line-card-position'] == idx]
}
if __name__ == '__main__':
# # # core1
# # dev = Device(
# # host='62.40.111.120',
# # port=10830,
# # user='ansible',
# # ssh_private_key_file='/Users/reid/.ssh/eve_ng_ansible_rsa')
#
# # core1
# dev = Device(
# host='62.40.111.120',
# port=10830,
# user='username',
# ssh_private_key_file='key-filename')
#
# # core2
# dev = Device(
# host='62.40.111.120',
# port=20830,
# user='username',
# ssh_private_key_file='key-filename')
#
# # core3
# dev = Device(
# host='62.40.111.120',
# port=30830,
# user='username',
# ssh_private_key_file='key-filename')
#
logging.basicConfig(level=logging.DEBUG)
for port in (10830, 20830, 30830):
fpcs = load_line_cards(
hostname='62.40.111.120',
port=port,
username='ansible',
key_filename='/Users/reid/.ssh/eve_ng_ansible_rsa')
fpcs = list(fpcs)
print(fpcs)
test/data/vmx-load-line-cards-output.json 0 → 100644
+ 47
− 0
View file @ 3c474fa9
[
{
"name": "FPC 0",
"ports": [
{
"name": "ge-0/0/0",
"line-card-position": 0
},
{
"name": "ge-0/0/1",
"line-card-position": 0
},
{
"name": "ge-0/0/2",
"line-card-position": 0
},
{
"name": "ge-0/0/3",
"line-card-position": 0
},
{
"name": "ge-0/0/4",
"line-card-position": 0
},
{
"name": "ge-0/0/5",
"line-card-position": 0
},
{
"name": "ge-0/0/6",
"line-card-position": 0
},
{
"name": "ge-0/0/7",
"line-card-position": 0
},
{
"name": "ge-0/0/8",
"line-card-position": 0
},
{
"name": "ge-0/0/9",
"line-card-position": 0
}
]
}
]
\ No newline at end of file
test/test_parse_router.py
+ 15
− 8
View file @ 3c474fa9
from resource_management.db import model
from unittest.mock import patch, MagicMock
from resource_management.hardware.test1 import load_line_cards
# Parse line card and ports from one router
def test_parse_router(load_router):
fqdn = "core1.router"
line_card, ports = load_router
for key, value in line_card.items():
print(key, value)
\ No newline at end of file
# This is not yet working ... try to find the right mocks that will
# allow us to run all lines of code in router.py
#
# another possible approach: try to mock the netconf responses and use
# the code from https://github.com/GIC-de/Juniper-PyEZ-Unit-Testing
@patch('resource_management.hardware.test1.PhyPortTable')
@patch('resource_management.hardware.test1.FpcHwTable')
@patch('resource_management.hardware.test1.Device')
def test_load_line_cards(Device, FpcHwTable, PhyPortTable):
PhyPortTable.return_value = MagicMock()
PhyPortTable.return_value.return_value = 'aaaa'
fpcs = load_line_cards(hostname='bogus', username='bogus', key_filename='no file')
assert len(list(fpcs)) == 1
\ No newline at end of file
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment