test_charge_point.py

from dataclasses import asdict

import pytest

from ocpp.charge_point import (
    camel_to_snake_case,
    remove_nones,
    serialize_as_dict,
    snake_to_camel_case,
)
from ocpp.messages import Call
from ocpp.routing import after, create_route_map, on
from ocpp.v16 import ChargePoint as cp_16
from ocpp.v16.call import BootNotification, GetConfiguration, MeterValues
from ocpp.v16.call_result import BootNotification as BootNotificationResult
from ocpp.v16.datatypes import MeterValue, SampledValue
from ocpp.v16.enums import Action, RegistrationStatus
from ocpp.v201 import ChargePoint as cp_201
from ocpp.v201 import call_result, datatypes, enums
from ocpp.v201.call import GetVariables as v201GetVariables
from ocpp.v201.call import SetNetworkProfile as v201SetNetworkProfile
from ocpp.v201.datatypes import (
    ComponentType,
    EVSEType,
    GetVariableDataType,
    NetworkConnectionProfileType,
    VariableType,
)
from ocpp.v201.enums import (
    OCPPInterfaceEnumType,
    OCPPTransportEnumType,
    OCPPVersionEnumType,
)


def test_getters_should_not_be_called_during_routemap_setup():
    class ChargePoint(cp_201):
        @property
        def foo(self):
            raise RuntimeError("this will be raised")

    try:
        ChargePoint("blah", None)
    except RuntimeError as e:
        assert str(e) == "this will be raised"
        pytest.fail("Getter was called during ChargePoint creation")


def test_multiple_classes_with_same_name_for_handler():
    class ChargerA(cp_201):
        @on(Action.heartbeat)
        def heartbeat(self, **kwargs):
            pass

    class ChargerB(cp_201):
        @on(Action.heartbeat)
        def heartbeat(self, **kwargs):
            pass

    A = ChargerA("A", None)
    B = ChargerB("B", None)
    route_mapA = create_route_map(A)
    route_mapB = create_route_map(B)
    assert route_mapA["Heartbeat"] != route_mapB["Heartbeat"]


@pytest.mark.parametrize(
    "test_input,expected",
    [
        ({"transactionId": "74563478"}, {"transaction_id": "74563478"}),
        ({"fullSoC": 100}, {"full_soc": 100}),
        ({"responderURL": "foo.com"}, {"responder_url": "foo.com"}),
        ({"url": "foo.com"}, {"url": "foo.com"}),
        ({"ocppCSMSURL": "foo.com"}, {"ocpp_csms_url": "foo.com"}),
        ({"CSMSRootCertificate": "foo.com"}, {"csms_root_certificate": "foo.com"}),
        ({"InvalidURL": "foo.com"}, {"invalid_url": "foo.com"}),
        ({"evMinV2XEnergyRequest": 200}, {"ev_min_v2x_energy_request": 200}),
        ({"v2xChargingCtrlr": 200}, {"v2x_charging_ctrlr": 200}),
        ({"webSocketPingInterval": 200}, {"web_socket_ping_interval": 200}),
        ({"signV2GCertificate": 200}, {"sign_v2g_certificate": 200}),
        (
            {"v2gCertificateInstallationEnabled": 200},
            {"v2g_certificate_installation_enabled": 200},
        ),
    ],
)
def test_camel_to_snake_case(test_input, expected):
    result = camel_to_snake_case(test_input)
    assert result == expected


@pytest.mark.parametrize(
    "test_input,expected",
    [
        ({"transaction_id": "74563478"}, {"transactionId": "74563478"}),
        ({"full_soc": 100}, {"fullSoC": 100}),
        ({"soc_limit_reached": 200}, {"SoCLimitReached": 200}),
        ({"ev_min_v2x_energy_request": 200}, {"evMinV2XEnergyRequest": 200}),
        ({"v2x_charging_ctrlr": 200}, {"v2xChargingCtrlr": 200}),
        ({"responder_url": "foo.com"}, {"responderURL": "foo.com"}),
        ({"url": "foo.com"}, {"url": "foo.com"}),
        ({"ocpp_csms_url": "foo.com"}, {"ocppCsmsUrl": "foo.com"}),
        ({"csms_root_certificate": "foo.com"}, {"CSMSRootCertificate": "foo.com"}),
        ({"invalid_url": "foo.com"}, {"invalidURL": "foo.com"}),
        ({"web_socket_ping_interval": 200}, {"webSocketPingInterval": 200}),
        ({"sign_v2g_certificate": 200}, {"signV2GCertificate": 200}),
        (
            {"v2g_certificate_installation_enabled": 200},
            {"v2gCertificateInstallationEnabled": 200},
        ),
    ],
)
def test_snake_to_camel_case(test_input, expected):
    result = snake_to_camel_case(test_input)
    assert result == expected


def test_remove_nones():
    expected_payload = {"charge_point_model": "foo", "charge_point_vendor": "bar"}

    payload = BootNotification(
        charge_point_model="foo",
        charge_point_vendor="bar",
        charge_box_serial_number=None,
    )
    payload = asdict(payload)

    assert expected_payload == remove_nones(payload)


