Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-20 02:48:57 +00:00
Code Activity

Add Portainer integration (#142875)

Browse Source 
Co-authored-by: Manu <4445816+tr4nt0r@users.noreply.github.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Erwin Douna
2025-09-10 21:01:41 +02:00
committed by GitHub
parent 7d471f9624
commit b496637bdd
24 changed files with 1638 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@@ -402,6 +402,7 @@ homeassistant.components.person.*
homeassistant.components.pi_hole.* homeassistant.components.pi_hole.*
homeassistant.components.ping.* homeassistant.components.ping.*
homeassistant.components.plugwise.* homeassistant.components.plugwise.*
homeassistant.components.portainer.*
homeassistant.components.powerfox.* homeassistant.components.powerfox.*
homeassistant.components.powerwall.* homeassistant.components.powerwall.*
homeassistant.components.private_ble_device.* homeassistant.components.private_ble_device.*

2
CODEOWNERS generated
View File

@@ -1191,6 +1191,8 @@ build.json @home-assistant/supervisor
/tests/components/pooldose/ @lmaertin /tests/components/pooldose/ @lmaertin
/homeassistant/components/poolsense/ @haemishkyd /homeassistant/components/poolsense/ @haemishkyd
/tests/components/poolsense/ @haemishkyd /tests/components/poolsense/ @haemishkyd
/homeassistant/components/portainer/ @erwindouna
/tests/components/portainer/ @erwindouna
/homeassistant/components/powerfox/ @klaasnicolaas /homeassistant/components/powerfox/ @klaasnicolaas
/tests/components/powerfox/ @klaasnicolaas /tests/components/powerfox/ @klaasnicolaas
/homeassistant/components/powerwall/ @bdraco @jrester @daniel-simpson /homeassistant/components/powerwall/ @bdraco @jrester @daniel-simpson

40
homeassistant/components/portainer/__init__.py Normal file
View File

@@ -0,0 +1,40 @@
"""The Portainer integration."""
from __future__ import annotations
from pyportainer import Portainer
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_HOST, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .coordinator import PortainerCoordinator
_PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR]
type PortainerConfigEntry = ConfigEntry[PortainerCoordinator]
async def async_setup_entry(hass: HomeAssistant, entry: PortainerConfigEntry) -> bool:
"""Set up Portainer from a config entry."""
session = async_create_clientsession(hass)
client = Portainer(
api_url=entry.data[CONF_HOST],
api_key=entry.data[CONF_API_KEY],
session=session,
)
coordinator = PortainerCoordinator(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: PortainerConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)

146
homeassistant/components/portainer/binary_sensor.py Normal file
View File

@@ -0,0 +1,146 @@
"""Binary sensor platform for Portainer."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from pyportainer.models.docker import DockerContainer
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import PortainerConfigEntry
from .coordinator import PortainerCoordinator
from .entity import (
PortainerContainerEntity,
PortainerCoordinatorData,
PortainerEndpointEntity,
)
@dataclass(frozen=True, kw_only=True)
class PortainerBinarySensorEntityDescription(BinarySensorEntityDescription):
"""Class to hold Portainer binary sensor description."""
state_fn: Callable[[Any], bool]
CONTAINER_SENSORS: tuple[PortainerBinarySensorEntityDescription, ...] = (
PortainerBinarySensorEntityDescription(
key="status",
translation_key="status",
state_fn=lambda data: data.state == "running",
device_class=BinarySensorDeviceClass.RUNNING,
entity_category=EntityCategory.DIAGNOSTIC,
),
)
ENDPOINT_SENSORS: tuple[PortainerBinarySensorEntityDescription, ...] = (
PortainerBinarySensorEntityDescription(
key="status",
translation_key="status",
state_fn=lambda data: data.endpoint.status == 1, # 1 = Running | 2 = Stopped
device_class=BinarySensorDeviceClass.RUNNING,
entity_category=EntityCategory.DIAGNOSTIC,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: PortainerConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Portainer binary sensors."""
coordinator = entry.runtime_data
entities: list[BinarySensorEntity] = []
for endpoint in coordinator.data.values():
entities.extend(
PortainerEndpointSensor(
coordinator,
entity_description,
endpoint,
)
for entity_description in ENDPOINT_SENSORS
)
entities.extend(
PortainerContainerSensor(
coordinator,
entity_description,
container,
endpoint,
)
for container in endpoint.containers.values()
for entity_description in CONTAINER_SENSORS
)
async_add_entities(entities)
class PortainerEndpointSensor(PortainerEndpointEntity, BinarySensorEntity):
"""Representation of a Portainer endpoint binary sensor entity."""
entity_description: PortainerBinarySensorEntityDescription
def __init__(
self,
coordinator: PortainerCoordinator,
entity_description: PortainerBinarySensorEntityDescription,
device_info: PortainerCoordinatorData,
) -> None:
"""Initialize Portainer endpoint binary sensor entity."""
self.entity_description = entity_description
super().__init__(device_info, coordinator)
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{device_info.id}_{entity_description.key}"
@property
def available(self) -> bool:
"""Return if the device is available."""
return super().available and self.device_id in self.coordinator.data
@property
def is_on(self) -> bool | None:
"""Return true if the binary sensor is on."""
return self.entity_description.state_fn(self.coordinator.data[self.device_id])
class PortainerContainerSensor(PortainerContainerEntity, BinarySensorEntity):
"""Representation of a Portainer container sensor."""
entity_description: PortainerBinarySensorEntityDescription
def __init__(
self,
coordinator: PortainerCoordinator,
entity_description: PortainerBinarySensorEntityDescription,
device_info: DockerContainer,
via_device: PortainerCoordinatorData,
) -> None:
"""Initialize the Portainer container sensor."""
self.entity_description = entity_description
super().__init__(device_info, coordinator, via_device)
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{device_info.id}_{entity_description.key}"
@property
def available(self) -> bool:
"""Return if the device is available."""
return super().available and self.endpoint_id in self.coordinator.data
@property
def is_on(self) -> bool | None:
"""Return true if the binary sensor is on."""
return self.entity_description.state_fn(
self.coordinator.data[self.endpoint_id].containers[self.device_id]
)

