return result
-def _parse_set_variable_specs(specs_str: str) -> list[dict]:
- return _parse_variable_specs(specs_str, require_value=True)
-
-
-def _parse_get_variable_specs(specs_str: str) -> list[dict]:
- return _parse_variable_specs(specs_str, require_value=False)
-
-
async def main():
parser = argparse.ArgumentParser(description="OCPP2 Server")
command_group = parser.add_mutually_exclusive_group()
if parsed_commands is not None and not parsed_commands:
parser.error("--commands must contain at least one CMD:DELAY entry")
parsed_set_variables = (
- _parse_set_variable_specs(args.set_variables)
+ _parse_variable_specs(args.set_variables, require_value=True)
if args.set_variables
else None
)
parsed_get_variables = (
- _parse_get_variable_specs(args.get_variables)
+ _parse_variable_specs(args.get_variables, require_value=False)
if args.get_variables
else None
)
ChargePoint,
ServerConfig,
_parse_commands,
- _parse_get_variable_specs,
- _parse_set_variable_specs,
+ _parse_variable_specs,
_random_request_id,
check_positive_number,
main,
"""Tests for multi-variable SetVariables/GetVariables CLI support."""
def test_parse_set_variable_specs_valid(self):
- result = _parse_set_variable_specs(
- "OCPPCommCtrlr.HeartbeatInterval=30,TxCtrlr.EVConnectionTimeOut=60"
+ result = _parse_variable_specs(
+ "OCPPCommCtrlr.HeartbeatInterval=30,TxCtrlr.EVConnectionTimeOut=60",
+ require_value=True,
)
assert len(result) == 2
assert result[0]["component"]["name"] == "OCPPCommCtrlr"
assert result[1]["attribute_value"] == "60"
def test_parse_get_variable_specs_valid(self):
- result = _parse_get_variable_specs(
- "ChargingStation.AvailabilityState,OCPPCommCtrlr.HeartbeatInterval"
+ result = _parse_variable_specs(
+ "ChargingStation.AvailabilityState,OCPPCommCtrlr.HeartbeatInterval",
+ require_value=False,
)
assert len(result) == 2
assert result[0]["component"]["name"] == "ChargingStation"
argparse.ArgumentTypeError,
match=r"expected 'Component\.Variable=Value'",
):
- _parse_set_variable_specs("NoComponentVariable=30")
+ _parse_variable_specs("NoComponentVariable=30", require_value=True)
async def test_send_set_variables_uses_custom_data(self, command_charge_point):
custom_data = [