mirrors/thatmattlove-hyperglass
1
0
Fork 1
mirror of https://github.com/thatmattlove/hyperglass.git synced 2026-04-17 21:38:27 +00:00
Code Issues Activity

Refactor code for improved readability and consistency across multiple files using black

Browse source
This commit is contained in:
Wilhelm Schonfeldt 2025-09-26 09:43:06 +02:00
parent 22c8f8310e
commit ae6a1a0bb8
18 changed files with 184 additions and 109 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
hyperglass/compat/_sshtunnel.py
View file

@ -272,7 +272,9 @@ class _ForwardServer(socketserver.TCPServer): # Not Threading
def handle_error(self, request, client_address):
(exc_class, exc, tb) = sys.exc_info()
self.logger.bind(source=request.getsockname()).error("Could not establish connection to remote side of the tunnel")
self.logger.bind(source=request.getsockname()).error(
"Could not establish connection to remote side of the tunnel"
)
self.tunnel_ok.put(False)
@property
@ -1023,7 +1025,7 @@ class SSHTunnelForwarder:
msg = template.format(self.ssh_host, self.ssh_port, e.args[0])
self.logger.error(msg)
return
for (rem, loc) in zip(self._remote_binds, self._local_binds):
for rem, loc in zip(self._remote_binds, self._local_binds):
try:
self._make_ssh_forward_server(rem, loc)
except BaseSSHTunnelForwarderError as e:
@ -1053,7 +1055,7 @@ class SSHTunnelForwarder:
bind_addresses = [bind_address]
if not is_remote:
# Add random port if missing in local bind
for (i, local_bind) in enumerate(bind_addresses):
for i, local_bind in enumerate(bind_addresses):
if isinstance(local_bind, tuple) and len(local_bind) == 1:
bind_addresses[i] = (local_bind[0], 0)
check_addresses(bind_addresses, is_remote)
@ -1400,9 +1402,11 @@ class SSHTunnelForwarder:
def __str__(self) -> str:
credentials = {
"password": self.ssh_password,
"pkeys": [(key.get_name(), hexlify(key.get_fingerprint())) for key in self.ssh_pkeys]
if any(self.ssh_pkeys)
else None,
"pkeys": (
[(key.get_name(), hexlify(key.get_fingerprint())) for key in self.ssh_pkeys]
if any(self.ssh_pkeys)
else None
),
}
_remove_none_values(credentials)
template = os.linesep.join(

9
hyperglass/constants.py
View file

@ -19,7 +19,14 @@ TARGET_FORMAT_SPACE = ("huawei", "huawei_vrpv8")
TARGET_JUNIPER_ASPATH = ("juniper", "juniper_junos")
SUPPORTED_STRUCTURED_OUTPUT = ("frr", "juniper", "arista_eos", "huawei", "mikrotik_routeros", "mikrotik_switchos")
SUPPORTED_STRUCTURED_OUTPUT = (
"frr",
"juniper",
"arista_eos",
"huawei",
"mikrotik_routeros",
"mikrotik_switchos",
)
CONFIG_EXTENSIONS = ("py", "yaml", "yml", "json", "toml")

8
hyperglass/defaults/directives/huawei.py
View file

@ -177,13 +177,13 @@ HuaweiBGPRouteTable = BuiltinDirective(
command="",
),
# Regra DENY AS PREFIXO
#RuleWithIPv4(
# RuleWithIPv4(
# condition="x.x.x.x/xx",
# ge="xx",
# le="32",
# action="deny",
# command="",
#),
# ),
RuleWithIPv4(
condition="0.0.0.0/0",
ge="8",
@ -232,13 +232,13 @@ HuaweiBGPRouteTable = BuiltinDirective(
command="",
),
# REGRA DENY AS PREFIXO
#RuleWithIPv6(
# RuleWithIPv6(
# condition="x.x.x.x/xx",
# ge="XX",
# le="128",
# action="deny",
# command="",
#),
# ),
RuleWithIPv6(
condition="::/0",
ge="10",

20
hyperglass/defaults/directives/mikrotik.py
View file

@ -35,7 +35,7 @@ Mikrotik_BGPRoute = BuiltinDirective(
# v7
command="routing route print detail without-paging where {target} in dst-address bgp and dst-address !=0.0.0.0/0",
# v6
#command="ip route print detail without-paging where {target} in dst-address bgp and dst-address !=0.0.0.0/0",
# command="ip route print detail without-paging where {target} in dst-address bgp and dst-address !=0.0.0.0/0",
),
RuleWithIPv6(
condition="::/0",
@ -43,7 +43,7 @@ Mikrotik_BGPRoute = BuiltinDirective(
# v7
command="routing route print detail without-paging where {target} in dst-address bgp and dst-address !=::/0",
# v6
#command="ipv6 route print detail without-paging where {target} in dst-address bgp and dst-address !=::/0",
# command="ipv6 route print detail without-paging where {target} in dst-address bgp and dst-address !=::/0",
),
],
field=Text(description="IP Address, Prefix, or Hostname"),
@ -66,7 +66,7 @@ Mikrotik_BGPASPath = BuiltinDirective(
)
],
field=Text(description="AS Path Regular Expression"),
plugins=["mikrotik_normalize_input","mikrotik_garbage_output", "bgp_routestr_mikrotik"],
plugins=["mikrotik_normalize_input", "mikrotik_garbage_output", "bgp_routestr_mikrotik"],
table_output="__hyperglass_mikrotik_bgp_aspath_table__",
platforms=PLATFORMS,
)
@ -85,7 +85,7 @@ Mikrotik_BGPCommunity = BuiltinDirective(
)
],
field=Text(description="BGP Community String"),
plugins=["mikrotik_normalize_input","mikrotik_garbage_output", "bgp_routestr_mikrotik"],
plugins=["mikrotik_normalize_input", "mikrotik_garbage_output", "bgp_routestr_mikrotik"],
table_output="__hyperglass_mikrotik_bgp_community_table__",
platforms=PLATFORMS,
)
@ -183,13 +183,13 @@ MikrotikBGPRouteTable = BuiltinDirective(
command="",
),
# Regra DENY AS PREFIXO
#RuleWithIPv4(
# RuleWithIPv4(
# condition="x.x.x.x/x",
# ge="xx",
# le="32",
# action="deny",
# command="",
#),
# ),
RuleWithIPv4(
condition="0.0.0.0/0",
ge="8",
@ -198,7 +198,7 @@ MikrotikBGPRouteTable = BuiltinDirective(
# v7
command="routing route print detail without-paging where {target} in dst-address bgp and dst-address !=0.0.0.0/0",
# v6
#command="ip route print detail without-paging where {target} in dst-address bgp and dst-address !=0.0.0.0/0",
# command="ip route print detail without-paging where {target} in dst-address bgp and dst-address !=0.0.0.0/0",
),
# REGRA DENY SITE LOCAL DEPRECIADO RFC 3879
RuleWithIPv6(
@ -241,20 +241,20 @@ MikrotikBGPRouteTable = BuiltinDirective(
command="",
),
# REGRA DENY AS PREFIXO
#RuleWithIPv6(
# RuleWithIPv6(
# condition="xxxx:xxxx::/xx",
# ge="xx",
# le="128",
# action="deny",
# command="",
#),
# ),
RuleWithIPv6(
condition="::/0",
action="permit",
# v7
command="routing route print detail without-paging where {target} in dst-address bgp and dst-address !=::/0",
# v6
#command="ipv6 route print detail without-paging where {target} in dst-address bgp and dst-address !=::/0",
# command="ipv6 route print detail without-paging where {target} in dst-address bgp and dst-address !=::/0",
),
],
field=Text(description="IP Address, Prefix, or Hostname"),

4
hyperglass/external/_base.py vendored
View file

@ -208,7 +208,9 @@ class BaseExternal:
data,
timeout,
response_required,
) = itemgetter(*kwargs.keys())(kwargs)
) = itemgetter(
*kwargs.keys()
)(kwargs)
if method.upper() not in supported_methods:
raise self._exception(

19
hyperglass/external/rpki.py vendored
View file

@ -13,20 +13,27 @@ if t.TYPE_CHECKING:
from ipaddress import IPv4Address, IPv6Address
RPKI_STATE_MAP = {
"Invalid": 0, "invalid": 0,
"Valid": 1, "valid": 1,
"NotFound": 2, "notfound": 2, "not_found": 2, "not-found": 2,
"Unknown": 2, "unknown": 2,
"DEFAULT": 3
"Invalid": 0,
"invalid": 0,
"Valid": 1,
"valid": 1,
"NotFound": 2,
"notfound": 2,
"not_found": 2,
"not-found": 2,
"Unknown": 2,
"unknown": 2,
"DEFAULT": 3,
}
RPKI_NAME_MAP = {v: k for k, v in RPKI_STATE_MAP.items()}
CACHE_KEY = "hyperglass.external.rpki"
def rpki_state(
prefix: t.Union["IPv4Address", "IPv6Address", str],
asn: t.Union[int, str],
backend: str = "cloudflare",
rpki_server_url: str = ""
rpki_server_url: str = "",
) -> int:
"""Get RPKI state and map to expected integer."""
_log = log.bind(prefix=prefix, asn=asn)

8
hyperglass/external/tests/test_rpki.py vendored
View file

@ -19,8 +19,8 @@ def test_rpki():
result = rpki_state(prefix, asn)
result_name = RPKI_NAME_MAP.get(result, "No Name")
expected_name = RPKI_NAME_MAP.get(expected, "No Name")
assert result == expected, (
"RPKI State for '{}' via AS{!s} '{}' ({}) instead of '{}' ({})".format(
prefix, asn, result, result_name, expected, expected_name
)
assert (
result == expected
), "RPKI State for '{}' via AS{!s} '{}' ({}) instead of '{}' ({})".format(
prefix, asn, result, result_name, expected, expected_name
)

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

@ -357,9 +357,9 @@ class Devices(MultiModel, model=Device, unique_by="id"):
"group": group,
"id": device.id,
"name": device.name,
"avatar": f"/images/{device.avatar.name}"
if device.avatar is not None
else None,
"avatar": (
f"/images/{device.avatar.name}" if device.avatar is not None else None
),
"description": device.description,
"directives": [d.frontend() for d in device.directives],
}

1
hyperglass/models/config/structured.py
View file

@ -24,6 +24,7 @@ class StructuredRpki(HyperglassModel):
backend: str = "cloudflare"
rpki_server_url: str = ""
class Structured(HyperglassModel):
"""Control structured data responses."""

58
hyperglass/models/parsing/huawei.py
View file

@ -21,10 +21,11 @@ RPKI_STATE_MAP = {
"unverified": 3,
}
def remove_prefix(text: str, prefix: str) -> str:
"""Remove prefix from text if it exists."""
if text.startswith(prefix):
return text[len(prefix):]
return text[len(prefix) :]
return text
@ -38,6 +39,7 @@ class HuaweiBase(HyperglassModel, extra="ignore"):
class HuaweiPaths(HuaweiBase):
"""BGP paths information."""
available: int = 0
best: int = 0
select: int = 0
@ -118,6 +120,7 @@ class HuaweiRouteEntry(HuaweiBase):
"""Get peer router ID."""
return self.from_addr
def _extract_paths(line: str) -> HuaweiPaths:
"""Extract paths information from line like 'Paths: 3 available, 1 best, 1 select, 0 best-external, 0 add-path'."""
paths_data = {
@ -150,7 +153,14 @@ def _extract_route_entries(lines: t.List[str]) -> t.List[HuaweiRouteEntry]:
# Split lines into route blocks using empty lines as separators
size = len(lines)
idx_list = [idx + 1 for idx, val in enumerate(lines) if val.strip() == ""]
entries = [lines[i:j] for i, j in zip([0] + idx_list, idx_list + ([size] if idx_list[-1] != size else []))] if idx_list else [lines]
entries = (
[
lines[i:j]
for i, j in zip([0] + idx_list, idx_list + ([size] if idx_list[-1] != size else []))
]
if idx_list
else [lines]
)
for route_entry in entries:
if not route_entry:
@ -188,7 +198,9 @@ def _extract_route_entries(lines: t.List[str]) -> t.List[HuaweiRouteEntry]:
continue
if info.startswith("BGP routing table entry information of"):
route_data["prefix"] = remove_prefix(info, "BGP routing table entry information of ").rstrip(":")
route_data["prefix"] = remove_prefix(
info, "BGP routing table entry information of "
).rstrip(":")
elif info.startswith("From:"):
route_data["from_addr"] = remove_prefix(info, "From: ").split(" (")[0]
elif info.startswith("Route Duration:"):
@ -205,7 +217,9 @@ def _extract_route_entries(lines: t.List[str]) -> t.List[HuaweiRouteEntry]:
minutes = int(m_match.group(1)) if m_match else 0
seconds = int(s_match.group(1)) if s_match else 0
route_data["duration"] = days * 24 * 60 * 60 + hours * 60 * 60 + minutes * 60 + seconds
route_data["duration"] = (
days * 24 * 60 * 60 + hours * 60 * 60 + minutes * 60 + seconds
)
except:
route_data["duration"] = 0
elif info.startswith("Direct Out-interface:"):
@ -215,23 +229,34 @@ def _extract_route_entries(lines: t.List[str]) -> t.List[HuaweiRouteEntry]:
elif info.startswith("Relay IP Nexthop:"):
route_data["relay_ip_next_hop"] = remove_prefix(info, "Relay IP Nexthop: ")
elif info.startswith("Relay IP Out-Interface:"):
route_data["relay_ip_out_interface"] = remove_prefix(info, "Relay IP Out-Interface: ")
route_data["relay_ip_out_interface"] = remove_prefix(
info, "Relay IP Out-Interface: "
)
elif info.startswith("Qos information :"):
route_data["qos"] = remove_prefix(info, "Qos information : ")
elif info.startswith("Community:"):
communities_str = remove_prefix(info, "Community: ")
if communities_str and communities_str.lower() != "none":
communities = [c.strip().replace("<", "").replace(">", "") for c in communities_str.split(", ")]
communities = [
c.strip().replace("<", "").replace(">", "")
for c in communities_str.split(", ")
]
route_data["communities"] = [c for c in communities if c]
elif info.startswith("Large-Community:"):
large_communities_str = remove_prefix(info, "Large-Community: ")
if large_communities_str and large_communities_str.lower() != "none":
large_communities = [c.strip().replace("<", "").replace(">", "") for c in large_communities_str.split(", ")]
large_communities = [
c.strip().replace("<", "").replace(">", "")
for c in large_communities_str.split(", ")
]
route_data["large_communities"] = [c for c in large_communities if c]
elif info.startswith("Ext-Community:"):
ext_communities_str = remove_prefix(info, "Ext-Community: ")
if ext_communities_str and ext_communities_str.lower() != "none":
ext_communities = [c.strip().replace("<", "").replace(">", "") for c in ext_communities_str.split(", ")]
ext_communities = [
c.strip().replace("<", "").replace(">", "")
for c in ext_communities_str.split(", ")
]
route_data["ext_communities"] = [c for c in ext_communities if c]
elif info.startswith("AS-path"):
values = info.split(",")
@ -240,7 +265,9 @@ def _extract_route_entries(lines: t.List[str]) -> t.List[HuaweiRouteEntry]:
if v.startswith("AS-path"):
as_path_str = remove_prefix(v, "AS-path ")
try:
route_data["as_path"] = [int(a) for a in as_path_str.split() if a.isdigit()]
route_data["as_path"] = [
int(a) for a in as_path_str.split() if a.isdigit()
]
except ValueError:
route_data["as_path"] = []
elif v.startswith("origin"):
@ -282,7 +309,9 @@ def _extract_route_entries(lines: t.List[str]) -> t.List[HuaweiRouteEntry]:
route = HuaweiRouteEntry(**route_data)
routes.append(route)
except Exception as e:
log.warning(f'Failed to create route entry for prefix {{route_data.get("prefix", "unknown")}}: {{e}}')
log.warning(
f'Failed to create route entry for prefix {{route_data.get("prefix", "unknown")}}: {{e}}'
)
continue
return routes
@ -323,7 +352,9 @@ class HuaweiBGPTable(HuaweiBase):
instance.local_router_id = remove_prefix(line, "BGP local router ID : ").strip()
elif "Local AS number" in line:
try:
instance.local_as_number = int(remove_prefix(line, "Local AS number : ").strip())
instance.local_as_number = int(
remove_prefix(line, "Local AS number : ").strip()
)
except ValueError:
instance.local_as_number = 0
elif line.strip().startswith("Paths:"):
@ -353,7 +384,9 @@ class HuaweiBGPTable(HuaweiBase):
"source_as": route.source_as,
"source_rid": route.source_rid,
"peer_rid": route.peer_rid,
"rpki_state": RPKI_STATE_MAP.get("unknown") if route.is_valid else RPKI_STATE_MAP.get("valid"),
"rpki_state": (
RPKI_STATE_MAP.get("unknown") if route.is_valid else RPKI_STATE_MAP.get("valid")
),
}
routes.append(route_data)
@ -363,4 +396,3 @@ class HuaweiBGPTable(HuaweiBase):
routes=routes,
winning_weight="high",
)

