starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-27 18:14:46 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add NexBlue integration (#178132)

Browse Source 
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Kristy
2026-08-26 15:34:39 +02:00
committed by GitHub
co-authored by Joost Lekkerkerker
parent 6c26c7211c
commit c7fa908fa7
22 changed files with 2347 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CODEOWNERS
Generated
+2
View File
@@ -1246,6 +1246,8 @@ CLAUDE.md @home-assistant/core
/tests/components/netio/ @agners
/homeassistant/components/network/ @home-assistant/core
/tests/components/network/ @home-assistant/core
/homeassistant/components/nexblue/ @nexblue-maintainer
/tests/components/nexblue/ @nexblue-maintainer
/homeassistant/components/nexia/ @bdraco
/tests/components/nexia/ @bdraco
/homeassistant/components/nextbus/ @vividboarder
homeassistant/components/nexblue/__init__.py
+25
View File
@@ -0,0 +1,25 @@
"""Integration for NexBlue EV chargers."""
from nexblue_api import NexBlueClient
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DEFAULT_API_URL, PLATFORMS
from .coordinator import NexBlueConfigEntry, NexBlueDataUpdateCoordinator
async def async_setup_entry(hass: HomeAssistant, entry: NexBlueConfigEntry) -> bool:
"""Set up NexBlue from a config entry."""
client = NexBlueClient(async_get_clientsession(hass), DEFAULT_API_URL)
coordinator = NexBlueDataUpdateCoordinator(hass, entry, client)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: NexBlueConfigEntry) -> bool:
"""Unload a NexBlue config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
homeassistant/components/nexblue/config_flow.py
+86
View File
@@ -0,0 +1,86 @@
"""Config flow for the NexBlue integration."""
from typing import Any, override
from nexblue_api import (
NexBlueAuthError,
NexBlueClient,
NexBlueConnectionError,
NexBlueError,
TokenBundle,
)
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONF_REFRESH_TOKEN, DEFAULT_API_URL, DOMAIN, LOGGER
AUTH_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
)
class NexBlueConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a NexBlue config flow."""
VERSION = 1
@override
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle initial setup from the user interface."""
errors: dict[str, str] = {}
if user_input is not None:
token, error = await self._async_validate_login(
user_input[CONF_USERNAME], user_input[CONF_PASSWORD]
)
if error:
errors["base"] = error
else:
assert token is not None
assert token.account_id is not None
assert token.refresh_token is not None
await self.async_set_unique_id(token.account_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"NexBlue ({user_input[CONF_USERNAME]})",
data={
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_REFRESH_TOKEN: token.refresh_token,
},
)
return self.async_show_form(
step_id="user", data_schema=AUTH_SCHEMA, errors=errors
)
async def _async_validate_login(
self, username: str, password: str
) -> tuple[TokenBundle | None, str | None]:
"""Validate credentials and return token data safe to persist."""
client = NexBlueClient(async_get_clientsession(self.hass), DEFAULT_API_URL)
try:
token = await client.async_login(username, password)
except NexBlueAuthError:
return None, "invalid_auth"
except NexBlueConnectionError:
return None, "cannot_connect"
except NexBlueError:
return None, "unknown"
except Exception: # noqa: BLE001
LOGGER.exception("Unexpected error validating NexBlue credentials")
return None, "unknown"
if not token.refresh_token:
return None, "invalid_auth"
if not token.account_id:
return None, "unknown"
return token, None
homeassistant/components/nexblue/const.py
+13
View File
@@ -0,0 +1,13 @@
"""Constants for the NexBlue integration."""
from datetime import timedelta
import logging
from homeassistant.const import Platform
DOMAIN = "nexblue"
CONF_REFRESH_TOKEN = "refresh_token"
DEFAULT_API_URL = "https://api.nexblue.com/third_party"
LOGGER = logging.getLogger(__package__)
PLATFORMS = [Platform.SENSOR]
UPDATE_INTERVAL = timedelta(minutes=1)
homeassistant/components/nexblue/coordinator.py
+99
View File
@@ -0,0 +1,99 @@
"""Data update coordinator for NexBlue."""
from typing import override
from nexblue_api import (
NexBlueAuthError,
NexBlueClient,
NexBlueConnectionError,
NexBlueDeviceOfflineError,
NexBlueError,
NexBlueRateLimitError,
)
from nexblue_api.models import ChargerStatus
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_REFRESH_TOKEN, LOGGER, UPDATE_INTERVAL
type NexBlueConfigEntry = ConfigEntry["NexBlueDataUpdateCoordinator"]
class NexBlueDataUpdateCoordinator(
DataUpdateCoordinator[dict[str, ChargerStatus | None]]
):
"""Fetch all charger telemetry in a coordinated update."""
config_entry: NexBlueConfigEntry
def __init__(
self,
hass: HomeAssistant,
entry: NexBlueConfigEntry,
client: NexBlueClient,
) -> None:
"""Initialize the coordinator."""
self.client = client
super().__init__(
hass,
LOGGER,
config_entry=entry,
name=f"NexBlue {entry.title}",
update_interval=UPDATE_INTERVAL,
)
@override
async def _async_update_data(self) -> dict[str, ChargerStatus | None]:
"""Fetch status for every charger visible to the configured account."""
try:
await self._async_ensure_authorized()
chargers = await self.client.async_list_chargers()
data: dict[str, ChargerStatus | None] = {}
for charger in chargers:
try:
data[
charger.serial_number
] = await self.client.async_get_charger_status(
charger.serial_number
)
except NexBlueAuthError, NexBlueConnectionError, NexBlueRateLimitError:
raise
except NexBlueDeviceOfflineError, NexBlueError:
data[charger.serial_number] = None
except NexBlueAuthError as err:
raise ConfigEntryAuthFailed from err
except (NexBlueConnectionError, NexBlueRateLimitError, NexBlueError) as err:
raise UpdateFailed("Unable to update NexBlue charger data") from err
return data
async def _async_ensure_authorized(self) -> None:
"""Refresh the access token, falling back to one saved-password login."""
try:
token = await self.client.async_ensure_access_token(
self.config_entry.data[CONF_REFRESH_TOKEN]
)
except NexBlueAuthError:
token = await self.client.async_login(
self.config_entry.data[CONF_USERNAME],
self.config_entry.data[CONF_PASSWORD],
)
if not token.refresh_token:
raise NexBlueAuthError from None
if (
token
and token.refresh_token
and token.refresh_token != self.config_entry.data[CONF_REFRESH_TOKEN]
):
self.hass.config_entries.async_update_entry(
self.config_entry,
data={
**self.config_entry.data,
CONF_REFRESH_TOKEN: token.refresh_token,
},
)
homeassistant/components/nexblue/icons.json
Executable
+42
View File
@@ -0,0 +1,42 @@
{
"entity": {
"sensor": {
"access_level": {
"default": "mdi:account-lock"
},
"brightness": {
"default": "mdi:brightness-percent"
},
"cable_current_limit": {
"default": "mdi:current-ac"
},
"cable_lock_mode": {
"default": "mdi:lock-clock"
},
"charging_state": {
"default": "mdi:ev-station"
},
"circuit_fuse": {
"default": "mdi:fuse"
},
"current": {
"default": "mdi:current-ac"
},
"current_limit": {
"default": "mdi:current-ac"
},
"lifetime_energy": {
"default": "mdi:counter"
},
"network_status": {
"default": "mdi:network-outline"
},
"phase_charging": {
"default": "mdi:sine-wave"
},
"voltage": {
"default": "mdi:sine-wave"
}
}
}
}
homeassistant/components/nexblue/manifest.json
+11
View File
@@ -0,0 +1,11 @@
{
"domain": "nexblue",
"name": "NexBlue",
"codeowners": ["@nexblue-maintainer"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/nexblue",
"integration_type": "hub",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["nexblue-api==0.1.3"]
}
homeassistant/components/nexblue/quality_scale.yaml
+74
View File
@@ -0,0 +1,74 @@
rules:
# Bronze
action-setup:
status: exempt
comment: The integration does not register custom actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: The integration does not provide custom actions.
docs-conditions:
status: exempt
comment: The integration does not provide conditions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
docs-triggers:
status: exempt
comment: The integration does not provide triggers.
entity-event-setup:
status: exempt
comment: The integration entities do not subscribe to events.
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions: todo
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: The integration does not have configuration parameters.
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: todo
parallel-updates: todo
reauthentication-flow: todo
test-coverage: todo
# Gold
devices: done
diagnostics: todo
discovery-update-info: todo
discovery: todo
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: done
entity-device-class: done
entity-disabled-by-default: todo
entity-translations: todo
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: todo
inject-websession: todo
strict-typing: todo
homeassistant/components/nexblue/sensor.py
+275
View File
@@ -0,0 +1,275 @@
"""Sensors for the NexBlue integration."""
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from typing import Literal, override
from nexblue_api.models import ChargerStatus
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
EntityCategory,
UnitOfElectricCurrent,
UnitOfElectricPotential,
UnitOfEnergy,
UnitOfPower,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import NexBlueConfigEntry, NexBlueDataUpdateCoordinator
CHARGING_STATE_MAP = {
0: "idle",
1: "connected",
2: "charging",
3: "finished",
4: "error",
5: "lb_waiting",
6: "delay_waiting",
7: "ev_waiting",
}
NETWORK_STATUS_MAP = {0: "none", 1: "wifi", 2: "modem", 3: "ethernet"}
CABLE_LOCK_MODE_MAP = {
0: "locked_while_charging",
1: "always_locked",
}
ACCESS_LEVEL_MAP = {
0: "authorized_users_only",
1: "no_restrictions",
}
PHASE_CHARGING_MAP = {0: "three_phase", 1: "single_phase"}
@dataclass(frozen=True, kw_only=True)
class NexBlueSensorEntityDescription(SensorEntityDescription):
"""Describe a NexBlue charger sensor."""
value_fn: Callable[[ChargerStatus], StateType]
phase: int | None = None
def _enum_options(values: dict[int, str]) -> list[str]:
"""Return the supported options for an enum sensor."""
return list(values.values())
def _phase_value(
values: Sequence[int | float],
phase: int,
) -> int | float | None:
"""Return a phase value when it is reported by the charger."""
if len(values) <= phase:
return None
return values[phase]
def _phase_value_fn(
metric: Literal["current", "voltage"], phase: int
) -> Callable[[ChargerStatus], StateType]:
"""Return a typed value function for a phase measurement."""
def value_fn(status: ChargerStatus) -> StateType:
values = status.current_a if metric == "current" else status.voltage_v
return _phase_value(values, phase)
return value_fn
SENSOR_DESCRIPTIONS: tuple[NexBlueSensorEntityDescription, ...] = (
NexBlueSensorEntityDescription(
key="charging_state",
translation_key="charging_state",
device_class=SensorDeviceClass.ENUM,
options=_enum_options(CHARGING_STATE_MAP),
value_fn=lambda status: CHARGING_STATE_MAP.get(status.charging_state),
),
NexBlueSensorEntityDescription(
key="cable_lock_mode",
translation_key="cable_lock_mode",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=_enum_options(CABLE_LOCK_MODE_MAP),
value_fn=lambda status: CABLE_LOCK_MODE_MAP.get(status.cable_lock_mode),
),
NexBlueSensorEntityDescription(
key="access_level",
translation_key="access_level",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=_enum_options(ACCESS_LEVEL_MAP),
value_fn=lambda status: ACCESS_LEVEL_MAP.get(status.access_level),
),
NexBlueSensorEntityDescription(
key="phase_charging",
translation_key="phase_charging",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=_enum_options(PHASE_CHARGING_MAP),
value_fn=lambda status: PHASE_CHARGING_MAP.get(status.phase_charging),
),
NexBlueSensorEntityDescription(
key="power",
device_class=SensorDeviceClass.POWER,
native_unit_of_measurement=UnitOfPower.KILO_WATT,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda status: status.power_kw,
),
NexBlueSensorEntityDescription(
key="energy",
device_class=SensorDeviceClass.ENERGY,
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
state_class=SensorStateClass.TOTAL,
value_fn=lambda status: status.energy_kwh,
),
NexBlueSensorEntityDescription(
key="lifetime_energy",
translation_key="lifetime_energy",
device_class=SensorDeviceClass.ENERGY,
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
state_class=SensorStateClass.TOTAL_INCREASING,
value_fn=lambda status: status.lifetime_energy_kwh,
),
NexBlueSensorEntityDescription(
key="current_limit",
translation_key="current_limit",
device_class=SensorDeviceClass.CURRENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda status: status.current_limit_a,
),
NexBlueSensorEntityDescription(
key="cable_current_limit",
translation_key="cable_current_limit",
device_class=SensorDeviceClass.CURRENT,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda status: status.cable_current_limit_a,
),
NexBlueSensorEntityDescription(
key="circuit_fuse",
translation_key="circuit_fuse",
device_class=SensorDeviceClass.CURRENT,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda status: status.circuit_fuse_a,
),
*(
NexBlueSensorEntityDescription(
key=f"current_{phase}",
translation_key="current",
phase=phase,
device_class=SensorDeviceClass.CURRENT,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=_phase_value_fn("current", phase - 1),
)
for phase in range(1, 4)
),
*(
NexBlueSensorEntityDescription(
key=f"voltage_{phase}",
translation_key="voltage",
phase=phase,
device_class=SensorDeviceClass.VOLTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
state_class=SensorStateClass.MEASUREMENT,
value_fn=_phase_value_fn("voltage", phase - 1),
)
for phase in range(1, 4)
),
NexBlueSensorEntityDescription(
key="network_status",
translation_key="network_status",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=_enum_options(NETWORK_STATUS_MAP),
value_fn=lambda status: NETWORK_STATUS_MAP.get(status.network_status),
),
NexBlueSensorEntityDescription(
key="brightness",
translation_key="brightness",
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda status: status.brightness_percent,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: NexBlueConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up NexBlue sensors for every discovered charger."""
coordinator = entry.runtime_data
async_add_entities(
NexBlueStatusSensor(coordinator, serial_number, description)
for serial_number in coordinator.data
for description in SENSOR_DESCRIPTIONS
)
class NexBlueStatusSensor(
CoordinatorEntity[NexBlueDataUpdateCoordinator], SensorEntity
):
"""Expose normalized NexBlue charger telemetry."""
_attr_has_entity_name = True
entity_description: NexBlueSensorEntityDescription
def __init__(
self,
coordinator: NexBlueDataUpdateCoordinator,
serial_number: str,
description: NexBlueSensorEntityDescription,
) -> None:
"""Initialize a sensor for one charger metric."""
super().__init__(coordinator)
self._serial_number = serial_number
self.entity_description = description
self._attr_unique_id = f"{serial_number}_{description.key}"
if description.phase is not None:
self._attr_translation_placeholders = {"phase": f"L{description.phase}"}
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, serial_number)},
manufacturer="NexBlue",
name=serial_number,
serial_number=serial_number,
)
@property
@override
def available(self) -> bool:
"""Return whether this charger is currently reachable."""
return (
super().available
and self.coordinator.data.get(self._serial_number) is not None
)
@property
@override
def native_value(self) -> StateType:
"""Return the sensor value from the latest coordinator data."""
status = self.coordinator.data[self._serial_number]
assert status is not None
return self.entity_description.value_fn(status)
homeassistant/components/nexblue/strings.json
+79
View File
@@ -0,0 +1,79 @@
{
"config": {
"abort": {
"already_configured": "This NexBlue account is already configured."
},
"error": {
"cannot_connect": "Cannot connect to NexBlue.",
"invalid_auth": "NexBlue authentication failed.",
"unknown": "NexBlue could not complete sign-in. Please try again later."
},
"step": {
"user": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
},
"data_description": {
"password": "The password for your NexBlue account.",
"username": "The email address or username used to sign in to NexBlue."
},
"description": "Sign in with your NexBlue end-user account."
}
}
},
"entity": {
"sensor": {
"access_level": {
"name": "Access level",
"state": {
"authorized_users_only": "Authorized users only",
"no_restrictions": "No restrictions"
}
},
"brightness": { "name": "LED brightness" },
"cable_current_limit": { "name": "Cable rating" },
"cable_lock_mode": {
"name": "Cable lock mode",
"state": {
"always_locked": "Always locked",
"locked_while_charging": "Locked while charging"
}
},
"charging_state": {
"name": "Charging state",
"state": {
"charging": "Charging",
"connected": "Ready to charge",
"delay_waiting": "Schedule waiting",
"error": "Charging unavailable",
"ev_waiting": "Waiting for car response",
"finished": "Charging complete",
"idle": "Connect cable to charge",
"lb_waiting": "Waiting for available power"
}
},
"circuit_fuse": { "name": "Circuit fuse" },
"current": { "name": "Current {phase}" },
"current_limit": { "name": "Current limit" },
"lifetime_energy": { "name": "Lifetime energy" },
"network_status": {
"name": "Network status",
"state": {
"ethernet": "Ethernet",
"modem": "4G",
"none": "None",
"wifi": "Wi-Fi"
}
},
"phase_charging": {
"name": "Phase charging",
"state": {
"single_phase": "Single-phase",
"three_phase": "Three-phase"
}
},
"voltage": { "name": "Voltage {phase}" }
}
}
}
homeassistant/generated/config_flows.py
Generated
+1
View File
@@ -524,6 +524,7 @@ FLOWS = {
"netgear",
"netgear_lte",
"netio",
"nexblue",
"nexia",
"nextbus",
"nextcloud",
homeassistant/generated/integrations.json
+6
View File
@@ -4847,6 +4847,12 @@
"config_flow": false,
"iot_class": "cloud_polling"
},
"nexblue": {
"name": "NexBlue",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"nexen": {
"name": "Nexen",
"integration_type": "virtual",
requirements_all.txt
Generated
+3
View File
@@ -1691,6 +1691,9 @@ nettigo-air-monitor==5.1.0
# homeassistant.components.neurio_energy
neurio==0.3.1
# homeassistant.components.nexblue
nexblue-api==0.1.3
# homeassistant.components.nexia
nexia==2.13.0
tests/components/nexblue/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Tests for the NexBlue integration."""
tests/components/nexblue/conftest.py
+74
View File
@@ -0,0 +1,74 @@
"""Fixtures for the NexBlue integration tests."""
from collections.abc import Generator
from unittest.mock import MagicMock, patch
from nexblue_api import Charger, ChargerStatus, TokenBundle
import pytest
from homeassistant.components.nexblue.const import CONF_REFRESH_TOKEN, DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_json_object_fixture
CHARGER = Charger.from_api(load_json_object_fixture("charger.json", DOMAIN))
CHARGER_STATUS = ChargerStatus.from_api(
CHARGER.serial_number, load_json_object_fixture("charger_status.json", DOMAIN)
)
TOKEN = TokenBundle.from_api(load_json_object_fixture("token.json", DOMAIN))
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a NexBlue config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="NexBlue (user@example.com)",
unique_id=TOKEN.account_id,
data={
CONF_USERNAME: "user@example.com",
CONF_PASSWORD: "password",
CONF_REFRESH_TOKEN: TOKEN.refresh_token,
},
)
@pytest.fixture
def mock_setup_entry() -> Generator[None]:
"""Prevent a config-flow test from setting up the integration."""
with patch("homeassistant.components.nexblue.async_setup_entry", return_value=True):
yield
@pytest.fixture
def mock_client() -> Generator[MagicMock]:
"""Return a mocked NexBlue API client."""
with (
patch(
"homeassistant.components.nexblue.NexBlueClient", autospec=True
) as client_mock,
patch(
"homeassistant.components.nexblue.config_flow.NexBlueClient",
new=client_mock,
),
):
client = client_mock.return_value
client.async_login.return_value = TOKEN
client.async_ensure_access_token.return_value = None
client.async_list_chargers.return_value = [CHARGER]
client.async_get_charger_status.return_value = CHARGER_STATUS
yield client
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> MockConfigEntry:
"""Set up the NexBlue integration."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry
tests/components/nexblue/fixtures/charger.json
+3
View File
@@ -0,0 +1,3 @@
{
"serial_number": "NB123456"
}
tests/components/nexblue/fixtures/charger_status.json
+20
View File
@@ -0,0 +1,20 @@
{
"protocol_version": "00.16.00",
"charging_state": 2,
"power": 7.2,
"energy": 1.5,
"lifetime_energy": 42.0,
"is_lock": true,
"network_status": 1,
"is_disable": false,
"cable_current_limit": 32,
"circuit_fuse": 32,
"current_limit": 16,
"cable_lock_mode": 0,
"access_level": 0,
"phase_charging": 1,
"brightness": 100,
"uk_reg": null,
"current_list": [0.05, 0.0, 0.0],
"voltage_list": [224, 0, 0]
}
tests/components/nexblue/fixtures/token.json
+5
View File
@@ -0,0 +1,5 @@
{
"access_token": "eyJhbGciOiJub25lIn0.eyJzdWIiOiIwMDAwMDAwMC0wMDAwLTAwMDAtMDAwMC0wMDAwMDAwMDAwMDEifQ.signature",
"refresh_token": "refresh-token",
"expires_in": 3600
}
tests/components/nexblue/snapshots/test_sensor.ambr
+1067
View File
File diff suppressed because it is too large Load Diff
tests/components/nexblue/test_config_flow.py
+236
View File
@@ -0,0 +1,236 @@
"""Tests for the NexBlue config flow."""
from unittest.mock import MagicMock
from nexblue_api import NexBlueAuthError, NexBlueConnectionError, NexBlueError
from nexblue_api.models import TokenBundle
import pytest
from homeassistant.components.nexblue.const import CONF_REFRESH_TOKEN, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import TOKEN
from tests.common import MockConfigEntry
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
@pytest.mark.usefixtures("mock_client")
async def test_user_flow(hass: HomeAssistant) -> None:
"""Test the full happy-path user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "User@Example.com",
CONF_PASSWORD: "password",
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
entry = result["result"]
assert entry.title == "NexBlue (User@Example.com)"
assert entry.unique_id == TOKEN.account_id
assert entry.data == {
CONF_USERNAME: "User@Example.com",
CONF_PASSWORD: "password",
CONF_REFRESH_TOKEN: TOKEN.refresh_token,
}
@pytest.mark.parametrize(
("side_effect", "expected_error"),
[
(NexBlueAuthError, "invalid_auth"),
(NexBlueConnectionError, "cannot_connect"),
(NexBlueError, "unknown"),
(Exception, "unknown"),
],
)
async def test_user_flow_errors(
hass: HomeAssistant,
mock_client: MagicMock,
side_effect: type[Exception],
expected_error: str,
) -> None:
"""Test the user flow can recover after an error."""
mock_client.async_login.side_effect = [side_effect, TOKEN]
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "user@example.com",
CONF_PASSWORD: "incorrect-password",
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": expected_error}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "user@example.com",
CONF_PASSWORD: "correct-password",
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.usefixtures("mock_client")
async def test_user_flow_duplicate_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test an account cannot be configured twice."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "USER@example.com",
CONF_PASSWORD: "password",
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_user_flow_retries_with_corrected_credentials(
hass: HomeAssistant,
mock_client: MagicMock,
) -> None:
"""Test the user can correct credentials without restarting the flow."""
mock_client.async_login.side_effect = [NexBlueAuthError, TOKEN]
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "user@example.com",
CONF_PASSWORD: "incorrect-password",
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "user@example.com",
CONF_PASSWORD: "correct-password",
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_user_flow_rejects_login_without_refresh_token(
hass: HomeAssistant,
mock_client: MagicMock,
) -> None:
"""Test the user flow can recover when a login response lacks a refresh token."""
mock_client.async_login.side_effect = [
TokenBundle(
access_token="access-token",
refresh_token=None,
expires_in=3600,
account_id="00000000-0000-0000-0000-000000000001",
),
TOKEN,
]
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "user@example.com",
CONF_PASSWORD: "password",
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "user@example.com",
CONF_PASSWORD: "password",
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_user_flow_rejects_login_without_account_id(
hass: HomeAssistant,
mock_client: MagicMock,
) -> None:
"""Test the user flow can recover when a login response lacks an account ID."""
mock_client.async_login.side_effect = [
TokenBundle(
access_token="access-token",
refresh_token="refresh-token",
expires_in=3600,
),
TOKEN,
]
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "user@example.com",
CONF_PASSWORD: "password",
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "unknown"}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_USERNAME: "user@example.com",
CONF_PASSWORD: "password",
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
tests/components/nexblue/test_init.py
+131
View File
@@ -0,0 +1,131 @@
"""Tests for NexBlue integration setup."""
from unittest.mock import MagicMock
from nexblue_api import NexBlueAuthError, NexBlueConnectionError
from nexblue_api.models import TokenBundle
import pytest
from homeassistant.components.nexblue.const import CONF_REFRESH_TOKEN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_setup_entry(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test a config entry sets up coordinator and sensors."""
assert init_integration.state is ConfigEntryState.LOADED
mock_client.async_ensure_access_token.assert_awaited_once()
mock_client.async_list_chargers.assert_awaited_once()
mock_client.async_get_charger_status.assert_awaited_once()
async def test_setup_entry_recovers_from_invalid_refresh_token(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test a saved password recovers an expired refresh token once."""
mock_client.async_ensure_access_token.side_effect = NexBlueAuthError
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_client.async_login.assert_awaited_once()
async def test_setup_entry_persists_rotated_refresh_token(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test setup persists a rotated refresh token."""
mock_client.async_ensure_access_token.return_value = TokenBundle(
access_token="access-token",
refresh_token="rotated-refresh-token",
expires_in=3600,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_config_entry.data[CONF_REFRESH_TOKEN] == "rotated-refresh-token"
@pytest.mark.parametrize(
("refresh_error", "login_error", "expected_state"),
[
(
NexBlueAuthError,
NexBlueAuthError,
ConfigEntryState.SETUP_ERROR,
),
(
NexBlueConnectionError,
None,
ConfigEntryState.SETUP_RETRY,
),
],
)
async def test_setup_entry_failure(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
refresh_error: type[Exception],
login_error: type[Exception] | None,
expected_state: ConfigEntryState,
) -> None:
"""Test setup does not load when NexBlue cannot authenticate or connect."""
mock_client.async_ensure_access_token.side_effect = refresh_error
mock_client.async_login.side_effect = login_error
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is expected_state
async def test_setup_entry_fails_when_fallback_login_has_no_refresh_token(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test fallback login without a refresh token stops setup."""
mock_client.async_ensure_access_token.side_effect = NexBlueAuthError
mock_client.async_login.return_value = TokenBundle(
access_token="access-token",
refresh_token=None,
expires_in=3600,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
mock_client.async_login.assert_awaited_once()
async def test_setup_entry_retries_when_charger_status_request_fails(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test a charger-status connection error retries setup."""
mock_client.async_get_charger_status.side_effect = NexBlueConnectionError
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
tests/components/nexblue/test_sensor.py
+94
View File
@@ -0,0 +1,94 @@
"""Tests for NexBlue sensors."""
from dataclasses import replace
from datetime import timedelta
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
from nexblue_api import NexBlueConnectionError, NexBlueDeviceOfflineError, NexBlueError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .conftest import CHARGER, CHARGER_STATUS
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
async def test_sensor_entities_snapshot(
hass: HomeAssistant,
init_integration: MockConfigEntry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test the complete NexBlue sensor platform through a snapshot."""
await snapshot_platform(
hass,
entity_registry,
snapshot,
init_integration.entry_id,
)
async def test_missing_phase_values_are_unknown(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
) -> None:
"""Test missing phase measurements are exposed as unknown."""
mock_client.async_get_charger_status.return_value = replace(
CHARGER_STATUS,
current_a=(16.0,),
voltage_v=(230,),
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("sensor.nb123456_current_l1").state == "16.0"
assert hass.states.get("sensor.nb123456_current_l2").state == STATE_UNKNOWN
assert hass.states.get("sensor.nb123456_voltage_l1").state == "230"
assert hass.states.get("sensor.nb123456_voltage_l3").state == STATE_UNKNOWN
@pytest.mark.parametrize("error", [NexBlueDeviceOfflineError, NexBlueError])
async def test_charger_error_does_not_block_other_chargers(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: MagicMock,
error: type[Exception],
) -> None:
"""Test a single charger error does not prevent other chargers updating."""
second_charger = type(CHARGER)(serial_number="NB654321")
mock_client.async_list_chargers.return_value = [CHARGER, second_charger]
mock_client.async_get_charger_status.side_effect = [
CHARGER_STATUS,
error,
]
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("sensor.nb123456_charging_state").state == "charging"
assert hass.states.get("sensor.nb654321_charging_state").state == STATE_UNAVAILABLE
async def test_sensors_unavailable_when_coordinator_update_fails(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_client: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test a failed coordinator update makes all charger entities unavailable."""
mock_client.async_list_chargers.side_effect = NexBlueConnectionError
freezer.tick(timedelta(minutes=1))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("sensor.nb123456_charging_state").state == STATE_UNAVAILABLE