Add arg skip_schema_validation to @on()

ocpp/charge_point.py

@@ -140,12 +140,11 @@ async def route_message(self, raw_msg):
             return
 
         if msg.message_type_id == MessageType.Call:
-            # Call's can be validated right away because the 'action' is know.
+            # Call's can be validated right away because the 'action' is known.
             # The 'action' is required to get the correct schema.
             #
             # CallResult's don't have an action field. The action must be
             # deducted from corresponding Call.
-            validate_payload(msg, self._ocpp_version)
             await self._handle_call(msg)
         elif msg.message_type_id in \
                 [MessageType.CallResult, MessageType.CallError]:
@@ -168,6 +167,9 @@ async def _handle_call(self, msg):
             raise NotImplementedError(f"No handler for '{msg.action}' "
                                       "registered.")
 
+        if not handlers.get('_skip_schema_validation', False):
+            validate_payload(msg, self._ocpp_version)
+
         # OCPP uses camelCase for the keys in the payload. It's more pythonic
         # to use snake_case for keyword arguments. Therefore the keys must be
         # 'translated'. Some examples:
@@ -205,7 +207,9 @@ async def _handle_call(self, msg):
         camel_case_payload = snake_to_camel_case(response_payload)
 
         response = msg.create_call_result(camel_case_payload)
-        validate_payload(response, self._ocpp_version)
+
+        if not handlers.get('_skip_schema_validation', False):
+            validate_payload(response, self._ocpp_version)
 
         await self._send(response.to_json())
 

ocpp/routing.py

@@ -1,7 +1,7 @@
 import functools
 
 
-def on(action):
+def on(action, *, skip_schema_validation=False):
     """ Function decorator to mark function as handler for specific action.
 
     This hook's argument are the data that is in the payload for the specific
@@ -20,13 +20,18 @@ def on_boot_notification(charge_point_model, charge_point_vendor, **kwargs): # n
                 status="Accepted",
             )
 
+    The decorator takes an optional argument `skip_schema_validation` which
+    defaults to False. Setting this argument to `True` will disable schema
+    validation of the request and the response of the specific route.
+
     """
     def decorator(func):
         @functools.wraps(func)
         def inner(*args, **kwargs):
             return func(*args, **kwargs)
 
         inner._on_action = action
+        inner._skip_schema_validation = skip_schema_validation
         return inner
     return decorator
 
@@ -79,20 +84,29 @@ def after_boot_notification(self, *args, **kwargs):
             Action.BootNotification: {
                 '_on_action': <reference to 'on_boot_notification'>,
                 '_after_action': <reference to 'after_boot_notification'>,
+                '_skip_schema_validation': False,
             },
         }
 
     """
     routes = {}
     for attr_name in dir(obj):
         attr = getattr(obj, attr_name)
-        for hook in ['_on_action', '_after_action']:
+        for option in ['_on_action', '_after_action']:
             try:
-                action = getattr(attr, hook)
+                action = getattr(attr, option)
 
                 if action not in routes:
                     routes[action] = {}
-                routes[action][hook] = attr
+
+                # Routes decorated with the `@on()` decorator can be configured
+                # to skip validation of the input and output. For more info see
+                # the docstring of `on()`.
+                if option == '_on_action':
+                    routes[action]['_skip_schema_validation'] = \
+                        getattr(attr, '_skip_schema_validation', False)
+
+                routes[action][option] = attr
 
             except AttributeError:
                 continue

tests/test_routing.py

@@ -9,7 +9,7 @@ def test_create_route_map():
 
     """
     class ChargePoint:
-        @on(Action.Heartbeat)
+        @on(Action.Heartbeat, skip_schema_validation=True)
         def on_heartbeat(self):
             pass
 
@@ -31,8 +31,10 @@ def undecorated(self):
         Action.Heartbeat: {
             '_on_action': cp.on_heartbeat,
             '_after_action': cp.after_heartbeat,
+            '_skip_schema_validation': True,
         },
         Action.MeterValues: {
             '_on_action': cp.meter_values,
+            '_skip_schema_validation': False,
         },
     }

tests/v16/test_v16_charge_point.py

@@ -1,6 +1,7 @@
 import json
 import pytest
 import asyncio
+from unittest import mock
 
 from ocpp.exceptions import NotImplementedError
 from ocpp.routing import on, after, create_route_map
@@ -52,6 +53,45 @@ def after_boot_notification(charge_point_model, charge_point_vendor,
     )
 
 
+@pytest.mark.asyncio
+async def test_route_message_without_validation(base_central_system):
+    @on(Action.BootNotification, skip_schema_validation=True)
+    def on_boot_notification(**kwargs):  # noqa
+        assert kwargs['firmware_version'] == "#1:3.4.0-2990#N:217H;1.0-223"
+
+        return call_result.BootNotificationPayload(
+            current_time='2018-05-29T17:37:05.495259',
+            interval=350,
+            # 'Yolo' is not a valid value for for field status.
+            status='Yolo',
+        )
+
+    setattr(base_central_system, 'on_boot_notification', on_boot_notification)
+    base_central_system.route_map = create_route_map(base_central_system)
+
+    await base_central_system.route_message(json.dumps([
+        2,
+        1,
+        "BootNotification",
+        {
+            # The payload is missing the required fields 'chargepointVendor'
+            # and 'chargePointModel'.
+            "firmwareVersion": "#1:3.4.0-2990#N:217H;1.0-223"
+        }
+    ]))
+
+    base_central_system._connection.send.call_args == \
+        mock.call(json.dumps([
+            3,
+            "1",
+            {
+                'currentTime': '2018-05-29T17:37:05.495259',
+                'interval': 350,
+                'status': 'Yolo',
+            }
+        ]))
+
+
 @pytest.mark.asyncio
 async def test_route_message_with_no_route(base_central_system,
                                            heartbeat_call):