diff --git a/oxi/interfaces/base.py b/oxi/interfaces/base.py index f3c287b..4077b92 100644 --- a/oxi/interfaces/base.py +++ b/oxi/interfaces/base.py @@ -12,7 +12,9 @@ class BaseDevice(ABC): def __init__(self, config: str): self.config: str = config + self._loaded_template = self._load_template() + self._declared_sections = None self._validate_template_groups() self.raw: dict = self._run_ttp() @@ -20,15 +22,17 @@ class BaseDevice(ABC): @abstractmethod def template(self) -> str: """ - Returns: + Expected structure: Название файла с парсером ttp + Returns: + None """ def vlans(self) -> list[dict]: """ Parse VLAN configuration from self.raw['vlans']. - Expected raw structure: + Expected structure: [{"id": 10, "description": "MGMT"}, {"id": 15, "name": "SSH"}, ...] Returns: @@ -64,16 +68,20 @@ class BaseDevice(ABC): """ return self.raw.get("system", None) - def _validate_contract(self): - optional_vlans = self.vlans() - if optional_vlans: - optional_vlans = [Vlans(**item) for item in optional_vlans] - return { - "system": System(**self.system()), - "interfaces": [Interfaces(**items) for items in self.interfaces()], - "vlans": optional_vlans, + def _validate_contract(self) -> dict: + system_data = self.system() + interfaces_data = self.interfaces() or [] + result = { + "system": System(**system_data), + "interfaces": [Interfaces(**item) for item in interfaces_data], + "vlans": [], } + if "vlans" in self._declared_sections: + vlans_data = self.vlans() or [] + result["vlans"] = [Vlans(**item) for item in vlans_data] + return result + def _load_template(self): """Подгрузка темплейтов из папки models/templates""" path = Path(__file__).parent / "models" / "templates" / self.template @@ -82,20 +90,20 @@ class BaseDevice(ABC): return path.read_text(encoding="utf-8") def _validate_template_groups(self) -> None: - """Проверка что TTP темлпейт имеет декларированные группы для всех требуемых и опциональных секций""" + """Проверяем только обязательные группы в template.""" try: root = ET.fromstring(self._loaded_template) except ET.ParseError: root = ET.fromstring(f"") declared = {g.get("name") for g in root.iter("group") if g.get("name")} - expected = self._REQUIRED_SECTIONS | self._OPTIONAL_SECTIONS - missing = expected - declared + self._declared_sections = declared - if missing: + missing_required = self._REQUIRED_SECTIONS - declared + if missing_required: raise ValueError( f"{self.__class__.__name__}: template '{self.template}' " - f"missing group declarations: {sorted(missing)}. " + f"missing required groups: {sorted(missing_required)}. " f"Declared groups: {sorted(declared)}" ) @@ -108,8 +116,8 @@ class BaseDevice(ABC): if missing: raise ValueError( f"{self.__class__.__name__}: TTP template '{self.template}' " - f"did not produce required sections: {sorted(missing)}. " - f"Got: {(raw.keys())}" + f"did not produce required groups: {sorted(missing)}. " + f"Return only: {(raw.keys())}" ) return raw diff --git a/oxi/interfaces/contract.py b/oxi/interfaces/contract.py index 21f0f9b..834c7a4 100644 --- a/oxi/interfaces/contract.py +++ b/oxi/interfaces/contract.py @@ -2,27 +2,41 @@ from ipaddress import IPv4Address from pydantic import BaseModel, ConfigDict, Field -class Interfaces(BaseModel): - name: str - ip_address: IPv4Address | None = None - mask: int | None = None - description: str | None = None +class Base(BaseModel): + model_config = ConfigDict(populate_by_name=True) class System(BaseModel): + """ + Requred + """ + model: str serial_number: str version: str -class Vlans(BaseModel): - model_config = ConfigDict(populate_by_name=True) +class Interfaces(Base): + """ + Requred + """ - vlan_id: int + name: str = Field(alias="interface") + ip_address: IPv4Address | None = None + mask: int | None = None + description: str | None = None + + +class Vlans(Base): + """ + Optional + """ + + vlan_id: int = Field(alias="id") name: str | None = Field(default=None, alias="description") class Device(BaseModel): system: System - interfaces: list[Interfaces] = [] + interfaces: list[Interfaces] vlans: list[Vlans] = [] diff --git a/oxi/interfaces/models/keenetic.py b/oxi/interfaces/models/keenetic.py index a550a49..2a692d8 100644 --- a/oxi/interfaces/models/keenetic.py +++ b/oxi/interfaces/models/keenetic.py @@ -1,7 +1,6 @@ from ipaddress import ip_interface from oxi.interfaces import register_parser from oxi.interfaces.base import BaseDevice -from oxi.interfaces.contract import Interfaces, Vlans @register_parser(["NDMS", "keenetic", "KeeneticOS"]) @@ -34,13 +33,13 @@ class Keenetic(BaseDevice): item["description"] = decoded return interfaces - def vlans(self): - vlans = self.raw["vlans"] - for item in vlans: - if item.get("description"): - decoded = self._decode_utf(item.get("description", "")) - item["description"] = decoded - return vlans + # def vlans(self): + # vlans = self.raw["vlans"] + # for item in vlans: + # if item.get("description"): + # decoded = self._decode_utf(item.get("description", "")) + # item["description"] = decoded + # return vlans if __name__ == "__main__":