95
homeassistant/components/portainer/config_flow.py Normal file
View File

@@ -0,0 +1,95 @@
"""Config flow for the portainer integration."""
from __future__ import annotations
import logging
from typing import Any
from pyportainer import (
Portainer,
PortainerAuthenticationError,
PortainerConnectionError,
PortainerTimeoutError,
)
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_KEY, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_API_KEY): str,
}
)
async def _validate_input(hass: HomeAssistant, data: dict[str, Any]) -> None:
"""Validate the user input allows us to connect."""
client = Portainer(
api_url=data[CONF_HOST],
api_key=data[CONF_API_KEY],
session=async_get_clientsession(hass),
)
try:
await client.get_endpoints()
except PortainerAuthenticationError:
raise InvalidAuth from None
except PortainerConnectionError as err:
raise CannotConnect from err
except PortainerTimeoutError as err:
raise PortainerTimeout from err
_LOGGER.debug("Connected to Portainer API: %s", data[CONF_HOST])
class PortainerConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Portainer."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
try:
await _validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
except PortainerTimeout:
errors["base"] = "timeout_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(user_input[CONF_API_KEY])
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input[CONF_HOST], data=user_input
)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""
class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""
class PortainerTimeout(HomeAssistantError):
"""Error to indicate a timeout occurred."""

4
homeassistant/components/portainer/const.py Normal file
View File

@@ -0,0 +1,4 @@
"""Constants for the Portainer integration."""
DOMAIN = "portainer"
DEFAULT_NAME = "Portainer"

137
homeassistant/components/portainer/coordinator.py Normal file
View File