44
hyperglass/models/parsing/mikrotik.py
View file

@ -21,22 +21,26 @@ RPKI_STATE_MAP = {
"unverified": 3,
}
def remove_prefix(text: str, prefix: str) -> str:
if text.startswith(prefix):
return text[len(prefix):]
return text[len(prefix) :]
return text
# Regex to find key=value pairs. The key can contain dots and hyphens.
# The value can be quoted or a single word.
TOKEN_RE = re.compile(r'([a-zA-Z0-9_.-]+)=(".*?"|\S+)')
# Regex to find flags at the beginning of a line (e.g., "Ab dst-address=...")
FLAGS_RE = re.compile(r'^\s*([DXIAcmsroivmyH\+b]+)\s+')
FLAGS_RE = re.compile(r"^\s*([DXIAcmsroivmyH\+b]+)\s+")
class MikrotikBase(HyperglassModel, extra="ignore"):
def __init__(self, **kwargs: t.Any) -> None:
super().__init__(**kwargs)
class MikrotikPaths(MikrotikBase):
available: int = 0
best: int = 0
@ -44,8 +48,10 @@ class MikrotikPaths(MikrotikBase):
best_external: int = 0
add_path: int = 0
class MikrotikRouteEntry(MikrotikBase):
"""MikroTik Route Entry."""
model_config = ConfigDict(validate_assignment=False)
prefix: str
@ -102,6 +108,7 @@ class MikrotikRouteEntry(MikrotikBase):
def peer_rid(self) -> str:
return self.gateway
def _extract_paths(lines: t.List[str]) -> MikrotikPaths:
"""Simple count based on lines with dst/dst-address and 'A' flag."""
available = 0
@ -114,6 +121,7 @@ def _extract_paths(lines: t.List[str]) -> MikrotikPaths:
best += 1
return MikrotikPaths(available=available, best=best, select=best)
def _process_kv(route: dict, key: str, val: str):
_log = log.bind(parser="MikrotikBGPTable")
"""Process a key-value pair and update the route dictionary."""
@ -125,7 +133,7 @@ def _process_kv(route: dict, key: str, val: str):
route["prefix"] = val
elif key in ("gateway", "nexthop"):
# Extract only the IP from gateway (e.g., 168.254.0.2%vlan-2000)
route["gateway"] = val.split('%')[0]
route["gateway"] = val.split("%")[0]
elif key == "distance":
route["distance"] = int(val) if val.isdigit() else route.get("distance", 0)
elif key == "scope":
@ -155,10 +163,11 @@ def _process_kv(route: dict, key: str, val: str):
if val and val.lower() != "none":
route["ext_communities"] = [c.strip() for c in val.split(",") if c.strip()]
elif key == "rpki":
#_log.debug(f"RPKI raw value: {val!r}")
# _log.debug(f"RPKI raw value: {val!r}")
clean_val = val.strip().strip('"').lower()
route["rpki_state"] = RPKI_STATE_MAP.get(clean_val, 2)
def _extract_route_entries(lines: t.List[str]) -> t.List[MikrotikRouteEntry]:
"""Extract route entries from a list of lines."""
routes: t.List[MikrotikRouteEntry] = []
@ -192,6 +201,7 @@ def _extract_route_entries(lines: t.List[str]) -> t.List[MikrotikRouteEntry]:
return routes
def _parse_route_block(block: t.List[str]) -> t.Optional[MikrotikRouteEntry]:
"""Parse a single route block and return a MikrotikRouteEntry."""
if not block:
@ -202,10 +212,21 @@ def _parse_route_block(block: t.List[str]) -> t.Optional[MikrotikRouteEntry]:
return None
rd = {
"prefix": "", "gateway": "", "distance": 20, "scope": 30, "target_scope": 10,
"as_path": [], "communities": [], "large_communities": [], "ext_communities": [],
"local_preference": 100, "metric": 0, "origin": "",
"is_active": False, "is_best": False, "is_valid": False,
"prefix": "",
"gateway": "",
"distance": 20,
"scope": 30,
"target_scope": 10,
"as_path": [],
"communities": [],
"large_communities": [],
"ext_communities": [],
"local_preference": 100,
"metric": 0,
"origin": "",
"is_active": False,
"is_best": False,
"is_valid": False,
"rpki_state": RPKI_STATE_MAP.get("unknown", 2),
}
@ -229,12 +250,14 @@ def _parse_route_block(block: t.List[str]) -> t.Optional[MikrotikRouteEntry]:
class MikrotikBGPRouteTable(BGPRouteTable):
"""Bypass validation to align with Huawei parser."""
def __init__(self, **kwargs):
object.__setattr__(self, "vrf", kwargs.get("vrf", "default"))
object.__setattr__(self, "count", kwargs.get("count", 0))
object.__setattr__(self, "routes", kwargs.get("routes", []))
object.__setattr__(self, "winning_weight", kwargs.get("winning_weight", "low"))
class MikrotikBGPTable(MikrotikBase):
"""MikroTik BGP Table in canonical format."""
@ -253,10 +276,7 @@ class MikrotikBGPTable(MikrotikBase):
return inst
# Filter out command echoes and header lines
lines = [
ln for ln in lines
if not ln.strip().startswith((">", "Flags:", "[", "#"))
]
lines = [ln for ln in lines if not ln.strip().startswith((">", "Flags:", "[", "#"))]
inst.paths = _extract_paths(lines)
inst.routes = _extract_route_entries(lines)

