jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 01:08:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Create and use config_entry fixture for Axis integration tests (#85865)

Browse Source
This commit is contained in:
Robert Svensson 2023-01-14 13:18:18 +01:00 committed by GitHub
parent 8fbcb93ab4
commit 768c3e163d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 153 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
tests/components/axis/conftest.py
View File

@ -6,8 +6,83 @@ from unittest.mock import patch
from axis.rtsp import Signal, State
import pytest
from homeassistant.components.axis.const import CONF_EVENTS, DOMAIN as AXIS_DOMAIN
from homeassistant.const import (
CONF_HOST,
CONF_MODEL,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
)
from tests.common import MockConfigEntry
from tests.components.light.conftest import mock_light_profiles # noqa: F401
MAC = "00408C123456"
FORMATTED_MAC = "00:40:8c:12:34:56"
MODEL = "model"
NAME = "name"
DEFAULT_HOST = "1.2.3.4"
ENTRY_OPTIONS = {CONF_EVENTS: True}
ENTRY_CONFIG = {
CONF_HOST: DEFAULT_HOST,
CONF_USERNAME: "root",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
CONF_MODEL: MODEL,
CONF_NAME: NAME,
}
@pytest.fixture(name="config_entry")
def config_entry_fixture(hass, config, options, config_entry_version):
"""Define a config entry fixture."""
entry = MockConfigEntry(
domain=AXIS_DOMAIN,
unique_id=FORMATTED_MAC,
data=config,
options=options,
version=config_entry_version,
)
entry.add_to_hass(hass)
return entry
@pytest.fixture(name="config_entry_version")
def config_entry_version_fixture(request):
"""Define a config entry version fixture.
@pytest.mark.config_entry_version(int)
"""
marker = request.node.get_closest_marker("config_entry_version")
version = 3
if marker:
version = marker.args[0]
return version
@pytest.fixture(name="config")
def config_fixture():
"""Define a config entry data fixture."""
return ENTRY_CONFIG.copy()
@pytest.fixture(name="options")
def options_fixture(request):
"""Define a config entry options fixture.
@pytest.mark.config_entry_options(dict)
"""
marker = request.node.get_closest_marker("config_entry_options")
options = ENTRY_OPTIONS.copy()
if marker:
options = marker.args[0]
return options
@pytest.fixture(autouse=True)
def mock_axis_rtspclient():

11
tests/components/axis/test_binary_sensor.py
View File

@ -8,7 +8,8 @@ from homeassistant.components.binary_sensor import (
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.setup import async_setup_component
from .test_device import NAME, setup_axis_integration
from .conftest import NAME
from .test_device import setup_axis_integration
async def test_platform_manually_configured(hass):
@ -25,16 +26,16 @@ async def test_platform_manually_configured(hass):
assert AXIS_DOMAIN not in hass.data
async def test_no_binary_sensors(hass):
async def test_no_binary_sensors(hass, config_entry):
"""Test that no sensors in Axis results in no sensor entities."""
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert not hass.states.async_entity_ids(BINARY_SENSOR_DOMAIN)
async def test_binary_sensors(hass, mock_rtsp_event):
async def test_binary_sensors(hass, config_entry, mock_rtsp_event):
"""Test that sensors are loaded properly."""
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
mock_rtsp_event(
topic="tns1:Device/tnsaxis:Sensor/PIR",

19
tests/components/axis/test_camera.py
View File

@ -2,6 +2,8 @@
from unittest.mock import patch
import pytest
from homeassistant.components import camera
from homeassistant.components.axis.const import (
CONF_STREAM_PROFILE,
@ -11,7 +13,8 @@ from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
from homeassistant.const import STATE_IDLE
from homeassistant.setup import async_setup_component
from .test_device import ENTRY_OPTIONS, NAME, setup_axis_integration
from .conftest import NAME
from .test_device import setup_axis_integration
async def test_platform_manually_configured(hass):
@ -26,9 +29,9 @@ async def test_platform_manually_configured(hass):
assert AXIS_DOMAIN not in hass.data
async def test_camera(hass):
async def test_camera(hass, config_entry):
"""Test that Axis camera platform is loaded properly."""
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert len(hass.states.async_entity_ids(CAMERA_DOMAIN)) == 1
@ -47,10 +50,10 @@ async def test_camera(hass):
)
async def test_camera_with_stream_profile(hass):
@pytest.mark.config_entry_options({CONF_STREAM_PROFILE: "profile_1"})
async def test_camera_with_stream_profile(hass, config_entry):
"""Test that Axis camera entity is using the correct path with stream profike."""
with patch.dict(ENTRY_OPTIONS, {CONF_STREAM_PROFILE: "profile_1"}):
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert len(hass.states.async_entity_ids(CAMERA_DOMAIN)) == 1
@ -72,9 +75,9 @@ async def test_camera_with_stream_profile(hass):
)
async def test_camera_disabled(hass):
async def test_camera_disabled(hass, config_entry):
"""Test that Axis camera platform is loaded properly but does not create camera entity."""
with patch("axis.vapix.vapix.Params.image_format", new=None):
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert len(hass.states.async_entity_ids(CAMERA_DOMAIN)) == 0

30
tests/components/axis/test_config_flow.py
View File

@ -32,14 +32,8 @@ from homeassistant.const import (
)
from homeassistant.data_entry_flow import FlowResultType
from .test_device import (
DEFAULT_HOST,
MAC,
MODEL,
NAME,
mock_default_vapix_requests,
setup_axis_integration,
)
from .conftest import DEFAULT_HOST, MAC, MODEL, NAME
from .test_device import mock_default_vapix_requests, setup_axis_integration
from tests.common import MockConfigEntry
@ -79,9 +73,9 @@ async def test_flow_manual_configuration(hass):
}
async def test_manual_configuration_update_configuration(hass):
async def test_manual_configuration_update_configuration(hass, config_entry):
"""Test that config flow fails on already configured device."""
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
result = await hass.config_entries.flow.async_init(
@ -211,9 +205,9 @@ async def test_flow_create_entry_multiple_existing_entries_of_same_model(hass):
assert result["data"][CONF_NAME] == "M1065-LW 2"
async def test_reauth_flow_update_configuration(hass):
async def test_reauth_flow_update_configuration(hass, config_entry):
"""Test that config flow fails on already configured device."""
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
result = await hass.config_entries.flow.async_init(
@ -383,10 +377,10 @@ async def test_discovery_flow(hass, source: str, discovery_info: dict):
],
)
async def test_discovered_device_already_configured(
hass, source: str, discovery_info: dict
hass, config_entry, source: str, discovery_info: dict
):
"""Test that discovery doesn't setup already configured devices."""
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert config_entry.data[CONF_HOST] == DEFAULT_HOST
result = await hass.config_entries.flow.async_init(
@ -439,10 +433,10 @@ async def test_discovered_device_already_configured(
],
)
async def test_discovery_flow_updated_configuration(
hass, source: str, discovery_info: dict, expected_port: int
hass, config_entry, source: str, discovery_info: dict, expected_port: int
):
"""Test that discovery flow update configuration with new parameters."""
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert config_entry.data == {
CONF_HOST: DEFAULT_HOST,
CONF_PORT: 80,
@ -573,9 +567,9 @@ async def test_discovery_flow_ignore_link_local_address(
assert result["reason"] == "link_local_address"
async def test_option_flow(hass):
async def test_option_flow(hass, config_entry):
"""Test config flow options."""
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
assert device.option_stream_profile == DEFAULT_STREAM_PROFILE
assert device.option_video_source == DEFAULT_VIDEO_SOURCE

77
tests/components/axis/test_device.py
View File

@ -8,41 +8,22 @@ import pytest
import respx
from homeassistant.components import axis, zeroconf
from homeassistant.components.axis.const import CONF_EVENTS, DOMAIN as AXIS_DOMAIN
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.config_entries import SOURCE_ZEROCONF
from homeassistant.const import (
CONF_HOST,
CONF_MODEL,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.helpers import device_registry as dr
from tests.common import MockConfigEntry, async_fire_mqtt_message
from .conftest import DEFAULT_HOST, ENTRY_CONFIG, FORMATTED_MAC, MAC, NAME
MAC = "00408C123456"
FORMATTED_MAC = "00:40:8c:12:34:56"
MODEL = "model"
NAME = "name"
DEFAULT_HOST = "1.2.3.4"
ENTRY_OPTIONS = {CONF_EVENTS: True}
ENTRY_CONFIG = {
CONF_HOST: DEFAULT_HOST,
CONF_USERNAME: "root",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
CONF_MODEL: MODEL,
CONF_NAME: NAME,
}
from tests.common import async_fire_mqtt_message
API_DISCOVERY_RESPONSE = {
"method": "getApiList",
@ -274,34 +255,22 @@ def mock_default_vapix_requests(respx: respx, host: str = DEFAULT_HOST) -> None:
respx.post(f"http://{host}:80/local/vmd/control.cgi").respond(json=VMD4_RESPONSE)
async def setup_axis_integration(
hass, config=ENTRY_CONFIG, options=ENTRY_OPTIONS, entry_version=3
):
async def setup_axis_integration(hass, config_entry):
"""Create the Axis device."""
config_entry = MockConfigEntry(
domain=AXIS_DOMAIN,
data=deepcopy(config),
options=deepcopy(options),
version=entry_version,
unique_id=FORMATTED_MAC,
)
config_entry.add_to_hass(hass)
with respx.mock:
mock_default_vapix_requests(respx)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
async def test_device_setup(hass):
async def test_device_setup(hass, config_entry):
"""Successful setup."""
with patch(
"homeassistant.config_entries.ConfigEntries.async_forward_entry_setup",
return_value=True,
) as forward_entry_setup:
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
assert device.api.vapix.firmware_version == "9.10.1"
@ -328,13 +297,13 @@ async def test_device_setup(hass):
assert device_entry.configuration_url == device.api.config.url
async def test_device_info(hass):
async def test_device_info(hass, config_entry):
"""Verify other path of device information works."""
api_discovery = deepcopy(API_DISCOVERY_RESPONSE)
api_discovery["data"]["apiList"].append(API_DISCOVERY_BASIC_DEVICE_INFO)
with patch.dict(API_DISCOVERY_RESPONSE, api_discovery):
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
assert device.api.vapix.firmware_version == "9.80.1"
@ -343,13 +312,13 @@ async def test_device_info(hass):
assert device.api.vapix.serial_number == "00408C123456"
async def test_device_support_mqtt(hass, mqtt_mock):
async def test_device_support_mqtt(hass, mqtt_mock, config_entry):
"""Successful setup."""
api_discovery = deepcopy(API_DISCOVERY_RESPONSE)
api_discovery["data"]["apiList"].append(API_DISCOVERY_MQTT)
with patch.dict(API_DISCOVERY_RESPONSE, api_discovery):
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
mqtt_mock.async_subscribe.assert_called_with(f"{MAC}/#", mock.ANY, 0, "utf-8")
@ -366,9 +335,9 @@ async def test_device_support_mqtt(hass, mqtt_mock):
assert pir.name == f"{NAME} PIR 0"
async def test_update_address(hass):
async def test_update_address(hass, config_entry):
"""Test update address works."""
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
assert device.api.config.host == "1.2.3.4"
@ -396,9 +365,11 @@ async def test_update_address(hass):
assert len(mock_setup_entry.mock_calls) == 1
async def test_device_unavailable(hass, mock_rtsp_event, mock_rtsp_signal_state):
async def test_device_unavailable(
hass, config_entry, mock_rtsp_event, mock_rtsp_signal_state
):
"""Successful setup."""
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
# Provide an entity that can be used to verify connection state on
mock_rtsp_event(
@ -430,35 +401,35 @@ async def test_device_unavailable(hass, mock_rtsp_event, mock_rtsp_signal_state)
assert hass.states.get(f"{BINARY_SENSOR_DOMAIN}.{NAME}_sound_1").state == STATE_OFF
async def test_device_reset(hass):
async def test_device_reset(hass, config_entry):
"""Successfully reset device."""
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
result = await device.async_reset()
assert result is True
async def test_device_not_accessible(hass):
async def test_device_not_accessible(hass, config_entry):
"""Failed setup schedules a retry of setup."""
with patch.object(axis, "get_axis_device", side_effect=axis.errors.CannotConnect):
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert hass.data[AXIS_DOMAIN] == {}
async def test_device_trigger_reauth_flow(hass):
async def test_device_trigger_reauth_flow(hass, config_entry):
"""Failed authentication trigger a reauthentication flow."""
with patch.object(
axis, "get_axis_device", side_effect=axis.errors.AuthenticationRequired
), patch.object(hass.config_entries.flow, "async_init") as mock_flow_init:
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
mock_flow_init.assert_called_once()
assert hass.data[AXIS_DOMAIN] == {}
async def test_device_unknown_error(hass):
async def test_device_unknown_error(hass, config_entry):
"""Unknown errors are handled."""
with patch.object(axis, "get_axis_device", side_effect=Exception):
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert hass.data[AXIS_DOMAIN] == {}

4
tests/components/axis/test_diagnostics.py
View File

@ -14,13 +14,13 @@ from .test_device import (
from tests.components.diagnostics import get_diagnostics_for_config_entry
async def test_entry_diagnostics(hass, hass_client):
async def test_entry_diagnostics(hass, hass_client, config_entry):
"""Test config entry diagnostics."""
api_discovery = deepcopy(API_DISCOVERY_RESPONSE)
api_discovery["data"]["apiList"].append(API_DISCOVERY_BASIC_DEVICE_INFO)
with patch.dict(API_DISCOVERY_RESPONSE, api_discovery):
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert await get_diagnostics_for_config_entry(hass, hass_client, config_entry) == {
"config": {

24
tests/components/axis/test_init.py
View File

@ -1,15 +1,14 @@
"""Test Axis component setup process."""
from unittest.mock import AsyncMock, Mock, patch
import pytest
from homeassistant.components import axis
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
from homeassistant.const import CONF_MAC
from homeassistant.setup import async_setup_component
from .test_device import setup_axis_integration
from tests.common import MockConfigEntry
async def test_setup_no_config(hass):
"""Test setup without configuration."""
@ -17,18 +16,15 @@ async def test_setup_no_config(hass):
assert AXIS_DOMAIN not in hass.data
async def test_setup_entry(hass):
async def test_setup_entry(hass, config_entry):
"""Test successful setup of entry."""
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert len(hass.data[AXIS_DOMAIN]) == 1
assert config_entry.entry_id in hass.data[AXIS_DOMAIN]
async def test_setup_entry_fails(hass):
async def test_setup_entry_fails(hass, config_entry):
"""Test successful setup of entry."""
config_entry = MockConfigEntry(domain=AXIS_DOMAIN, data={CONF_MAC: "0123"})
config_entry.add_to_hass(hass)
mock_device = Mock()
mock_device.async_setup = AsyncMock(return_value=False)
@ -40,20 +36,18 @@ async def test_setup_entry_fails(hass):
assert not hass.data[AXIS_DOMAIN]
async def test_unload_entry(hass):
async def test_unload_entry(hass, config_entry):
"""Test successful unload of entry."""
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert hass.data[AXIS_DOMAIN]
assert await hass.config_entries.async_unload(config_entry.entry_id)
assert not hass.data[AXIS_DOMAIN]
async def test_migrate_entry(hass):
@pytest.mark.config_entry_version(1)
async def test_migrate_entry(hass, config_entry):
"""Test successful migration of entry data."""
config_entry = MockConfigEntry(domain=AXIS_DOMAIN, version=1)
config_entry.add_to_hass(hass)
assert config_entry.version == 1
mock_device = Mock()

14
tests/components/axis/test_light.py
View File

@ -14,10 +14,10 @@ from homeassistant.const import (
)
from homeassistant.setup import async_setup_component
from .conftest import NAME
from .test_device import (
API_DISCOVERY_RESPONSE,
LIGHT_CONTROL_RESPONSE,
NAME,
setup_axis_integration,
)
@ -37,15 +37,15 @@ async def test_platform_manually_configured(hass):
assert AXIS_DOMAIN not in hass.data
async def test_no_lights(hass):
async def test_no_lights(hass, config_entry):
"""Test that no light events in Axis results in no light entities."""
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert not hass.states.async_entity_ids(LIGHT_DOMAIN)
async def test_no_light_entity_without_light_control_representation(
hass, mock_rtsp_event
hass, config_entry, mock_rtsp_event
):
"""Verify no lights entities get created without light control representation."""
api_discovery = deepcopy(API_DISCOVERY_RESPONSE)
@ -57,7 +57,7 @@ async def test_no_light_entity_without_light_control_representation(
with patch.dict(API_DISCOVERY_RESPONSE, api_discovery), patch.dict(
LIGHT_CONTROL_RESPONSE, light_control
):
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
mock_rtsp_event(
topic="tns1:Device/tnsaxis:Light/Status",
@ -71,13 +71,13 @@ async def test_no_light_entity_without_light_control_representation(
assert not hass.states.async_entity_ids(LIGHT_DOMAIN)
async def test_lights(hass, mock_rtsp_event):
async def test_lights(hass, config_entry, mock_rtsp_event):
"""Test that lights are loaded properly."""
api_discovery = deepcopy(API_DISCOVERY_RESPONSE)
api_discovery["data"]["apiList"].append(API_DISCOVERY_LIGHT_CONTROL)
with patch.dict(API_DISCOVERY_RESPONSE, api_discovery):
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
# Add light
with patch(

16
tests/components/axis/test_switch.py
View File

@ -14,10 +14,10 @@ from homeassistant.const import (
)
from homeassistant.setup import async_setup_component
from .conftest import NAME
from .test_device import (
API_DISCOVERY_PORT_MANAGEMENT,
API_DISCOVERY_RESPONSE,
NAME,
setup_axis_integration,
)
@ -31,16 +31,16 @@ async def test_platform_manually_configured(hass):
assert AXIS_DOMAIN not in hass.data
async def test_no_switches(hass):
async def test_no_switches(hass, config_entry):
"""Test that no output events in Axis results in no switch entities."""
await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
assert not hass.states.async_entity_ids(SWITCH_DOMAIN)
async def test_switches_with_port_cgi(hass, mock_rtsp_event):
async def test_switches_with_port_cgi(hass, config_entry, mock_rtsp_event):
"""Test that switches are loaded properly using port.cgi."""
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device.api.vapix.ports = {"0": AsyncMock(), "1": AsyncMock()}
@ -94,15 +94,13 @@ async def test_switches_with_port_cgi(hass, mock_rtsp_event):
device.api.vapix.ports["0"].open.assert_called_once()
async def test_switches_with_port_management(
hass, mock_axis_rtspclient, mock_rtsp_event
):
async def test_switches_with_port_management(hass, config_entry, mock_rtsp_event):
"""Test that switches are loaded properly using port management."""
api_discovery = deepcopy(API_DISCOVERY_RESPONSE)
api_discovery["data"]["apiList"].append(API_DISCOVERY_PORT_MANAGEMENT)
with patch.dict(API_DISCOVERY_RESPONSE, api_discovery):
config_entry = await setup_axis_integration(hass)
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device.api.vapix.ports = {"0": AsyncMock(), "1": AsyncMock()}