Merge branch 'main' into url_monitoring

catalystwan/abstractions.py

@@ -24,10 +24,10 @@ def text(self) -> str:
     def content(self) -> bytes:
         ...
 
-    def dataobj(self, cls: Type[T], sourcekey: Optional[str]) -> T:
+    def dataobj(self, cls: Type[T], sourcekey: Optional[str], validate: bool) -> T:
         ...
 
-    def dataseq(self, cls: Type[T], sourcekey: Optional[str]) -> DataSequence[T]:
+    def dataseq(self, cls: Type[T], sourcekey: Optional[str], validate: bool) -> DataSequence[T]:
         ...
 
     def json(self) -> dict:
@@ -52,3 +52,7 @@ def api_version(self) -> Optional[Version]:
     @property
     def session_type(self) -> Optional[SessionType]:
         ...
+
+    @property
+    def validate_responses(self) -> bool:
+        ...

catalystwan/endpoints/__init__.py

@@ -565,9 +565,15 @@ def wrapper(*args, **kwargs):
                     pass
                 elif issubclass(self.return_spec.payload_type, BaseModel):
                     if self.return_spec.sequence_type == DataSequence:
-                        return response.dataseq(self.return_spec.payload_type, self.resp_json_key)
+                        return response.dataseq(
+                            cls=self.return_spec.payload_type,
+                            sourcekey=self.resp_json_key,
+                            validate=_self._client.validate_responses,
+                        )
                     else:
-                        return response.dataobj(self.return_spec.payload_type, self.resp_json_key)
+                        return response.dataobj(
+                            self.return_spec.payload_type, self.resp_json_key, validate=_self._client.validate_responses
+                        )
                 elif issubclass(self.return_spec.payload_type, str):
                     return response.text
                 elif issubclass(self.return_spec.payload_type, bytes):

catalystwan/response.py

@@ -174,7 +174,7 @@ def info(self, history: bool = False) -> str:
             return response_history_debug(self, None)
         return response_debug(self, None)
 
-    def dataseq(self, cls: Type[T], sourcekey: Optional[str] = "data") -> DataSequence[T]:
+    def dataseq(self, cls: Type[T], sourcekey: Optional[str] = "data", validate: bool = True) -> DataSequence[T]:
         """Returns data contents from JSON payload parsed as DataSequence of Dataclass/BaseModel instances
         Args:
             cls: Dataclass/BaseModel subtype (eg. Devices)
@@ -195,10 +195,12 @@ def dataseq(self, cls: Type[T], sourcekey: Optional[str] = "data") -> DataSequen
             sequence = [cast(dict, data)]
 
         if issubclass(cls, BaseModel):
-            return DataSequence(cls, [cls.model_validate(item) for item in sequence])  # type: ignore
+            if validate:
+                return DataSequence(cls, [cls.model_validate(item) for item in sequence])  # type: ignore
+            return DataSequence(cls, [cls.model_construct(**item) for item in sequence])  # type: ignore
         return DataSequence(cls, [create_dataclass(cls, item) for item in sequence])
 
-    def dataobj(self, cls: Type[T], sourcekey: Optional[str] = "data") -> T:
+    def dataobj(self, cls: Type[T], sourcekey: Optional[str] = "data", validate: bool = True) -> T:
         """Returns data contents from JSON payload parsed as Dataclass/BaseModel instance
         Args:
             cls: Dataclass/BaseModel subtype (eg. Devices)
@@ -214,7 +216,9 @@ def dataobj(self, cls: Type[T], sourcekey: Optional[str] = "data") -> T:
             data = self.payload.json.get(sourcekey)
 
         if issubclass(cls, BaseModel):
-            return cls.model_validate(data)  # type: ignore[return-value]
+            if validate:
+                return cls.model_validate(data)  # type: ignore[return-value]
+            return cls.model_construct(**data)  # type: ignore[return-value]
         return create_dataclass(cls, data)
 
     def get_error_info(self) -> ManagerErrorInfo:

catalystwan/session.py

@@ -153,6 +153,7 @@ def __init__(
         port: Optional[int] = None,
         subdomain: Optional[str] = None,
         auth: Optional[AuthBase] = None,
+        validate_responses: bool = True,
     ):
         self.url = url
         self.port = port