6
hyperglass/models/tests/test_util.py
View file

@ -19,9 +19,9 @@ def test_check_legacy_fields():
test1_expected.keys()
), "legacy field not replaced"
assert set(check_legacy_fields(model="Device", data=test2).keys()) == set(test2.keys()), (
"new field not left unmodified"
)
assert set(check_legacy_fields(model="Device", data=test2).keys()) == set(
test2.keys()
), "new field not left unmodified"
with pytest.raises(ValueError):
check_legacy_fields(model="Device", data=test3)

2
hyperglass/plugins/_builtin/bgp_routestr_huawei.py
View file

@ -35,7 +35,7 @@ def parse_huawei(output: Sequence[str]) -> "OutputDataModel":
_log.debug(f"Combined output length: {len(combined_output)}")
# Debug: log the first few lines to understand the format
lines = combined_output.split('\n')[:10]
lines = combined_output.split("\n")[:10]
_log.debug(f"First 10 lines: {lines}")
for response in output:

5
hyperglass/plugins/_builtin/mikrotik_garbage_output.py
View file

@ -70,7 +70,7 @@ class MikrotikGarbageOutput(OutputPlugin):
# Iniciar a detecção da seção de Flags
if stripped_line.startswith("Flags:"):
in_flags_section = True
continue # Pula a própria linha "Flags:"
continue # Pula a própria linha "Flags:"
# Se estivermos na seção de flags, verificar se a linha ainda é parte dela.
# Uma linha de dados de rota real geralmente começa com flags (ex: "Ab") ou é indentada.
@ -84,7 +84,7 @@ class MikrotikGarbageOutput(OutputPlugin):
if "=" in stripped_line:
in_flags_section = False
else:
continue # Pula as linhas da legenda de flags
continue # Pula as linhas da legenda de flags
filtered_lines.append(line)
@ -94,4 +94,3 @@ class MikrotikGarbageOutput(OutputPlugin):
log.debug(f"MikrotikGarbageOutput cleaned {len(output)} output blocks.")
return tuple(cleaned_outputs)

