diff --git a/verifiers/src/yang/verifiers/__init__.py b/verifiers/src/yang/verifiers/__init__.py index cd78d16a..fa459b31 100644 --- a/verifiers/src/yang/verifiers/__init__.py +++ b/verifiers/src/yang/verifiers/__init__.py @@ -11,10 +11,11 @@ __contact__ = 'yang-python@cisco.com' __copyright__ = 'Cisco Systems, Inc.' -from .count_verifier import CountVerifier from .base_verifier import BaseVerifier +from .count_verifier import NetconfCountVerifier, GnmiCountVerifier __all__ = ( 'BaseVerifier', - 'CountVerifier', + 'NetconfCountVerifier', + 'GnmiCountVerifier' ) diff --git a/verifiers/src/yang/verifiers/count_verifier.py b/verifiers/src/yang/verifiers/count_verifier.py index 5fa404b0..badb9a7e 100644 --- a/verifiers/src/yang/verifiers/count_verifier.py +++ b/verifiers/src/yang/verifiers/count_verifier.py @@ -1,19 +1,19 @@ -from typing import List +from typing import Any, List from dataclasses import field, dataclass from google.protobuf import json_format # Import base classes. For non pyats installation you can use class provided within this module try: - from genie.libs.sdk.triggers.blitz.verifiers import GnmiDefaultVerifier + from genie.libs.sdk.triggers.blitz.verifiers import GnmiDefaultVerifier, NetconfDefaultVerifier except ImportError: from yang.verifiers.base_verifier import BaseVerifier as GnmiDefaultVerifier -class CountVerifier(GnmiDefaultVerifier): +class GnmiCountVerifier(GnmiDefaultVerifier): from genie.libs.sdk.triggers.blitz.rpcverify import OptFields @dataclass - class MyCustomReturns(OptFields): + class CustomReturns(OptFields): ''' Create a custom returns class to be used by the verifier by adding new fields to the default returns dataclass @@ -23,15 +23,15 @@ class MyCustomReturns(OptFields): found_items: int = 0 @property - def returns(self) -> List[MyCustomReturns]: + def returns(self) -> List[CustomReturns]: return self._returns @returns.setter - def returns(self, value: List[dict]) -> List[MyCustomReturns]: + def returns(self, value: List[dict]) -> List[CustomReturns]: ''' Register our custom returns class ''' - self._returns = [self.MyCustomReturns(**r) for r in value] + self._returns = [self.CustomReturns(**r) for r in value] def decode(self, response, namespace: dict = None, method: str = 'get') -> List[dict]: from genie.libs.sdk.triggers.blitz.gnmi_util import GnmiMessage @@ -61,6 +61,50 @@ def end_subscription(self, errors): if errors: return False for ret in self.returns: - if ret.count != ret.found_items or ret.found_items < self.kwargs['min_count']: + if (ret.count != ret.found_items or + ret.found_items < self.format['verifier']['min_count']): + return False + return True + + +class NetconfCountVerifier(NetconfDefaultVerifier): + from genie.libs.sdk.triggers.blitz.rpcverify import OptFields + + @dataclass + class CustomReturns(OptFields): + ''' + Create a custom returns class to be used by the verifier + by adding new fields to the default returns dataclass + ''' + cli_return: dict = field(default_factory=dict) + count: int = 0 + found_items: int = 0 + + @property + def returns(self) -> List[CustomReturns]: + return self._returns + + @returns.setter + def returns(self, value: List[dict]) -> List[CustomReturns]: + ''' + Register our custom returns class + ''' + self._returns = [self.CustomReturns(**r) for r in value] + + def get_config_verify(self, raw_response: Any) -> bool: + # Here we have raw netconf xml response + # We are decoding it using default netconf decoder + # but you can use your own decode method + try: + decoded_response = self.decode(raw_response) + except self.DecodeError: + return False + for response in decoded_response: + for ret in self.returns: + if response[1] == ret.xpath: + ret.found_items += 1 + for ret in self.returns: + if (ret.count != ret.found_items or + ret.found_items < self.format['verifier']['min_count']): return False return True