diff --git a/hyperglass/api/__init__.py b/hyperglass/api/__init__.py index 5684629..65458f9 100644 --- a/hyperglass/api/__init__.py +++ b/hyperglass/api/__init__.py @@ -1,4 +1,5 @@ """hyperglass API.""" + # Standard Library import logging diff --git a/hyperglass/cli/echo.py b/hyperglass/cli/echo.py index 642c1ce..4f7ced0 100644 --- a/hyperglass/cli/echo.py +++ b/hyperglass/cli/echo.py @@ -1,4 +1,5 @@ """Helper functions for CLI message printing.""" + # Standard Library import typing as t diff --git a/hyperglass/cli/main.py b/hyperglass/cli/main.py index 2ee2107..7cc06f6 100644 --- a/hyperglass/cli/main.py +++ b/hyperglass/cli/main.py @@ -33,7 +33,7 @@ def run(): def _version( version: t.Optional[bool] = typer.Option( None, "--version", help="hyperglass version", callback=_version - ) + ), ) -> None: """hyperglass""" pass @@ -77,7 +77,6 @@ def _build_ui(timeout: int = typer.Option(180, help="Timeout in seconds")) -> No with echo._console.status( f"Starting new UI build with a {timeout} second timeout...", spinner="aesthetic" ): - _build_ui(timeout=120) @@ -140,7 +139,7 @@ def _clear_cache(): @cli.command(name="devices") def _devices( - search: t.Optional[str] = typer.Argument(None, help="Device ID or Name Search Pattern") + search: t.Optional[str] = typer.Argument(None, help="Device ID or Name Search Pattern"), ): """Show all configured devices""" # Third Party @@ -189,7 +188,7 @@ def _devices( @cli.command(name="directives") def _directives( - search: t.Optional[str] = typer.Argument(None, help="Directive ID or Name Search Pattern") + search: t.Optional[str] = typer.Argument(None, help="Directive ID or Name Search Pattern"), ): """Show all configured devices""" # Third Party @@ -280,7 +279,7 @@ def _plugins( def _params( path: t.Optional[str] = typer.Argument( None, help="Parameter Object Path, for example 'messages.no_input'" - ) + ), ): """Show configuration parameters""" # Standard Library @@ -312,7 +311,7 @@ def _params( ) raise typer.Exit(0) except AttributeError: - echo.error(f"{'params.'+path!r} does not exist") + echo.error(f"{'params.' + path!r} does not exist") raise typer.Exit(1) panel = Inspect( diff --git a/hyperglass/exceptions/_common.py b/hyperglass/exceptions/_common.py index 9963324..1f04f94 100644 --- a/hyperglass/exceptions/_common.py +++ b/hyperglass/exceptions/_common.py @@ -2,7 +2,7 @@ # Standard Library import json as _json -from typing import Any, Dict, List, Union, Literal, Optional, Set +from typing import Any, Set, Dict, List, Union, Literal, Optional # Third Party from pydantic import ValidationError @@ -72,7 +72,7 @@ class HyperglassError(Exception): for err in errors: loc = " → ".join(str(loc) for loc in err["loc"]) - errs += (f'Field: {loc}\n Error: {err["msg"]}\n',) + errs += (f"Field: {loc}\n Error: {err['msg']}\n",) return "\n".join(errs) diff --git a/hyperglass/execution/drivers/_construct.py b/hyperglass/execution/drivers/_construct.py index 86032bc..c1d1e2b 100644 --- a/hyperglass/execution/drivers/_construct.py +++ b/hyperglass/execution/drivers/_construct.py @@ -94,7 +94,7 @@ class Construct: for key in [k for k in keys if k != "target" and k != "mask"]: if key not in attrs: raise ConfigError( - ("Command '{c}' has attribute '{k}', " "which is missing from device '{d}'"), + ("Command '{c}' has attribute '{k}', which is missing from device '{d}'"), level="danger", c=self.directive.name, k=key, @@ -224,4 +224,4 @@ class Formatter: def _bird_bgp_community(self, target: str) -> str: """Convert from standard community format to BIRD format.""" parts = target.split(":") - return f'({",".join(parts)})' + return f"({','.join(parts)})" diff --git a/hyperglass/execution/drivers/ssh.py b/hyperglass/execution/drivers/ssh.py index 13e070e..25d81b4 100644 --- a/hyperglass/execution/drivers/ssh.py +++ b/hyperglass/execution/drivers/ssh.py @@ -44,9 +44,9 @@ class SSHConnection(Connection): if proxy.credential._method == "encrypted_key": # If the key is encrypted, use the password field as the # private key password. - tunnel_kwargs[ - "ssh_private_key_password" - ] = proxy.credential.password.get_secret_value() + tunnel_kwargs["ssh_private_key_password"] = ( + proxy.credential.password.get_secret_value() + ) try: return open_tunnel(proxy._target, proxy.port, **tunnel_kwargs) diff --git a/hyperglass/external/_base.py b/hyperglass/external/_base.py index 60c35da..3954c51 100644 --- a/hyperglass/external/_base.py +++ b/hyperglass/external/_base.py @@ -212,7 +212,7 @@ class BaseExternal: if method.upper() not in supported_methods: raise self._exception( - f'Method must be one of {", ".join(supported_methods)}. ' f"Got: {str(method)}" + f"Method must be one of {', '.join(supported_methods)}. Got: {str(method)}" ) endpoint = "/".join( @@ -284,7 +284,7 @@ class BaseExternal: status = httpx.codes(response.status_code) error = self._parse_response(response) raise self._exception( - f'{status.name.replace("_", " ")}: {error}', level="danger" + f"{status.name.replace('_', ' ')}: {error}", level="danger" ) from None except httpx.HTTPError as http_err: @@ -340,7 +340,7 @@ class BaseExternal: status = httpx.codes(response.status_code) error = self._parse_response(response) raise self._exception( - f'{status.name.replace("_", " ")}: {error}', level="danger" + f"{status.name.replace('_', ' ')}: {error}", level="danger" ) from None except httpx.HTTPError as http_err: diff --git a/hyperglass/external/bgptools.py b/hyperglass/external/bgptools.py index 1ea1cee..f2fd7f6 100644 --- a/hyperglass/external/bgptools.py +++ b/hyperglass/external/bgptools.py @@ -35,7 +35,7 @@ def default_ip_targets(*targets: str) -> t.Tuple[TargetData, t.Tuple[str, ...]]: default_data = {} query = () for target in targets: - detail: TargetDetail = {k: "None" for k in DEFAULT_KEYS} + detail: TargetDetail = dict.fromkeys(DEFAULT_KEYS, "None") try: valid: t.Union[IPv4Address, IPv6Address] = ip_address(target) @@ -139,7 +139,7 @@ async def network_info(*targets: str) -> TargetData: cache = use_state("cache") # Set default data structure. - query_data = {t: {k: "" for k in DEFAULT_KEYS} for t in query_targets} + query_data = {t: dict.fromkeys(DEFAULT_KEYS, "") for t in query_targets} # Get all cached bgp.tools data. cached = cache.get_map(CACHE_KEY) or {} diff --git a/hyperglass/external/tests/test_base.py b/hyperglass/external/tests/test_base.py index 5f23594..6604f4e 100644 --- a/hyperglass/external/tests/test_base.py +++ b/hyperglass/external/tests/test_base.py @@ -1,4 +1,5 @@ """Test external http client.""" + # Standard Library import asyncio diff --git a/hyperglass/external/tests/test_bgptools.py b/hyperglass/external/tests/test_bgptools.py index ca8f98d..542c1dc 100644 --- a/hyperglass/external/tests/test_bgptools.py +++ b/hyperglass/external/tests/test_bgptools.py @@ -16,7 +16,6 @@ WHOIS_OUTPUT = """AS | IP | BGP Prefix | CC | Registry | Allocated | AS # Ignore asyncio deprecation warning about loop @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_network_info(): - checks = ( ("192.0.2.1", {"asn": "None", "rir": "Private Address"}), ("127.0.0.1", {"asn": "None", "rir": "Loopback Address"}), diff --git a/hyperglass/external/tests/test_rpki.py b/hyperglass/external/tests/test_rpki.py index 66da2e4..01438e7 100644 --- a/hyperglass/external/tests/test_rpki.py +++ b/hyperglass/external/tests/test_rpki.py @@ -1,4 +1,5 @@ """Test RPKI data fetching.""" + # Third Party import pytest @@ -18,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 + ) ) diff --git a/hyperglass/models/api/__init__.py b/hyperglass/models/api/__init__.py index 70769f4..8f4bfcd 100644 --- a/hyperglass/models/api/__init__.py +++ b/hyperglass/models/api/__init__.py @@ -1,4 +1,5 @@ """Query & Response Validation Models.""" + # Local from .query import Query from .response import ( diff --git a/hyperglass/models/api/query.py b/hyperglass/models/api/query.py index 571311f..3b941fa 100644 --- a/hyperglass/models/api/query.py +++ b/hyperglass/models/api/query.py @@ -69,7 +69,7 @@ class Query(BaseModel): self.validate_query_target() except InputValidationError as err: raise InputInvalid(**err.kwargs) from err - + self.query_target = self.transform_query_target() def summary(self) -> SimpleQuery: diff --git a/hyperglass/models/config/cache.py b/hyperglass/models/config/cache.py index 0871b94..f3c3e93 100644 --- a/hyperglass/models/config/cache.py +++ b/hyperglass/models/config/cache.py @@ -1,6 +1,5 @@ """Validation model for cache config.""" - # Local from ..main import HyperglassModel diff --git a/hyperglass/models/data/bgp_route.py b/hyperglass/models/data/bgp_route.py index dc417ed..a00af4e 100644 --- a/hyperglass/models/data/bgp_route.py +++ b/hyperglass/models/data/bgp_route.py @@ -6,7 +6,7 @@ import typing as t from ipaddress import ip_network # Third Party -from pydantic import field_validator, ValidationInfo +from pydantic import ValidationInfo, field_validator # Project from hyperglass.state import use_state diff --git a/hyperglass/models/parsing/frr.py b/hyperglass/models/parsing/frr.py index 2dbac8a..dacd82c 100644 --- a/hyperglass/models/parsing/frr.py +++ b/hyperglass/models/parsing/frr.py @@ -50,7 +50,7 @@ class FRRPath(_FRRBase): aspath: t.List[int] aggregator_as: int = 0 aggregator_id: str = "" - loc_prf: int = 100 # 100 is the default value for local preference + loc_prf: int = 100 # 100 is the default value for local preference metric: int = 0 med: int = 0 weight: int = 0 @@ -66,7 +66,7 @@ class FRRPath(_FRRBase): """Extract meaningful data from FRR response.""" new = values.copy() new["aspath"] = values["aspath"]["segments"][0]["list"] - community = values.get("community", {'list': []}) + community = values.get("community", {"list": []}) new["community"] = community["list"] new["lastUpdate"] = values["lastUpdate"]["epoch"] bestpath = values.get("bestpath", {}) diff --git a/hyperglass/models/tests/test_util.py b/hyperglass/models/tests/test_util.py index bb133d6..f990d57 100644 --- a/hyperglass/models/tests/test_util.py +++ b/hyperglass/models/tests/test_util.py @@ -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) diff --git a/hyperglass/plugins/_builtin/__init__.py b/hyperglass/plugins/_builtin/__init__.py index 8f8bd0f..5e87f5b 100644 --- a/hyperglass/plugins/_builtin/__init__.py +++ b/hyperglass/plugins/_builtin/__init__.py @@ -1,12 +1,12 @@ """Built-in hyperglass plugins.""" # Local +from .bgp_route_frr import BGPRoutePluginFrr from .remove_command import RemoveCommand from .bgp_route_arista import BGPRoutePluginArista -from .bgp_route_frr import BGPRoutePluginFrr +from .bgp_route_huawei import BGPRoutePluginHuawei from .bgp_route_juniper import BGPRoutePluginJuniper from .mikrotik_garbage_output import MikrotikGarbageOutput -from .bgp_route_huawei import BGPRoutePluginHuawei __all__ = ( "BGPRoutePluginArista", diff --git a/hyperglass/plugins/_builtin/bgp_route_frr.py b/hyperglass/plugins/_builtin/bgp_route_frr.py index 627fad2..7130b20 100644 --- a/hyperglass/plugins/_builtin/bgp_route_frr.py +++ b/hyperglass/plugins/_builtin/bgp_route_frr.py @@ -62,14 +62,13 @@ def parse_frr(output: t.Sequence[str]) -> "OutputDataModel": return result + class BGPRoutePluginFrr(OutputPlugin): """Coerce a FRR route table in JSON format to a standard BGP Table structure.""" _hyperglass_builtin: bool = PrivateAttr(True) platforms: t.Sequence[str] = ("frr",) - directives: t.Sequence[str] = ( - "__hyperglass_frr_bgp_route_table__", - ) + directives: t.Sequence[str] = ("__hyperglass_frr_bgp_route_table__",) def process(self, *, output: "OutputType", query: "Query") -> "OutputType": """Parse FRR response if data is a string (and is therefore unparsed).""" diff --git a/hyperglass/plugins/_builtin/bgp_route_huawei.py b/hyperglass/plugins/_builtin/bgp_route_huawei.py index 86e70a3..172d906 100644 --- a/hyperglass/plugins/_builtin/bgp_route_huawei.py +++ b/hyperglass/plugins/_builtin/bgp_route_huawei.py @@ -1,36 +1,41 @@ -from ipaddress import ip_network -from .._input import InputPlugin - # Standard Library import typing as t +from ipaddress import ip_network # Third Party from pydantic import PrivateAttr +# Local +from .._input import InputPlugin + 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",) + 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) - + 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 @@ -38,5 +43,5 @@ class BGPRoutePluginHuawei(InputPlugin): return target target_network = ip_network(target) - - return f"{target_network.network_address!s} {target_network.prefixlen!s}" \ No newline at end of file + + return f"{target_network.network_address!s} {target_network.prefixlen!s}" diff --git a/hyperglass/plugins/tests/test_bgp_community.py b/hyperglass/plugins/tests/test_bgp_community.py index d5f8834..3541244 100644 --- a/hyperglass/plugins/tests/test_bgp_community.py +++ b/hyperglass/plugins/tests/test_bgp_community.py @@ -1,4 +1,5 @@ """Test BGP Community validation.""" + # Standard Library import typing as t diff --git a/hyperglass/state/tests/test_hooks.py b/hyperglass/state/tests/test_hooks.py index aaa9d5c..e82f1a4 100644 --- a/hyperglass/state/tests/test_hooks.py +++ b/hyperglass/state/tests/test_hooks.py @@ -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}" diff --git a/hyperglass/ui/pnpm-workspace.yaml b/hyperglass/ui/pnpm-workspace.yaml new file mode 100644 index 0000000..d649e81 --- /dev/null +++ b/hyperglass/ui/pnpm-workspace.yaml @@ -0,0 +1,3 @@ +onlyBuiltDependencies: + - '@biomejs/biome' + - esbuild