From 1bcca1204e23f6a74b6473d069be9d8a3bf9c322 Mon Sep 17 00:00:00 2001 From: Matt Rayner Date: Sun, 7 Apr 2024 21:52:46 +0100 Subject: [PATCH] Add charge override deletion and fetching connection status * Add getting connection status from API: * `Client.async_get_connection_status` - @mattrayner * Add support for charge override deletion: * Add `Client.async_delete_charge_override` - @mattrayner --- CHANGELOG.md | 11 +- example.py | 21 +++ podpointclient/client.py | 36 ++++- podpointclient/connectivity_status.py | 166 ++++++++++++++++++++++ podpointclient/endpoints.py | 7 +- podpointclient/factories.py | 10 ++ podpointclient/version.py | 2 +- tests/fixtures/connectivity_status.json | 28 ++++ tests/helpers.py | 3 + tests/test_client.py | 157 +++++++++++++++++++- tests/test_connectivity_status.py | 86 +++++++++++ tests/test_connectivity_status_factory.py | 16 +++ 12 files changed, 533 insertions(+), 10 deletions(-) create mode 100644 podpointclient/connectivity_status.py create mode 100644 tests/fixtures/connectivity_status.json create mode 100644 tests/test_connectivity_status.py create mode 100644 tests/test_connectivity_status_factory.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a988109..8672d7f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Pod Point Client Changelog +## v1.6.0 + +* Add getting connection status from API: + * `Client.async_get_connection_status` - @mattrayner +* Add support for charge override deletion: + * Add `Client.async_delete_charge_override` - @mattrayner + ## v1.5.0 * Add support for refreshing expired tokens, rather than grabbing new ones each time @@ -61,7 +68,7 @@ * Added additional testing dependencies - @mattrayner * Add CD pipeline, when a new tag/release is pushed, auto-publish to PyPi - @mattrayner -## v0.3.0 +## v0.3.0 * Add http_debug flag - @mattrayner * When enabled, complete response bodies will be sent to logger.debug @@ -69,7 +76,7 @@ * Completed a pylon pass to standardize the code base - @mattrayner * Improved test coverage - @mattrayner -## v0.2.2 +## v0.2.2 * Make timestamp=XXX optional, and off by default * Greatly improve test coverage diff --git a/example.py b/example.py index 936698b..f3b8053 100644 --- a/example.py +++ b/example.py @@ -63,6 +63,27 @@ async def main(username: str, password: str, http_debug: bool = False, loop=None energy_used = charges[0].kwh_used print(f" kW charged: {energy_used}") + # Set charge override + print(f"Setting 'Charge now' for pod {pod.ppid}") + override = await client.async_set_charge_override(pod=pod, hours=1) + print(f" Override until: {override.ends_at}") + + # Delete override + print(f"Deleting 'Charge now' for pod {pod.ppid}") + await client.async_delete_charge_override(pod=pod) + print(" Done") + + # Get charge override + print(f"Attempting to get charge override for pod {pod.ppid}") + override = await client.async_get_charge_override(pod=pod) + print(f" Override ends at: {override.ends_at}") + + # Get connectivity status + print(f"Getting connectivity status for pod {pod.ppid}") + connectivity = await client.async_get_connectivity_status(pod=pod) + print(f" Connectivity status: {connectivity.evses[0].connectivity_state.connectivity_status}") + print(f" Last message at: {connectivity.evses[0].connectivity_state.last_message_at}") + # Expire token and exchange a refresh print("Expiring token and refreshing...") client.auth.access_token_expiry = datetime.now() - timedelta(minutes=10) diff --git a/podpointclient/client.py b/podpointclient/client.py index 6cd3965..ab5ed20 100644 --- a/podpointclient/client.py +++ b/podpointclient/client.py @@ -5,15 +5,16 @@ import aiohttp -from .endpoints import API_BASE_URL, CHARGE_SCHEDULES, PODS, UNITS, USERS, CHARGES, FIRMWARE, AUTH, CHARGE_OVERRIDE +from .endpoints import API_BASE_URL, CHARGE_SCHEDULES, PODS, UNITS, USERS, CHARGES, FIRMWARE, AUTH, CHARGE_OVERRIDE, CHARGERS, CONNECTIVITY_STATUS, MOBILE_API_BASE_URL from .helpers.auth import Auth from .helpers.functions import auth_headers from .helpers.api_wrapper import APIWrapper -from .factories import PodFactory, ScheduleFactory, ChargeFactory, FirmwareFactory, UserFactory, ChargeOverrideFactory +from .factories import PodFactory, ScheduleFactory, ChargeFactory, FirmwareFactory, UserFactory, ChargeOverrideFactory, ConnectivityStatusFactory from .pod import Pod, Firmware from .charge import Charge from .charge_mode import ChargeMode from .charge_override import ChargeOverride +from .connectivity_status import ConnectivityStatus from .schedule import Schedule from .user import User from .errors import ChargeOverrideValidationError @@ -251,6 +252,33 @@ async def async_get_charge_override(self, pod: Pod) -> Union[None, ChargeOverrid return ChargeOverrideFactory().build_charge_override(charge_override_response=json) + async def async_delete_charge_override(self, pod:Pod) -> bool: + await self.auth.async_update_access_token() + + response = await self.api_wrapper.delete( + url=self._url_from_path( + path=f"{UNITS}/{pod.unit_id}{CHARGE_OVERRIDE}"), + params=self._generate_complete_params(params=None), + headers=auth_headers(access_token=self.auth.access_token) + ) + + return response.status == 204 + + async def async_get_connectivity_status(self, pod:Pod) -> ConnectivityStatus: + await self.auth.async_update_access_token() + + response = await self.api_wrapper.get( + url=self._url_from_path( + path=f"{CHARGERS}/{pod.ppid}{CONNECTIVITY_STATUS}", + base=MOBILE_API_BASE_URL + ), + params=self._generate_complete_params(params=None), + headers=auth_headers(access_token=self.auth.access_token) + ) + + json = await self._handle_json_response(response=response) + + return ConnectivityStatusFactory().build_connectivity_status(connectivity_status_response=json) async def async_set_charge_override(self, pod:Pod, hours:int=0, minutes:int=0, seconds:int=0) -> ChargeOverride: await self.auth.async_update_access_token() @@ -348,9 +376,9 @@ def _schedule_data(self, enabled: bool) -> Dict[str, Any]: return {"data": d_list} - def _url_from_path(self, path: str) -> str: + def _url_from_path(self, path: str, base: str = API_BASE_URL) -> str: """Given a path, return a complete API URL""" - return f"{API_BASE_URL}{path}" + return f"{base}{path}" def _generate_complete_params(self, params: Union[None, Dict[str, Any]]) -> Dict[str, any]: """Given a params object, add optional params if required""" diff --git a/podpointclient/connectivity_status.py b/podpointclient/connectivity_status.py new file mode 100644 index 0000000..f1bd31b --- /dev/null +++ b/podpointclient/connectivity_status.py @@ -0,0 +1,166 @@ +"""Connectivity State class, represents a 'Connectivity State' from the podpoint apis""" + +from datetime import datetime, timedelta +from typing import Dict, Any, List +from dataclasses import dataclass, field +from .helpers.functions import lazy_convert_to_datetime, lazy_iso_format_datetime +import json + +# { +# "ppid": "PSL-266056", +# "evses": [{ +# "id": 1, +# "connectivityState": { +# "protocol": "POW", +# "connectivityStatus": "ONLINE", +# "signalStrength": -68, +# "lastMessageAt": "2024-04-05T18:36:29Z", +# "connectionStartedAt": "2024-04-05T18:26:26.819Z", +# "connectionQuality": 3 +# }, +# "connectors": [{ +# "id": 1, +# "door": "A", +# "chargingState": "SUSPENDED_EV" +# }], +# "architecture": "arch3", +# "energyOfferStatus": { +# "isOfferingEnergy": true, +# "reason": "CHARGE_SCHEDULE", +# "until": null, +# "randomDelay": null, +# "doNotCache": false +# } +# }], +# "connectedComponents": ["evses"] +# } + +@dataclass +class Evse: + """Represents a Location within a charge from pod point""" + + def __init__(self, data: Dict[str, Any]): + self.id: int = data.get('id', None) + self.architecture: str = data.get('architecture', None) + + connectivity_state_data = data.get('connectivityState', {}) + self.connectivity_state = self.ConnectivityState(data=connectivity_state_data) + + connectors_data = data.get('connectors', []) + self.connectors = [] + for connector in connectors_data: + self.connectors.append(self.Connector(data=connector)) + + energy_offer_status_data = data.get('energyOfferStatus', {}) + self.energy_offer_status = self.EnergyOfferStatus(data=energy_offer_status_data) + + @property + def dict(self): + return { + "id": self.id, + "architecture": self.architecture, + "connectivityState": self.connectivity_state.dict, + "connectors": [connector.dict for connector in self.connectors], + "energyOfferStatus": self.energy_offer_status.dict + } + + def to_json(self): + """JSON representation of a ConnectivityState object""" + return json.dumps(self.dict, ensure_ascii=False) + + @dataclass + class ConnectivityState: + """Represents a Location within a charge from pod point""" + + def __init__(self, data: Dict[str, Any]): + self.protocol: str = data.get('protocol', None) + self.connectivity_status: str = data.get('connectivityStatus', None) + self.signal_strength:int = data.get('signalStrength', None) + self.last_message_at: datetime = lazy_convert_to_datetime(data.get('lastMessageAt', None)) + self.connection_started_at: datetime = lazy_convert_to_datetime(data.get('connectionStartedAt', None)) + self.connection_quality: int = data.get('connectionQuality', None) + + @property + def dict(self): + return { + "protocol": self.protocol, + "connectivityStatus": self.connectivity_status, + "signalStrength": self.signal_strength, + "lastMessageAt": lazy_iso_format_datetime(self.last_message_at), + "connectionStartedAt": lazy_iso_format_datetime(self.connection_started_at), + "connectionQuality": self.connection_quality + } + + def to_json(self): + """JSON representation of a ConnectivityState object""" + return json.dumps(self.dict, ensure_ascii=False) + + @dataclass + class Connector: + """Represents a Location within a charge from pod point""" + + def __init__(self, data: Dict[str, Any]): + self.id: int = data.get('id', None) + self.door: str = data.get('door', None) + self.charging_state: str = data.get('chargingState', None) + + @property + def dict(self): + return { + "id": self.id, + "door": self.door, + "chargingState": self.charging_state + } + + def to_json(self): + """JSON representation of a ConnectivityState object""" + return json.dumps(self.dict, ensure_ascii=False) + + @dataclass + class EnergyOfferStatus: + """Represents a Location within a charge from pod point""" + + def __init__(self, data: Dict[str, Any]): + self.is_offering_energy: bool = data.get('isOfferingEnergy', None) + self.reason: str = data.get('reason', None) + self.until: datetime = lazy_convert_to_datetime(data.get('until', None)) + self.random_delay = data.get('randomDelay', None) + self.do_not_cache: bool = data.get('doNotCache', None) + + @property + def dict(self): + return { + "isOfferingEnergy": self.is_offering_energy, + "reason": self.reason, + "until": lazy_iso_format_datetime(self.until), + "randomDelay": self.random_delay, + "doNotCache": self.do_not_cache + } + + def to_json(self): + """JSON representation of a ConnectivityState object""" + return json.dumps(self.dict, ensure_ascii=False) + + +class ConnectivityStatus: + """Representation of a Connectivity State from pod point""" + + def __init__(self, data: Dict[str, Any]): + self.ppid: int = data.get('ppid', None) + self.connected_components: List[str] = data.get('connectedComponents', []) + + self.evses: List[Evse] = [] + for evse in data.get('evses', []): + self.evses.append(Evse(data=evse)) + + @property + def dict(self) -> Dict[str, Any]: + return { + "ppid": self.ppid, + "connected_components": self.connected_components, + "evses": [evse.dict for evse in self.evses] + } + + def to_json(self): + """JSON representation of a ConnectivityState object""" + return json.dumps(self.dict, ensure_ascii=False) diff --git a/podpointclient/endpoints.py b/podpointclient/endpoints.py index cec3cbe..748cbcd 100644 --- a/podpointclient/endpoints.py +++ b/podpointclient/endpoints.py @@ -8,12 +8,17 @@ CHARGE_SCHEDULES = '/charge-schedules' CHARGES = '/charges' CHARGE_OVERRIDE = '/charge-override' +CHARGERS = '/chargers' +CONNECTIVITY_STATUS = '/connectivity-status' FIRMWARE = '/firmware' -API_BASE = 'mobile-api.pod-point.com/api3/' +MOBILE_API_BASE = 'mobile-api.pod-point.com' +API_BASE = f"{MOBILE_API_BASE}/api3/" API_VERSION = 'v5' API_BASE_URL = f"https://{API_BASE}{API_VERSION}" +MOBILE_API_BASE_URL = f"https://{MOBILE_API_BASE}" + """Google endpoint, used for auth""" GOOGLE_KEY = '?key=AIzaSyCwhF8IOl_7qHXML0pOd5HmziYP46IZAGU' PASSWORD_VERIFY = f"/verifyPassword{GOOGLE_KEY}" diff --git a/podpointclient/factories.py b/podpointclient/factories.py index 7cfd310..7f2ac83 100644 --- a/podpointclient/factories.py +++ b/podpointclient/factories.py @@ -5,6 +5,7 @@ from .schedule import Schedule, ScheduleStatus from .charge import Charge from .charge_override import ChargeOverride +from .connectivity_status import ConnectivityStatus class PodFactory: """Factory for creating Pod objects""" @@ -96,3 +97,12 @@ def build_user(self, user_response: Dict[str, Any]) -> User: return None return User(data=user_data) + +class ConnectivityStatusFactory: + """Factory for creating ConnectivityStatus objects""" + def build_connectivity_status(self, connectivity_status_response: Dict[str, Any]): + """Build a ConnectivityStatus object based off of a response from pod point""" + if connectivity_status_response is None: + return None + + return ConnectivityStatus(data=connectivity_status_response) diff --git a/podpointclient/version.py b/podpointclient/version.py index de4174c..953841f 100644 --- a/podpointclient/version.py +++ b/podpointclient/version.py @@ -1,3 +1,3 @@ """Version for the podpointclient library""" -__version__ = "1.5.0" +__version__ = "1.6.0a1" diff --git a/tests/fixtures/connectivity_status.json b/tests/fixtures/connectivity_status.json new file mode 100644 index 0000000..c56da5f --- /dev/null +++ b/tests/fixtures/connectivity_status.json @@ -0,0 +1,28 @@ +{ + "ppid": "PSL-123456", + "evses": [{ + "id": 1, + "connectivityState": { + "protocol": "POW", + "connectivityStatus": "ONLINE", + "signalStrength": -68, + "lastMessageAt": "2024-04-05T18:26:28Z", + "connectionStartedAt": "2024-04-05T18:26:26.819Z", + "connectionQuality": 3 + }, + "connectors": [{ + "id": 1, + "door": "A", + "chargingState": "SUSPENDED_EV" + }], + "architecture": "arch3", + "energyOfferStatus": { + "isOfferingEnergy": true, + "reason": "CHARGE_SCHEDULE", + "until": null, + "randomDelay": null, + "doNotCache": false + } + }], + "connectedComponents": ["evses"] +} \ No newline at end of file diff --git a/tests/helpers.py b/tests/helpers.py index 1debf66..9a2b796 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -81,6 +81,9 @@ def firmware_response(self): def user_response(self): return self.__json_load_fixture('complete_user') + def connectivity_status_response(self): + return self.__json_load_fixture('connectivity_status') + def __json_load_fixture(self, fixture_name: str): file_location = os.path.dirname(__file__) path = f'fixtures/{fixture_name}.json' diff --git a/tests/test_client.py b/tests/test_client.py index ed1cdb2..f75a153 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -7,15 +7,17 @@ from podpointclient.pod import Pod, Firmware from podpointclient.charge import Charge from podpointclient.charge_override import ChargeOverride +from podpointclient.connectivity_status import ConnectivityStatus, Evse from podpointclient.user import User from podpointclient.errors import ChargeOverrideValidationError import pytest +from datetime import datetime, timezone from freezegun import freeze_time import json import pytz from datetime import timedelta -from podpointclient.endpoints import GOOGLE_BASE_URL, PASSWORD_VERIFY, API_BASE_URL, AUTH, CHARGE_SCHEDULES, CHARGES, FIRMWARE, PODS, SESSIONS, UNITS, USERS, CHARGE_OVERRIDE +from podpointclient.endpoints import GOOGLE_BASE_URL, PASSWORD_VERIFY, API_BASE_URL, AUTH, CHARGE_SCHEDULES, CHARGES, FIRMWARE, PODS, SESSIONS, UNITS, USERS, CHARGE_OVERRIDE, MOBILE_API_BASE_URL, CHARGERS, CONNECTIVITY_STATUS @pytest.mark.asyncio @freeze_time("Jan 1st, 2022") @@ -791,7 +793,158 @@ async def test_async_set_charge_override_with_an_invalid_time_set(): with pytest.raises(ChargeOverrideValidationError): await client.async_set_charge_override(pod=Pod(data={"unit_id": 1234}), hours=-3, minutes=0, seconds="a") - + + +@pytest.mark.asyncio +@freeze_time("Jan 1st, 2022") +async def test_async_delete_charge_override_with_a_204_response(): + auth_response = { + "idToken": "1234", + "expiresIn": "1234", + "refreshToken": "1234" + } + session_response = { + "sessions": { + "id": "1234", + "user_id": "1234" + } + } + + with aioresponses() as m: + m.post(f'{GOOGLE_BASE_URL}{PASSWORD_VERIFY}', payload=auth_response) + m.post(f'{API_BASE_URL}{SESSIONS}', payload=session_response) + m.delete(f'{API_BASE_URL}{UNITS}/1234{CHARGE_OVERRIDE}?timestamp=1640995200.0', status=204) + + async with aiohttp.ClientSession() as session: + client = PodPointClient(username="1233", password="1234", session=session, include_timestamp=True) + + response = await client.async_delete_charge_override(pod=Pod(data={"unit_id": 1234})) + assert response is True + + +@pytest.mark.asyncio +@freeze_time("Jan 1st, 2022") +async def test_async_delete_charge_override_with_non_204_response(): + auth_response = { + "idToken": "1234", + "expiresIn": "1234", + "refreshToken": "1234" + } + session_response = { + "sessions": { + "id": "1234", + "user_id": "1234" + } + } + + with aioresponses() as m: + m.post(f'{GOOGLE_BASE_URL}{PASSWORD_VERIFY}', payload=auth_response) + m.post(f'{API_BASE_URL}{SESSIONS}', payload=session_response) + m.delete(f'{API_BASE_URL}{UNITS}/1234{CHARGE_OVERRIDE}?timestamp=1640995200.0', status=201) + + async with aiohttp.ClientSession() as session: + client = PodPointClient(username="1233", password="1234", session=session, include_timestamp=True) + + response = await client.async_delete_charge_override(pod=Pod(data={"unit_id": 1234})) + assert response is False + + +@pytest.mark.asyncio +@freeze_time("Jan 1st, 2022") +async def test_async_get_connectivity_status(): + auth_response = { + "idToken": "1234", + "expiresIn": "1234", + "refreshToken": "1234" + } + session_response = { + "sessions": { + "id": "1234", + "user_id": "1234" + } + } + connectivity_status_response = { + "ppid": "PSL-123456", + "evses": [ + { + "id": 1, + "connectivityState": { + "protocol": "POW", + "connectivityStatus": "ONLINE", + "signalStrength": -68, + "lastMessageAt": "2024-04-05T18:36:29Z", + "connectionStartedAt": "2024-04-05T18:26:26.819Z", + "connectionQuality": 3 + }, + "connectors": [{ + "id": 1, + "door": "A", + "chargingState": "SUSPENDED_EV" + }], + "architecture": "arch3", + "energyOfferStatus": { + "isOfferingEnergy": True, + "reason": "CHARGE_SCHEDULE", + "until": None, + "randomDelay": None, + "doNotCache": False + } + } + ], + "connectedComponents": ["evse"] + } + + with aioresponses() as m: + m.post(f'{GOOGLE_BASE_URL}{PASSWORD_VERIFY}', payload=auth_response) + m.post(f'{API_BASE_URL}{SESSIONS}', payload=session_response) + m.get(f'{MOBILE_API_BASE_URL}{CHARGERS}/PSL-123456{CONNECTIVITY_STATUS}?timestamp=1640995200.0', payload=connectivity_status_response) + + async with aiohttp.ClientSession() as session: + client = PodPointClient(username="1233", password="1234", session=session, include_timestamp=True) + + connectivity_status_response = await client.async_get_connectivity_status(pod=Pod(data={"ppid": "PSL-123456"})) + + assert connectivity_status_response is not None + assert ConnectivityStatus == type(connectivity_status_response) + assert connectivity_status_response.ppid == "PSL-123456" + assert connectivity_status_response.connected_components == ["evse"] + assert len(connectivity_status_response.evses) == 1 + assert Evse == type(connectivity_status_response.evses[0]) + assert connectivity_status_response.evses[0].id == 1 + assert connectivity_status_response.evses[0].connectivity_state.protocol == "POW" + assert connectivity_status_response.evses[0].connectivity_state.connectivity_status == "ONLINE" + assert connectivity_status_response.evses[0].connectivity_state.signal_strength == -68 + assert connectivity_status_response.evses[0].connectivity_state.last_message_at == datetime( + year=2024, + month=4, + day=5, + hour=18, + minute=36, + second=29, + tzinfo=timezone.utc + ) + assert connectivity_status_response.evses[0].connectivity_state.connection_started_at == datetime( + year=2024, + month=4, + day=5, + hour=18, + minute=26, + second=26, + microsecond=819000, + tzinfo=timezone.utc + ) + assert connectivity_status_response.evses[0].connectivity_state.connection_quality == 3 + assert connectivity_status_response.evses[0].connectors[0].id == 1 + assert connectivity_status_response.evses[0].connectors[0].door == "A" + assert connectivity_status_response.evses[0].connectors[0].charging_state == "SUSPENDED_EV" + assert connectivity_status_response.evses[0].architecture == "arch3" + assert connectivity_status_response.evses[0].energy_offer_status.is_offering_energy is True + assert connectivity_status_response.evses[0].energy_offer_status.reason == "CHARGE_SCHEDULE" + assert connectivity_status_response.evses[0].energy_offer_status.until is None + assert connectivity_status_response.evses[0].energy_offer_status.random_delay is None + assert connectivity_status_response.evses[0].energy_offer_status.do_not_cache == False + + @pytest.mark.asyncio @freeze_time("Jan 1st, 2022") async def test_async_set_charge_mode_manual_with_expected_response(): diff --git a/tests/test_connectivity_status.py b/tests/test_connectivity_status.py new file mode 100644 index 0000000..69afbde --- /dev/null +++ b/tests/test_connectivity_status.py @@ -0,0 +1,86 @@ +import pytest +from podpointclient.connectivity_status import ConnectivityStatus, Evse +from datetime import datetime, timezone + + +@pytest.fixture +def connectivity_state_data(): + return { + "ppid": "PSL-266056", + "evses": [{ + "id": 1, + "connectivityState": { + "protocol": "POW", + "connectivityStatus": "ONLINE", + "signalStrength": -68, + "lastMessageAt": "2024-04-05T18:36:29Z", + "connectionStartedAt": "2024-04-05T18:26:26.819Z", + "connectionQuality": 3 + }, + "connectors": [{ + "id": 1, + "door": "A", + "chargingState": "SUSPENDED_EV" + }], + "architecture": "arch3", + "energyOfferStatus": { + "isOfferingEnergy": True, + "reason": "CHARGE_SCHEDULE", + "until": None, + "randomDelay": None, + "doNotCache": False + } + }], + "connectedComponents": ["evses"] + } + + +def test_connectivity_state_initialization(connectivity_state_data): + cs = ConnectivityStatus(connectivity_state_data) + assert cs.ppid == "PSL-266056" + assert cs.connected_components == ["evses"] + assert len(cs.evses) == 1 + + assert Evse == type(cs.evses[0]) + assert cs.evses[0].id == 1 + assert cs.evses[0].connectivity_state.protocol == "POW" + assert cs.evses[0].connectivity_state.connectivity_status == "ONLINE" + assert cs.evses[0].connectivity_state.signal_strength == -68 + assert cs.evses[0].connectivity_state.last_message_at == datetime( + year=2024, + month=4, + day=5, + hour=18, + minute=36, + second=29, + tzinfo=timezone.utc + ) + assert cs.evses[0].connectivity_state.connection_started_at == datetime( + year=2024, + month=4, + day=5, + hour=18, + minute=26, + second=26, + microsecond=819000, + tzinfo=timezone.utc + ) + assert cs.evses[0].connectivity_state.connection_quality == 3 + assert cs.evses[0].connectors[0].id == 1 + assert cs.evses[0].connectors[0].door == "A" + assert cs.evses[0].connectors[0].charging_state == "SUSPENDED_EV" + assert cs.evses[0].architecture == "arch3" + + print(cs.evses[0].energy_offer_status) + + assert cs.evses[0].energy_offer_status.is_offering_energy is True + assert cs.evses[0].energy_offer_status.reason == "CHARGE_SCHEDULE" + assert cs.evses[0].energy_offer_status.until is None + assert cs.evses[0].energy_offer_status.random_delay is None + assert cs.evses[0].energy_offer_status.do_not_cache is False + + +def test_connectivity_state_to_json(connectivity_state_data): + cs = ConnectivityStatus(connectivity_state_data) + json_data = cs.to_json() + assert isinstance(json_data, str) diff --git a/tests/test_connectivity_status_factory.py b/tests/test_connectivity_status_factory.py new file mode 100644 index 0000000..f5a5376 --- /dev/null +++ b/tests/test_connectivity_status_factory.py @@ -0,0 +1,16 @@ +from podpointclient.connectivity_status import ConnectivityStatus +from podpointclient.factories import ConnectivityStatusFactory +from helpers import Mocks + +def test_charge_factory_charge_array_creation(): + mocks = Mocks() + connectivity_status_response = mocks.connectivity_status_response() + + factory = ConnectivityStatusFactory() + connectivity_status = factory.build_connectivity_status(connectivity_status_response) + + assert ConnectivityStatus == type(connectivity_status) + +def test_charge_factory_with_none_passed(): + factory = ConnectivityStatusFactory() + assert factory.build_connectivity_status(None) == None