@@ -0,0 +1,137 @@
"""Data Update Coordinator for Portainer."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from pyportainer import (
Portainer,
PortainerAuthenticationError,
PortainerConnectionError,
PortainerTimeoutError,
)
from pyportainer.models.docker import DockerContainer
from pyportainer.models.portainer import Endpoint
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
type PortainerConfigEntry = ConfigEntry[PortainerCoordinator]
_LOGGER = logging.getLogger(__name__)
DEFAULT_SCAN_INTERVAL = timedelta(seconds=60)
@dataclass
class PortainerCoordinatorData:
"""Data class for Portainer Coordinator."""
id: int
name: str | None
endpoint: Endpoint
containers: dict[str, DockerContainer]
class PortainerCoordinator(DataUpdateCoordinator[dict[int, PortainerCoordinatorData]]):
"""Data Update Coordinator for Portainer."""
config_entry: PortainerConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: PortainerConfigEntry,
portainer: Portainer,
) -> None:
"""Initialize the Portainer Data Update Coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=DOMAIN,
update_interval=DEFAULT_SCAN_INTERVAL,
)
self.portainer = portainer
async def _async_setup(self) -> None:
"""Set up the Portainer Data Update Coordinator."""
try:
await self.portainer.get_endpoints()
except PortainerAuthenticationError as err:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="invalid_auth",
translation_placeholders={"error": repr(err)},
) from err
except PortainerConnectionError as err:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": repr(err)},
) from err
except PortainerTimeoutError as err:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="timeout_connect",
translation_placeholders={"error": repr(err)},
) from err
async def _async_update_data(self) -> dict[int, PortainerCoordinatorData]:
"""Fetch data from Portainer API."""
_LOGGER.debug(
"Fetching data from Portainer API: %s", self.config_entry.data[CONF_HOST]
)
try:
endpoints = await self.portainer.get_endpoints()
except PortainerAuthenticationError as err:
_LOGGER.error("Authentication error: %s", repr(err))
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="invalid_auth",
translation_placeholders={"error": repr(err)},
) from err
except PortainerConnectionError as err:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": repr(err)},
) from err
else:
_LOGGER.debug("Fetched endpoints: %s", endpoints)
mapped_endpoints: dict[int, PortainerCoordinatorData] = {}
for endpoint in endpoints:
try:
containers = await self.portainer.get_containers(endpoint.id)
except PortainerConnectionError as err:
_LOGGER.exception("Connection error")
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": repr(err)},
) from err
except PortainerAuthenticationError as err:
_LOGGER.exception("Authentication error")
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="invalid_auth",
translation_placeholders={"error": repr(err)},
) from err
mapped_endpoints[endpoint.id] = PortainerCoordinatorData(
id=endpoint.id,
name=endpoint.name,
endpoint=endpoint,
containers={container.id: container for container in containers},
)
return mapped_endpoints

73
homeassistant/components/portainer/entity.py Normal file
View File

@@ -0,0 +1,73 @@
"""Base class for Portainer entities."""
from pyportainer.models.docker import DockerContainer
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DEFAULT_NAME, DOMAIN
from .coordinator import PortainerCoordinator, PortainerCoordinatorData
class PortainerCoordinatorEntity(CoordinatorEntity[PortainerCoordinator]):
"""Base class for Portainer entities."""
_attr_has_entity_name = True
class PortainerEndpointEntity(PortainerCoordinatorEntity):
"""Base implementation for Portainer endpoint."""
def __init__(
self,
device_info: PortainerCoordinatorData,
coordinator: PortainerCoordinator,
) -> None:
"""Initialize a Portainer endpoint."""
super().__init__(coordinator)
self._device_info = device_info
self.device_id = device_info.endpoint.id
self._attr_device_info = DeviceInfo(
identifiers={
(DOMAIN, f"{coordinator.config_entry.entry_id}_{self.device_id}")
},
manufacturer=DEFAULT_NAME,
model="Endpoint",
name=device_info.endpoint.name,
)
class PortainerContainerEntity(PortainerCoordinatorEntity):
"""Base implementation for Portainer container."""
def __init__(
self,
device_info: DockerContainer,
coordinator: PortainerCoordinator,
via_device: PortainerCoordinatorData,
) -> None:
"""Initialize a Portainer container."""
super().__init__(coordinator)
self._device_info = device_info
self.device_id = self._device_info.id
self.endpoint_id = via_device.endpoint.id
device_name = (
self._device_info.names[0].replace("/", " ").strip()
if self._device_info.names
else None
)
self._attr_device_info = DeviceInfo(
identifiers={
(DOMAIN, f"{self.coordinator.config_entry.entry_id}_{self.device_id}")
},
manufacturer=DEFAULT_NAME,
model="Container",
name=device_name,
via_device=(
DOMAIN,
f"{self.coordinator.config_entry.entry_id}_{self.endpoint_id}",
),
translation_key=None if device_name else "unknown_container",
)

10
homeassistant/components/portainer/manifest.json Normal file
View File

