witine/lookingglass
1
0
Fork 0
forked from mirrors/thatmattlove-hyperglass
Code Issues Pull requests Releases Activity

Refactor nos naming to type/device_type

Browse source
This commit is contained in:
thatmattlove 2021-09-13 10:00:21 -07:00
parent e3be569322
commit 723048d1d1
9 changed files with 91 additions and 81 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
hyperglass/exceptions/private.py
View file

@ -19,8 +19,8 @@ class ExternalError(PrivateHyperglassError):
class UnsupportedDevice(PrivateHyperglassError):
"""Raised when an input NOS is not in the supported NOS list."""
def __init__(self, nos: str) -> None:
"""Show the unsupported NOS and a list of supported drivers."""
def __init__(self, device_type: str) -> None:
"""Show the unsupported device type and a list of supported drivers."""
# Third Party
from netmiko.ssh_dispatcher import CLASS_MAPPER # type: ignore
@ -29,7 +29,7 @@ class UnsupportedDevice(PrivateHyperglassError):
drivers = ("", *[*DRIVER_MAP.keys(), *CLASS_MAPPER.keys()].sort())
driver_list = "\n - ".join(drivers)
super().__init__(message=f"'{nos}' is not supported. Must be one of:{driver_list}")
super().__init__(message=f"'{device_type}' is not supported. Must be one of:{driver_list}")
class InputValidationError(PrivateHyperglassError):

14
hyperglass/execution/drivers/_construct.py
View file

@ -41,14 +41,14 @@ class Construct:
# Set transport method based on NOS type
self.transport = "scrape"
if self.device.nos in TRANSPORT_REST:
if self.device.type in TRANSPORT_REST:
self.transport = "rest"
# Remove slashes from target for required platforms
if self.device.nos in TARGET_FORMAT_SPACE:
if self.device.type in TARGET_FORMAT_SPACE:
self.target = re.sub(r"\/", r" ", str(self.query.query_target))
with Formatter(self.device.nos, self.query.query_type) as formatter:
with Formatter(self.device.type, self.query.query_type) as formatter:
self.target = formatter(self.query.query_target)
def json(self, afi):
@ -100,9 +100,9 @@ class Construct:
class Formatter:
"""Modify query target based on the device's NOS requirements and the query type."""
def __init__(self, nos: str, query_type: str) -> None:
def __init__(self, device_type: str, query_type: str) -> None:
"""Initialize target formatting."""
self.nos = nos
self.device_type = device_type
self.query_type = query_type
def __enter__(self):
@ -116,10 +116,10 @@ class Formatter:
pass
def _get_formatter(self):
if self.nos in ("juniper", "juniper_junos"):
if self.device_type in ("juniper", "juniper_junos"):
if self.query_type == "bgp_aspath":
return self._juniper_bgp_aspath
if self.nos in ("bird", "bird_ssh"):
if self.device_type in ("bird", "bird_ssh"):
if self.query_type == "bgp_aspath":
return self._bird_bgp_aspath
elif self.query_type == "bgp_community":

12
hyperglass/execution/drivers/ssh_netmiko.py
View file