def test_nested_remove_nones():
    expected_payload = {
        "configuration_slot": 1,
        "connection_data": {
            "ocpp_version": "OCPP20",
            "ocpp_transport": "JSON",
            "ocpp_csms_url": "wss://localhost:9000",
            "message_timeout": 60,
            "security_profile": 1,
            "ocpp_interface": "Wired0",
        },
    }

    connection_data = NetworkConnectionProfileType(
        ocpp_version=OCPPVersionEnumType.ocpp20,
        ocpp_transport=OCPPTransportEnumType.json,
        ocpp_csms_url="wss://localhost:9000",
        message_timeout=60,
        security_profile=1,
        ocpp_interface=OCPPInterfaceEnumType.wired0,
        vpn=None,
        apn=None,
    )

    payload = v201SetNetworkProfile(
        configuration_slot=1, connection_data=connection_data
    )
    payload = asdict(payload)

    assert expected_payload == remove_nones(payload)


def test_nested_list_remove_nones():
    expected_payload = {
        "connector_id": 3,
        "meter_value": [
            {
                "timestamp": "2017-08-17T07:08:06.186748+00:00",
                "sampled_value": [
                    {
                        "value": "10",
                        "context": "Sample.Periodic",
                        "measurand": "Power.Active.Import",
                        "unit": "W",
                    },
                    {
                        "value": "50000",
                        "context": "Sample.Periodic",
                        "measurand": "Power.Active.Import",
                        "phase": "L1",
                        "unit": "W",
                    },
                ],
            },
            {
                "timestamp": "2017-08-17T07:07:07.186748+00:00",
                "sampled_value": [
                    {
                        "value": "10",
                        "context": "Sample.Periodic",
                        "measurand": "Power.Active.Import",
                        "unit": "W",
                    },
                    {
                        "value": "50000",
                        "context": "Sample.Periodic",
                        "measurand": "Power.Active.Import",
                        "phase": "L1",
                        "unit": "W",
                    },
                ],
            },
        ],
        "transaction_id": 5,
    }

    payload = MeterValues(
        connector_id=3,
        meter_value=[
            MeterValue(
                timestamp="2017-08-17T07:08:06.186748+00:00",
                sampled_value=[
                    SampledValue(
                        value="10",
                        context="Sample.Periodic",
                        format=None,
                        measurand="Power.Active.Import",
                        phase=None,
                        location=None,
                        unit="W",
                    ),
                    SampledValue(
                        value="50000",
                        context="Sample.Periodic",
                        format=None,
                        measurand="Power.Active.Import",
                        phase="L1",
                        location=None,
                        unit="W",
                    ),
                ],
            ),
            MeterValue(
                timestamp="2017-08-17T07:07:07.186748+00:00",
                sampled_value=[
                    SampledValue(
                        value="10",
                        context="Sample.Periodic",
                        format=None,
                        measurand="Power.Active.Import",
                        phase=None,
                        location=None,
                        unit="W",
                    ),
                    SampledValue(
                        value="50000",
                        context="Sample.Periodic",
                        format=None,
                        measurand="Power.Active.Import",
                        phase="L1",
                        location=None,
                        unit="W",
                    ),
                ],
            ),
        ],
        transaction_id=5,
    )

    payload = asdict(payload)
    assert expected_payload == remove_nones(payload)


def test_remove_nones_with_list_of_strings():
    """Make sure that `remove_nones()` doesn't crash when it encounters an
    iterable other than a list or dict. See
    https://github.com/mobilityhouse/ocpp/issues/289.
    """
    payload = asdict(
        GetConfiguration(key=["ClockAlignedDataInterval", "ConnectionTimeOut"])
    )

    assert remove_nones(payload) == {
        "key": ["ClockAlignedDataInterval", "ConnectionTimeOut"]
    }


def test_serialize_as_dict():
    """
    Test recursively serializing a dataclasses as a dictionary.
    """
    # Setup
    expected = camel_to_snake_case(
        {
            "getVariableData": [
                {
                    "component": {
                        "name": "Component",
                        "instance": None,
                        "evse": {
                            "id": 1,
                            "connectorId": None,
                        },
                    },
                    "variable": {
                        "name": "Variable",
                        "instance": None,
                    },
                    "attributeType": None,
                }
            ],
            "customData": None,
        }
    )

    payload = v201GetVariables(
        get_variable_data=[
            GetVariableDataType(
                component=ComponentType(
                    name="Component",
                    evse=EVSEType(id=1),
                ),
                variable=VariableType(name="Variable"),
            )
        ]
    )

    # Execute / Assert
    assert serialize_as_dict(payload) == expected


