From a58016e994a8f43064f503cb2c608565d4cfac5d Mon Sep 17 00:00:00 2001 From: rikroe Date: Fri, 4 Aug 2023 20:14:07 +0200 Subject: [PATCH] Refactor tests, make remote service change state --- test/__init__.py | 13 +- test/common.py | 313 +++++++++++++++ test/conftest.py | 47 +++ test/test_account.py | 506 +++++++++---------------- test/test_api.py | 63 ++- test/test_deprecated_vehicle.py | 15 +- test/test_deprecated_vehicle_status.py | 75 ++-- test/test_remote_services.py | 313 +++++++-------- test/test_utils.py | 11 +- test/test_vehicle.py | 62 +-- test/test_vehicle_status.py | 106 +++--- 11 files changed, 867 insertions(+), 657 deletions(-) create mode 100644 test/common.py create mode 100644 test/conftest.py diff --git a/test/__init__.py b/test/__init__.py index 97e9f208..56e3653f 100644 --- a/test/__init__.py +++ b/test/__init__.py @@ -23,14 +23,21 @@ VIN_I01_REX = "WBY00000000REXI01" VIN_I20 = "WBA00000000DEMO01" -ALL_FINGERPRINTS: Dict[str, List[Dict]] = {brand.value: [] for brand in CarBrands} +ALL_VEHICLES: Dict[str, List[Dict]] = {brand.value: [] for brand in CarBrands} ALL_STATES: Dict[str, Dict] = {} ALL_CHARGING_SETTINGS: Dict[str, Dict] = {} +REMOTE_SERVICE_RESPONSE_INITIATED = RESPONSE_DIR / "remote_services" / "eadrax_service_initiated.json" +REMOTE_SERVICE_RESPONSE_PENDING = RESPONSE_DIR / "remote_services" / "eadrax_service_pending.json" +REMOTE_SERVICE_RESPONSE_DELIVERED = RESPONSE_DIR / "remote_services" / "eadrax_service_delivered.json" +REMOTE_SERVICE_RESPONSE_EXECUTED = RESPONSE_DIR / "remote_services" / "eadrax_service_executed.json" +REMOTE_SERVICE_RESPONSE_ERROR = RESPONSE_DIR / "remote_services" / "eadrax_service_error.json" +REMOTE_SERVICE_RESPONSE_EVENTPOSITION = RESPONSE_DIR / "remote_services" / "eadrax_service_eventposition.json" + def get_fingerprint_count() -> int: """Return number of loaded vehicles.""" - return sum([len(vehicles) for vehicles in ALL_FINGERPRINTS.values()]) + return sum([len(vehicles) for vehicles in ALL_VEHICLES.values()]) def load_response(path: Union[Path, str]) -> Any: @@ -44,7 +51,7 @@ def load_response(path: Union[Path, str]) -> Any: for fingerprint in RESPONSE_DIR.rglob("*-eadrax-vcs_v4_vehicles.json"): brand = fingerprint.stem.split("-")[0] for vehicle in load_response(fingerprint): - ALL_FINGERPRINTS[brand].append(vehicle) + ALL_VEHICLES[brand].append(vehicle) for state in RESPONSE_DIR.rglob("*-eadrax-vcs_v4_vehicles_state_*.json"): ALL_STATES[state.stem.split("_")[-1]] = load_response(state) diff --git a/test/common.py b/test/common.py new file mode 100644 index 00000000..468dc96e --- /dev/null +++ b/test/common.py @@ -0,0 +1,313 @@ +"""Fixtures for BMW tests.""" + +import json +from collections import defaultdict +from copy import deepcopy +from pathlib import Path +from typing import Dict, List, Optional +from uuid import uuid4 + +from bimmer_connected.vehicle.climate import ClimateActivityState +from bimmer_connected.vehicle.fuel_and_battery import ChargingState + +try: + from unittest import mock + + if not hasattr(mock, "AsyncMock"): + # AsyncMock was only introduced with Python3.8, so we have to use the backported module + raise ImportError() +except ImportError: + import mock # type: ignore[import,no-redef] # noqa: UP026 + +import httpx +import respx + +from bimmer_connected.const import Regions +from bimmer_connected.models import ChargingSettings +from bimmer_connected.vehicle.remote_services import MAP_CHARGING_MODE_TO_REMOTE_SERVICE + +from . import ( + ALL_VEHICLES, + REMOTE_SERVICE_RESPONSE_DELIVERED, + REMOTE_SERVICE_RESPONSE_EVENTPOSITION, + REMOTE_SERVICE_RESPONSE_EXECUTED, + REMOTE_SERVICE_RESPONSE_INITIATED, + REMOTE_SERVICE_RESPONSE_PENDING, + RESPONSE_DIR, + load_response, +) + +POI_DATA = { + "lat": 37.4028943, + "lon": -121.9700289, + "name": "49ers", + "street": "4949 Marie P DeBartolo Way", + "city": "Santa Clara", + "postal_code": "CA 95054", + "country": "United States", +} + +CHARGING_SETTINGS = {"target_soc": 75, "ac_limit": 16} + +STATUSREMOTE_SERVICE_RESPONSE_ORDER = [ + REMOTE_SERVICE_RESPONSE_PENDING, + REMOTE_SERVICE_RESPONSE_DELIVERED, + REMOTE_SERVICE_RESPONSE_EXECUTED, +] +STATUSREMOTE_SERVICE_RESPONSE_DICT: Dict[str, List[Path]] = defaultdict( + lambda: deepcopy(STATUSREMOTE_SERVICE_RESPONSE_ORDER) +) + +MAP_REMOTE_SERVICE_CHARGING_MODE_TO_STATE = {v: k.value for k, v in MAP_CHARGING_MODE_TO_REMOTE_SERVICE.items()} + +LOCAL_STATES: Dict[str, Dict] = {} +LOCAL_CHARGING_SETTINGS: Dict[str, Dict] = {} + + +class MyBMWMockRouter(respx.MockRouter): + """Stateful MockRouter for MyBMW APIs.""" + + def __init__( + self, + vehicles_to_load: Optional[List[str]] = None, + states: Optional[Dict[str, Dict]] = None, + charging_settings: Optional[Dict[str, Dict]] = None, + ) -> None: + """Initialize the MyBMWMockRouter with clean responses.""" + super().__init__(assert_all_called=False) + self.vehicles_to_load = vehicles_to_load or [] + self.states = deepcopy(states) if states else {} + self.charging_settings = deepcopy(charging_settings) if charging_settings else {} + + self.add_login_routes() + self.add_vehicle_routes() + self.add_remote_service_routes() + + # # # # # # # # # # # # # # # # # # # # # # # # + # Routes + # # # # # # # # # # # # # # # # # # # # # # # # + + def add_login_routes(self) -> None: + """Add routes for login.""" + + # Login to north_america and rest_of_world + self.get("/eadrax-ucs/v1/presentation/oauth/config").respond( + 200, json=load_response(RESPONSE_DIR / "auth" / "oauth_config.json") + ) + self.post("/gcdm/oauth/authenticate").mock(side_effect=self.authenticate_sideeffect) + self.post("/gcdm/oauth/token").respond(200, json=load_response(RESPONSE_DIR / "auth" / "auth_token.json")) + + # Login to china + self.get("/eadrax-coas/v1/cop/publickey").respond( + 200, json=load_response(RESPONSE_DIR / "auth" / "auth_cn_publickey.json") + ) + self.post("/eadrax-coas/v2/cop/slider-captcha").respond( + 200, json=load_response(RESPONSE_DIR / "auth" / "auth_slider_captcha.json") + ) + + self.post("/eadrax-coas/v1/cop/check-captcha").mock( + side_effect=[ + httpx.Response(422), + httpx.Response(201, json=load_response(RESPONSE_DIR / "auth" / "auth_slider_captcha_check.json")), + ] + ) + + self.post("/eadrax-coas/v2/login/pwd").respond( + 200, json=load_response(RESPONSE_DIR / "auth" / "auth_cn_login_pwd.json") + ) + self.post("/eadrax-coas/v2/oauth/token").respond( + 200, json=load_response(RESPONSE_DIR / "auth" / "auth_token.json") + ) + + def add_vehicle_routes(self) -> None: + """Add routes for vehicle requests.""" + + self.get("/eadrax-vcs/v4/vehicles").mock(side_effect=self.vehicles_sideeffect) + self.get("/eadrax-vcs/v4/vehicles/state").mock(side_effect=self.vehicle_state_sideeffect) + self.get("/eadrax-crccs/v2/vehicles").mock(side_effect=self.vehicle_charging_settings_sideeffect) + + def add_remote_service_routes(self) -> None: + """Add routes for remote services.""" + + self.post(path__regex=r"/eadrax-vrccs/v3/presentation/remote-commands/(?P.+)/(?P.+)$").mock( + side_effect=self.service_trigger_sideeffect + ) + self.post(path__regex=r"/eadrax-crccs/v1/vehicles/(?P.+)/(?P(start|stop)-charging)$").mock( + side_effect=self.service_trigger_sideeffect + ) + self.post(path__regex=r"/eadrax-crccs/v1/vehicles/(?P.+)/charging-settings$").mock( + side_effect=self.charging_settings_sideeffect + ) + self.post(path__regex=r"/eadrax-crccs/v1/vehicles/(?P.+)/charging-profile$").mock( + side_effect=self.charging_profile_sideeffect + ) + self.post("/eadrax-vrccs/v3/presentation/remote-commands/eventStatus", params={"eventId": mock.ANY}).mock( + side_effect=self.service_status_sideeffect + ) + + self.post("/eadrax-dcs/v1/send-to-car/send-to-car").mock(side_effect=self.poi_sideeffect) + self.post("/eadrax-vrccs/v3/presentation/remote-commands/eventPosition", params={"eventId": mock.ANY}).respond( + 200, + json=load_response(REMOTE_SERVICE_RESPONSE_EVENTPOSITION), + ) + + # # # # # # # # # # # # # # # # # # # # # # # # + # Authentication sideeffects + # # # # # # # # # # # # # # # # # # # # # # # # + + @staticmethod + def authenticate_sideeffect(request: httpx.Request) -> httpx.Response: + """Return /oauth/authentication response based on request.""" + request_text = request.read().decode("UTF-8") + if "username" in request_text and "password" in request_text and "grant_type" in request_text: + return httpx.Response(200, json=load_response(RESPONSE_DIR / "auth" / "authorization_response.json")) + return httpx.Response( + 302, + headers={ + "Location": "com.mini.connected://oauth?code=CODE&state=STATE&client_id=CLIENT_ID&nonce=login_nonce", + }, + ) + + # # # # # # # # # # # # # # # # # # # # # # # # + # Vehicle state sideeffects + # # # # # # # # # # # # # # # # # # # # # # # # + + def vehicles_sideeffect(self, request: httpx.Request) -> httpx.Response: + """Return /vehicles response based on x-user-agent.""" + x_user_agent = request.headers.get("x-user-agent", "").split(";") + if len(x_user_agent) == 4: + brand = x_user_agent[1] + else: + raise ValueError("x-user-agent not configured correctly!") + + # Test if given region is valid + _ = Regions(x_user_agent[3]) + + fingerprints = ALL_VEHICLES.get(brand, []) + if self.vehicles_to_load: + fingerprints = [f for f in fingerprints if f["vin"] in self.vehicles_to_load] + + return httpx.Response(200, json=fingerprints) + + def vehicle_state_sideeffect(self, request: httpx.Request) -> httpx.Response: + """Return /vehicles response based on x-user-agent.""" + x_user_agent = request.headers.get("x-user-agent", "").split(";") + assert len(x_user_agent) == 4 + + try: + return httpx.Response(200, json=self.states[request.headers["bmw-vin"]]) + except KeyError: + return httpx.Response(404) + + def vehicle_charging_settings_sideeffect(self, request: httpx.Request) -> httpx.Response: + """Return /vehicles response based on x-user-agent.""" + x_user_agent = request.headers.get("x-user-agent", "").split(";") + assert len(x_user_agent) == 4 + assert "fields" in request.url.params + assert "has_charging_settings_capabilities" in request.url.params + + try: + return httpx.Response(200, json=self.charging_settings[request.headers["bmw-vin"]]) + except KeyError: + return httpx.Response(404) + + # # # # # # # # # # # # # # # # # # # # # # # # + # Remote service sideeffects + # # # # # # # # # # # # # # # # # # # # # # # # + + def service_trigger_sideeffect( + self, request: httpx.Request, vin: str, service: Optional[str] = None + ) -> httpx.Response: + """Return specific eventId for each remote function.""" + + if service in ["door-lock", "door-unlock"]: + new_state = "LOCKED" if service == "door-lock" else "UNLOCKED" + self.states[vin]["state"]["doorsState"]["combinedSecurityState"] = new_state + + elif service in ["climate-now"]: + new_state = ( + ClimateActivityState.COOLING + if request.url.params["action"] == "START" + else ClimateActivityState.STANDBY + ) + self.states[vin]["state"]["climateControlState"]["activity"] = new_state + + elif service in ["start-charging", "stop-charging"]: + new_state = ChargingState.PLUGGED_IN if "stop" in service else ChargingState.CHARGING + self.states[vin]["state"]["electricChargingState"]["chargingStatus"] = new_state + + json_data = load_response(REMOTE_SERVICE_RESPONSE_INITIATED) + json_data["eventId"] = str(uuid4()) + + return httpx.Response(200, json=json_data) + + def charging_settings_sideeffect(self, request: httpx.Request, vin: str) -> httpx.Response: + """Check if payload is a valid charging settings payload and return evendId.""" + cs = ChargingSettings(**json.loads(request.content)) + self.states[vin]["state"]["electricChargingState"]["chargingTarget"] = cs.chargingTarget + self.states[vin]["state"]["chargingProfile"]["chargingSettings"]["acCurrentLimit"] = cs.acLimitValue + + return self.service_trigger_sideeffect(request, vin) + + def charging_profile_sideeffect(self, request: httpx.Request, vin: str) -> httpx.Response: + """Check if payload is a valid charging settings payload and return evendId.""" + + data = json.loads(request.content) + + if {"chargingMode", "departureTimer", "isPreconditionForDepartureActive", "servicePack"} != set(data): + return httpx.Response(500) + if ( + data["chargingMode"]["chargingPreference"] == "NO_PRESELECTION" + and data["chargingMode"]["type"] != "CHARGING_IMMEDIATELY" + ): + return httpx.Response(500) + if ( + data["chargingMode"]["chargingPreference"] == "CHARGING_WINDOW" + and data["chargingMode"]["type"] != "TIME_SLOT" + ): + return httpx.Response(500) + + if not isinstance(data["isPreconditionForDepartureActive"], bool): + return httpx.Response(500) + + # separate charging profile endpoint + self.charging_settings[vin]["chargeAndClimateTimerDetail"]["chargingMode"]["chargingPreference"] = data[ + "chargingMode" + ]["chargingPreference"] + self.charging_settings[vin]["chargeAndClimateTimerDetail"]["chargingMode"]["type"] = data["chargingMode"][ + "type" + ] + self.charging_settings[vin]["chargeAndClimateTimerDetail"]["isPreconditionForDepartureActive"] = data[ + "isPreconditionForDepartureActive" + ] + + # state endpoint + self.states[vin]["state"]["chargingProfile"]["chargingPreference"] = data["chargingMode"]["chargingPreference"] + self.states[vin]["state"]["chargingProfile"]["chargingMode"] = MAP_REMOTE_SERVICE_CHARGING_MODE_TO_STATE[ + data["chargingMode"]["type"] + ] + self.states[vin]["state"]["chargingProfile"]["climatisationOn"] = data["isPreconditionForDepartureActive"] + + return self.service_trigger_sideeffect(request, vin) + + @staticmethod + def service_status_sideeffect(request: httpx.Request) -> httpx.Response: + """Return all 3 eventStatus responses per function.""" + response_data = STATUSREMOTE_SERVICE_RESPONSE_DICT[request.url.params["eventId"]].pop(0) + return httpx.Response(200, json=load_response(response_data)) + + @staticmethod + def poi_sideeffect(request: httpx.Request) -> httpx.Response: + """Check if payload is a valid POI.""" + data = json.loads(request.content) + tests = all( + [ + len(data["vin"]) == 17, + isinstance(data["location"]["coordinates"]["latitude"], float), + isinstance(data["location"]["coordinates"]["longitude"], float), + len(data["location"]["name"]) > 0, + ] + ) + if not tests: + return httpx.Response(400) + return httpx.Response(201) diff --git a/test/conftest.py b/test/conftest.py new file mode 100644 index 00000000..9e3d86f2 --- /dev/null +++ b/test/conftest.py @@ -0,0 +1,47 @@ +"""Fixtures for BMW tests.""" +from typing import Optional + +try: + from unittest import mock + + if not hasattr(mock, "AsyncMock"): + # AsyncMock was only introduced with Python3.8, so we have to use the backported module + raise ImportError() +except ImportError: + import mock # type: ignore[import,no-redef] # noqa: UP026 + +import pytest +import respx + +from bimmer_connected.account import MyBMWAccount +from bimmer_connected.const import Regions + +from . import ( + ALL_CHARGING_SETTINGS, + ALL_STATES, + TEST_PASSWORD, + TEST_REGION, + TEST_USERNAME, +) +from .common import MyBMWMockRouter + + +@pytest.fixture +def bmw_fixture(request: pytest.FixtureRequest) -> respx.MockRouter: + """Patch MyBMW login API calls.""" + # Now we can start patching the API calls + router = MyBMWMockRouter( + vehicles_to_load=getattr(request, "param", []), + states=ALL_STATES, + charging_settings=ALL_CHARGING_SETTINGS, + ) + + with router: + yield router + + +async def prepare_account_with_vehicles(region: Optional[Regions] = None, metric: bool = True): + """Initialize account and get vehicles.""" + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, region or TEST_REGION, use_metric_units=metric) + await account.get_vehicles() + return account diff --git a/test/test_account.py b/test/test_account.py index 0ddf9c52..96dd0205 100644 --- a/test/test_account.py +++ b/test/test_account.py @@ -3,7 +3,6 @@ import datetime import logging from pathlib import Path -from typing import Optional try: from unittest import mock @@ -22,13 +21,10 @@ from bimmer_connected.api.authentication import MyBMWAuthentication, MyBMWLoginRetry from bimmer_connected.api.client import MyBMWClient from bimmer_connected.api.regions import get_region_from_name -from bimmer_connected.const import ATTR_CAPABILITIES, Regions +from bimmer_connected.const import ATTR_CAPABILITIES from bimmer_connected.models import GPSPosition, MyBMWAPIError, MyBMWAuthError, MyBMWQuotaError from . import ( - ALL_CHARGING_SETTINGS, - ALL_FINGERPRINTS, - ALL_STATES, RESPONSE_DIR, TEST_PASSWORD, TEST_REGION, @@ -39,125 +35,19 @@ get_fingerprint_count, load_response, ) +from .conftest import prepare_account_with_vehicles -def authenticate_sideeffect(request: httpx.Request) -> httpx.Response: - """Return /oauth/authentication response based on request.""" - request_text = request.read().decode("UTF-8") - if "username" in request_text and "password" in request_text and "grant_type" in request_text: - return httpx.Response(200, json=load_response(RESPONSE_DIR / "auth" / "authorization_response.json")) - return httpx.Response( - 302, - headers={ - "Location": "com.mini.connected://oauth?code=CODE&state=STATE&client_id=CLIENT_ID&nonce=login_nonce", - }, - ) - - -def vehicles_sideeffect(request: httpx.Request) -> httpx.Response: - """Return /vehicles response based on x-user-agent.""" - x_user_agent = request.headers.get("x-user-agent", "").split(";") - if len(x_user_agent) == 4: - brand = x_user_agent[1] - else: - raise ValueError("x-user-agent not configured correctly!") - - # Test if given region is valid - _ = Regions(x_user_agent[3]) - - return httpx.Response(200, json=ALL_FINGERPRINTS.get(brand, [])) - - -def vehicle_state_sideeffect(request: httpx.Request) -> httpx.Response: - """Return /vehicles response based on x-user-agent.""" - x_user_agent = request.headers.get("x-user-agent", "").split(";") - assert len(x_user_agent) == 4 - - try: - return httpx.Response(200, json=ALL_STATES[request.headers["bmw-vin"]]) - except KeyError: - return httpx.Response(404) - - -def vehicle_charging_settings_sideeffect(request: httpx.Request) -> httpx.Response: - """Return /vehicles response based on x-user-agent.""" - x_user_agent = request.headers.get("x-user-agent", "").split(";") - assert len(x_user_agent) == 4 - assert "fields" in request.url.params - assert "has_charging_settings_capabilities" in request.url.params - - try: - return httpx.Response(200, json=ALL_CHARGING_SETTINGS[request.headers["bmw-vin"]]) - except KeyError: - return httpx.Response(404) - - -def account_mock(): - """Return mocked adapter for auth.""" - router = respx.mock(assert_all_called=False) - - # Login to north_america and rest_of_world - router.get("/eadrax-ucs/v1/presentation/oauth/config").respond( - 200, json=load_response(RESPONSE_DIR / "auth" / "oauth_config.json") - ) - router.post("/gcdm/oauth/authenticate").mock(side_effect=authenticate_sideeffect) - router.post("/gcdm/oauth/token").respond(200, json=load_response(RESPONSE_DIR / "auth" / "auth_token.json")) - - # Login to china - router.get("/eadrax-coas/v1/cop/publickey").respond( - 200, json=load_response(RESPONSE_DIR / "auth" / "auth_cn_publickey.json") - ) - router.post("/eadrax-coas/v2/cop/slider-captcha").respond( - 200, json=load_response(RESPONSE_DIR / "auth" / "auth_slider_captcha.json") - ) - - router.post("/eadrax-coas/v1/cop/check-captcha").mock( - side_effect=[ - httpx.Response(422), - httpx.Response(201, json=load_response(RESPONSE_DIR / "auth" / "auth_slider_captcha_check.json")), - ] - ) - - router.post("/eadrax-coas/v2/login/pwd").respond( - 200, json=load_response(RESPONSE_DIR / "auth" / "auth_cn_login_pwd.json") - ) - router.post("/eadrax-coas/v2/oauth/token").respond( - 200, json=load_response(RESPONSE_DIR / "auth" / "auth_token.json") - ) - - # Get all vehicle fingerprints - router.get("/eadrax-vcs/v4/vehicles").mock(side_effect=vehicles_sideeffect) - router.get("/eadrax-vcs/v4/vehicles/state").mock(side_effect=vehicle_state_sideeffect) - router.get("/eadrax-crccs/v2/vehicles").mock(side_effect=vehicle_charging_settings_sideeffect) - - return router - - -def get_account(region: Optional[Regions] = None, metric: bool = True): - """Return account without token and vehicles (sync).""" - return MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, region or TEST_REGION, use_metric_units=metric) - - -async def get_mocked_account(region: Optional[Regions] = None, metric: bool = True): - """Return pre-mocked account.""" - with account_mock(): - account = get_account(region, metric) - await account.get_vehicles() - return account - - -@account_mock() @pytest.mark.asyncio -async def test_login_row_na(): +async def test_login_row_na(bmw_fixture: respx.Router): """Test the login flow.""" account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name(TEST_REGION_STRING)) await account.get_vehicles() assert account is not None -@account_mock() @pytest.mark.asyncio -async def test_login_refresh_token_row_na_expired(): +async def test_login_refresh_token_row_na_expired(bmw_fixture: respx.Router): """Test the login flow using refresh_token.""" with mock.patch("bimmer_connected.api.authentication.EXPIRES_AT_OFFSET", datetime.timedelta(seconds=30000)): account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name(TEST_REGION_STRING)) @@ -176,61 +66,58 @@ async def test_login_refresh_token_row_na_expired(): @pytest.mark.asyncio -async def test_login_refresh_token_row_na_401(): +async def test_login_refresh_token_row_na_401(bmw_fixture: respx.Router): """Test the login flow using refresh_token.""" - with account_mock() as mock_api: - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name(TEST_REGION_STRING)) - await account.get_vehicles() - with mock.patch( - "bimmer_connected.api.authentication.MyBMWAuthentication._refresh_token_row_na", - wraps=account.config.authentication._refresh_token_row_na, - ) as mock_listener: - mock_api.get("/eadrax-vcs/v4/vehicles/state").mock( - side_effect=[httpx.Response(401), *([httpx.Response(200, json={ATTR_CAPABILITIES: {}})] * 10)] - ) - mock_listener.reset_mock() - await account.get_vehicles() + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name(TEST_REGION_STRING)) + await account.get_vehicles() - assert mock_listener.call_count == 1 - assert account.config.authentication.refresh_token is not None + with mock.patch( + "bimmer_connected.api.authentication.MyBMWAuthentication._refresh_token_row_na", + wraps=account.config.authentication._refresh_token_row_na, + ) as mock_listener: + bmw_fixture.get("/eadrax-vcs/v4/vehicles/state").mock( + side_effect=[httpx.Response(401), *([httpx.Response(200, json={ATTR_CAPABILITIES: {}})] * 10)] + ) + mock_listener.reset_mock() + await account.get_vehicles() + + assert mock_listener.call_count == 1 + assert account.config.authentication.refresh_token is not None @pytest.mark.asyncio -async def test_login_refresh_token_row_na_invalid(caplog): +async def test_login_refresh_token_row_na_invalid(caplog, bmw_fixture: respx.Router): """Test the login flow using refresh_token.""" - with account_mock() as mock_api: - mock_api.post("/gcdm/oauth/token").mock( - side_effect=[ - httpx.Response(400), - httpx.Response(200, json=load_response(RESPONSE_DIR / "auth" / "auth_token.json")), - ] - ) + bmw_fixture.post("/gcdm/oauth/token").mock( + side_effect=[ + httpx.Response(400), + httpx.Response(200, json=load_response(RESPONSE_DIR / "auth" / "auth_token.json")), + ] + ) - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name(TEST_REGION_STRING)) - account.set_refresh_token("INVALID") + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name(TEST_REGION_STRING)) + account.set_refresh_token("INVALID") - caplog.set_level(logging.DEBUG) - await account.get_vehicles() + caplog.set_level(logging.DEBUG) + await account.get_vehicles() - debug_messages = [r.message for r in caplog.records if r.name.startswith("bimmer_connected")] - assert "Authenticating with refresh token for North America & Rest of World." in debug_messages - assert "Unable to get access token using refresh token, falling back to username/password." in debug_messages - assert "Authenticating with MyBMW flow for North America & Rest of World." in debug_messages + debug_messages = [r.message for r in caplog.records if r.name.startswith("bimmer_connected")] + assert "Authenticating with refresh token for North America & Rest of World." in debug_messages + assert "Unable to get access token using refresh token, falling back to username/password." in debug_messages + assert "Authenticating with MyBMW flow for North America & Rest of World." in debug_messages -@account_mock() @pytest.mark.asyncio -async def test_login_china(): +async def test_login_china(bmw_fixture: respx.Router): """Test the login flow for region `china`.""" account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name("china")) await account.get_vehicles() assert account is not None -@account_mock() @pytest.mark.asyncio -async def test_login_refresh_token_china_expired(): +async def test_login_refresh_token_china_expired(bmw_fixture: respx.Router): """Test the login flow using refresh_token for region `china`.""" with mock.patch("bimmer_connected.api.authentication.EXPIRES_AT_OFFSET", datetime.timedelta(seconds=30000)): account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name("china")) @@ -249,52 +136,49 @@ async def test_login_refresh_token_china_expired(): @pytest.mark.asyncio -async def test_login_refresh_token_china_401(): +async def test_login_refresh_token_china_401(bmw_fixture: respx.Router): """Test the login flow using refresh_token.""" - with account_mock() as mock_api: - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name("china")) - await account.get_vehicles() + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name("china")) + await account.get_vehicles() - with mock.patch( - "bimmer_connected.api.authentication.MyBMWAuthentication._refresh_token_china", - wraps=account.config.authentication._refresh_token_china, - ) as mock_listener: - mock_api.get("/eadrax-vcs/v4/vehicles/state").mock( - side_effect=[httpx.Response(401), *([httpx.Response(200, json={ATTR_CAPABILITIES: {}})] * 10)] - ) - mock_listener.reset_mock() - await account.get_vehicles() + with mock.patch( + "bimmer_connected.api.authentication.MyBMWAuthentication._refresh_token_china", + wraps=account.config.authentication._refresh_token_china, + ) as mock_listener: + bmw_fixture.get("/eadrax-vcs/v4/vehicles/state").mock( + side_effect=[httpx.Response(401), *([httpx.Response(200, json={ATTR_CAPABILITIES: {}})] * 10)] + ) + mock_listener.reset_mock() + await account.get_vehicles() - assert mock_listener.call_count == 1 - assert account.config.authentication.refresh_token is not None + assert mock_listener.call_count == 1 + assert account.config.authentication.refresh_token is not None @pytest.mark.asyncio -async def test_login_refresh_token_china_invalid(caplog): +async def test_login_refresh_token_china_invalid(caplog, bmw_fixture: respx.Router): """Test the login flow using refresh_token.""" - with account_mock() as mock_api: - mock_api.post("/eadrax-coas/v2/oauth/token").mock( - side_effect=[ - httpx.Response(400), - httpx.Response(200, json=load_response(RESPONSE_DIR / "auth" / "auth_token.json")), - ] - ) + bmw_fixture.post("/eadrax-coas/v2/oauth/token").mock( + side_effect=[ + httpx.Response(400), + httpx.Response(200, json=load_response(RESPONSE_DIR / "auth" / "auth_token.json")), + ] + ) - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name("china")) - account.set_refresh_token("INVALID") + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name("china")) + account.set_refresh_token("INVALID") - caplog.set_level(logging.DEBUG) - await account.get_vehicles() + caplog.set_level(logging.DEBUG) + await account.get_vehicles() - debug_messages = [r.message for r in caplog.records if r.name.startswith("bimmer_connected")] - assert "Authenticating with refresh token for China." in debug_messages - assert "Unable to get access token using refresh token, falling back to username/password." in debug_messages - assert "Authenticating with MyBMW flow for China." in debug_messages + debug_messages = [r.message for r in caplog.records if r.name.startswith("bimmer_connected")] + assert "Authenticating with refresh token for China." in debug_messages + assert "Unable to get access token using refresh token, falling back to username/password." in debug_messages + assert "Authenticating with MyBMW flow for China." in debug_messages -@account_mock() @pytest.mark.asyncio -async def test_vehicles(): +async def test_vehicles(bmw_fixture: respx.Router): """Test the login flow.""" account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name("china")) await account.get_vehicles() @@ -303,14 +187,14 @@ async def test_vehicles(): assert get_fingerprint_count() == len(account.vehicles) vehicle = account.get_vehicle(VIN_G26) + assert vehicle is not None assert VIN_G26 == vehicle.vin assert account.get_vehicle("invalid_vin") is None -@account_mock() @pytest.mark.asyncio -async def test_vehicle_init(): +async def test_vehicle_init(bmw_fixture: respx.Router): """Test vehicle initialization.""" account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) with mock.patch( @@ -335,45 +219,42 @@ async def test_vehicle_init(): @pytest.mark.asyncio -async def test_invalid_password(): +async def test_invalid_password(bmw_fixture: respx.Router): """Test parsing the results of an invalid password.""" - with account_mock() as mock_api: - mock_api.post("/gcdm/oauth/authenticate").respond( - 401, json=load_response(RESPONSE_DIR / "auth" / "auth_error_wrong_password.json") - ) - with pytest.raises(MyBMWAuthError): - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) - await account.get_vehicles() + bmw_fixture.post("/gcdm/oauth/authenticate").respond( + 401, json=load_response(RESPONSE_DIR / "auth" / "auth_error_wrong_password.json") + ) + with pytest.raises(MyBMWAuthError): + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) + await account.get_vehicles() @pytest.mark.asyncio -async def test_invalid_password_china(): +async def test_invalid_password_china(bmw_fixture: respx.Router): """Test parsing the results of an invalid password.""" - with account_mock() as mock_api: - mock_api.post("/eadrax-coas/v2/login/pwd").respond( - 422, json=load_response(RESPONSE_DIR / "auth" / "auth_cn_login_error.json") - ) - with pytest.raises(MyBMWAPIError): - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name("china")) - await account.get_vehicles() + bmw_fixture.post("/eadrax-coas/v2/login/pwd").respond( + 422, json=load_response(RESPONSE_DIR / "auth" / "auth_cn_login_error.json") + ) + with pytest.raises(MyBMWAPIError): + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, get_region_from_name("china")) + await account.get_vehicles() @pytest.mark.asyncio -async def test_server_error(): +async def test_server_error(bmw_fixture: respx.Router): """Test parsing the results of a server error.""" - with account_mock() as mock_api: - mock_api.post("/gcdm/oauth/authenticate").respond( - 500, text=load_response(RESPONSE_DIR / "auth" / "auth_error_internal_error.txt") - ) - with pytest.raises(MyBMWAPIError): - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) - await account.get_vehicles() + bmw_fixture.post("/gcdm/oauth/authenticate").respond( + 500, text=load_response(RESPONSE_DIR / "auth" / "auth_error_internal_error.txt") + ) + with pytest.raises(MyBMWAPIError): + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) + await account.get_vehicles() @pytest.mark.asyncio -async def test_vehicle_search_case(): +async def test_vehicle_search_case(bmw_fixture: respx.Router): """Check if the search for the vehicle by VIN is NOT case sensitive.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vin = account.vehicles[1].vin assert vin == account.get_vehicle(vin).vin @@ -382,17 +263,16 @@ async def test_vehicle_search_case(): @pytest.mark.asyncio -async def test_get_fingerprints(): +async def test_get_fingerprints(bmw_fixture: respx.Router): """Test getting fingerprints.""" - with account_mock() as mock_api: - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION, log_responses=True) - await account.get_vehicles() + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION, log_responses=True) + await account.get_vehicles() - mock_api.get("/eadrax-vcs/v4/vehicles/state").respond( - 500, text=load_response(RESPONSE_DIR / "auth" / "auth_error_internal_error.txt") - ) - with pytest.raises(MyBMWAPIError): - await account.get_vehicles() + bmw_fixture.get("/eadrax-vcs/v4/vehicles/state").respond( + 500, text=load_response(RESPONSE_DIR / "auth" / "auth_error_internal_error.txt") + ) + with pytest.raises(MyBMWAPIError): + await account.get_vehicles() filenames = [Path(f.filename) for f in account.get_stored_responses()] json_files = [f for f in filenames if f.suffix == ".json"] @@ -403,9 +283,9 @@ async def test_get_fingerprints(): @pytest.mark.asyncio -async def test_set_observer_value(): +async def test_set_observer_value(bmw_fixture: respx.Router): """Test set_observer_position with valid arguments.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() account.set_observer_position(1.0, 2.0) @@ -413,9 +293,9 @@ async def test_set_observer_value(): @pytest.mark.asyncio -async def test_set_observer_not_set(): +async def test_set_observer_not_set(bmw_fixture: respx.Router): """Test set_observer_position with no arguments.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() assert account.config.observer_position is None @@ -425,9 +305,9 @@ async def test_set_observer_not_set(): @pytest.mark.asyncio -async def test_set_observer_invalid_values(): +async def test_set_observer_invalid_values(bmw_fixture: respx.Router): """Test set_observer_position with invalid arguments.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() with pytest.raises(TypeError): account.set_observer_position(None, 2.0) @@ -459,9 +339,8 @@ async def test_set_use_metric_units(): assert imperial_client.generate_default_header()["bmw-units-preferences"] == "d=MI;v=G" -@account_mock() @pytest.mark.asyncio -async def test_deprecated_account(caplog): +async def test_deprecated_account(caplog, bmw_fixture: respx.Router): """Test deprecation warning for ConnectedDriveAccount.""" account = ConnectedDriveAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) await account.get_vehicles() @@ -470,9 +349,8 @@ async def test_deprecated_account(caplog): assert 1 == len(get_deprecation_warning_count(caplog)) -@account_mock() @pytest.mark.asyncio -async def test_refresh_token_getset(): +async def test_refresh_token_getset(bmw_fixture: respx.Router): """Test getting/setting the refresh_token and gcid.""" account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) assert account.refresh_token is None @@ -494,134 +372,128 @@ async def test_refresh_token_getset(): @pytest.mark.asyncio -async def test_429_retry_ok_login(caplog): +async def test_429_retry_ok_login(caplog, bmw_fixture: respx.Router): """Test the login flow using refresh_token.""" - with account_mock() as mock_api: - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) - json_429 = {"statusCode": 429, "message": "Rate limit is exceeded. Try again in 2 seconds."} + json_429 = {"statusCode": 429, "message": "Rate limit is exceeded. Try again in 2 seconds."} - mock_api.get("/eadrax-ucs/v1/presentation/oauth/config").mock( - side_effect=[ - httpx.Response(429, json=json_429), - httpx.Response(429, json=json_429), - httpx.Response(200, json=load_response(RESPONSE_DIR / "auth" / "oauth_config.json")), - ] - ) - caplog.set_level(logging.DEBUG) + bmw_fixture.get("/eadrax-ucs/v1/presentation/oauth/config").mock( + side_effect=[ + httpx.Response(429, json=json_429), + httpx.Response(429, json=json_429), + httpx.Response(200, json=load_response(RESPONSE_DIR / "auth" / "oauth_config.json")), + ] + ) + caplog.set_level(logging.DEBUG) - with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock): - await account.get_vehicles() + with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock): + await account.get_vehicles() - log_429 = [ - r - for r in caplog.records - if r.module == "authentication" and "seconds due to 429 Too Many Requests" in r.message - ] - assert len(log_429) == 2 + log_429 = [ + r + for r in caplog.records + if r.module == "authentication" and "seconds due to 429 Too Many Requests" in r.message + ] + assert len(log_429) == 2 @pytest.mark.asyncio -async def test_429_retry_raise_login(caplog): +async def test_429_retry_raise_login(caplog, bmw_fixture: respx.Router): """Test the login flow using refresh_token.""" - with account_mock() as mock_api: - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) - json_429 = {"statusCode": 429, "message": "Rate limit is exceeded. Try again in 2 seconds."} + json_429 = {"statusCode": 429, "message": "Rate limit is exceeded. Try again in 2 seconds."} - mock_api.get("/eadrax-ucs/v1/presentation/oauth/config").mock(return_value=httpx.Response(429, json=json_429)) - caplog.set_level(logging.DEBUG) + bmw_fixture.get("/eadrax-ucs/v1/presentation/oauth/config").mock(return_value=httpx.Response(429, json=json_429)) + caplog.set_level(logging.DEBUG) - with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock): - with pytest.raises(MyBMWAPIError): - await account.get_vehicles() + with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock): + with pytest.raises(MyBMWAPIError): + await account.get_vehicles() - log_429 = [ - r - for r in caplog.records - if r.module == "authentication" and "seconds due to 429 Too Many Requests" in r.message - ] - assert len(log_429) == 3 + log_429 = [ + r + for r in caplog.records + if r.module == "authentication" and "seconds due to 429 Too Many Requests" in r.message + ] + assert len(log_429) == 3 @pytest.mark.asyncio -async def test_429_retry_ok_vehicles(caplog): +async def test_429_retry_ok_vehicles(caplog, bmw_fixture: respx.Router): """Test waiting on 429 for vehicles.""" - with account_mock() as mock_api: - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) - json_429 = {"statusCode": 429, "message": "Rate limit is exceeded. Try again in 2 seconds."} + json_429 = {"statusCode": 429, "message": "Rate limit is exceeded. Try again in 2 seconds."} - mock_api.get("/eadrax-vcs/v4/vehicles").mock( - side_effect=[ - httpx.Response(429, json=json_429), - httpx.Response(429, json=json_429), - *[httpx.Response(200, json=[])] * 2, - ] - ) - caplog.set_level(logging.DEBUG) + bmw_fixture.get("/eadrax-vcs/v4/vehicles").mock( + side_effect=[ + httpx.Response(429, json=json_429), + httpx.Response(429, json=json_429), + *[httpx.Response(200, json=[])] * 2, + ] + ) + caplog.set_level(logging.DEBUG) - with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock): - await account.get_vehicles() + with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock): + await account.get_vehicles() - log_429 = [ - r - for r in caplog.records - if r.module == "authentication" and "seconds due to 429 Too Many Requests" in r.message - ] - assert len(log_429) == 2 + log_429 = [ + r + for r in caplog.records + if r.module == "authentication" and "seconds due to 429 Too Many Requests" in r.message + ] + assert len(log_429) == 2 @pytest.mark.asyncio -async def test_429_retry_raise_vehicles(caplog): +async def test_429_retry_raise_vehicles(caplog, bmw_fixture: respx.Router): """Test waiting on 429 for vehicles and fail if it happens too often.""" - with account_mock() as mock_api: - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) - json_429 = {"statusCode": 429, "message": "Rate limit is exceeded. Try again in 2 seconds."} + json_429 = {"statusCode": 429, "message": "Rate limit is exceeded. Try again in 2 seconds."} - mock_api.get("/eadrax-vcs/v4/vehicles").mock(return_value=httpx.Response(429, json=json_429)) - caplog.set_level(logging.DEBUG) + bmw_fixture.get("/eadrax-vcs/v4/vehicles").mock(return_value=httpx.Response(429, json=json_429)) + caplog.set_level(logging.DEBUG) - with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock): - with pytest.raises(MyBMWQuotaError): - await account.get_vehicles() + with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock): + with pytest.raises(MyBMWQuotaError): + await account.get_vehicles() - log_429 = [ - r - for r in caplog.records - if r.module == "authentication" and "seconds due to 429 Too Many Requests" in r.message - ] - assert len(log_429) == 3 + log_429 = [ + r + for r in caplog.records + if r.module == "authentication" and "seconds due to 429 Too Many Requests" in r.message + ] + assert len(log_429) == 3 @pytest.mark.asyncio -async def test_403_quota_exceeded_vehicles_usa(caplog): +async def test_403_quota_exceeded_vehicles_usa(caplog, bmw_fixture: respx.Router): """Test 403 quota issues for vehicle state and fail if it happens too often.""" - with account_mock() as mock_api: - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) - # get vehicles once - await account.get_vehicles() + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION) + # get vehicles once + await account.get_vehicles() - mock_api.get("/eadrax-vcs/v4/vehicles/state").mock( - return_value=httpx.Response( - 403, - json={"statusCode": 403, "message": "Out of call volume quota. Quota will be replenished in 02:12:20."}, - ) + bmw_fixture.get("/eadrax-vcs/v4/vehicles/state").mock( + return_value=httpx.Response( + 403, + json={"statusCode": 403, "message": "Out of call volume quota. Quota will be replenished in 02:12:20."}, ) - caplog.set_level(logging.DEBUG) + ) + caplog.set_level(logging.DEBUG) - with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock): - with pytest.raises(MyBMWQuotaError): - await account.get_vehicles() + with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock): + with pytest.raises(MyBMWQuotaError): + await account.get_vehicles() - log_quota = [r for r in caplog.records if "quota" in r.message] - assert len(log_quota) == 1 + log_quota = [r for r in caplog.records if "quota" in r.message] + assert len(log_quota) == 1 -@account_mock() @pytest.mark.asyncio -async def test_client_async_only(): +async def test_client_async_only(bmw_fixture: respx.Router): """Test that the Authentication providers only work async.""" with httpx.Client(auth=MyBMWAuthentication(TEST_USERNAME, TEST_PASSWORD, TEST_REGION)) as client: diff --git a/test/test_api.py b/test/test_api.py index 6411383f..dd1967b0 100644 --- a/test/test_api.py +++ b/test/test_api.py @@ -2,6 +2,7 @@ import json import pytest +import respx from bimmer_connected.account import MyBMWAccount from bimmer_connected.api.regions import get_region_from_name, valid_regions @@ -10,7 +11,6 @@ from bimmer_connected.utils import log_response_store_to_file from . import RESPONSE_DIR, TEST_PASSWORD, TEST_REGION, TEST_USERNAME, get_fingerprint_count, load_response -from .test_account import account_mock def test_valid_regions(): @@ -54,17 +54,16 @@ def test_anonymize_data(): @pytest.mark.asyncio -async def test_storing_fingerprints(tmp_path): +async def test_storing_fingerprints(tmp_path, bmw_fixture: respx.Router): """Test storing fingerprints to file.""" - with account_mock() as mock_api: - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION, log_responses=True) - await account.get_vehicles() + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION, log_responses=True) + await account.get_vehicles() - mock_api.get("/eadrax-vcs/v4/vehicles/state").respond( - 500, text=load_response(RESPONSE_DIR / "auth" / "auth_error_internal_error.txt") - ) - with pytest.raises(MyBMWAPIError): - await account.get_vehicles() + bmw_fixture.get("/eadrax-vcs/v4/vehicles/state").respond( + 500, text=load_response(RESPONSE_DIR / "auth" / "auth_error_internal_error.txt") + ) + with pytest.raises(MyBMWAPIError): + await account.get_vehicles() log_response_store_to_file(account.get_stored_responses(), tmp_path) @@ -77,26 +76,26 @@ async def test_storing_fingerprints(tmp_path): @pytest.mark.asyncio -async def test_fingerprint_deque(): +async def test_fingerprint_deque(bmw_fixture: respx.Router): """Test storing fingerprints to file.""" - with account_mock() as mock_api: - account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION, log_responses=True) - await account.get_vehicles() - await account.get_vehicles() - - # More than 10 calls were made, but only last 10 are stored - assert len([c for c in mock_api.calls if c.request.url.path.startswith("/eadrax-vcs")]) > 10 - assert len(account.get_stored_responses()) == 10 - - # Stored responses are reset - account.config.set_log_responses(False) - assert len(account.get_stored_responses()) == 0 - - # No new responses are getting added - await account.get_vehicles() - assert len(account.get_stored_responses()) == 0 - - # Get responses again - account.config.set_log_responses(True) - await account.get_vehicles() - assert len(account.get_stored_responses()) == 10 + # with bmw_fixture as mock_api: + account = MyBMWAccount(TEST_USERNAME, TEST_PASSWORD, TEST_REGION, log_responses=True) + await account.get_vehicles() + await account.get_vehicles() + + # More than 10 calls were made, but only last 10 are stored + assert len([c for c in bmw_fixture.calls if c.request.url.path.startswith("/eadrax-vcs")]) > 10 + assert len(account.get_stored_responses()) == 10 + + # Stored responses are reset + account.config.set_log_responses(False) + assert len(account.get_stored_responses()) == 0 + + # No new responses are getting added + await account.get_vehicles() + assert len(account.get_stored_responses()) == 0 + + # Get responses again + account.config.set_log_responses(True) + await account.get_vehicles() + assert len(account.get_stored_responses()) == 10 diff --git a/test/test_deprecated_vehicle.py b/test/test_deprecated_vehicle.py index 879e789f..3bfa5255 100644 --- a/test/test_deprecated_vehicle.py +++ b/test/test_deprecated_vehicle.py @@ -1,5 +1,6 @@ """Tests for deprecated MyBMWVehicle.""" import pytest +import respx from bimmer_connected.const import CarBrands from bimmer_connected.vehicle.vehicle import ConnectedDriveVehicle @@ -15,7 +16,7 @@ VIN_I20, get_deprecation_warning_count, ) -from .test_account import get_mocked_account +from .conftest import prepare_account_with_vehicles ATTRIBUTE_MAPPING = { "remainingFuel": "remaining_fuel", @@ -36,9 +37,9 @@ @pytest.mark.asyncio -async def test_parsing_attributes(caplog): +async def test_parsing_attributes(caplog, bmw_fixture: respx.router): """Test parsing different attributes of the vehicle.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() for vehicle in account.vehicles: print(vehicle.name) @@ -54,9 +55,9 @@ async def test_parsing_attributes(caplog): @pytest.mark.asyncio -async def test_drive_train_attributes(caplog): +async def test_drive_train_attributes(caplog, bmw_fixture: respx.router): """Test parsing different attributes of the vehicle.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle_drivetrains = { VIN_F31: (True, False, False), @@ -78,9 +79,9 @@ async def test_drive_train_attributes(caplog): @pytest.mark.asyncio -async def test_deprecated_vehicle(caplog): +async def test_deprecated_vehicle(caplog, bmw_fixture: respx.router): """Test deprecation warning for ConnectedDriveVehicle.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() deprecated_vehicle = ConnectedDriveVehicle(account, account.vehicles[0].data) diff --git a/test/test_deprecated_vehicle_status.py b/test/test_deprecated_vehicle_status.py index 91828ff7..14bd1e31 100644 --- a/test/test_deprecated_vehicle_status.py +++ b/test/test_deprecated_vehicle_status.py @@ -3,6 +3,7 @@ import datetime import pytest +import respx import time_machine from bimmer_connected.api.regions import get_region_from_name @@ -12,13 +13,13 @@ from bimmer_connected.vehicle.vehicle import ConnectedDriveVehicle from . import VIN_F31, VIN_G01, VIN_G20, VIN_G26, VIN_I01_NOREX, VIN_I01_REX, VIN_I20, get_deprecation_warning_count -from .test_account import get_mocked_account +from .conftest import prepare_account_with_vehicles @pytest.mark.asyncio -async def test_generic(caplog): +async def test_generic(caplog, bmw_fixture: respx.router): """Test generic attributes.""" - status = (await get_mocked_account()).get_vehicle(VIN_G26).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G26).status expected = datetime.datetime(year=2023, month=1, day=4, hour=14, minute=57, second=6, tzinfo=datetime.timezone.utc) assert expected == status.timestamp @@ -39,9 +40,9 @@ async def test_generic(caplog): @pytest.mark.asyncio -async def test_range_combustion_no_info(caplog): +async def test_range_combustion_no_info(caplog, bmw_fixture: respx.router): """Test if the parsing of mileage and range is working.""" - status = (await get_mocked_account()).get_vehicle(VIN_F31).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_F31).status assert (14, "L") == status.remaining_fuel assert status.remaining_range_fuel == (None, None) @@ -56,9 +57,9 @@ async def test_range_combustion_no_info(caplog): @pytest.mark.asyncio -async def test_range_combustion(caplog): +async def test_range_combustion(caplog, bmw_fixture: respx.router): """Test if the parsing of mileage and range is working.""" - status = (await get_mocked_account()).get_vehicle(VIN_G20).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G20).status assert (40, "L") == status.remaining_fuel assert (629, "km") == status.remaining_range_fuel @@ -73,9 +74,9 @@ async def test_range_combustion(caplog): @pytest.mark.asyncio -async def test_range_phev(caplog): +async def test_range_phev(caplog, bmw_fixture: respx.router): """Test if the parsing of mileage and range is working.""" - status = (await get_mocked_account()).get_vehicle(VIN_G01).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01).status assert (40, "L") == status.remaining_fuel assert (436, "km") == status.remaining_range_fuel @@ -92,9 +93,9 @@ async def test_range_phev(caplog): @pytest.mark.asyncio -async def test_range_rex(caplog): +async def test_range_rex(caplog, bmw_fixture: respx.router): """Test if the parsing of mileage and range is working.""" - status = (await get_mocked_account()).get_vehicle(VIN_I01_REX).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_I01_REX).status assert (6, "L") == status.remaining_fuel assert (105, "km") == status.remaining_range_fuel @@ -111,9 +112,9 @@ async def test_range_rex(caplog): @pytest.mark.asyncio -async def test_range_electric(caplog): +async def test_range_electric(caplog, bmw_fixture: respx.router): """Test if the parsing of mileage and range is working.""" - status = (await get_mocked_account()).get_vehicle(VIN_G26).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G26).status assert status.remaining_fuel == (None, None) assert status.remaining_range_fuel == (None, None) @@ -129,9 +130,9 @@ async def test_range_electric(caplog): @time_machine.travel("2021-11-28 21:28:59 +0000", tick=False) @pytest.mark.asyncio -async def test_charging_end_time(caplog): +async def test_charging_end_time(caplog, bmw_fixture: respx.router): """Test if the parsing of mileage and range is working.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() status = account.get_vehicle(VIN_I01_NOREX).status assert datetime.datetime(2021, 11, 28, 23, 27, 59, tzinfo=datetime.timezone.utc) == status.charging_end_time @@ -140,9 +141,9 @@ async def test_charging_end_time(caplog): @pytest.mark.asyncio -async def test_charging_time_label(caplog): +async def test_charging_time_label(caplog, bmw_fixture: respx.router): """Test if the parsing of mileage and range is working.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() status = account.get_vehicle(VIN_I20).status assert None is status.charging_time_label @@ -150,10 +151,10 @@ async def test_charging_time_label(caplog): @pytest.mark.asyncio -async def test_plugged_in_waiting_for_charge_window(caplog): +async def test_plugged_in_waiting_for_charge_window(caplog, bmw_fixture: respx.router): """G01 is plugged in but not charging, as its waiting for charging window.""" # Should be None on G01 as it is only "charging" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_I01_REX) assert vehicle.status.charging_end_time is None @@ -164,9 +165,9 @@ async def test_plugged_in_waiting_for_charge_window(caplog): @pytest.mark.asyncio -async def test_condition_based_services(caplog): +async def test_condition_based_services(caplog, bmw_fixture: respx.router): """Test condition based service messages.""" - status = (await get_mocked_account()).get_vehicle(VIN_G26).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G26).status cbs = status.condition_based_services assert 5 == len(cbs) @@ -191,9 +192,9 @@ async def test_condition_based_services(caplog): @pytest.mark.asyncio -async def test_parse_f31_no_position(caplog): +async def test_parse_f31_no_position(caplog, bmw_fixture: respx.router): """Test parsing of F31 data with position tracking disabled in the vehicle.""" - status = (await get_mocked_account()).get_vehicle(VIN_F31).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_F31).status assert status.gps_position is None assert status.gps_heading is None @@ -202,9 +203,9 @@ async def test_parse_f31_no_position(caplog): @pytest.mark.asyncio -async def test_parse_gcj02_position(caplog): +async def test_parse_gcj02_position(caplog, bmw_fixture: respx.router): """Test conversion of GCJ02 to WGS84 for china.""" - account = await get_mocked_account(get_region_from_name("china")) + account = await prepare_account_with_vehicles(get_region_from_name("china")) vehicle = account.get_vehicle(VIN_G01) vehicle = ConnectedDriveVehicle(account, vehicle.data) vehicle.update_state( @@ -228,15 +229,15 @@ async def test_parse_gcj02_position(caplog): @pytest.mark.asyncio -async def test_lids(caplog): +async def test_lids(caplog, bmw_fixture: respx.router): """Test features around lids.""" - # status = (await get_mocked_account()).get_vehicle(VIN_G30).status + # status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G30).status # assert 6 == len(list(status.lids)) # assert 3 == len(list(status.open_lids)) # assert status.all_lids_closed is False - status = (await get_mocked_account()).get_vehicle(VIN_G26).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G26).status for lid in status.lids: assert LidState.CLOSED == lid.state @@ -248,9 +249,9 @@ async def test_lids(caplog): @pytest.mark.asyncio -async def test_windows_g31(caplog): +async def test_windows_g31(caplog, bmw_fixture: respx.router): """Test features around windows.""" - status = (await get_mocked_account()).get_vehicle(VIN_G01).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01).status for window in status.windows: assert LidState.CLOSED == window.state @@ -263,13 +264,13 @@ async def test_windows_g31(caplog): @pytest.mark.asyncio -async def test_door_locks(caplog): +async def test_door_locks(caplog, bmw_fixture: respx.router): """Test the door locks.""" - status = (await get_mocked_account()).get_vehicle(VIN_G01).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01).status assert LockState.LOCKED == status.door_lock_state - status = (await get_mocked_account()).get_vehicle(VIN_I01_REX).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_I01_REX).status assert LockState.UNLOCKED == status.door_lock_state @@ -277,13 +278,13 @@ async def test_door_locks(caplog): @pytest.mark.asyncio -async def test_check_control_messages(caplog): +async def test_check_control_messages(caplog, bmw_fixture: respx.router): """Test handling of check control messages. F11 is the only vehicle with active Check Control Messages, so we only expect to get something there. However we have no vehicle with issues in check control. """ - vehicle = (await get_mocked_account()).get_vehicle(VIN_G01) + vehicle = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01) assert vehicle.status.has_check_control_messages is True ccms = vehicle.status.check_control_messages @@ -297,9 +298,9 @@ async def test_check_control_messages(caplog): @pytest.mark.asyncio -async def test_functions_without_data(caplog): +async def test_functions_without_data(caplog, bmw_fixture: respx.router): """Test functions that do not return any result anymore.""" - status = (await get_mocked_account()).get_vehicle(VIN_G01).status + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01).status assert status.last_charging_end_result is None assert status.parking_lights is None diff --git a/test/test_remote_services.py b/test/test_remote_services.py index 639808ce..7c1b642a 100644 --- a/test/test_remote_services.py +++ b/test/test_remote_services.py @@ -1,14 +1,4 @@ """Test for remote_services.""" -import json -from collections import defaultdict -from copy import deepcopy -from pathlib import Path -from typing import Dict, List - -from bimmer_connected.api.client import MyBMWClient -from bimmer_connected.models import MyBMWAPIError, MyBMWRemoteServiceError -from bimmer_connected.vehicle.charging_profile import ChargingMode - try: from unittest import mock @@ -22,133 +12,48 @@ import httpx import pytest +import respx import time_machine -from bimmer_connected.models import ChargingSettings, PointOfInterest +from bimmer_connected.api.client import MyBMWClient +from bimmer_connected.models import MyBMWAPIError, MyBMWRemoteServiceError, PointOfInterest from bimmer_connected.vehicle import remote_services +from bimmer_connected.vehicle.charging_profile import ChargingMode +from bimmer_connected.vehicle.climate import ClimateActivityState +from bimmer_connected.vehicle.doors_windows import LockState +from bimmer_connected.vehicle.fuel_and_battery import ChargingState from bimmer_connected.vehicle.remote_services import ExecutionState, RemoteServiceStatus -from . import RESPONSE_DIR, VIN_F31, VIN_G01, VIN_G26, VIN_I01_NOREX, VIN_I20, load_response -from .test_account import account_mock, get_mocked_account - -_RESPONSE_INITIATED = RESPONSE_DIR / "remote_services" / "eadrax_service_initiated.json" -_RESPONSE_PENDING = RESPONSE_DIR / "remote_services" / "eadrax_service_pending.json" -_RESPONSE_DELIVERED = RESPONSE_DIR / "remote_services" / "eadrax_service_delivered.json" -_RESPONSE_EXECUTED = RESPONSE_DIR / "remote_services" / "eadrax_service_executed.json" -_RESPONSE_ERROR = RESPONSE_DIR / "remote_services" / "eadrax_service_error.json" -_RESPONSE_EVENTPOSITION = RESPONSE_DIR / "remote_services" / "eadrax_service_eventposition.json" +from . import ( + REMOTE_SERVICE_RESPONSE_DELIVERED, + REMOTE_SERVICE_RESPONSE_ERROR, + REMOTE_SERVICE_RESPONSE_EXECUTED, + REMOTE_SERVICE_RESPONSE_PENDING, + VIN_F31, + VIN_G01, + VIN_G26, + VIN_I01_NOREX, + VIN_I20, + load_response, +) +from .common import ( + CHARGING_SETTINGS, + POI_DATA, +) +from .conftest import prepare_account_with_vehicles remote_services._POLLING_CYCLE = 0 -POI_DATA = { - "lat": 37.4028943, - "lon": -121.9700289, - "name": "49ers", - "street": "4949 Marie P DeBartolo Way", - "city": "Santa Clara", - "postal_code": "CA 95054", - "country": "United States", -} - -CHARGING_SETTINGS = {"target_soc": 75, "ac_limit": 16} - -STATUS_RESPONSE_ORDER = [_RESPONSE_PENDING, _RESPONSE_DELIVERED, _RESPONSE_EXECUTED] -STATUS_RESPONSE_DICT: Dict[str, List[Path]] = defaultdict(lambda: deepcopy(STATUS_RESPONSE_ORDER)) - - -def service_trigger_sideeffect(request: httpx.Request) -> httpx.Response: - """Return specific eventId for each remote function.""" - json_data = load_response(_RESPONSE_INITIATED) - json_data["eventId"] = str(uuid4()) - return httpx.Response(200, json=json_data) - - -def charging_settings_sideeffect(request: httpx.Request) -> httpx.Response: - """Check if payload is a valid charging settings payload and return evendId.""" - _ = ChargingSettings(**json.loads(request.content)) - return service_trigger_sideeffect(request) - - -def charging_profile_sideeffect(request: httpx.Request) -> httpx.Response: - """Check if payload is a valid charging settings payload and return evendId.""" - - data = json.loads(request.content) - - if {"chargingMode", "departureTimer", "isPreconditionForDepartureActive", "servicePack"} != set(data): - return httpx.Response(500) - if ( - data["chargingMode"]["chargingPreference"] == "NO_PRESELECTION" - and data["chargingMode"]["type"] != "CHARGING_IMMEDIATELY" - ): - return httpx.Response(500) - if data["chargingMode"]["chargingPreference"] == "CHARGING_WINDOW" and data["chargingMode"]["type"] != "TIME_SLOT": - return httpx.Response(500) - - if not isinstance(data["isPreconditionForDepartureActive"], bool): - return httpx.Response(500) - - return service_trigger_sideeffect(request) - - -def service_status_sideeffect(request: httpx.Request) -> httpx.Response: - """Return all 3 eventStatus responses per function.""" - response_data = STATUS_RESPONSE_DICT[request.url.params["eventId"]].pop(0) - return httpx.Response(200, json=load_response(response_data)) - - -def poi_sideeffect(request: httpx.Request) -> httpx.Response: - """Check if payload is a valid POI.""" - data = json.loads(request.content) - tests = all( - [ - len(data["vin"]) == 17, - isinstance(data["location"]["coordinates"]["latitude"], float), - isinstance(data["location"]["coordinates"]["longitude"], float), - len(data["location"]["name"]) > 0, - ] - ) - if not tests: - return httpx.Response(400) - return httpx.Response(201) - - -def remote_services_mock(): - """Return mocked adapter for auth.""" - router = account_mock() - - router.post(path__regex=r"/eadrax-vrccs/v3/presentation/remote-commands/.+/.+$").mock( - side_effect=service_trigger_sideeffect - ) - router.post(path__regex=r"/eadrax-crccs/v1/vehicles/.+/(start|stop)-charging$").mock( - side_effect=service_trigger_sideeffect - ) - router.post(path__regex=r"/eadrax-crccs/v1/vehicles/.+/charging-settings$").mock( - side_effect=charging_settings_sideeffect - ) - router.post(path__regex=r"/eadrax-crccs/v1/vehicles/.+/charging-profile$").mock( - side_effect=charging_profile_sideeffect - ) - router.post("/eadrax-vrccs/v3/presentation/remote-commands/eventStatus", params={"eventId": mock.ANY}).mock( - side_effect=service_status_sideeffect - ) - - router.post("/eadrax-dcs/v1/send-to-car/send-to-car").mock(side_effect=poi_sideeffect) - router.post("/eadrax-vrccs/v3/presentation/remote-commands/eventPosition", params={"eventId": mock.ANY}).respond( - 200, - json=load_response(_RESPONSE_EVENTPOSITION), - ) - return router - def test_states(): """Test parsing the different response types.""" - rss = RemoteServiceStatus(load_response(_RESPONSE_PENDING)) + rss = RemoteServiceStatus(load_response(REMOTE_SERVICE_RESPONSE_PENDING)) assert ExecutionState.PENDING == rss.state - rss = RemoteServiceStatus(load_response(_RESPONSE_DELIVERED)) + rss = RemoteServiceStatus(load_response(REMOTE_SERVICE_RESPONSE_DELIVERED)) assert ExecutionState.DELIVERED == rss.state - rss = RemoteServiceStatus(load_response(_RESPONSE_EXECUTED)) + rss = RemoteServiceStatus(load_response(REMOTE_SERVICE_RESPONSE_EXECUTED)) assert ExecutionState.EXECUTED == rss.state @@ -167,13 +72,12 @@ def test_states(): } -@remote_services_mock() @pytest.mark.asyncio @pytest.mark.filterwarnings("ignore:coroutine 'AsyncMockMixin._execute_mock_call' was never awaited:RuntimeWarning") -async def test_trigger_remote_services(): +async def test_trigger_remote_services(bmw_fixture: respx.Router): """Test executing a remote light flash.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_I20) for service in ALL_SERVICES.values(): @@ -182,7 +86,7 @@ async def test_trigger_remote_services(): ) as mock_listener: mock_listener.reset_mock() - response = await getattr(vehicle.remote_services, service["call"])( + response = await getattr(vehicle.remote_services, service["call"])( # type: ignore[call-overload] *service.get("args", []), **service.get("kwargs", {}) ) assert ExecutionState.EXECUTED == response.state @@ -194,36 +98,86 @@ async def test_trigger_remote_services(): @pytest.mark.asyncio -async def test_get_remote_service_status(): +async def test_get_remote_service_status(bmw_fixture: respx.Router): """Test get_remove_service_status method.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_G26) client = MyBMWClient(account.config) - with remote_services_mock() as mock_api: - mock_api.post("/eadrax-vrccs/v3/presentation/remote-commands/eventStatus", params={"eventId": mock.ANY}).mock( - side_effect=[ - httpx.Response(500), - httpx.Response(200, text="You can't parse this..."), - httpx.Response(200, json=load_response(_RESPONSE_ERROR)), - ], - ) - - with pytest.raises(MyBMWAPIError): - await vehicle.remote_services._block_until_done(client, uuid4()) - with pytest.raises(ValueError): - await vehicle.remote_services._block_until_done(client, uuid4()) - with pytest.raises(MyBMWRemoteServiceError): - await vehicle.remote_services._block_until_done(client, uuid4()) + bmw_fixture.post("/eadrax-vrccs/v3/presentation/remote-commands/eventStatus", params={"eventId": mock.ANY}).mock( + side_effect=[ + httpx.Response(500), + httpx.Response(200, text="You can't parse this..."), + httpx.Response(200, json=load_response(REMOTE_SERVICE_RESPONSE_ERROR)), + ], + ) + + with pytest.raises(MyBMWAPIError): + await vehicle.remote_services._block_until_done(client, uuid4()) + with pytest.raises(ValueError): + await vehicle.remote_services._block_until_done(client, uuid4()) + with pytest.raises(MyBMWRemoteServiceError): + await vehicle.remote_services._block_until_done(client, uuid4()) + + +@pytest.mark.asyncio +async def test_set_lock_result(bmw_fixture: respx.Router): + """Test locking/unlocking a car.""" + + account = await prepare_account_with_vehicles() + + vehicle = account.get_vehicle(VIN_I01_NOREX) + # check current state, unlock vehicle, check changed state + assert vehicle.doors_and_windows.door_lock_state == LockState.UNLOCKED + await vehicle.remote_services.trigger_remote_door_lock() + assert vehicle.doors_and_windows.door_lock_state == LockState.LOCKED + + # now lock vehicle again, check changed state + await vehicle.remote_services.trigger_remote_door_unlock() + assert vehicle.doors_and_windows.door_lock_state == LockState.UNLOCKED -@remote_services_mock() @pytest.mark.asyncio -async def test_set_charging_settings(): +async def test_set_climate_result(bmw_fixture: respx.Router): + """Test starting/stopping climatization.""" + + account = await prepare_account_with_vehicles() + + vehicle = account.get_vehicle(VIN_G01) + # check current state, unlock vehicle, check changed state + assert vehicle.climate.activity == ClimateActivityState.STANDBY + await vehicle.remote_services.trigger_remote_air_conditioning() + assert vehicle.climate.activity in [ClimateActivityState.COOLING, ClimateActivityState.HEATING] + + # now lock vehicle again, check changed state + await vehicle.remote_services.trigger_remote_air_conditioning_stop() + assert vehicle.climate.activity == ClimateActivityState.STANDBY + + +@pytest.mark.asyncio +async def test_charging_start_stop(bmw_fixture: respx.Router): + """Test starting/stopping climatization.""" + + account = await prepare_account_with_vehicles() + + vehicle = account.get_vehicle(VIN_I20) + + # check current state, unlock vehicle, check changed state + assert vehicle.fuel_and_battery.charging_status == ChargingState.CHARGING + await vehicle.remote_services.trigger_charge_stop() + assert vehicle.fuel_and_battery.charging_status == ChargingState.PLUGGED_IN + + # now lock vehicle again, check changed state + await vehicle.remote_services.trigger_charge_start() + assert vehicle.fuel_and_battery.charging_status == ChargingState.CHARGING + + +@pytest.mark.asyncio +async def test_set_charging_settings(bmw_fixture: respx.Router): """Test setting the charging settings on a car.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() # Errors on old electric vehicles, combustion engines and PHEV for vin in [VIN_I01_NOREX, VIN_F31, VIN_G01]: @@ -233,9 +187,16 @@ async def test_set_charging_settings(): with pytest.raises(ValueError): await vehicle.remote_services.trigger_charging_settings_update(ac_limit=16) - # This shouldn't fail + # This should work vehicle = account.get_vehicle(VIN_G26) - await vehicle.remote_services.trigger_charging_settings_update(target_soc=80, ac_limit=16) + # Test current state + assert vehicle.charging_profile.ac_current_limit == 16 + assert vehicle.fuel_and_battery.charging_target == 80 + # Update settings + await vehicle.remote_services.trigger_charging_settings_update(target_soc=75, ac_limit=12) + # Test changed state + assert vehicle.charging_profile.ac_current_limit == 12 + assert vehicle.fuel_and_battery.charging_target == 75 # But these are not allowed with pytest.raises(ValueError): @@ -252,12 +213,11 @@ async def test_set_charging_settings(): await vehicle.remote_services.trigger_charging_settings_update(ac_limit="asdf") -@remote_services_mock() @pytest.mark.asyncio -async def test_set_charging_profile(): +async def test_set_charging_profile(bmw_fixture: respx.Router): """Test setting the charging profile on a car.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() # Errors on combustion engines vehicle = account.get_vehicle(VIN_F31) @@ -266,20 +226,31 @@ async def test_set_charging_profile(): # This shouldn't fail even on older EV vehicle = account.get_vehicle(VIN_I01_NOREX) + # check current state + assert vehicle.charging_profile.charging_mode == ChargingMode.IMMEDIATE_CHARGING + assert vehicle.charging_profile.is_pre_entry_climatization_enabled is True + + # update two settings await vehicle.remote_services.trigger_charging_profile_update( - charging_mode=ChargingMode.IMMEDIATE_CHARGING, precondition_climate=True + charging_mode=ChargingMode.DELAYED_CHARGING, precondition_climate=False ) + assert vehicle.charging_profile.charging_mode == ChargingMode.DELAYED_CHARGING + assert vehicle.charging_profile.is_pre_entry_climatization_enabled is False + # change back only charging mode await vehicle.remote_services.trigger_charging_profile_update(charging_mode=ChargingMode.IMMEDIATE_CHARGING) + assert vehicle.charging_profile.charging_mode == ChargingMode.IMMEDIATE_CHARGING + + # change back only climatization await vehicle.remote_services.trigger_charging_profile_update(precondition_climate=True) + assert vehicle.charging_profile.is_pre_entry_climatization_enabled is True -@remote_services_mock() @pytest.mark.asyncio -async def test_vehicles_without_enabled_services(): +async def test_vehicles_without_enabled_services(bmw_fixture: respx.Router): """Test setting the charging profile on a car.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() # Errors on combustion engines vehicle = account.get_vehicle(VIN_F31) @@ -288,17 +259,16 @@ async def test_vehicles_without_enabled_services(): for service in ALL_SERVICES.values(): with pytest.raises(ValueError): - await getattr(vehicle.remote_services, service["call"])( + await getattr(vehicle.remote_services, service["call"])( # type: ignore[call-overload] *service.get("args", []), **service.get("kwargs", {}) ) -@remote_services_mock() @pytest.mark.asyncio -async def test_trigger_charge_start_stop_warnings(caplog): +async def test_trigger_charge_start_stop_warnings(caplog, bmw_fixture: respx.Router): """Test if warnings are produced correctly with the charge start/stop services.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_I20) fixture_not_connected = { @@ -331,12 +301,11 @@ async def test_trigger_charge_start_stop_warnings(caplog): caplog.clear() -@remote_services_mock() @pytest.mark.asyncio -async def test_get_remote_position(): +async def test_get_remote_position(bmw_fixture: respx.Router): """Test getting position from remote service.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() account.set_observer_position(1.0, 0.0) vehicle = account.get_vehicle(VIN_G26) status = vehicle.status @@ -356,12 +325,11 @@ async def test_get_remote_position(): assert 121 == status.gps_heading -@remote_services_mock() @pytest.mark.asyncio -async def test_get_remote_position_fail_without_observer(caplog): +async def test_get_remote_position_fail_without_observer(caplog, bmw_fixture: respx.Router): """Test getting position from remote service.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_G26) await vehicle.remote_services.trigger_remote_vehicle_finder() @@ -374,14 +342,13 @@ async def test_get_remote_position_fail_without_observer(caplog): assert len(errors) == 1 -@remote_services_mock() @pytest.mark.asyncio -async def test_fail_with_timeout(): +async def test_fail_with_timeout(bmw_fixture: respx.Router): """Test failing after timeout was reached.""" remote_services._POLLING_CYCLE = 1 remote_services._POLLING_TIMEOUT = 2 - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_G26) with pytest.raises(MyBMWRemoteServiceError): @@ -389,12 +356,11 @@ async def test_fail_with_timeout(): @time_machine.travel("2020-01-01", tick=False) -@remote_services_mock() @pytest.mark.asyncio -async def test_get_remote_position_too_old(): +async def test_get_remote_position_too_old(bmw_fixture: respx.Router): """Test remote service position being ignored as vehicle status is newer.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_G26) status = vehicle.status @@ -404,12 +370,11 @@ async def test_get_remote_position_too_old(): assert 180 == status.gps_heading -@remote_services_mock() @pytest.mark.asyncio -async def test_poi(): +async def test_poi(bmw_fixture: respx.Router): """Test get_remove_service_status method.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_G26) await vehicle.remote_services.trigger_send_poi({"lat": 12.34, "lon": 12.34}) diff --git a/test/test_utils.py b/test/test_utils.py index 477930f1..06aa2bdd 100644 --- a/test/test_utils.py +++ b/test/test_utils.py @@ -8,19 +8,20 @@ from backports import zoneinfo # type: ignore[import, no-redef] import pytest +import respx import time_machine from bimmer_connected.models import ChargingSettings, ValueWithUnit from bimmer_connected.utils import MyBMWJSONEncoder, get_class_property_names, parse_datetime from . import VIN_G26 -from .test_account import get_mocked_account +from .conftest import prepare_account_with_vehicles @pytest.mark.asyncio -async def test_drive_train(): +async def test_drive_train(bmw_fixture: respx.Router): """Tests available attribute.""" - vehicle = (await get_mocked_account()).get_vehicle(VIN_G26) + vehicle = (await prepare_account_with_vehicles()).get_vehicle(VIN_G26) assert [ "available_attributes", "brand", @@ -77,9 +78,9 @@ def test_parse_datetime(caplog): tick=False, ) @pytest.mark.asyncio -async def test_account_timezone(): +async def test_account_timezone(bmw_fixture: respx.Router): """Test the timezone in MyBMWAccount.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() assert account.utcdiff == 960 diff --git a/test/test_vehicle.py b/test/test_vehicle.py index 3e186a27..fc001abb 100644 --- a/test/test_vehicle.py +++ b/test/test_vehicle.py @@ -1,5 +1,6 @@ """Tests for MyBMWVehicle.""" import pytest +import respx from bimmer_connected.const import ATTR_ATTRIBUTES, ATTR_STATE, CarBrands from bimmer_connected.models import GPSPosition, StrEnum, VehicleDataBase @@ -18,7 +19,7 @@ VIN_I20, get_deprecation_warning_count, ) -from .test_account import account_mock, get_mocked_account +from .conftest import prepare_account_with_vehicles ATTRIBUTE_MAPPING = { "remainingFuel": "remaining_fuel", @@ -39,30 +40,31 @@ @pytest.mark.asyncio -async def test_drive_train(caplog): +async def test_drive_train(caplog, bmw_fixture: respx.Router): """Tests around drive_train attribute.""" - vehicle = (await get_mocked_account()).get_vehicle(VIN_F31) + account = await prepare_account_with_vehicles() + vehicle = account.get_vehicle(VIN_F31) assert DriveTrainType.COMBUSTION == vehicle.drive_train - vehicle = (await get_mocked_account()).get_vehicle(VIN_G01) + vehicle = account.get_vehicle(VIN_G01) assert DriveTrainType.PLUGIN_HYBRID == vehicle.drive_train - vehicle = (await get_mocked_account()).get_vehicle(VIN_G26) + vehicle = account.get_vehicle(VIN_G26) assert DriveTrainType.ELECTRIC == vehicle.drive_train - vehicle = (await get_mocked_account()).get_vehicle(VIN_I01_NOREX) + vehicle = account.get_vehicle(VIN_I01_NOREX) assert DriveTrainType.ELECTRIC == vehicle.drive_train - vehicle = (await get_mocked_account()).get_vehicle(VIN_I01_REX) + vehicle = account.get_vehicle(VIN_I01_REX) assert DriveTrainType.ELECTRIC_WITH_RANGE_EXTENDER == vehicle.drive_train assert len(get_deprecation_warning_count(caplog)) == 0 @pytest.mark.asyncio -async def test_parsing_attributes(caplog): +async def test_parsing_attributes(caplog, bmw_fixture: respx.Router): """Test parsing different attributes of the vehicle.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() for vehicle in account.vehicles: print(vehicle.name) @@ -78,9 +80,9 @@ async def test_parsing_attributes(caplog): @pytest.mark.asyncio -async def test_drive_train_attributes(caplog): +async def test_drive_train_attributes(caplog, bmw_fixture: respx.Router): """Test parsing different attributes of the vehicle.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle_drivetrains = { VIN_F31: (True, False), @@ -101,9 +103,9 @@ async def test_drive_train_attributes(caplog): @pytest.mark.asyncio -async def test_parsing_of_lsc_type(caplog): +async def test_parsing_of_lsc_type(caplog, bmw_fixture: respx.Router): """Test parsing the lsc type field.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() for vehicle in account.vehicles: assert vehicle.lsc_type is not None @@ -111,7 +113,7 @@ async def test_parsing_of_lsc_type(caplog): assert len(get_deprecation_warning_count(caplog)) == 0 -def test_car_brand(caplog): +def test_car_brand(caplog, bmw_fixture: respx.Router): """Test CarBrand enum.""" assert CarBrands("BMW") == CarBrands("bmw") @@ -122,21 +124,22 @@ def test_car_brand(caplog): @pytest.mark.asyncio -async def test_get_is_tracking_enabled(caplog): +async def test_get_is_tracking_enabled(caplog, bmw_fixture: respx.Router): """Test setting observer position.""" - vehicle = (await get_mocked_account()).get_vehicle(VIN_I01_REX) + account = await prepare_account_with_vehicles() + vehicle = account.get_vehicle(VIN_I01_REX) assert vehicle.is_vehicle_tracking_enabled is False - vehicle = (await get_mocked_account()).get_vehicle(VIN_F31) + vehicle = account.get_vehicle(VIN_F31) assert vehicle.is_vehicle_tracking_enabled is True assert len(get_deprecation_warning_count(caplog)) == 0 @pytest.mark.asyncio -async def test_available_attributes(caplog): +async def test_available_attributes(caplog, bmw_fixture: respx.Router): """Check that available_attributes returns exactly the arguments we have in our test data.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_F31) assert ["gps_position", "vin"] == vehicle.available_attributes @@ -204,25 +207,24 @@ async def test_available_attributes(caplog): @pytest.mark.asyncio -async def test_vehicle_image(caplog): +async def test_vehicle_image(caplog, bmw_fixture: respx.Router): """Test vehicle image request.""" - vehicle = (await get_mocked_account()).get_vehicle(VIN_G01) + vehicle = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01) - with account_mock() as mock_api: - mock_api.get( - path__regex=r"(.*)/eadrax-ics/v3/presentation/vehicles/\w*/images", - params={"carView": "FrontView"}, - headers={"accept": "image/png"}, - ).respond(200, content="png_image") - assert b"png_image" == await vehicle.get_vehicle_image(VehicleViewDirection.FRONT) + bmw_fixture.get( + path__regex=r"(.*)/eadrax-ics/v3/presentation/vehicles/\w*/images", + params={"carView": "FrontView"}, + headers={"accept": "image/png"}, + ).respond(200, content="png_image") + assert b"png_image" == await vehicle.get_vehicle_image(VehicleViewDirection.FRONT) assert len(get_deprecation_warning_count(caplog)) == 0 @pytest.mark.asyncio -async def test_no_timestamp(): +async def test_no_timestamp(bmw_fixture: respx.Router): """Test no timestamp available.""" - vehicle = (await get_mocked_account()).get_vehicle(VIN_F31) + vehicle = (await prepare_account_with_vehicles()).get_vehicle(VIN_F31) vehicle.data[ATTR_STATE].pop("lastFetched") vehicle.data[ATTR_ATTRIBUTES].pop("lastFetched") diff --git a/test/test_vehicle_status.py b/test/test_vehicle_status.py index 50dd68f4..ff20ab6e 100644 --- a/test/test_vehicle_status.py +++ b/test/test_vehicle_status.py @@ -3,6 +3,7 @@ import datetime import pytest +import respx import time_machine from bimmer_connected.api.regions import get_region_from_name @@ -24,15 +25,16 @@ VIN_I20, get_deprecation_warning_count, ) -from .test_account import get_mocked_account +from .conftest import prepare_account_with_vehicles UTC = datetime.timezone.utc @pytest.mark.asyncio -async def test_generic(caplog): +@pytest.mark.parametrize("bmw_fixture", [[VIN_G26]], indirect=True) +async def test_generic(caplog, bmw_fixture: respx.Router): """Test generic attributes.""" - status = (await get_mocked_account()).get_vehicle(VIN_G26) + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G26) expected = datetime.datetime(year=2023, month=1, day=4, hour=14, minute=57, second=6, tzinfo=UTC) assert expected == status.timestamp @@ -44,9 +46,9 @@ async def test_generic(caplog): @pytest.mark.asyncio -async def test_range_combustion_no_info(caplog): +async def test_range_combustion_no_info(caplog, bmw_fixture: respx.Router): """Test if the parsing of mileage and range is working.""" - vehicle = (await get_mocked_account()).get_vehicle(VIN_F31) + vehicle = (await prepare_account_with_vehicles()).get_vehicle(VIN_F31) status = vehicle.fuel_and_battery assert (14, "L") == status.remaining_fuel @@ -67,10 +69,10 @@ async def test_range_combustion_no_info(caplog): @pytest.mark.asyncio -async def test_range_combustion(caplog): +async def test_range_combustion(caplog, bmw_fixture: respx.Router): """Test if the parsing of mileage and range is working.""" # Metric units - status = (await get_mocked_account()).get_vehicle(VIN_G20).fuel_and_battery + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G20).fuel_and_battery assert (40, "L") == status.remaining_fuel assert (629, "km") == status.remaining_range_fuel @@ -84,7 +86,7 @@ async def test_range_combustion(caplog): assert len(get_deprecation_warning_count(caplog)) == 0 # Imperial units - status = (await get_mocked_account(metric=False)).get_vehicle(VIN_G20).fuel_and_battery + status = (await prepare_account_with_vehicles(metric=False)).get_vehicle(VIN_G20).fuel_and_battery assert (40, "gal") == status.remaining_fuel assert (629, "mi") == status.remaining_range_fuel @@ -99,10 +101,10 @@ async def test_range_combustion(caplog): @pytest.mark.asyncio -async def test_range_phev(caplog): +async def test_range_phev(caplog, bmw_fixture: respx.Router): """Test if the parsing of mileage and range is working.""" # Metric units - status = (await get_mocked_account()).get_vehicle(VIN_G01).fuel_and_battery + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01).fuel_and_battery assert (40, "L") == status.remaining_fuel assert (436, "km") == status.remaining_range_fuel @@ -118,7 +120,7 @@ async def test_range_phev(caplog): assert len(get_deprecation_warning_count(caplog)) == 0 # Imperial units - status = (await get_mocked_account(metric=False)).get_vehicle(VIN_G01).fuel_and_battery + status = (await prepare_account_with_vehicles(metric=False)).get_vehicle(VIN_G01).fuel_and_battery assert (40, "gal") == status.remaining_fuel assert (436, "mi") == status.remaining_range_fuel @@ -135,10 +137,10 @@ async def test_range_phev(caplog): @pytest.mark.asyncio -async def test_range_rex(caplog): +async def test_range_rex(caplog, bmw_fixture: respx.Router): """Test if the parsing of mileage and range is working.""" # Metric units - status = (await get_mocked_account()).get_vehicle(VIN_I01_REX).fuel_and_battery + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_I01_REX).fuel_and_battery assert (6, "L") == status.remaining_fuel assert (105, "km") == status.remaining_range_fuel @@ -154,7 +156,7 @@ async def test_range_rex(caplog): assert len(get_deprecation_warning_count(caplog)) == 0 # Imperial units - status = (await get_mocked_account(metric=False)).get_vehicle(VIN_I01_REX).fuel_and_battery + status = (await prepare_account_with_vehicles(metric=False)).get_vehicle(VIN_I01_REX).fuel_and_battery assert (6, "gal") == status.remaining_fuel assert (105, "mi") == status.remaining_range_fuel @@ -171,10 +173,10 @@ async def test_range_rex(caplog): @pytest.mark.asyncio -async def test_range_electric(caplog): +async def test_range_electric(caplog, bmw_fixture: respx.Router): """Test if the parsing of mileage and range is working.""" # Metric units - status = (await get_mocked_account()).get_vehicle(VIN_I20).fuel_and_battery + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_I20).fuel_and_battery assert status.remaining_fuel == (None, None) assert status.remaining_range_fuel == (None, None) @@ -188,7 +190,7 @@ async def test_range_electric(caplog): assert len(get_deprecation_warning_count(caplog)) == 0 # Imperial units - status = (await get_mocked_account(metric=False)).get_vehicle(VIN_I20).fuel_and_battery + status = (await prepare_account_with_vehicles(metric=False)).get_vehicle(VIN_I20).fuel_and_battery assert status.remaining_fuel == (None, None) assert status.remaining_range_fuel == (None, None) @@ -204,9 +206,9 @@ async def test_range_electric(caplog): @time_machine.travel("2021-11-28 21:28:59 +0000", tick=False) @pytest.mark.asyncio -async def test_charging_end_time(caplog): +async def test_charging_end_time(caplog, bmw_fixture: respx.Router): """Test charging end time.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_I01_NOREX) assert vehicle.fuel_and_battery.charging_end_time.astimezone(UTC) == datetime.datetime( @@ -221,9 +223,9 @@ async def test_charging_end_time(caplog): @time_machine.travel("2021-11-28 17:28:59 +0000", tick=False) @pytest.mark.asyncio -async def test_plugged_in_waiting_for_charge_window(caplog): +async def test_plugged_in_waiting_for_charge_window(caplog, bmw_fixture: respx.Router): """I01_REX is plugged in but not charging, as its waiting for charging window.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() vehicle = account.get_vehicle(VIN_I01_REX) assert vehicle.fuel_and_battery.charging_end_time is None @@ -238,9 +240,9 @@ async def test_plugged_in_waiting_for_charge_window(caplog): @pytest.mark.asyncio -async def test_condition_based_services(caplog): +async def test_condition_based_services(caplog, bmw_fixture: respx.Router): """Test condition based service messages.""" - vehicle = (await get_mocked_account()).get_vehicle(VIN_G26) + vehicle = (await prepare_account_with_vehicles()).get_vehicle(VIN_G26) cbs = vehicle.condition_based_services.messages assert 5 == len(cbs) @@ -265,9 +267,9 @@ async def test_condition_based_services(caplog): @pytest.mark.asyncio -async def test_position_generic(caplog): +async def test_position_generic(caplog, bmw_fixture: respx.Router): """Test generic attributes.""" - status = (await get_mocked_account()).get_vehicle(VIN_G26) + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G26) assert (48.177334, 11.556274) == status.vehicle_location.location assert 180 == status.vehicle_location.heading @@ -280,9 +282,9 @@ async def test_position_generic(caplog): @pytest.mark.asyncio -async def test_vehicle_active(caplog): +async def test_vehicle_active(caplog, bmw_fixture: respx.Router): """Test that vehicle_active is always False.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() for vehicle in account.vehicles: assert vehicle.is_vehicle_active is False @@ -291,9 +293,9 @@ async def test_vehicle_active(caplog): @pytest.mark.asyncio -async def test_parse_f31_no_position(caplog): +async def test_parse_f31_no_position(caplog, bmw_fixture: respx.Router): """Test parsing of F31 data with position tracking disabled in the vehicle.""" - vehicle = (await get_mocked_account()).get_vehicle(VIN_F31) + vehicle = (await prepare_account_with_vehicles()).get_vehicle(VIN_F31) assert vehicle.vehicle_location.location is None assert vehicle.vehicle_location.heading is None @@ -302,9 +304,9 @@ async def test_parse_f31_no_position(caplog): @pytest.mark.asyncio -async def test_parse_gcj02_position(caplog): +async def test_parse_gcj02_position(caplog, bmw_fixture: respx.Router): """Test conversion of GCJ02 to WGS84 for china.""" - account = await get_mocked_account(get_region_from_name("china")) + account = await prepare_account_with_vehicles(get_region_from_name("china")) vehicle = account.get_vehicle(VIN_G01) vehicle_test_data = { @@ -332,22 +334,22 @@ async def test_parse_gcj02_position(caplog): @pytest.mark.asyncio -async def test_lids(caplog): +async def test_lids(caplog, bmw_fixture: respx.Router): """Test features around lids.""" - # status = (await get_mocked_account()).get_vehicle(VIN_G01).doors_and_windows + # status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01).doors_and_windows # assert 6 == len(list(status.lids)) # assert 3 == len(list(status.open_lids)) # assert status.all_lids_closed is False - status = (await get_mocked_account()).get_vehicle(VIN_G26).doors_and_windows + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G26).doors_and_windows for lid in status.lids: assert LidState.CLOSED == lid.state assert status.all_lids_closed is True assert 6 == len(list(status.lids)) - status = (await get_mocked_account()).get_vehicle(VIN_I01_REX).doors_and_windows + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_I01_REX).doors_and_windows for lid in status.lids: assert LidState.CLOSED == lid.state @@ -360,9 +362,9 @@ async def test_lids(caplog): @pytest.mark.asyncio -async def test_windows_g01(caplog): +async def test_windows_g01(caplog, bmw_fixture: respx.Router): """Test features around windows.""" - status = (await get_mocked_account()).get_vehicle(VIN_G01).doors_and_windows + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01).doors_and_windows for window in status.windows: assert LidState.CLOSED == window.state @@ -375,13 +377,13 @@ async def test_windows_g01(caplog): @pytest.mark.asyncio -async def test_door_locks(caplog): +async def test_door_locks(caplog, bmw_fixture: respx.Router): """Test the door locks.""" - status = (await get_mocked_account()).get_vehicle(VIN_G01).doors_and_windows + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01).doors_and_windows assert LockState.LOCKED == status.door_lock_state - status = (await get_mocked_account()).get_vehicle(VIN_I01_REX).doors_and_windows + status = (await prepare_account_with_vehicles()).get_vehicle(VIN_I01_REX).doors_and_windows assert LockState.UNLOCKED == status.door_lock_state @@ -389,13 +391,13 @@ async def test_door_locks(caplog): @pytest.mark.asyncio -async def test_check_control_messages(caplog): +async def test_check_control_messages(caplog, bmw_fixture: respx.Router): """Test handling of check control messages. F11 is the only vehicle with active Check Control Messages, so we only expect to get something there. However we have no vehicle with issues in check control. """ - vehicle = (await get_mocked_account()).get_vehicle(VIN_G01) + vehicle = (await prepare_account_with_vehicles()).get_vehicle(VIN_G01) assert vehicle.check_control_messages.has_check_control_messages is True ccms = vehicle.check_control_messages.messages @@ -405,7 +407,7 @@ async def test_check_control_messages(caplog): assert "ENGINE_OIL" == ccms[1].description_short assert None is ccms[1].description_long - vehicle = (await get_mocked_account()).get_vehicle(VIN_G20) + vehicle = (await prepare_account_with_vehicles()).get_vehicle(VIN_G20) assert vehicle.check_control_messages.has_check_control_messages is False ccms = vehicle.check_control_messages.messages @@ -419,10 +421,10 @@ async def test_check_control_messages(caplog): @pytest.mark.asyncio -async def test_charging_profile(caplog): +async def test_charging_profile(caplog, bmw_fixture: respx.Router): """Test parsing of the charging profile.""" - charging_profile = (await get_mocked_account()).get_vehicle(VIN_I01_REX).charging_profile + charging_profile = (await prepare_account_with_vehicles()).get_vehicle(VIN_I01_REX).charging_profile assert charging_profile.is_pre_entry_climatization_enabled is False departure_timer = charging_profile.departure_times[0] @@ -439,7 +441,7 @@ async def test_charging_profile(caplog): assert charging_profile.ac_available_limits is None - charging_settings = (await get_mocked_account()).get_vehicle(VIN_G26).charging_profile + charging_settings = (await prepare_account_with_vehicles()).get_vehicle(VIN_G26).charging_profile assert charging_settings.ac_current_limit == 16 assert charging_settings.ac_available_limits == [6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 20, 32] @@ -447,9 +449,9 @@ async def test_charging_profile(caplog): @pytest.mark.asyncio -async def test_charging_profile_format_for_remote_service(caplog): +async def test_charging_profile_format_for_remote_service(caplog, bmw_fixture: respx.Router): """Test formatting of the charging profile.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() for vin in ALL_CHARGING_SETTINGS: vehicle = account.get_vehicle(vin) @@ -470,9 +472,9 @@ async def test_charging_profile_format_for_remote_service(caplog): @pytest.mark.asyncio -async def test_tires(): +async def test_tires(bmw_fixture: respx.Router): """Test tire status.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() # Older vehicles do not provide tire status assert account.get_vehicle(VIN_F31).tires is None @@ -501,9 +503,9 @@ async def test_tires(): @time_machine.travel("2021-11-28 21:28:59 +0000", tick=False) @pytest.mark.asyncio -async def test_climate(): +async def test_climate(bmw_fixture: respx.Router): """Test climate status.""" - account = await get_mocked_account() + account = await prepare_account_with_vehicles() # Older vehicles do not provide climate status climate = account.get_vehicle(VIN_I01_REX).climate