@@ -0,0 +1,10 @@
{
"domain": "portainer",
"name": "Portainer",
"codeowners": ["@erwindouna"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/portainer",
"iot_class": "local_polling",
"quality_scale": "bronze",
"requirements": ["pyportainer==0.1.7"]
}

80
homeassistant/components/portainer/quality_scale.yaml Normal file
View File

@@ -0,0 +1,80 @@
rules:
# Bronze
action-setup: done
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
No custom actions are defined.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: |
No explicit event subscriptions.
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:
status: exempt
comment: |
No custom actions are defined.
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates:
status: exempt
comment: |
No explicit parallel updates are defined.
reauthentication-flow:
status: todo
comment: |
No reauthentication flow is defined. It will be done in a next iteration.
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: |
No discovery is implemented, since it's software based.
discovery:
status: exempt
comment: |
No discovery is implemented, since it's software based.
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: todo
entity-device-class: todo
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: done
strict-typing: done

49
homeassistant/components/portainer/strings.json Normal file
View File

@@ -0,0 +1,49 @@
{
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"api_key": "[%key:common::config_flow::data::api_key%]"
},
"data_description": {
"host": "The host/URL, including the port, of your Portainer instance",
"api_key": "The API key for authenticating with Portainer"
},
"description": "You can create an API key in the Portainer UI. Go to **My account > API keys** and select **Add API key**"
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"timeout_connect": "[%key:common::config_flow::error::timeout_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"device": {
"unknown_container": {
"name": "Unknown container"
}
},
"entity": {
"binary_sensor": {
"status": {
"name": "Status"
}
}
},
"exceptions": {
"cannot_connect": {
"message": "An error occurred while trying to connect to the Portainer instance: {error}"
},
"invalid_auth": {
"message": "An error occurred while trying authenticate: {error}"
},
"timeout_connect": {
"message": "A timeout occurred while trying to connect to the Portainer instance: {error}"
}
}
}

1
homeassistant/generated/config_flows.py generated
View File