def test_serialization_of_collection_of_multiple_elements():
    """This test validates that bug #635 is fixed.
    That bug incorrectly serialized payloads that contain a collection
    of elements.

    This test serializes a call_result.SetVariables that contains 2
    SetVariableResultTypes.

    https://github.com/mobilityhouse/ocpp/issues/635
    """
    payload = call_result.SetVariables(
        set_variable_result=[
            datatypes.SetVariableResultType(
                attribute_status=enums.SetVariableStatusEnumType.accepted,
                component={
                    "name": "TemperatureSensor",
                    "instance": "First",
                    "evse": {"id": 1, "connector_id": 1},
                },
                variable={"name": "DisplayUnit", "instance": "Main"},
                attribute_type="Actual",
                attribute_status_info=None,
            ),
            datatypes.SetVariableResultType(
                attribute_status="Accepted",
                component={"name": "TxCtrlr"},
                variable={"name": "TxStopPoint"},
                attribute_type="Actual",
                attribute_status_info=None,
            ),
        ],
        custom_data=None,
    )

    expected = {
        "custom_data": None,
        "set_variable_result": [
            {
                "attribute_status": "Accepted",
                "attribute_status_info": None,
                "attribute_type": "Actual",
                "component": {
                    "evse": {"connector_id": 1, "id": 1},
                    "instance": "First",
                    "name": "TemperatureSensor",
                },
                "variable": {"instance": "Main", "name": "DisplayUnit"},
            },
            {
                "attribute_status": "Accepted",
                "attribute_status_info": None,
                "attribute_type": "Actual",
                "component": {"name": "TxCtrlr"},
                "variable": {"name": "TxStopPoint"},
            },
        ],
    }
    # Execute / Assert
    assert serialize_as_dict(payload) == expected


@pytest.mark.asyncio
async def test_call_unique_id_added_to_handler_args_correctly(connection):
    """
    This test ensures that the `call_unique_id` is getting passed to the
    `on` and `after` handlers only if it is explicitly set in the handler signature.

    To cover all possible cases, we define two chargers:

    ChargerA:
    `call_unique_id` not required on `on` handler but required on `after` handler.

    ChargerB:
    `call_unique_id` required on `on` handler but not required on `after` handler.

    Each handler verifies a set of asserts to verify that the `call_unique_id`
    is passed correctly.
    To confirm that the handlers are actually being called and hence the asserts
    are being ran, we introduce a set of counters that increase each time a specific
    handler runs.
    """
    charger_a_test_call_unique_id = "charger_a_1234"
    charger_b_test_call_unique_id = "charger_b_5678"
    payload_a = {"chargePointVendor": "foo_a", "chargePointModel": "bar_a"}
    payload_b = {"chargePointVendor": "foo_b", "chargePointModel": "bar_b"}

    class ChargerA(cp_16):
        on_boot_notification_call_count = 0
        after_boot_notification_call_count = 0

        @on(Action.boot_notification)
        def on_boot_notification(self, *args, **kwargs):
            # call_unique_id should not be passed as arg nor kwarg
            assert kwargs == camel_to_snake_case(payload_a)
            assert args == ()
            ChargerA.on_boot_notification_call_count += 1
            return BootNotificationResult(
                current_time="foo", interval=1, status=RegistrationStatus.accepted
            )

        @after(Action.boot_notification)
        def after_boot_notification(self, call_unique_id, *args, **kwargs):
            assert call_unique_id == charger_a_test_call_unique_id
            assert kwargs == camel_to_snake_case(payload_a)
            # call_unique_id should not be passed as arg
            assert args == ()
            ChargerA.after_boot_notification_call_count += 1
            return BootNotificationResult(
                current_time="foo", interval=1, status=RegistrationStatus.accepted
            )

    class ChargerB(cp_16):
        on_boot_notification_call_count = 0
        after_boot_notification_call_count = 0

        @on(Action.boot_notification)
        def on_boot_notification(self, call_unique_id, *args, **kwargs):
            assert call_unique_id == charger_b_test_call_unique_id
            assert kwargs == camel_to_snake_case(payload_b)
            # call_unique_id should not be passed as arg
            assert args == ()
            ChargerB.on_boot_notification_call_count += 1
            return BootNotificationResult(
                current_time="foo", interval=1, status=RegistrationStatus.accepted
            )

        @after(Action.boot_notification)
        def after_boot_notification(self, *args, **kwargs):
            # call_unique_id should not be passed as arg nor kwarg
            assert kwargs == camel_to_snake_case(payload_b)
            assert args == ()
            ChargerB.after_boot_notification_call_count += 1
            return BootNotificationResult(
                current_time="foo", interval=1, status=RegistrationStatus.accepted
            )

    charger_a = ChargerA("charger_a_id", connection)
    charger_b = ChargerB("charger_b_id", connection)

    msg_a = Call(
        unique_id=charger_a_test_call_unique_id,
        action=Action.boot_notification.value,
        payload=payload_a,
    )
    await charger_a._handle_call(msg_a)

    msg_b = Call(
        unique_id=charger_b_test_call_unique_id,
        action=Action.boot_notification.value,
        payload=payload_b,
    )
    await charger_b._handle_call(msg_b)

    assert ChargerA.on_boot_notification_call_count == 1
    assert ChargerA.after_boot_notification_call_count == 1
    assert ChargerB.on_boot_notification_call_count == 1
    assert ChargerB.after_boot_notification_call_count == 1