Use fixtures to setup Axis integration in tests (#86034)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Robert Svensson 2023-01-22 17:33:40 +01:00 committed by GitHub
parent 8e117ee499
commit 332d3e0f19
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 491 additions and 485 deletions

View File

@ -1,10 +1,12 @@
"""Axis conftest."""
from __future__ import annotations
from copy import deepcopy
from unittest.mock import patch
from axis.rtsp import Signal, State
import pytest
import respx
from homeassistant.components.axis.const import CONF_EVENTS, DOMAIN as AXIS_DOMAIN
from homeassistant.const import (
@ -16,26 +18,30 @@ from homeassistant.const import (
CONF_USERNAME,
)
from .const import (
API_DISCOVERY_RESPONSE,
APPLICATIONS_LIST_RESPONSE,
BASIC_DEVICE_INFO_RESPONSE,
BRAND_RESPONSE,
DEFAULT_HOST,
FORMATTED_MAC,
IMAGE_RESPONSE,
MODEL,
MQTT_CLIENT_RESPONSE,
NAME,
PORT_MANAGEMENT_RESPONSE,
PORTS_RESPONSE,
PROPERTIES_RESPONSE,
PTZ_RESPONSE,
STREAM_PROFILES_RESPONSE,
VIEW_AREAS_RESPONSE,
VMD4_RESPONSE,
)
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,
}
# Config entry fixtures
@pytest.fixture(name="config_entry")
@ -61,13 +67,138 @@ def config_entry_version_fixture(request):
@pytest.fixture(name="config")
def config_fixture():
"""Define a config entry data fixture."""
return ENTRY_CONFIG.copy()
return {
CONF_HOST: DEFAULT_HOST,
CONF_USERNAME: "root",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
CONF_MODEL: MODEL,
CONF_NAME: NAME,
}
@pytest.fixture(name="options")
def options_fixture(request):
"""Define a config entry options fixture."""
return ENTRY_OPTIONS.copy()
return {CONF_EVENTS: True}
# Axis API fixtures
@pytest.fixture(name="mock_vapix_requests")
def default_request_fixture(respx_mock):
"""Mock default Vapix requests responses."""
def __mock_default_requests(host):
path = f"http://{host}:80"
if host != DEFAULT_HOST:
respx.post(f"{path}/axis-cgi/apidiscovery.cgi").respond(
json=API_DISCOVERY_RESPONSE,
)
respx.post(f"{path}/axis-cgi/basicdeviceinfo.cgi").respond(
json=BASIC_DEVICE_INFO_RESPONSE,
)
respx.post(f"{path}/axis-cgi/io/portmanagement.cgi").respond(
json=PORT_MANAGEMENT_RESPONSE,
)
respx.post(f"{path}/axis-cgi/mqtt/client.cgi").respond(
json=MQTT_CLIENT_RESPONSE,
)
respx.post(f"{path}/axis-cgi/streamprofile.cgi").respond(
json=STREAM_PROFILES_RESPONSE,
)
respx.post(f"{path}/axis-cgi/viewarea/info.cgi").respond(
json=VIEW_AREAS_RESPONSE
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.Brand").respond(
text=BRAND_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.Image").respond(
text=IMAGE_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.Input").respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.IOPort").respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.Output").respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"{path}/axis-cgi/param.cgi?action=list&group=root.Properties"
).respond(
text=PROPERTIES_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(f"{path}/axis-cgi/param.cgi?action=list&group=root.PTZ").respond(
text=PTZ_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"{path}/axis-cgi/param.cgi?action=list&group=root.StreamProfile"
).respond(
text=STREAM_PROFILES_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.post(f"{path}/axis-cgi/applications/list.cgi").respond(
text=APPLICATIONS_LIST_RESPONSE,
headers={"Content-Type": "text/xml"},
)
respx.post(f"{path}/local/vmd/control.cgi").respond(json=VMD4_RESPONSE)
yield __mock_default_requests
@pytest.fixture()
def api_discovery_items():
"""Additional Apidiscovery items."""
return {}
@pytest.fixture(autouse=True)
def api_discovery_fixture(api_discovery_items):
"""Apidiscovery mock response."""
data = deepcopy(API_DISCOVERY_RESPONSE)
if api_discovery_items:
data["data"]["apiList"].append(api_discovery_items)
respx.post(f"http://{DEFAULT_HOST}:80/axis-cgi/apidiscovery.cgi").respond(json=data)
@pytest.fixture(name="setup_default_vapix_requests")
def default_vapix_requests_fixture(mock_vapix_requests):
"""Mock default Vapix requests responses."""
mock_vapix_requests(DEFAULT_HOST)
@pytest.fixture(name="prepare_config_entry")
async def prep_config_entry_fixture(hass, config_entry, setup_default_vapix_requests):
"""Fixture factory to set up Axis network device."""
async def __mock_setup_config_entry():
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
yield __mock_setup_config_entry
@pytest.fixture(name="setup_config_entry")
async def setup_config_entry_fixture(hass, config_entry, setup_default_vapix_requests):
"""Define a fixture to set up Axis network device."""
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
yield config_entry
# RTSP fixtures
@pytest.fixture(autouse=True)

View File

@ -0,0 +1,141 @@
"""Constants for Axis integration tests."""
MAC = "00408C123456"
FORMATTED_MAC = "00:40:8c:12:34:56"
MODEL = "model"
NAME = "name"
DEFAULT_HOST = "1.2.3.4"
API_DISCOVERY_RESPONSE = {
"method": "getApiList",
"apiVersion": "1.0",
"data": {
"apiList": [
{"id": "api-discovery", "version": "1.0", "name": "API Discovery Service"},
{"id": "param-cgi", "version": "1.0", "name": "Legacy Parameter Handling"},
]
},
}
API_DISCOVERY_BASIC_DEVICE_INFO = {
"id": "basic-device-info",
"version": "1.1",
"name": "Basic Device Information",
}
API_DISCOVERY_MQTT = {"id": "mqtt-client", "version": "1.0", "name": "MQTT Client API"}
API_DISCOVERY_PORT_MANAGEMENT = {
"id": "io-port-management",
"version": "1.0",
"name": "IO Port Management",
}
APPLICATIONS_LIST_RESPONSE = """<reply result="ok">
<application Name="vmd" NiceName="AXIS Video Motion Detection" Vendor="Axis Communications" Version="4.2-0" ApplicationID="143440" License="None" Status="Running" ConfigurationPage="local/vmd/config.html" VendorHomePage="http://www.axis.com" />
</reply>"""
BASIC_DEVICE_INFO_RESPONSE = {
"apiVersion": "1.1",
"data": {
"propertyList": {
"ProdNbr": "M1065-LW",
"ProdType": "Network Camera",
"SerialNumber": MAC,
"Version": "9.80.1",
}
},
}
MQTT_CLIENT_RESPONSE = {
"apiVersion": "1.0",
"context": "some context",
"method": "getClientStatus",
"data": {"status": {"state": "active", "connectionStatus": "Connected"}},
}
PORT_MANAGEMENT_RESPONSE = {
"apiVersion": "1.0",
"method": "getPorts",
"data": {
"numberOfPorts": 1,
"items": [
{
"port": "0",
"configurable": False,
"usage": "",
"name": "PIR sensor",
"direction": "input",
"state": "open",
"normalState": "open",
}
],
},
}
VMD4_RESPONSE = {
"apiVersion": "1.4",
"method": "getConfiguration",
"context": "Axis library",
"data": {
"cameras": [{"id": 1, "rotation": 0, "active": True}],
"profiles": [
{"filters": [], "camera": 1, "triggers": [], "name": "Profile 1", "uid": 1}
],
},
}
BRAND_RESPONSE = """root.Brand.Brand=AXIS
root.Brand.ProdFullName=AXIS M1065-LW Network Camera
root.Brand.ProdNbr=M1065-LW
root.Brand.ProdShortName=AXIS M1065-LW
root.Brand.ProdType=Network Camera
root.Brand.ProdVariant=
root.Brand.WebURL=http://www.axis.com
"""
IMAGE_RESPONSE = """root.Image.I0.Enabled=yes
root.Image.I0.Name=View Area 1
root.Image.I0.Source=0
root.Image.I1.Enabled=no
root.Image.I1.Name=View Area 2
root.Image.I1.Source=0
"""
PORTS_RESPONSE = """root.Input.NbrOfInputs=1
root.IOPort.I0.Configurable=no
root.IOPort.I0.Direction=input
root.IOPort.I0.Input.Name=PIR sensor
root.IOPort.I0.Input.Trig=closed
root.Output.NbrOfOutputs=0
"""
PROPERTIES_RESPONSE = f"""root.Properties.API.HTTP.Version=3
root.Properties.API.Metadata.Metadata=yes
root.Properties.API.Metadata.Version=1.0
root.Properties.EmbeddedDevelopment.Version=2.16
root.Properties.Firmware.BuildDate=Feb 15 2019 09:42
root.Properties.Firmware.BuildNumber=26
root.Properties.Firmware.Version=9.10.1
root.Properties.Image.Format=jpeg,mjpeg,h264
root.Properties.Image.NbrOfViews=2
root.Properties.Image.Resolution=1920x1080,1280x960,1280x720,1024x768,1024x576,800x600,640x480,640x360,352x240,320x240
root.Properties.Image.Rotation=0,180
root.Properties.System.SerialNumber={MAC}
"""
PTZ_RESPONSE = ""
STREAM_PROFILES_RESPONSE = """root.StreamProfile.MaxGroups=26
root.StreamProfile.S0.Description=profile_1_description
root.StreamProfile.S0.Name=profile_1
root.StreamProfile.S0.Parameters=videocodec=h264
root.StreamProfile.S1.Description=profile_2_description
root.StreamProfile.S1.Name=profile_2
root.StreamProfile.S1.Parameters=videocodec=h265
"""
VIEW_AREAS_RESPONSE = {"apiVersion": "1.0", "method": "list", "data": {"viewAreas": []}}

View File

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

View File

@ -13,8 +13,7 @@ from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
from homeassistant.const import STATE_IDLE
from homeassistant.setup import async_setup_component
from .conftest import NAME
from .test_device import setup_axis_integration
from .const import NAME
async def test_platform_manually_configured(hass):
@ -29,10 +28,8 @@ async def test_platform_manually_configured(hass):
assert AXIS_DOMAIN not in hass.data
async def test_camera(hass, config_entry):
async def test_camera(hass, setup_config_entry):
"""Test that Axis camera platform is loaded properly."""
await setup_axis_integration(hass, config_entry)
assert len(hass.states.async_entity_ids(CAMERA_DOMAIN)) == 1
entity_id = f"{CAMERA_DOMAIN}.{NAME}"
@ -51,10 +48,8 @@ async def test_camera(hass, config_entry):
@pytest.mark.parametrize("options", [{CONF_STREAM_PROFILE: "profile_1"}])
async def test_camera_with_stream_profile(hass, config_entry):
async def test_camera_with_stream_profile(hass, setup_config_entry):
"""Test that Axis camera entity is using the correct path with stream profike."""
await setup_axis_integration(hass, config_entry)
assert len(hass.states.async_entity_ids(CAMERA_DOMAIN)) == 1
entity_id = f"{CAMERA_DOMAIN}.{NAME}"
@ -75,9 +70,9 @@ async def test_camera_with_stream_profile(hass, config_entry):
)
async def test_camera_disabled(hass, config_entry):
async def test_camera_disabled(hass, prepare_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, config_entry)
await prepare_config_entry()
assert len(hass.states.async_entity_ids(CAMERA_DOMAIN)) == 0

View File

@ -2,7 +2,6 @@
from unittest.mock import patch
import pytest
import respx
from homeassistant.components import dhcp, ssdp, zeroconf
from homeassistant.components.axis import config_flow
@ -32,13 +31,12 @@ from homeassistant.const import (
)
from homeassistant.data_entry_flow import FlowResultType
from .conftest import DEFAULT_HOST, MAC, MODEL, NAME
from .test_device import mock_default_vapix_requests, setup_axis_integration
from .const import DEFAULT_HOST, MAC, MODEL, NAME
from tests.common import MockConfigEntry
async def test_flow_manual_configuration(hass):
async def test_flow_manual_configuration(hass, setup_default_vapix_requests):
"""Test that config flow works."""
MockConfigEntry(domain=AXIS_DOMAIN, source=SOURCE_IGNORE).add_to_hass(hass)
@ -49,17 +47,15 @@ async def test_flow_manual_configuration(hass):
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == SOURCE_USER
with respx.mock:
mock_default_vapix_requests(respx)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == f"M1065-LW - {MAC}"
@ -73,10 +69,11 @@ async def test_flow_manual_configuration(hass):
}
async def test_manual_configuration_update_configuration(hass, config_entry):
async def test_manual_configuration_update_configuration(
hass, setup_config_entry, mock_vapix_requests
):
"""Test that config flow fails on already configured device."""
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN, context={"source": SOURCE_USER}
@ -86,10 +83,9 @@ async def test_manual_configuration_update_configuration(hass, config_entry):
assert result["step_id"] == SOURCE_USER
with patch(
"homeassistant.components.axis.async_setup_entry",
return_value=True,
) as mock_setup_entry, respx.mock:
mock_default_vapix_requests(respx, "2.3.4.5")
"homeassistant.components.axis.async_setup_entry", return_value=True
) as mock_setup_entry:
mock_vapix_requests("2.3.4.5")
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
@ -159,7 +155,9 @@ async def test_flow_fails_cannot_connect(hass):
assert result["errors"] == {"base": "cannot_connect"}
async def test_flow_create_entry_multiple_existing_entries_of_same_model(hass):
async def test_flow_create_entry_multiple_existing_entries_of_same_model(
hass, setup_default_vapix_requests
):
"""Test that create entry can generate a name with other entries."""
entry = MockConfigEntry(
domain=AXIS_DOMAIN,
@ -179,17 +177,15 @@ async def test_flow_create_entry_multiple_existing_entries_of_same_model(hass):
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == SOURCE_USER
with respx.mock:
mock_default_vapix_requests(respx)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == f"M1065-LW - {MAC}"
@ -205,32 +201,32 @@ 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, config_entry):
async def test_reauth_flow_update_configuration(
hass, setup_config_entry, mock_vapix_requests
):
"""Test that config flow fails on already configured device."""
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN,
context={"source": SOURCE_REAUTH},
data=config_entry.data,
data=setup_config_entry.data,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == SOURCE_USER
with respx.mock:
mock_default_vapix_requests(respx, "2.3.4.5")
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "2.3.4.5",
CONF_USERNAME: "user2",
CONF_PASSWORD: "pass2",
CONF_PORT: 80,
},
)
await hass.async_block_till_done()
mock_vapix_requests("2.3.4.5")
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "2.3.4.5",
CONF_USERNAME: "user2",
CONF_PASSWORD: "pass2",
CONF_PORT: 80,
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
@ -303,7 +299,9 @@ async def test_reauth_flow_update_configuration(hass, config_entry):
),
],
)
async def test_discovery_flow(hass, source: str, discovery_info: dict):
async def test_discovery_flow(
hass, setup_default_vapix_requests, source: str, discovery_info: dict
):
"""Test the different discovery flows for new devices work."""
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN, data=discovery_info, context={"source": source}
@ -316,17 +314,15 @@ async def test_discovery_flow(hass, source: str, discovery_info: dict):
assert len(flows) == 1
assert flows[0].get("context", {}).get("configuration_url") == "http://1.2.3.4:80"
with respx.mock:
mock_default_vapix_requests(respx)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == f"M1065-LW - {MAC}"
@ -380,11 +376,10 @@ async def test_discovery_flow(hass, source: str, discovery_info: dict):
],
)
async def test_discovered_device_already_configured(
hass, config_entry, source: str, discovery_info: dict
hass, setup_config_entry, source: str, discovery_info: dict
):
"""Test that discovery doesn't setup already configured devices."""
await setup_axis_integration(hass, config_entry)
assert config_entry.data[CONF_HOST] == DEFAULT_HOST
assert setup_config_entry.data[CONF_HOST] == DEFAULT_HOST
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN, data=discovery_info, context={"source": source}
@ -392,7 +387,7 @@ async def test_discovered_device_already_configured(
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert config_entry.data[CONF_HOST] == DEFAULT_HOST
assert setup_config_entry.data[CONF_HOST] == DEFAULT_HOST
@pytest.mark.parametrize(
@ -436,11 +431,15 @@ async def test_discovered_device_already_configured(
],
)
async def test_discovery_flow_updated_configuration(
hass, config_entry, source: str, discovery_info: dict, expected_port: int
hass,
setup_config_entry,
mock_vapix_requests,
source: str,
discovery_info: dict,
expected_port: int,
):
"""Test that discovery flow update configuration with new parameters."""
await setup_axis_integration(hass, config_entry)
assert config_entry.data == {
assert setup_config_entry.data == {
CONF_HOST: DEFAULT_HOST,
CONF_PORT: 80,
CONF_USERNAME: "root",
@ -450,10 +449,9 @@ async def test_discovery_flow_updated_configuration(
}
with patch(
"homeassistant.components.axis.async_setup_entry",
return_value=True,
) as mock_setup_entry, respx.mock:
mock_default_vapix_requests(respx, "2.3.4.5")
"homeassistant.components.axis.async_setup_entry", return_value=True
) as mock_setup_entry:
mock_vapix_requests("2.3.4.5")
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN, data=discovery_info, context={"source": source}
)
@ -461,7 +459,7 @@ async def test_discovery_flow_updated_configuration(
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert config_entry.data == {
assert setup_config_entry.data == {
CONF_HOST: "2.3.4.5",
CONF_PORT: expected_port,
CONF_USERNAME: "root",
@ -570,16 +568,13 @@ async def test_discovery_flow_ignore_link_local_address(
assert result["reason"] == "link_local_address"
async def test_option_flow(hass, config_entry):
async def test_option_flow(hass, setup_config_entry):
"""Test config flow options."""
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
assert device.option_stream_profile == DEFAULT_STREAM_PROFILE
assert device.option_video_source == DEFAULT_VIDEO_SOURCE
with respx.mock:
mock_default_vapix_requests(respx)
result = await hass.config_entries.options.async_init(config_entry.entry_id)
result = await hass.config_entries.options.async_init(setup_config_entry.entry_id)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "configure_stream"

View File

@ -1,11 +1,9 @@
"""Test Axis device."""
from copy import deepcopy
from unittest import mock
from unittest.mock import Mock, patch
import axis as axislib
import pytest
import respx
from homeassistant.components import axis, zeroconf
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
@ -21,257 +19,27 @@ from homeassistant.const import (
)
from homeassistant.helpers import device_registry as dr
from .conftest import DEFAULT_HOST, ENTRY_CONFIG, FORMATTED_MAC, MAC, NAME
from .const import (
API_DISCOVERY_BASIC_DEVICE_INFO,
API_DISCOVERY_MQTT,
FORMATTED_MAC,
MAC,
NAME,
)
from tests.common import async_fire_mqtt_message
API_DISCOVERY_RESPONSE = {
"method": "getApiList",
"apiVersion": "1.0",
"data": {
"apiList": [
{"id": "api-discovery", "version": "1.0", "name": "API Discovery Service"},
{"id": "param-cgi", "version": "1.0", "name": "Legacy Parameter Handling"},
]
},
}
API_DISCOVERY_BASIC_DEVICE_INFO = {
"id": "basic-device-info",
"version": "1.1",
"name": "Basic Device Information",
}
API_DISCOVERY_MQTT = {"id": "mqtt-client", "version": "1.0", "name": "MQTT Client API"}
API_DISCOVERY_PORT_MANAGEMENT = {
"id": "io-port-management",
"version": "1.0",
"name": "IO Port Management",
}
APPLICATIONS_LIST_RESPONSE = """<reply result="ok">
<application Name="vmd" NiceName="AXIS Video Motion Detection" Vendor="Axis Communications" Version="4.2-0" ApplicationID="143440" License="None" Status="Running" ConfigurationPage="local/vmd/config.html" VendorHomePage="http://www.axis.com" />
</reply>"""
BASIC_DEVICE_INFO_RESPONSE = {
"apiVersion": "1.1",
"data": {
"propertyList": {
"ProdNbr": "M1065-LW",
"ProdType": "Network Camera",
"SerialNumber": MAC,
"Version": "9.80.1",
}
},
}
LIGHT_CONTROL_RESPONSE = {
"apiVersion": "1.1",
"method": "getLightInformation",
"data": {
"items": [
{
"lightID": "led0",
"lightType": "IR",
"enabled": True,
"synchronizeDayNightMode": True,
"lightState": False,
"automaticIntensityMode": False,
"automaticAngleOfIlluminationMode": False,
"nrOfLEDs": 1,
"error": False,
"errorInfo": "",
}
]
},
}
MQTT_CLIENT_RESPONSE = {
"apiVersion": "1.0",
"context": "some context",
"method": "getClientStatus",
"data": {"status": {"state": "active", "connectionStatus": "Connected"}},
}
PORT_MANAGEMENT_RESPONSE = {
"apiVersion": "1.0",
"method": "getPorts",
"data": {
"numberOfPorts": 1,
"items": [
{
"port": "0",
"configurable": False,
"usage": "",
"name": "PIR sensor",
"direction": "input",
"state": "open",
"normalState": "open",
}
],
},
}
VMD4_RESPONSE = {
"apiVersion": "1.4",
"method": "getConfiguration",
"context": "Axis library",
"data": {
"cameras": [{"id": 1, "rotation": 0, "active": True}],
"profiles": [
{"filters": [], "camera": 1, "triggers": [], "name": "Profile 1", "uid": 1}
],
},
}
BRAND_RESPONSE = """root.Brand.Brand=AXIS
root.Brand.ProdFullName=AXIS M1065-LW Network Camera
root.Brand.ProdNbr=M1065-LW
root.Brand.ProdShortName=AXIS M1065-LW
root.Brand.ProdType=Network Camera
root.Brand.ProdVariant=
root.Brand.WebURL=http://www.axis.com
"""
IMAGE_RESPONSE = """root.Image.I0.Enabled=yes
root.Image.I0.Name=View Area 1
root.Image.I0.Source=0
root.Image.I1.Enabled=no
root.Image.I1.Name=View Area 2
root.Image.I1.Source=0
"""
PORTS_RESPONSE = """root.Input.NbrOfInputs=1
root.IOPort.I0.Configurable=no
root.IOPort.I0.Direction=input
root.IOPort.I0.Input.Name=PIR sensor
root.IOPort.I0.Input.Trig=closed
root.Output.NbrOfOutputs=0
"""
PROPERTIES_RESPONSE = f"""root.Properties.API.HTTP.Version=3
root.Properties.API.Metadata.Metadata=yes
root.Properties.API.Metadata.Version=1.0
root.Properties.EmbeddedDevelopment.Version=2.16
root.Properties.Firmware.BuildDate=Feb 15 2019 09:42
root.Properties.Firmware.BuildNumber=26
root.Properties.Firmware.Version=9.10.1
root.Properties.Image.Format=jpeg,mjpeg,h264
root.Properties.Image.NbrOfViews=2
root.Properties.Image.Resolution=1920x1080,1280x960,1280x720,1024x768,1024x576,800x600,640x480,640x360,352x240,320x240
root.Properties.Image.Rotation=0,180
root.Properties.System.SerialNumber={MAC}
"""
PTZ_RESPONSE = ""
@pytest.fixture(name="forward_entry_setup")
def hass_mock_forward_entry_setup(hass):
"""Mock async_forward_entry_setup."""
with patch.object(hass.config_entries, "async_forward_entry_setup") as forward_mock:
yield forward_mock
STREAM_PROFILES_RESPONSE = """root.StreamProfile.MaxGroups=26
root.StreamProfile.S0.Description=profile_1_description
root.StreamProfile.S0.Name=profile_1
root.StreamProfile.S0.Parameters=videocodec=h264
root.StreamProfile.S1.Description=profile_2_description
root.StreamProfile.S1.Name=profile_2
root.StreamProfile.S1.Parameters=videocodec=h265
"""
VIEW_AREAS_RESPONSE = {"apiVersion": "1.0", "method": "list", "data": {"viewAreas": []}}
def mock_default_vapix_requests(respx: respx, host: str = DEFAULT_HOST) -> None:
"""Mock default Vapix requests responses."""
respx.post(f"http://{host}:80/axis-cgi/apidiscovery.cgi").respond(
json=API_DISCOVERY_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/basicdeviceinfo.cgi").respond(
json=BASIC_DEVICE_INFO_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/io/portmanagement.cgi").respond(
json=PORT_MANAGEMENT_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/lightcontrol.cgi").respond(
json=LIGHT_CONTROL_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/mqtt/client.cgi").respond(
json=MQTT_CLIENT_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/streamprofile.cgi").respond(
json=STREAM_PROFILES_RESPONSE,
)
respx.post(f"http://{host}:80/axis-cgi/viewarea/info.cgi").respond(
json=VIEW_AREAS_RESPONSE
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.Brand"
).respond(
text=BRAND_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.Image"
).respond(
text=IMAGE_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.Input"
).respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.IOPort"
).respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.Output"
).respond(
text=PORTS_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.Properties"
).respond(
text=PROPERTIES_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.PTZ"
).respond(
text=PTZ_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.get(
f"http://{host}:80/axis-cgi/param.cgi?action=list&group=root.StreamProfile"
).respond(
text=STREAM_PROFILES_RESPONSE,
headers={"Content-Type": "text/plain"},
)
respx.post(f"http://{host}:80/axis-cgi/applications/list.cgi").respond(
text=APPLICATIONS_LIST_RESPONSE,
headers={"Content-Type": "text/xml"},
)
respx.post(f"http://{host}:80/local/vmd/control.cgi").respond(json=VMD4_RESPONSE)
async def setup_axis_integration(hass, config_entry):
"""Create the Axis device."""
with respx.mock:
mock_default_vapix_requests(respx)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async def test_device_setup(hass, config_entry):
async def test_device_setup(hass, forward_entry_setup, config, setup_config_entry):
"""Successful setup."""
with patch(
"homeassistant.config_entries.ConfigEntries.async_forward_entry_setup",
return_value=True,
) as forward_entry_setup:
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
assert device.api.vapix.firmware_version == "9.10.1"
assert device.api.vapix.product_number == "M1065-LW"
@ -279,14 +47,14 @@ async def test_device_setup(hass, config_entry):
assert device.api.vapix.serial_number == "00408C123456"
assert len(forward_entry_setup.mock_calls) == 4
assert forward_entry_setup.mock_calls[0][1] == (config_entry, "binary_sensor")
assert forward_entry_setup.mock_calls[1][1] == (config_entry, "camera")
assert forward_entry_setup.mock_calls[2][1] == (config_entry, "light")
assert forward_entry_setup.mock_calls[3][1] == (config_entry, "switch")
assert forward_entry_setup.mock_calls[0][1][1] == "binary_sensor"
assert forward_entry_setup.mock_calls[1][1][1] == "camera"
assert forward_entry_setup.mock_calls[2][1][1] == "light"
assert forward_entry_setup.mock_calls[3][1][1] == "switch"
assert device.host == ENTRY_CONFIG[CONF_HOST]
assert device.model == ENTRY_CONFIG[CONF_MODEL]
assert device.name == ENTRY_CONFIG[CONF_NAME]
assert device.host == config[CONF_HOST]
assert device.model == config[CONF_MODEL]
assert device.name == config[CONF_NAME]
assert device.unique_id == FORMATTED_MAC
device_registry = dr.async_get(hass)
@ -297,14 +65,10 @@ async def test_device_setup(hass, config_entry):
assert device_entry.configuration_url == device.api.config.url
async def test_device_info(hass, config_entry):
@pytest.mark.parametrize("api_discovery_items", [API_DISCOVERY_BASIC_DEVICE_INFO])
async def test_device_info(hass, setup_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):
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
assert device.api.vapix.firmware_version == "9.80.1"
assert device.api.vapix.product_number == "M1065-LW"
@ -312,14 +76,9 @@ async def test_device_info(hass, config_entry):
assert device.api.vapix.serial_number == "00408C123456"
async def test_device_support_mqtt(hass, mqtt_mock, config_entry):
@pytest.mark.parametrize("api_discovery_items", [API_DISCOVERY_MQTT])
async def test_device_support_mqtt(hass, mqtt_mock, setup_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, config_entry)
mqtt_mock.async_subscribe.assert_called_with(f"{MAC}/#", mock.ANY, 0, "utf-8")
topic = f"{MAC}/event/tns:onvif/Device/tns:axis/Sensor/PIR/$source/sensor/0"
@ -338,17 +97,15 @@ async def test_device_support_mqtt(hass, mqtt_mock, config_entry):
assert pir.name == f"{NAME} PIR 0"
async def test_update_address(hass, config_entry):
async def test_update_address(hass, setup_config_entry, mock_vapix_requests):
"""Test update address works."""
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
assert device.api.config.host == "1.2.3.4"
with patch(
"homeassistant.components.axis.async_setup_entry",
return_value=True,
) as mock_setup_entry, respx.mock:
mock_default_vapix_requests(respx, "2.3.4.5")
"homeassistant.components.axis.async_setup_entry", return_value=True
) as mock_setup_entry:
mock_vapix_requests("2.3.4.5")
await hass.config_entries.flow.async_init(
AXIS_DOMAIN,
data=zeroconf.ZeroconfServiceInfo(
@ -369,11 +126,9 @@ async def test_update_address(hass, config_entry):
async def test_device_unavailable(
hass, config_entry, mock_rtsp_event, mock_rtsp_signal_state
hass, setup_config_entry, mock_rtsp_event, mock_rtsp_signal_state
):
"""Successful setup."""
await setup_axis_integration(hass, config_entry)
# Provide an entity that can be used to verify connection state on
mock_rtsp_event(
topic="tns1:AudioSource/tnsaxis:TriggerLevel",
@ -404,43 +159,47 @@ async def test_device_unavailable(
assert hass.states.get(f"{BINARY_SENSOR_DOMAIN}.{NAME}_sound_1").state == STATE_OFF
async def test_device_reset(hass, config_entry):
async def test_device_reset(hass, setup_config_entry):
"""Successfully reset device."""
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
result = await device.async_reset()
assert result is True
async def test_device_not_accessible(hass, config_entry):
async def test_device_not_accessible(hass, config_entry, setup_default_vapix_requests):
"""Failed setup schedules a retry of setup."""
with patch.object(axis, "get_axis_device", side_effect=axis.errors.CannotConnect):
await setup_axis_integration(hass, config_entry)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.data[AXIS_DOMAIN] == {}
async def test_device_trigger_reauth_flow(hass, config_entry):
async def test_device_trigger_reauth_flow(
hass, config_entry, setup_default_vapix_requests
):
"""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, config_entry)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
mock_flow_init.assert_called_once()
assert hass.data[AXIS_DOMAIN] == {}
async def test_device_unknown_error(hass, config_entry):
async def test_device_unknown_error(hass, config_entry, setup_default_vapix_requests):
"""Unknown errors are handled."""
with patch.object(axis, "get_axis_device", side_effect=Exception):
await setup_axis_integration(hass, config_entry)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.data[AXIS_DOMAIN] == {}
async def test_shutdown():
async def test_shutdown(config):
"""Successful shutdown."""
hass = Mock()
entry = Mock()
entry.data = ENTRY_CONFIG
entry.data = config
axis_device = axis.device.AxisNetworkDevice(hass, entry, Mock())
@ -449,25 +208,25 @@ async def test_shutdown():
assert len(axis_device.api.stream.stop.mock_calls) == 1
async def test_get_device_fails(hass):
async def test_get_device_fails(hass, config):
"""Device unauthorized yields authentication required error."""
with patch(
"axis.vapix.vapix.Vapix.request", side_effect=axislib.Unauthorized
), pytest.raises(axis.errors.AuthenticationRequired):
await axis.device.get_axis_device(hass, ENTRY_CONFIG)
await axis.device.get_axis_device(hass, config)
async def test_get_device_device_unavailable(hass):
async def test_get_device_device_unavailable(hass, config):
"""Device unavailable yields cannot connect error."""
with patch(
"axis.vapix.vapix.Vapix.request", side_effect=axislib.RequestError
), pytest.raises(axis.errors.CannotConnect):
await axis.device.get_axis_device(hass, ENTRY_CONFIG)
await axis.device.get_axis_device(hass, config)
async def test_get_device_unknown_error(hass):
async def test_get_device_unknown_error(hass, config):
"""Device yield unknown error."""
with patch(
"axis.vapix.vapix.Vapix.request", side_effect=axislib.AxisException
), pytest.raises(axis.errors.AuthenticationRequired):
await axis.device.get_axis_device(hass, ENTRY_CONFIG)
await axis.device.get_axis_device(hass, config)

View File

@ -1,30 +1,22 @@
"""Test Axis diagnostics."""
from copy import deepcopy
from unittest.mock import patch
import pytest
from homeassistant.components.diagnostics import REDACTED
from .test_device import (
API_DISCOVERY_BASIC_DEVICE_INFO,
API_DISCOVERY_RESPONSE,
setup_axis_integration,
)
from .const import API_DISCOVERY_BASIC_DEVICE_INFO
from tests.components.diagnostics import get_diagnostics_for_config_entry
async def test_entry_diagnostics(hass, hass_client, config_entry):
@pytest.mark.parametrize("api_discovery_items", [API_DISCOVERY_BASIC_DEVICE_INFO])
async def test_entry_diagnostics(hass, hass_client, setup_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):
await setup_axis_integration(hass, config_entry)
assert await get_diagnostics_for_config_entry(hass, hass_client, config_entry) == {
assert await get_diagnostics_for_config_entry(
hass, hass_client, setup_config_entry
) == {
"config": {
"entry_id": config_entry.entry_id,
"entry_id": setup_config_entry.entry_id,
"version": 3,
"domain": "axis",
"title": "Mock Title",

View File

@ -7,8 +7,6 @@ from homeassistant.components import axis
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
from homeassistant.setup import async_setup_component
from .test_device import setup_axis_integration
async def test_setup_no_config(hass):
"""Test setup without configuration."""
@ -16,11 +14,10 @@ async def test_setup_no_config(hass):
assert AXIS_DOMAIN not in hass.data
async def test_setup_entry(hass, config_entry):
async def test_setup_entry(hass, setup_config_entry):
"""Test successful setup of entry."""
await setup_axis_integration(hass, config_entry)
assert len(hass.data[AXIS_DOMAIN]) == 1
assert config_entry.entry_id in hass.data[AXIS_DOMAIN]
assert setup_config_entry.entry_id in hass.data[AXIS_DOMAIN]
async def test_setup_entry_fails(hass, config_entry):
@ -36,12 +33,11 @@ async def test_setup_entry_fails(hass, config_entry):
assert not hass.data[AXIS_DOMAIN]
async def test_unload_entry(hass, config_entry):
async def test_unload_entry(hass, setup_config_entry):
"""Test successful unload of entry."""
await setup_axis_integration(hass, config_entry)
assert hass.data[AXIS_DOMAIN]
assert await hass.config_entries.async_unload(config_entry.entry_id)
assert await hass.config_entries.async_unload(setup_config_entry.entry_id)
assert not hass.data[AXIS_DOMAIN]

View File

@ -1,8 +1,10 @@
"""Axis light platform tests."""
from copy import deepcopy
from unittest.mock import patch
import pytest
import respx
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
from homeassistant.components.light import ATTR_BRIGHTNESS, DOMAIN as LIGHT_DOMAIN
from homeassistant.const import (
@ -14,12 +16,7 @@ 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,
setup_axis_integration,
)
from .const import DEFAULT_HOST, NAME
API_DISCOVERY_LIGHT_CONTROL = {
"id": "light-control",
@ -28,6 +25,38 @@ API_DISCOVERY_LIGHT_CONTROL = {
}
@pytest.fixture()
def light_control_items():
"""Available lights."""
return [
{
"lightID": "led0",
"lightType": "IR",
"enabled": True,
"synchronizeDayNightMode": True,
"lightState": False,
"automaticIntensityMode": False,
"automaticAngleOfIlluminationMode": False,
"nrOfLEDs": 1,
"error": False,
"errorInfo": "",
}
]
@pytest.fixture(autouse=True)
def light_control_fixture(light_control_items):
"""Light control mock response."""
data = {
"apiVersion": "1.1",
"method": "getLightInformation",
"data": {"items": light_control_items},
}
respx.post(f"http://{DEFAULT_HOST}:80/axis-cgi/lightcontrol.cgi").respond(
json=data,
)
async def test_platform_manually_configured(hass):
"""Test that nothing happens when platform is manually configured."""
assert await async_setup_component(
@ -37,28 +66,17 @@ async def test_platform_manually_configured(hass):
assert AXIS_DOMAIN not in hass.data
async def test_no_lights(hass, config_entry):
async def test_no_lights(hass, setup_config_entry):
"""Test that no light events in Axis results in no light entities."""
await setup_axis_integration(hass, config_entry)
assert not hass.states.async_entity_ids(LIGHT_DOMAIN)
@pytest.mark.parametrize("api_discovery_items", [API_DISCOVERY_LIGHT_CONTROL])
@pytest.mark.parametrize("light_control_items", [[]])
async def test_no_light_entity_without_light_control_representation(
hass, config_entry, mock_rtsp_event
hass, setup_config_entry, mock_rtsp_event
):
"""Verify no lights entities get created without light control representation."""
api_discovery = deepcopy(API_DISCOVERY_RESPONSE)
api_discovery["data"]["apiList"].append(API_DISCOVERY_LIGHT_CONTROL)
light_control = deepcopy(LIGHT_CONTROL_RESPONSE)
light_control["data"]["items"] = []
with patch.dict(API_DISCOVERY_RESPONSE, api_discovery), patch.dict(
LIGHT_CONTROL_RESPONSE, light_control
):
await setup_axis_integration(hass, config_entry)
mock_rtsp_event(
topic="tns1:Device/tnsaxis:Light/Status",
data_type="state",
@ -71,14 +89,9 @@ async def test_no_light_entity_without_light_control_representation(
assert not hass.states.async_entity_ids(LIGHT_DOMAIN)
async def test_lights(hass, config_entry, mock_rtsp_event):
@pytest.mark.parametrize("api_discovery_items", [API_DISCOVERY_LIGHT_CONTROL])
async def test_lights(hass, setup_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, config_entry)
# Add light
with patch(
"axis.vapix.interfaces.light_control.LightControl.get_current_intensity",

View File

@ -1,7 +1,8 @@
"""Axis switch platform tests."""
from copy import deepcopy
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
@ -14,12 +15,7 @@ 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,
setup_axis_integration,
)
from .const import API_DISCOVERY_PORT_MANAGEMENT, NAME
async def test_platform_manually_configured(hass):
@ -31,17 +27,14 @@ async def test_platform_manually_configured(hass):
assert AXIS_DOMAIN not in hass.data
async def test_no_switches(hass, config_entry):
async def test_no_switches(hass, setup_config_entry):
"""Test that no output events in Axis results in no switch entities."""
await setup_axis_integration(hass, config_entry)
assert not hass.states.async_entity_ids(SWITCH_DOMAIN)
async def test_switches_with_port_cgi(hass, config_entry, mock_rtsp_event):
async def test_switches_with_port_cgi(hass, setup_config_entry, mock_rtsp_event):
"""Test that switches are loaded properly using port.cgi."""
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
device.api.vapix.ports = {"0": AsyncMock(), "1": AsyncMock()}
device.api.vapix.ports["0"].name = "Doorbell"
@ -94,14 +87,10 @@ async def test_switches_with_port_cgi(hass, config_entry, mock_rtsp_event):
device.api.vapix.ports["0"].open.assert_called_once()
async def test_switches_with_port_management(hass, config_entry, mock_rtsp_event):
@pytest.mark.parametrize("api_discovery_items", [API_DISCOVERY_PORT_MANAGEMENT])
async def test_switches_with_port_management(hass, setup_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):
await setup_axis_integration(hass, config_entry)
device = hass.data[AXIS_DOMAIN][config_entry.entry_id]
device = hass.data[AXIS_DOMAIN][setup_config_entry.entry_id]
device.api.vapix.ports = {"0": AsyncMock(), "1": AsyncMock()}
device.api.vapix.ports["0"].name = "Doorbell"