@ -22,14 +22,14 @@ from hyperglass.exceptions.public import AuthError, DeviceTimeout, ResponseEmpty
# Local
from .ssh import SSHConnection
netmiko_nos_globals = {
netmiko_device_globals = {
# Netmiko doesn't currently handle Mikrotik echo verification well,
# see ktbyers/netmiko#1600
"mikrotik_routeros": {"global_cmd_verify": False},
"mikrotik_switchos": {"global_cmd_verify": False},
}
netmiko_nos_send_args = {
netmiko_device_send_args = {
# Netmiko doesn't currently handle the Mikrotik prompt properly, see
# ktbyers/netmiko#1956
"mikrotik_routeros": {"expect_string": r"\S+\s\>\s$"},
@ -56,14 +56,14 @@ class NetmikoConnection(SSHConnection):
else:
log.debug("Connecting directly to {}", self.device.name)
global_args = netmiko_nos_globals.get(self.device.nos, {})
global_args = netmiko_device_globals.get(self.device.type, {})
send_args = netmiko_nos_send_args.get(self.device.nos, {})
send_args = netmiko_device_send_args.get(self.device.type, {})
driver_kwargs = {
"host": host or self.device._target,
"port": port or self.device.port,
"device_type": self.device.nos,
"device_type": self.device.type,
"username": self.device.credential.username,
"global_delay_factor": params.netmiko_delay_factor,
"timeout": math.floor(params.request_timeout * 1.25),
@ -71,7 +71,7 @@ class NetmikoConnection(SSHConnection):
**global_args,
}
if "_telnet" in self.device.nos:
if "_telnet" in self.device.type:
# Telnet devices with a low delay factor (default) tend to
# throw login errors.
driver_kwargs["global_delay_factor"] = 2

10
hyperglass/execution/drivers/ssh_scrapli.py
View file

@ -55,10 +55,10 @@ driver_global_args = {
}
def _map_driver(nos: str) -> AsyncGenericDriver:
driver = SCRAPLI_DRIVER_MAP.get(nos)
def _map_driver(_type: str) -> AsyncGenericDriver:
driver = SCRAPLI_DRIVER_MAP.get(_type)
if driver is None:
raise UnsupportedDevice("{nos} is not supported by scrapli.", nos=nos)
raise UnsupportedDevice(_type)
return driver
@ -71,7 +71,7 @@ class ScrapliConnection(SSHConnection):
Directly connects to the router via Netmiko library, returns the
command output.
"""
driver = _map_driver(self.device.nos)
driver = _map_driver(self.device.type)
if host is not None:
log.debug(
@ -83,7 +83,7 @@ class ScrapliConnection(SSHConnection):
else:
log.debug("Connecting directly to {}", self.device.name)
global_args = driver_global_args.get(self.device.nos, {})
global_args = driver_global_args.get(self.device.type, {})
driver_kwargs = {
"host": host or self.device._target,

21
hyperglass/models/commands/__init__.py
View file

@ -17,7 +17,7 @@ from .nokia_sros import NokiaSROSCommands
from .mikrotik_routeros import MikrotikRouterOS
from .mikrotik_switchos import MikrotikSwitchOS
_NOS_MAP = {
_DEVICE_TYPE_MAP = {
"arista_eos": AristaEOSCommands,
"bird": BIRDCommands,
"cisco_ios": CiscoIOSCommands,
@ -53,19 +53,10 @@ class Commands(HyperglassModel, extra="allow", validate_all=False):
@classmethod
def import_params(cls, **input_params):
"""Import loaded YAML, initialize per-command definitions.
Dynamically set attributes for the command class.
Arguments:
input_params {dict} -- Unvalidated command definitions
Returns:
{object} -- Validated commands object
"""
"""Import loaded YAML, initialize per-command definitions."""
obj = Commands()
for nos, cmds in input_params.items():
nos_cmd_set = _NOS_MAP.get(nos, CommandGroup)
nos_cmds = nos_cmd_set(**cmds)
setattr(obj, nos, nos_cmds)
for device_type, cmds in input_params.items():
cmd_set = _DEVICE_TYPE_MAP.get(device_type, CommandGroup)
cmds = cmd_set(**cmds)
setattr(obj, device_type, cmds)
return obj

58
hyperglass/models/config/devices.py
View file

@ -8,11 +8,11 @@ from pathlib import Path
from ipaddress import IPv4Address, IPv6Address
# Third Party
from pydantic import StrictInt, StrictStr, StrictBool, validator, root_validator
from pydantic import StrictInt, StrictStr, StrictBool, validator, root_validator, Field
# Project
from hyperglass.log import log
from hyperglass.util import get_driver, get_fmt_keys, validate_nos, resolve_hostname
from hyperglass.util import get_driver, get_fmt_keys, validate_device_type, resolve_hostname
from hyperglass.constants import SCRAPE_HELPERS, SUPPORTED_STRUCTURED_OUTPUT
from hyperglass.exceptions.private import ConfigError, UnsupportedDevice
from hyperglass.models.commands.generic import Directive
@ -39,7 +39,7 @@ class Device(HyperglassModelWithId, extra="allow"):
display_name: Optional[StrictStr]
port: StrictInt = 22
ssl: Optional[Ssl]
nos: StrictStr
type: StrictStr = Field(..., alias="nos")
commands: List[Directive]
structured_output: Optional[StrictBool]
driver: Optional[SupportedDriver]
@ -132,20 +132,34 @@ class Device(HyperglassModelWithId, extra="allow"):
)
return value
@validator("type", pre=True, always=True)
def validate_type(cls: "Device", value: Any, values: Dict[str, Any]) -> str:
"""Validate device type."""
legacy = values.pop("nos", None)
if legacy is not None and value is None:
log.warning(
"The 'nos' field on device '{}' has been deprecated and will be removed in a future release. Use the 'type' field moving forward.",
values.get("name", values.get("display_name", "Unknown")),
)
return legacy
if value is not None:
return value
raise ValueError("type is missing")
@validator("structured_output", pre=True, always=True)
def validate_structured_output(cls, value: bool, values: Dict) -> bool:
"""Validate structured output is supported on the device & set a default."""
if value is True:
if values["nos"] not in SUPPORTED_STRUCTURED_OUTPUT:
if values["type"] not in SUPPORTED_STRUCTURED_OUTPUT:
raise ConfigError(
"The 'structured_output' field is set to 'true' on device '{d}' with "
+ "NOS '{n}', which does not support structured output",
d=values["name"],
n=values["nos"],
n=values["type"],
)
return value
elif value is None and values["nos"] in SUPPORTED_STRUCTURED_OUTPUT:
elif value is None and values["type"] in SUPPORTED_STRUCTURED_OUTPUT:
value = True
else:
value = False
@ -166,32 +180,32 @@ class Device(HyperglassModelWithId, extra="allow"):
return value
@root_validator(pre=True)
def validate_nos_commands(cls, values: Dict) -> Dict:
"""Validate & rewrite NOS, set default commands."""
def validate_device_commands(cls, values: Dict) -> Dict:
"""Validate & rewrite device type, set default commands."""
nos = values.get("nos", "")
if not nos:
# Ensure nos is defined.
_type = values.get("type", values.get("nos"))
if not _type:
# Ensure device type is defined.
raise ValueError(
f"Device {values['name']} is missing a 'nos' (Network Operating System) property."
f"Device {values['name']} is missing a 'type' (Network Operating System) property."
)
if nos in SCRAPE_HELPERS.keys():
if _type in SCRAPE_HELPERS.keys():
# Rewrite NOS to helper value if needed.
nos = SCRAPE_HELPERS[nos]
_type = SCRAPE_HELPERS[_type]
# Verify NOS is supported by hyperglass.
supported, _ = validate_nos(nos)
# Verify device type is supported by hyperglass.
supported, _ = validate_device_type(_type)
if not supported:
raise UnsupportedDevice(nos=nos)
raise UnsupportedDevice(_type)
values["nos"] = nos
values["type"] = _type
commands = values.get("commands")
if commands is None:
# If no commands are defined, set commands to the NOS.
inferred = values["nos"]
inferred = values["type"]
# If the _telnet prefix is added, remove it from the command
# profile so the commands are the same regardless of
@ -206,7 +220,7 @@ class Device(HyperglassModelWithId, extra="allow"):
@validator("driver")
def validate_driver(cls, value: Optional[str], values: Dict) -> Dict:
"""Set the correct driver and override if supported."""
return get_driver(values["nos"], value)
return get_driver(values["type"], value)
class Devices(HyperglassModel, extra="allow"):
@ -215,7 +229,6 @@ class Devices(HyperglassModel, extra="allow"):
ids: List[StrictStr] = []
hostnames: List[StrictStr] = []
objects: List[Device] = []
all_nos: List[StrictStr] = []
def __init__(self, input_params: List[Dict]) -> None:
"""Import loaded YAML, initialize per-network definitions.
@ -224,7 +237,6 @@ class Devices(HyperglassModel, extra="allow"):
set attributes for the devices class. Builds lists of common
attributes for easy access in other modules.
"""
all_nos = set()
objects = set()
hostnames = set()
ids = set()
@ -242,13 +254,11 @@ class Devices(HyperglassModel, extra="allow"):
hostnames.add(device.name)
ids.add(device.id)
objects.add(device)
all_nos.add(device.nos)
# Convert the de-duplicated sets to a standard list, add lists
# as class attributes. Sort router list by router name attribute
init_kwargs["ids"] = list(ids)
init_kwargs["hostnames"] = list(hostnames)
init_kwargs["all_nos"] = list(all_nos)
init_kwargs["objects"] = sorted(objects, key=lambda x: x.name)
super().__init__(**init_kwargs)

35
hyperglass/models/config/proxy.py
View file

@ -1,13 +1,14 @@
"""Validate SSH proxy configuration variables."""
# Standard Library
from typing import Union
from typing import Union, Any, Dict
from ipaddress import IPv4Address, IPv6Address
# Third Party
from pydantic import StrictInt, StrictStr, validator
from pydantic import StrictInt, StrictStr, validator, Field
# Project
from hyperglass.log import log
from hyperglass.util import resolve_hostname
from hyperglass.exceptions.private import ConfigError, UnsupportedDevice
@ -23,7 +24,7 @@ class Proxy(HyperglassModel):
address: Union[IPv4Address, IPv6Address, StrictStr]
port: StrictInt = 22
credential: Credential
nos: StrictStr = "linux_ssh"
type: StrictStr = Field("linux_ssh", alias="nos")
@property
def _target(self):
@ -42,14 +43,22 @@ class Proxy(HyperglassModel):
)
return value
@validator("nos")
def supported_nos(cls, value, values):
"""Verify NOS is supported by hyperglass."""
if not value == "linux_ssh":
raise UnsupportedDevice(
"Proxy '{p}' uses NOS '{n}', which is currently unsupported.",
p=values["name"],
n=value,
@validator("type", pre=True, always=True)
def validate_type(cls: "Proxy", value: Any, values: Dict[str, Any]) -> str:
"""Validate device type."""
legacy = values.pop("nos", None)
if legacy is not None and value is None:
log.warning(
"The 'nos' field on proxy '{}' has been deprecated and will be removed in a future release. Use the 'type' field moving forward.",
values.get("name", "Unknown"),
)
return value
return legacy
if value is not None:
if value != "linux_ssh":
raise UnsupportedDevice(
"Proxy '{p}' uses type '{t}', which is currently unsupported.",
p=values["name"],
t=value,
)
return value
raise ValueError("type is missing")

2
hyperglass/plugins/tests/test_bgp_route_juniper.py
View file

@ -28,7 +28,7 @@ def _tester(sample: str):
address="127.0.0.1",
network={"name": "Test Network", "display_name": "Test Network"},
credential={"username": "", "password": ""},
nos="juniper",
type="juniper",
structured_output=True,
commands=[{"id": "test", "name": "Test", "rules": []}],
)

14
hyperglass/util/__init__.py
View file

@ -31,7 +31,7 @@ from netmiko.ssh_dispatcher import CLASS_MAPPER # type: ignore
from hyperglass.log import log
from hyperglass.constants import DRIVER_MAP
ALL_NOS = {*DRIVER_MAP.keys(), *CLASS_MAPPER.keys()}
ALL_DEVICE_TYPES = {*DRIVER_MAP.keys(), *CLASS_MAPPER.keys()}
ALL_DRIVERS = {*DRIVER_MAP.values(), "netmiko"}
DeepConvert = TypeVar("DeepConvert", bound=Dict[str, Any])
@ -260,24 +260,24 @@ def make_repr(_class):
return f'{_class.__name__}({", ".join(_process_attrs(dir(_class)))})'
def validate_nos(nos):
"""Validate device NOS is supported."""
def validate_device_type(_type: str) -> Tuple[bool, Union[None, str]]:
"""Validate device type is supported."""
result = (False, None)
if nos in ALL_NOS:
result = (True, DRIVER_MAP.get(nos, "netmiko"))
if _type in ALL_DEVICE_TYPES:
result = (True, DRIVER_MAP.get(_type, "netmiko"))
return result
def get_driver(nos: str, driver: Optional[str]) -> str:
def get_driver(_type: str, driver: Optional[str]) -> str:
"""Determine the appropriate driver for a device."""
if driver is None:
# If no driver is set, use the driver map with netmiko as
# fallback.
return DRIVER_MAP.get(nos, "netmiko")
return DRIVER_MAP.get(_type, "netmiko")
elif driver in ALL_DRIVERS:
# If a driver is set and it is valid, allow it.
return driver