@@ -177,6 +178,7 @@ def __init__(
         self._state: ManagerSessionState = ManagerSessionState.OPERATIVE
         self.restart_timeout: int = 1200
         self.polling_requests_timeout: int = 10
+        self._validate_responses = validate_responses
 
     @property
     def state(self) -> ManagerSessionState:
@@ -236,7 +238,7 @@ def login(self) -> ManagerSession:
         try:
             server_info = self.server()
         except DefaultPasswordError:
-            server_info = ServerInfo.parse_obj({})
+            server_info = ServerInfo.model_construct(**{})
 
         self.server_name = server_info.server
 
@@ -404,7 +406,7 @@ def get_tenant_id(self) -> str:
         Returns:
             Tenant UUID.
         """
-        tenants = self.get("dataservice/tenant").dataseq(Tenant)
+        tenants = self.get("dataservice/tenant").dataseq(Tenant, validate=False)
         tenant = tenants.filter(subdomain=self.subdomain).single_or_default()
 
         if not tenant or not tenant.tenant_id:
@@ -480,6 +482,14 @@ def platform_version(self, version: str):
     def api_version(self) -> Version:
         return self._api_version
 
+    @property
+    def validate_responses(self) -> bool:
+        return self._validate_responses
+
+    @validate_responses.setter
+    def validate_responses(self, value: bool):
+        self._validate_responses = value
+
     def __str__(self) -> str:
         return f"{self.username}@{self.base_url}"
 

catalystwan/tests/test_response.py

@@ -6,7 +6,7 @@
 
 from attr import define, field  # type: ignore
 from parameterized import parameterized  # type: ignore
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ValidationError
 
 from catalystwan.dataclasses import DataclassBase
 from catalystwan.response import ManagerErrorInfo, ManagerResponse
@@ -26,6 +26,11 @@ class ParsedDataTypePydanticV2(BaseModel):
     key3: Optional[float] = Field(default=None)
 
 
+class DataForValidateTest(BaseModel):
+    important: str
+    not_important: ParsedDataTypePydanticV2
+
+
 PARSE_DATASEQ_TEST_DATA: List = [
     (True, None, 0, "data"),
     (True, "string", 0, "data"),
@@ -56,6 +61,11 @@ class ParsedDataTypePydanticV2(BaseModel):
     (True, {"data": [{"key1": "string", "key2": 66}, {"key1": "required", "key2": 18, "key3": 0.1}]}, "data"),
 ]
 
+VALIDATE_DATASEQ_TEST_DATA = [
+    {"important": "correct1", "not_important": {"invalid-key1": "string", "key2": "not int"}},
+    {"important": "correct2", "not_important": {}},
+]
+
 
 class TestResponse(unittest.TestCase):
     @patch("requests.Response")
@@ -136,3 +146,28 @@ def test_get_error(self, empty_error: bool, json: Any):
             assert error_info.message is None
             assert error_info.details is None
             assert error_info.code is None
+
+    def test_dataobj_optional_validate(self):
+        # Arrange
+        self.response_mock.json.return_value = VALIDATE_DATASEQ_TEST_DATA[0]
+        vmng_response = ManagerResponse(self.response_mock)
+        # Act
+        data = vmng_response.dataobj(DataForValidateTest, sourcekey=None, validate=False)
+        # Assert
+        assert isinstance(data, DataForValidateTest)
+        assert data.important == VALIDATE_DATASEQ_TEST_DATA[0]["important"]
+        with self.assertRaises(ValidationError):
+            vmng_response.dataobj(DataForValidateTest, sourcekey=None, validate=True)
+
+    def test_dataseq_optional_validate(self):
+        self.response_mock.json.return_value = VALIDATE_DATASEQ_TEST_DATA
+        vmng_response = ManagerResponse(self.response_mock)
+        # Act
+        dataseq = vmng_response.dataseq(DataForValidateTest, sourcekey=None, validate=False)
+        # Assert
+        assert isinstance(dataseq, DataSequence)
+        for i, data in enumerate(dataseq):
+            assert isinstance(data, DataForValidateTest)
+            assert data.important == VALIDATE_DATASEQ_TEST_DATA[i]["important"]
+        with self.assertRaises(ValidationError):
+            vmng_response.dataseq(DataForValidateTest, sourcekey=None, validate=True)