8
hyperglass/plugins/_builtin/mikrotik_normalize_input.py
View file

@ -7,6 +7,7 @@ from pydantic import PrivateAttr
# Project
from hyperglass.log import log
# REMOVIDA a importação direta de Query para evitar o erro circular
# from hyperglass.models.api.query import Query
@ -20,6 +21,7 @@ class MikrotikTargetNormalizerInput(InputPlugin):
This ensures that queries for different IPs within the same subnet
resolve to the same cache key.
"""
_hyperglass_builtin: bool = PrivateAttr(False)
name: str = "mikrotik_normalizer"
platforms: t.Sequence[str] = ("mikrotik_routeros", "mikrotik_switchos", "mikrotik")
@ -28,9 +30,9 @@ class MikrotikTargetNormalizerInput(InputPlugin):
# INÍCIO DA MODIFICAÇÃO: Usar 't.Any' em vez de 'Query'
# #############################################################
def validate(self, query: t.Any) -> InputPluginValidationReturn:
# #############################################################
# FIM DA MODIFICAÇÃO
# #############################################################
# #############################################################
# FIM DA MODIFICAÇÃO
# #############################################################
"""
Takes the query object and modifies the target if it's a BGP Route query.
"""

6
hyperglass/state/tests/test_hooks.py
View file

@ -96,7 +96,7 @@ def test_use_state_caching(state):
instance = use_state(attr)
if i == 0:
first = instance
assert isinstance(instance, model), (
f"{instance!r} is not an instance of '{model.__name__}'"
)
assert isinstance(
instance, model
), f"{instance!r} is not an instance of '{model.__name__}'"
assert instance == first, f"{instance!r} is not equal to {first!r}"

1
hyperglass/util/tests/test_typing.py
View file

@ -1,4 +1,5 @@
"""Test typing utilities."""
# flake8: noqa
# Standard Library