mirror of
https://github.com/thatmattlove/hyperglass.git
synced 2026-01-17 00:38:06 +00:00
feat: add Huawei BGP Route Input Plugin
Create builtin plugin to transform input field before passing to Huawei device
This commit is contained in:
parent
44f8faa1e5
commit
f67c676a2d
2 changed files with 44 additions and 0 deletions
|
|
@ -6,11 +6,13 @@ from .bgp_route_arista import BGPRoutePluginArista
|
|||
from .bgp_route_frr import BGPRoutePluginFrr
|
||||
from .bgp_route_juniper import BGPRoutePluginJuniper
|
||||
from .mikrotik_garbage_output import MikrotikGarbageOutput
|
||||
from .bgp_route_huawei import BGPRoutePluginHuawei
|
||||
|
||||
__all__ = (
|
||||
"BGPRoutePluginArista",
|
||||
"BGPRoutePluginFrr",
|
||||
"BGPRoutePluginJuniper",
|
||||
"BGPRoutePluginHuawei",
|
||||
"MikrotikGarbageOutput",
|
||||
"RemoveCommand",
|
||||
)
|
||||
|
|
|
|||
42
hyperglass/plugins/_builtin/bgp_route_huawei.py
Normal file
42
hyperglass/plugins/_builtin/bgp_route_huawei.py
Normal file
|
|
@ -0,0 +1,42 @@
|
|||
from ipaddress import ip_network
|
||||
from .._input import InputPlugin
|
||||
|
||||
# Standard Library
|
||||
import typing as t
|
||||
|
||||
# Third Party
|
||||
from pydantic import PrivateAttr
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
# Project
|
||||
from hyperglass.models.api.query import Query
|
||||
|
||||
InputPluginTransformReturn = t.Union[t.Sequence[str], str]
|
||||
|
||||
class BGPRoutePluginHuawei(InputPlugin):
|
||||
_hyperglass_builtin: bool = PrivateAttr(True)
|
||||
platforms: t.Sequence[str] =("huawei", "huawei_vrpv8",)
|
||||
directives: t.Sequence[str] = ("__hyperglass_huawei_bgp_route__",)
|
||||
|
||||
"""
|
||||
Huawei BGP Route Input Plugin
|
||||
|
||||
This plugin transforms a query target into a network address and prefix length
|
||||
ex.: 192.0.2.0/24 -> 192.0.2.0 24
|
||||
ex.: 2001:db8::/32 -> 2001:db8:: 32
|
||||
"""
|
||||
def transform(self, query: "Query") -> InputPluginTransformReturn:
|
||||
(target := query.query_target)
|
||||
|
||||
if not target or not isinstance(target, list) or len(target) == 0:
|
||||
return None
|
||||
|
||||
target = target[0].strip()
|
||||
|
||||
# Check for the / in the query target
|
||||
if target.find("/") == -1:
|
||||
return target
|
||||
|
||||
target_network = ip_network(target)
|
||||
|
||||
return f"{target_network.network_address!s} {target_network.prefixlen!s}"
|
||||
Loading…
Add table
Reference in a new issue