Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

test: Unit test for RemoteStartTransaction nested dataclass #5

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/source/central_system.rst
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ Remove the `on_connect()` handler from the code above and replace it by the foll


class MyChargePoint(cp):
@on(Action.BootNotification)
@on(Action.boot_notification)
async def on_boot_notification(self, charge_point_vendor, charge_point_model, **kwargs):
return call_result.BootNotificationPayload(
current_time=datetime.utcnow().isoformat(),
Expand Down
2 changes: 1 addition & 1 deletion examples/v16/central_system.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@


class ChargePoint(cp):
@on(Action.BootNotification)
@on(Action.boot_notification)
def on_boot_notification(
self, charge_point_vendor: str, charge_point_model: str, **kwargs
):
Expand Down
5 changes: 3 additions & 2 deletions examples/v201/central_system.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,18 +16,19 @@
from ocpp.routing import on
from ocpp.v201 import ChargePoint as cp
from ocpp.v201 import call_result
from ocpp.v201.enums import Action

logging.basicConfig(level=logging.INFO)


class ChargePoint(cp):
@on("BootNotification")
@on(Action.boot_notification)
def on_boot_notification(self, charging_station, reason, **kwargs):
return call_result.BootNotificationPayload(
current_time=datetime.utcnow().isoformat(), interval=10, status="Accepted"
)

@on("Heartbeat")
@on(Action.heartbeat)
def on_heartbeat(self):
print("Got a Heartbeat!")
return call_result.HeartbeatPayload(
Expand Down
10 changes: 6 additions & 4 deletions ocpp/charge_point.py
Original file line number Diff line number Diff line change
Expand Up @@ -170,20 +170,22 @@ def _raise_key_error(action, version):
from ocpp.v201.enums import Action as v201_Action

if version == "1.6":
if hasattr(v16_Action, action):
try:
v16_Action(action)
raise NotImplementedError(
details={"cause": f"No handler for {action} registered."}
)
else:
except ValueError:
raise NotSupportedError(
details={"cause": f"{action} not supported by OCPP{version}."}
)
elif version in ["2.0", "2.0.1"]:
if hasattr(v201_Action, action):
try:
v201_Action(action)
raise NotImplementedError(
details={"cause": f"No handler for {action} registered."}
)
else:
except ValueError:
raise NotSupportedError(
details={"cause": f"{action} not supported by OCPP{version}."}
)
Expand Down
10 changes: 5 additions & 5 deletions ocpp/routing.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ def on(action, *, skip_schema_validation=False):

```
class MyChargePoint(cp):
@on(Action.BootNotification):
@on(Action.boot_notification):
async def on_boot_notification(
self,
charge_point_model,
Expand Down Expand Up @@ -64,7 +64,7 @@ def after(action):

It can be used like so:

@after(Action.BootNotification):
@after(Action.boot_notification):
def after_boot_notification():
pass

Expand Down Expand Up @@ -93,19 +93,19 @@ def create_route_map(obj):

class ChargePoint:

@on(Action.BootNotification)
@on(Action.boot_notification)
def on_boot_notification(self, *args, **kwargs):
pass

@after(Action.BootNotification)
@after(Action.boot_notification)
def after_boot_notification(self, *args, **kwargs):
pass


In this case this returns:

{
Action.BootNotification: {
Action.boot_notification: {
'_on_action': <reference to 'on_boot_notification'>,
'_after_action': <reference to 'after_boot_notification'>,
'_skip_schema_validation': False,
Expand Down
7 changes: 4 additions & 3 deletions ocpp/v16/call.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import warnings
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from typing import Dict, List, Optional, Union

from ocpp.v16.datatypes import ChargingProfile
from ocpp.v16.enums import (
AvailabilityType,
CertificateUse,
Expand Down Expand Up @@ -133,7 +134,7 @@ class InstallCertificate:
class RemoteStartTransaction:
id_tag: str
connector_id: Optional[int] = None
charging_profile: Optional[Dict] = None
charging_profile: Optional[Union[Dict, ChargingProfile]] = None


@dataclass
Expand Down Expand Up @@ -165,7 +166,7 @@ class SendLocalList:
@dataclass
class SetChargingProfile:
connector_id: int
cs_charging_profiles: Dict
cs_charging_profiles: Union[ChargingProfile, Dict]


@dataclass
Expand Down
52 changes: 0 additions & 52 deletions ocpp/v16/enums.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,3 @@
from warnings import warn

try:
# breaking change introduced in python 3.11
from enum import StrEnum
Expand All @@ -13,56 +11,6 @@ class StrEnum(str, Enum): # pragma: no cover
class Action(StrEnum):
"""An Action is a required part of a Call message."""

def __init__(self, *args, **kwargs):
warn(
message="Action enum contains deprecated members and will be removed in "
"the next major release, please use snake case members.",
category=DeprecationWarning,
)

# --------- Soon to be deprecated ---------------------
Authorize = "Authorize"
BootNotification = "BootNotification"
CancelReservation = "CancelReservation"
CertificateSigned = "CertificateSigned"
ChangeAvailability = "ChangeAvailability"
ChangeConfiguration = "ChangeConfiguration"
ClearCache = "ClearCache"
ClearChargingProfile = "ClearChargingProfile"
DataTransfer = "DataTransfer"
DeleteCertificate = "DeleteCertificate"
DiagnosticsStatusNotification = "DiagnosticsStatusNotification"
ExtendedTriggerMessage = "ExtendedTriggerMessage"
FirmwareStatusNotification = "FirmwareStatusNotification"
GetCompositeSchedule = "GetCompositeSchedule"
GetConfiguration = "GetConfiguration"
GetDiagnostics = "GetDiagnostics"
GetInstalledCertificateIds = "GetInstalledCertificateIds"
GetLocalListVersion = "GetLocalListVersion"
GetLog = "GetLog"
Heartbeat = "Heartbeat"
InstallCertificate = "InstallCertificate"
LogStatusNotification = "LogStatusNotification"
MeterValues = "MeterValues"
RemoteStartTransaction = "RemoteStartTransaction"
RemoteStopTransaction = "RemoteStopTransaction"
ReserveNow = "ReserveNow"
Reset = "Reset"
SecurityEventNotification = "SecurityEventNotification"
SendLocalList = "SendLocalList"
SetChargingProfile = "SetChargingProfile"
SignCertificate = "SignCertificate"
SignedFirmwareStatusNotification = "SignedFirmwareStatusNotification"
SignedUpdateFirmware = "SignedUpdateFirmware"
StartTransaction = "StartTransaction"
StatusNotification = "StatusNotification"
StopTransaction = "StopTransaction"
TriggerMessage = "TriggerMessage"
UnlockConnector = "UnlockConnector"
UpdateFirmware = "UpdateFirmware"

# --------------------------------------------------------

authorize = "Authorize"
boot_notification = "BootNotification"
cancel_reservation = "CancelReservation"
Expand Down
Loading