Skip schema validation on call (#650)

Implements schema validation skipping when running `call`.

`skip_schema_validation` is defaulted to `False`, and once set to
`True`, will skip schema validation of both request and response.

Implements #315.

Author: esiebert

ocpp/charge_point.py

@@ -361,7 +361,9 @@ async def _handle_call(self, msg):
             pass
         return response
 
-    async def call(self, payload, suppress=True, unique_id=None):
+    async def call(
+        self, payload, suppress=True, unique_id=None, skip_schema_validation=False
+    ):
         """
         Send Call message to client and return payload of response.
 
@@ -382,6 +384,9 @@ async def call(self, payload, suppress=True, unique_id=None):
         set to False, an exception will be raised for users to handle this
         CallError.
 
+        Schema validation can be skipped for the request and the response
+        for this call by setting `skip_schema_validation` to `True`.
+
         """
         camel_case_payload = snake_to_camel_case(serialize_as_dict(payload))
 
@@ -400,7 +405,8 @@ async def call(self, payload, suppress=True, unique_id=None):
             payload=remove_nones(camel_case_payload),
         )
 
-        validate_payload(call, self._ocpp_version)
+        if not skip_schema_validation:
+            validate_payload(call, self._ocpp_version)
 
         # Use a lock to prevent make sure that only 1 message can be send at a
         # a time.
@@ -421,7 +427,7 @@ async def call(self, payload, suppress=True, unique_id=None):
             if suppress:
                 return
             raise response.to_exception()
-        else:
+        elif not skip_schema_validation:
             response.action = call.action
             validate_payload(response, self._ocpp_version)
 

tests/v16/conftest.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from unittest.mock import AsyncMock
 
 import pytest
@@ -50,6 +51,15 @@ def mock_boot_request():
     )
 
 
+@pytest.fixture
+def mock_invalid_boot_request():
+    @dataclass
+    class BootNotification:
+        custom_field: str
+
+    return BootNotification(custom_field="custom_field")
+
+
 @pytest.fixture
 def mock_base_central_system(base_central_system):
     mock_result_call = CallResult(

tests/v16/test_v16_charge_point.py

@@ -263,3 +263,29 @@ async def test_call_without_unique_id_should_return_a_random_value(
     ) = mock_base_central_system._get_specific_response.call_args_list[0][0]
     # Check the actual unique id is equals to the one internally generated
     assert actual_unique_id == expected_unique_id
+
+
+@pytest.mark.asyncio
+async def test_call_skip_schema_validation(
+    mock_invalid_boot_request, mock_base_central_system
+):
+    """
+    Test that schema validation is skipped for an invalid boot notification request.
+
+    """
+
+    expected_unique_id = "12345"
+    # Call the method being tested with an invalid boot notification request
+    # and a unique_id as a parameter
+    await mock_base_central_system.call(
+        mock_invalid_boot_request,
+        unique_id=expected_unique_id,
+        skip_schema_validation=True,
+    )
+    (
+        actual_unique_id,
+        _,
+    ) = mock_base_central_system._get_specific_response.call_args_list[0][0]
+
+    # Check the actual unique id is equals to the one passed to the call method
+    assert actual_unique_id == expected_unique_id