1
0
Fork 1
mirror of https://github.com/thatmattlove/hyperglass.git synced 2026-01-17 08:48:05 +00:00

Merge branch 'Computroniks-computroniks/bug/#311-fix-field-validation'

This commit is contained in:
Jason D. Hall 2025-09-27 22:39:23 -04:00
commit 43d3c66ed7
3 changed files with 21 additions and 13 deletions

View file

@ -166,6 +166,7 @@ your-directive:
- condition: null
command: show ip route {target}
field:
description: IP of target
validation: '[0-9a-f\.\:]+'
```
@ -178,6 +179,7 @@ your-directive:
- condition: null
command: show ip bgp community {target}
field:
description: BGP community to show
options:
- value: "65001:1"
description: Provider A Routes

View file

@ -7,7 +7,8 @@ import secrets
from datetime import datetime
# Third Party
from pydantic import Field, BaseModel, ConfigDict, field_validator
from pydantic import BaseModel, ConfigDict, field_validator, StringConstraints
from typing_extensions import Annotated
# Project
from hyperglass.log import log
@ -21,6 +22,11 @@ from hyperglass.exceptions.private import InputValidationError
from ..config.devices import Device
QueryLocation = Annotated[str, StringConstraints(strict=True, min_length=1, strip_whitespace=True)]
QueryTarget = Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]
QueryType = Annotated[str, StringConstraints(strict=True, min_length=1, strip_whitespace=True)]
class SimpleQuery(BaseModel):
"""A simple representation of a post-validated query."""
@ -39,12 +45,12 @@ class Query(BaseModel):
model_config = ConfigDict(extra="allow", alias_generator=snake_to_camel, populate_by_name=True)
# Device `name` field
query_location: str = Field(strict=True, min_length=1, strip_whitespace=True)
query_location: QueryLocation
query_target: t.Union[t.List[str], str] = Field(min_length=1, strip_whitespace=True)
query_target: t.Union[t.List[QueryTarget], QueryTarget]
# Directive `id` field
query_type: str = Field(strict=True, min_length=1, strip_whitespace=True)
query_type: QueryType
_kwargs: t.Dict[str, t.Any]
def __init__(self, **data) -> None:

View file

@ -170,7 +170,7 @@ class Device(HyperglassModelWithId, extra="allow"):
@field_validator("address")
def validate_address(
cls, value: t.Union[IPv4Address, IPv6Address, str], values: t.Dict[str, t.Any]
cls, value: t.Union[IPv4Address, IPv6Address, str], info: ValidationInfo
) -> t.Union[IPv4Address, IPv6Address, str]:
"""Ensure a hostname is resolvable."""
@ -178,14 +178,14 @@ class Device(HyperglassModelWithId, extra="allow"):
if not any(resolve_hostname(value)):
raise ConfigError(
"Device '{d}' has an address of '{a}', which is not resolvable.",
d=values["name"],
d=info.data["name"],
a=value,
)
return value
@field_validator("avatar")
def validate_avatar(
cls, value: t.Union[FilePath, None], values: t.Dict[str, t.Any]
cls, value: t.Union[FilePath, None], info: ValidationInfo
) -> t.Union[FilePath, None]:
"""Migrate avatar to static directory."""
if value is not None:
@ -198,7 +198,7 @@ class Device(HyperglassModelWithId, extra="allow"):
target = Settings.static_path / "images" / value.name
copied = shutil.copy2(value, target)
log.bind(
device=values["name"],
device=info.data["name"],
source=str(value),
destination=str(target),
).debug("Copied device avatar")
@ -210,24 +210,24 @@ class Device(HyperglassModelWithId, extra="allow"):
return value
@field_validator("platform", mode="before")
def validate_platform(cls: "Device", value: t.Any, values: t.Dict[str, t.Any]) -> str:
def validate_platform(cls: "Device", value: t.Any, info: ValidationInfo) -> str:
"""Validate & rewrite device platform, set default `directives`."""
if value == "http":
if values.get("http") is None:
if info.data.get("http") is None:
raise ConfigError(
"Device '{device}' has platform 'http' configured, but no http parameters are defined.",
device=values["name"],
device=info.data["name"],
)
if value is None:
if values.get("http") is not None:
if info.data.get("http") is not None:
value = "http"
else:
# Ensure device platform is defined.
raise ConfigError(
"Device '{device}' is missing a 'platform' (Network Operating System) property",
device=values["name"],
device=info.data["name"],
)
if value in SCRAPE_HELPERS.keys():