@@ -495,6 +495,7 @@ FLOWS = {
"point", "point",
"pooldose", "pooldose",
"poolsense", "poolsense",
"portainer",
"powerfox", "powerfox",
"powerwall", "powerwall",
"private_ble_device", "private_ble_device",

6
homeassistant/generated/integrations.json
View File

@@ -5072,6 +5072,12 @@
"config_flow": true, "config_flow": true,
"iot_class": "cloud_polling" "iot_class": "cloud_polling"
}, },
"portainer": {
"name": "Portainer",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling"
},
"portlandgeneral": { "portlandgeneral": {
"name": "Portland General Electric (PGE)", "name": "Portland General Electric (PGE)",
"integration_type": "virtual", "integration_type": "virtual",

10
mypy.ini generated
View File

@@ -3776,6 +3776,16 @@ disallow_untyped_defs = true
warn_return_any = true warn_return_any = true
warn_unreachable = true warn_unreachable = true
[mypy-homeassistant.components.portainer.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.powerfox.*] [mypy-homeassistant.components.powerfox.*]
check_untyped_defs = true check_untyped_defs = true
disallow_incomplete_defs = true disallow_incomplete_defs = true

3
requirements_all.txt generated
View File

@@ -2266,6 +2266,9 @@ pyplaato==0.0.19
# homeassistant.components.point # homeassistant.components.point
pypoint==3.0.0 pypoint==3.0.0
# homeassistant.components.portainer
pyportainer==0.1.7
# homeassistant.components.probe_plus # homeassistant.components.probe_plus
pyprobeplus==1.0.1 pyprobeplus==1.0.1

3
requirements_test_all.txt generated
View File

@@ -1890,6 +1890,9 @@ pyplaato==0.0.19
# homeassistant.components.point # homeassistant.components.point
pypoint==3.0.0 pypoint==3.0.0
# homeassistant.components.portainer
pyportainer==0.1.7
# homeassistant.components.probe_plus # homeassistant.components.probe_plus
pyprobeplus==1.0.1 pyprobeplus==1.0.1

13
tests/components/portainer/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
"""Tests for the Portainer integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Fixture for setting up the component."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

63
tests/components/portainer/conftest.py Normal file
View File

@@ -0,0 +1,63 @@
"""Common fixtures for the portainer tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from pyportainer.models.docker import DockerContainer
from pyportainer.models.portainer import Endpoint
import pytest
from homeassistant.components.portainer.const import DOMAIN
from homeassistant.const import CONF_API_KEY, CONF_HOST
from tests.common import MockConfigEntry, load_json_array_fixture
MOCK_TEST_CONFIG = {
CONF_HOST: "https://127.0.0.1:9000/",
CONF_API_KEY: "test_api_key",
}
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.portainer.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_portainer_client() -> Generator[AsyncMock]:
"""Mock Portainer client with dynamic exception injection support."""
with (
patch(
"homeassistant.components.portainer.Portainer", autospec=True
) as mock_client,
patch(
"homeassistant.components.portainer.config_flow.Portainer", new=mock_client
),
):
client = mock_client.return_value
client.get_endpoints.return_value = [
Endpoint.from_dict(endpoint)
for endpoint in load_json_array_fixture("endpoints.json", DOMAIN)
]
client.get_containers.return_value = [
DockerContainer.from_dict(container)
for container in load_json_array_fixture("containers.json", DOMAIN)
]
yield client
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock a config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="Portainer test",
data=MOCK_TEST_CONFIG,
entry_id="portainer_test_entry_123",
)

166
tests/components/portainer/fixtures/containers.json Normal file
View File

@@ -0,0 +1,166 @@
[
{
"Id": "aa86eacfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf",
"Names": ["/funny_chatelet"],
"Image": "docker.io/library/ubuntu:latest",
"ImageID": "sha256:72297848456d5d37d1262630108ab308d3e9ec7ed1c3286a32fe09856619a782",
"ImageManifestDescriptor": {
"mediaType": "application/vnd.oci.image.manifest.v1+json",
"digest": "sha256:c0537ff6a5218ef531ece93d4984efc99bbf3f7497c0a7726c88e2bb7584dc96",
"size": 424,
"urls": ["http://example.com"],
"annotations": {
"com.docker.official-images.bashbrew.arch": "amd64",
"org.opencontainers.image.base.digest": "sha256:0d0ef5c914d3ea700147da1bd050c59edb8bb12ca312f3800b29d7c8087eabd8",
"org.opencontainers.image.base.name": "scratch",
"org.opencontainers.image.created": "2025-01-27T00:00:00Z",
"org.opencontainers.image.revision": "9fabb4bad5138435b01857e2fe9363e2dc5f6a79",
"org.opencontainers.image.source": "https://git.launchpad.net/cloud-images/+oci/ubuntu-base",
"org.opencontainers.image.url": "https://hub.docker.com/_/ubuntu",
"org.opencontainers.image.version": "24.04"
},
"data": null,
"platform": {
"architecture": "arm",
"os": "windows",
"os.version": "10.0.19041.1165",
"os.features": ["win32k"],
"variant": "v7"
},
"artifactType": null
},
"Command": "/bin/bash",
"Created": "1739811096",
"Ports": [
{
"PrivatePort": 8080,
"PublicPort": 80,
"Type": "tcp"
}
],
"SizeRw": "122880",
"SizeRootFs": "1653948416",
"Labels": {
"com.example.vendor": "Acme",
"com.example.license": "GPL",
"com.example.version": "1.0"
},
"State": "running",
"Status": "Up 4 days",
"HostConfig": {
"NetworkMode": "mynetwork",
"Annotations": {
"io.kubernetes.docker.type": "container",
"io.kubernetes.sandbox.id": "3befe639bed0fd6afdd65fd1fa84506756f59360ec4adc270b0fdac9be22b4d3"
}
},
"NetworkSettings": {
"Networks": {
"property1": {
"IPAMConfig": {
"IPv4Address": "172.20.30.33",
"IPv6Address": "2001:db8:abcd::3033",
"LinkLocalIPs": ["169.254.34.68", "fe80::3468"]
},
"Links": ["container_1", "container_2"],
"MacAddress": "02:42:ac:11:00:04",
"Aliases": ["server_x", "server_y"],
"DriverOpts": {
"com.example.some-label": "some-value",
"com.example.some-other-label": "some-other-value"
},
"GwPriority": [10],
"NetworkID": "08754567f1f40222263eab4102e1c733ae697e8e354aa9cd6e18d7402835292a",
"EndpointID": "b88f5b905aabf2893f3cbc4ee42d1ea7980bbc0a92e2c8922b1e1795298afb0b",
"Gateway": "172.17.0.1",
"IPAddress": "172.17.0.4",
"IPPrefixLen": 16,
"IPv6Gateway": "2001:db8:2::100",
"GlobalIPv6Address": "2001:db8::5689",
"GlobalIPv6PrefixLen": 64,
"DNSNames": ["foobar", "server_x", "server_y", "my.ctr"]
}
}
},
"Mounts": [
{
"Type": "volume",
"Name": "myvolume",
"Source": "/var/lib/docker/volumes/myvolume/_data",
"Destination": "/usr/share/nginx/html/",
"Driver": "local",
"Mode": "z",
"RW": true,
"Propagation": ""
}
]
},
{
"Id": "bb97facfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf",
"Names": ["/serene_banach"],
"Image": "docker.io/library/nginx:latest",
"ImageID": "sha256:3f8a4339a5c5d5d37d1262630108ab308d3e9ec7ed1c3286a32fe09856619a782",
"Command": "nginx -g 'daemon off;'",
"Created": "1739812096",
"Ports": [
{
"PrivatePort": 80,
"PublicPort": 8081,
"Type": "tcp"
}
],
"State": "running",
"Status": "Up 2 days"
},
{
"Id": "cc08facfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf",
"Names": ["/stoic_turing"],
"Image": "docker.io/library/postgres:15",
"ImageID": "sha256:4f8a4339a5c5d5d37d1262630108ab308d3e9ec7ed1c3286a32fe09856619a782",
"Command": "postgres",
"Created": "1739813096",
"Ports": [
{
"PrivatePort": 5432,
"PublicPort": 5432,
"Type": "tcp"
}
],
"State": "running",
"Status": "Up 1 day"
},
{
"Id": "dd19facfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf",
"Names": ["/focused_einstein"],
"Image": "docker.io/library/redis:7",
"ImageID": "sha256:5f8a4339a5c5d5d37d1262630108ab308d3e9ec7ed1c3286a32fe09856619a782",
"Command": "redis-server",
"Created": "1739814096",
"Ports": [
{
"PrivatePort": 6379,
"PublicPort": 6379,
"Type": "tcp"
}
],
"State": "running",
"Status": "Up 12 hours"
},
{
"Id": "ee20facfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf",
"Names": ["/practical_morse"],
"Image": "docker.io/library/python:3.13-slim",
"ImageID": "sha256:6f8a4339a5c5d5d37d1262630108ab308d3e9ec7ed1c3286a32fe09856619a782",
"Command": "python3 -m http.server",
"Created": "1739815096",
"Ports": [
{
"PrivatePort": 8000,
"PublicPort": 8000,
"Type": "tcp"
}
],
"State": "running",
"Status": "Up 6 hours"
}
]

195
tests/components/portainer/fixtures/endpoints.json Normal file
View File

@@ -0,0 +1,195 @@
[
{
"AMTDeviceGUID": "4c4c4544-004b-3910-8037-b6c04f504633",
"AuthorizedTeams": [1],
"AuthorizedUsers": [1],
"AzureCredentials": {
"ApplicationID": "eag7cdo9-o09l-9i83-9dO9-f0b23oe78db4",
"AuthenticationKey": "cOrXoK/1D35w8YQ8nH1/8ZGwzz45JIYD5jxHKXEQknk=",
"TenantID": "34ddc78d-4fel-2358-8cc1-df84c8o839f5"
},
"ComposeSyntaxMaxVersion": "3.8",
"ContainerEngine": "docker",
"EdgeCheckinInterval": 5,
"EdgeID": "string",
"EdgeKey": "string",
"EnableGPUManagement": true,
"Gpus": [
{
"name": "name",
"value": "value"
}
],
"GroupId": 1,
"Heartbeat": true,
"Id": 1,
"IsEdgeDevice": true,
"Kubernetes": {
"Configuration": {
"AllowNoneIngressClass": true,
"EnableResourceOverCommit": true,
"IngressAvailabilityPerNamespace": true,
"IngressClasses": [
{
"Blocked": true,
"BlockedNamespaces": ["string"],
"Name": "string",
"Type": "string"
}
],
"ResourceOverCommitPercentage": 0,
"RestrictDefaultNamespace": true,
"StorageClasses": [
{
"AccessModes": ["string"],
"AllowVolumeExpansion": true,
"Name": "string",
"Provisioner": "string"
}
],
"UseLoadBalancer": true,
"UseServerMetrics": true
},
"Flags": {
"IsServerIngressClassDetected": true,
"IsServerMetricsDetected": true,
"IsServerStorageDetected": true
},
"Snapshots": [
{
"DiagnosticsData": {
"DNS": {
"additionalProp1": "string",
"additionalProp2": "string",
"additionalProp3": "string"
},
"Log": "string",
"Proxy": {
"additionalProp1": "string",
"additionalProp2": "string",
"additionalProp3": "string"
},
"Telnet": {
"additionalProp1": "string",
"additionalProp2": "string",
"additionalProp3": "string"
}
},
"KubernetesVersion": "string",
"NodeCount": 0,
"Time": 0,
"TotalCPU": 0,
"TotalMemory": 0
}
]
},
"Name": "my-environment",
"PostInitMigrations": {
"MigrateGPUs": true,
"MigrateIngresses": true
},
"PublicURL": "docker.mydomain.tld:2375",
"Snapshots": [
{
"ContainerCount": 0,
"DiagnosticsData": {
"DNS": {
"additionalProp1": "string",
"additionalProp2": "string",
"additionalProp3": "string"
},
"Log": "string",
"Proxy": {
"additionalProp1": "string",
"additionalProp2": "string",
"additionalProp3": "string"
},
"Telnet": {
"additionalProp1": "string",
"additionalProp2": "string",
"additionalProp3": "string"
}
},
"DockerSnapshotRaw": {},
"DockerVersion": "string",
"GpuUseAll": true,
"GpuUseList": ["string"],
"HealthyContainerCount": 0,
"ImageCount": 0,
"IsPodman": true,
"NodeCount": 0,
"RunningContainerCount": 0,
"ServiceCount": 0,
"StackCount": 0,
"StoppedContainerCount": 0,
"Swarm": true,
"Time": 0,
"TotalCPU": 0,
"TotalMemory": 0,
"UnhealthyContainerCount": 0,
"VolumeCount": 0
}
],
"Status": 1,
"TLS": true,
"TLSCACert": "string",
"TLSCert": "string",
"TLSConfig": {
"TLS": true,
"TLSCACert": "/data/tls/ca.pem",
"TLSCert": "/data/tls/cert.pem",
"TLSKey": "/data/tls/key.pem",
"TLSSkipVerify": false
},
"TLSKey": "string",
"TagIds": [1],
"Tags": ["string"],
"TeamAccessPolicies": {
"additionalProp1": {
"RoleId": 1
},
"additionalProp2": {
"RoleId": 1
},
"additionalProp3": {
"RoleId": 1
}
},
"Type": 1,
"URL": "docker.mydomain.tld:2375",
"UserAccessPolicies": {
"additionalProp1": {
"RoleId": 1
},
"additionalProp2": {
"RoleId": 1
},
"additionalProp3": {
"RoleId": 1
}
},
"UserTrusted": true,
"agent": {
"version": "1.0.0"
},
"edge": {
"CommandInterval": 60,
"PingInterval": 60,
"SnapshotInterval": 60,
"asyncMode": true
},
"lastCheckInDate": 0,
"queryDate": 0,
"securitySettings": {
"allowBindMountsForRegularUsers": false,
"allowContainerCapabilitiesForRegularUsers": true,
"allowDeviceMappingForRegularUsers": true,
"allowHostNamespaceForRegularUsers": true,
"allowPrivilegedModeForRegularUsers": false,
"allowStackManagementForRegularUsers": true,
"allowSysctlSettingForRegularUsers": true,
"allowVolumeBrowserForRegularUsers": true,
"enableHostManagementFeatures": true
}
}
]

295
tests/components/portainer/snapshots/test_binary_sensor.ambr Normal file
View File

@@ -0,0 +1,295 @@
# serializer version: 1
# name: test_all_entities[binary_sensor.focused_einstein_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.focused_einstein_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': None,
'original_name': 'Status',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'status',
'unique_id': 'portainer_test_entry_123_dd19facfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf_status',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.focused_einstein_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'focused_einstein Status',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.focused_einstein_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[binary_sensor.funny_chatelet_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.funny_chatelet_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': None,
'original_name': 'Status',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'status',
'unique_id': 'portainer_test_entry_123_aa86eacfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf_status',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.funny_chatelet_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'funny_chatelet Status',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.funny_chatelet_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[binary_sensor.my_environment_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.my_environment_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': None,
'original_name': 'Status',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'status',
'unique_id': 'portainer_test_entry_123_1_status',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.my_environment_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'my-environment Status',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.my_environment_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[binary_sensor.practical_morse_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.practical_morse_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': None,
'original_name': 'Status',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'status',
'unique_id': 'portainer_test_entry_123_ee20facfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf_status',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.practical_morse_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'practical_morse Status',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.practical_morse_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[binary_sensor.serene_banach_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.serene_banach_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': None,
'original_name': 'Status',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'status',
'unique_id': 'portainer_test_entry_123_bb97facfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf_status',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.serene_banach_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'serene_banach Status',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.serene_banach_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[binary_sensor.stoic_turing_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.stoic_turing_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.RUNNING: 'running'>,
'original_icon': None,
'original_name': 'Status',
'platform': 'portainer',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'status',
'unique_id': 'portainer_test_entry_123_cc08facfb3b3ed4cd362c1e88fc89a53908ad05fb3a4103bca3f9b28292d14bf_status',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[binary_sensor.stoic_turing_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'running',
'friendly_name': 'stoic_turing Status',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.stoic_turing_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

81
tests/components/portainer/test_binary_sensor.py Normal file
View File

@@ -0,0 +1,81 @@
"""Tests for the Portainer binary sensor platform."""
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
from pyportainer.exceptions import (
PortainerAuthenticationError,
PortainerConnectionError,
PortainerTimeoutError,
)
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.portainer.coordinator import DEFAULT_SCAN_INTERVAL
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from . import setup_integration
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_portainer_client: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test all entities."""
with patch(
"homeassistant.components.portainer._PLATFORMS",
[Platform.BINARY_SENSOR],
):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(
hass, entity_registry, snapshot, mock_config_entry.entry_id
)
@pytest.mark.parametrize(
("exception"),
[
PortainerAuthenticationError("bad creds"),
PortainerConnectionError("cannot connect"),
PortainerTimeoutError("timeout"),
],
)
async def test_refresh_exceptions(
hass: HomeAssistant,
mock_portainer_client: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
exception: Exception,
) -> None:
"""Test entities go unavailable after coordinator refresh failures."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_portainer_client.get_endpoints.side_effect = exception
freezer.tick(DEFAULT_SCAN_INTERVAL)
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.practical_morse_status")
assert state.state == STATE_UNAVAILABLE
# Reset endpoints; fail on containers fetch
mock_portainer_client.get_endpoints.side_effect = None
mock_portainer_client.get_containers.side_effect = exception
freezer.tick(DEFAULT_SCAN_INTERVAL)
async_fire_time_changed(hass, dt_util.utcnow())
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.practical_morse_status")
assert state.state == STATE_UNAVAILABLE

127
tests/components/portainer/test_config_flow.py Normal file
View File

@@ -0,0 +1,127 @@
"""Test the Portainer config flow."""
from unittest.mock import AsyncMock, MagicMock
from pyportainer.exceptions import (
PortainerAuthenticationError,
PortainerConnectionError,
PortainerTimeoutError,
)
import pytest
from homeassistant.components.portainer.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_API_KEY, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import MOCK_TEST_CONFIG
from tests.common import MockConfigEntry
MOCK_USER_SETUP = {
CONF_HOST: "https://127.0.0.1:9000/",
CONF_API_KEY: "test_api_key",
}
async def test_form(
hass: HomeAssistant,
mock_portainer_client: MagicMock,
) -> None:
"""Test we get the form."""
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=MOCK_USER_SETUP,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "https://127.0.0.1:9000/"
assert result["data"] == MOCK_TEST_CONFIG
@pytest.mark.parametrize(
("exception", "reason"),
[
(
PortainerAuthenticationError,
"invalid_auth",
),
(
PortainerConnectionError,
"cannot_connect",
),
(
PortainerTimeoutError,
"timeout_connect",
),
(
Exception("Some other error"),
"unknown",
),
],
)
async def test_form_exceptions(
hass: HomeAssistant,
mock_portainer_client: AsyncMock,
exception: Exception,
reason: str,
) -> None:
"""Test we handle all exceptions."""
mock_portainer_client.get_endpoints.side_effect = exception
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=MOCK_USER_SETUP,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": reason}
mock_portainer_client.get_endpoints.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=MOCK_USER_SETUP,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "https://127.0.0.1:9000/"
assert result["data"] == MOCK_TEST_CONFIG
async def test_duplicate_entry(
hass: HomeAssistant,
mock_portainer_client: AsyncMock,
mock_setup_entry: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test we handle duplicate entries."""
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=MOCK_USER_SETUP,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

38
tests/components/portainer/test_init.py Normal file
View File

@@ -0,0 +1,38 @@
"""Test the Portainer initial specific behavior."""
from unittest.mock import AsyncMock
from pyportainer.exceptions import (
PortainerAuthenticationError,
PortainerConnectionError,
PortainerTimeoutError,
)
import pytest
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("exception", "expected_state"),
[
(PortainerAuthenticationError("bad creds"), ConfigEntryState.SETUP_ERROR),
(PortainerConnectionError("cannot connect"), ConfigEntryState.SETUP_RETRY),
(PortainerTimeoutError("timeout"), ConfigEntryState.SETUP_RETRY),
],
)
async def test_setup_exceptions(
hass: HomeAssistant,
mock_portainer_client: AsyncMock,
mock_config_entry: MockConfigEntry,
exception: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test the _async_setup."""
mock_portainer_client.get_endpoints.side_effect = exception
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state == expected_state