diff --git a/homeassistant/components/axis/hub/api.py b/homeassistant/components/axis/hub/api.py index f33e925929c..2bfce19bae5 100644 --- a/homeassistant/components/axis/hub/api.py +++ b/homeassistant/components/axis/hub/api.py @@ -15,7 +15,7 @@ from homeassistant.const import ( CONF_USERNAME, ) from homeassistant.core import HomeAssistant -from homeassistant.helpers.httpx_client import get_async_client +from homeassistant.helpers.aiohttp_client import async_get_clientsession from ..const import LOGGER from ..errors import AuthenticationRequired, CannotConnect @@ -26,7 +26,7 @@ async def get_axis_api( config: Mapping[str, Any], ) -> axis.AxisDevice: """Create a Axis device API.""" - session = get_async_client(hass, verify_ssl=False) + session = async_get_clientsession(hass, verify_ssl=False) api = axis.AxisDevice( Configuration( diff --git a/tests/components/axis/conftest.py b/tests/components/axis/conftest.py index d2693a83f05..8a452109d1d 100644 --- a/tests/components/axis/conftest.py +++ b/tests/components/axis/conftest.py @@ -4,13 +4,14 @@ from __future__ import annotations from collections.abc import Callable, Coroutine, Generator from copy import deepcopy +import re from types import MappingProxyType from typing import Any, Protocol from unittest.mock import AsyncMock, patch from axis.rtsp import Signal, State import pytest -import respx +from yarl import URL from homeassistant.components.axis.const import DOMAIN from homeassistant.const import ( @@ -45,6 +46,7 @@ from .const import ( ) from tests.common import MockConfigEntry +from tests.test_util.aiohttp import AiohttpClientMocker, AiohttpClientMockResponse type ConfigEntryFactoryType = Callable[[], Coroutine[Any, Any, MockConfigEntry]] type RtspStateType = Callable[[bool], None] @@ -129,15 +131,15 @@ def fixture_config_entry_options() -> MappingProxyType[str, Any]: @pytest.fixture(autouse=True) -def reset_mock_requests() -> Generator[None]: - """Reset respx mock routes after the test.""" +def reset_mock_requests(aioclient_mock: AiohttpClientMocker) -> Generator[None]: + """Reset mocked HTTP routes after the test.""" yield - respx.mock.clear() + aioclient_mock.clear_requests() @pytest.fixture(name="mock_requests") def fixture_request( - respx_mock: respx.MockRouter, + aioclient_mock: AiohttpClientMocker, port_management_payload: dict[str, Any], param_properties_payload: str, param_ports_payload: str, @@ -146,90 +148,91 @@ def fixture_request( """Mock default Vapix requests responses.""" def __mock_default_requests(host: str) -> None: - respx_mock(base_url=f"http://{host}:80") + def _url_pattern(path: str) -> re.Pattern[str]: + return re.compile(rf"^https?://{re.escape(host)}(?::\d+)?{path}$") + + def _text_response(url: URL, text: str) -> AiohttpClientMockResponse: + return AiohttpClientMockResponse( + "post", + url, + text=text, + headers={"Content-Type": "text/plain"}, + ) + + async def _param_cgi_response( + _method: str, url: URL, data: dict[str, Any] | None + ) -> AiohttpClientMockResponse: + group = (data or {}).get("group") + if group == "root.Brand": + return _text_response(url, BRAND_RESPONSE) + if group == "root.Image": + return _text_response(url, IMAGE_RESPONSE) + if group == "root.Input": + return _text_response(url, PORTS_RESPONSE) + if group == "root.IOPort": + return _text_response(url, param_ports_payload) + if group == "root.Output": + return _text_response(url, PORTS_RESPONSE) + if group == "root.Properties": + return _text_response(url, param_properties_payload) + if group == "root.PTZ": + return _text_response(url, PTZ_RESPONSE) + if group == "root.StreamProfile": + return _text_response(url, STREAM_PROFILES_RESPONSE) + return _text_response(url, "") if host != DEFAULT_HOST: - respx.post("/axis-cgi/apidiscovery.cgi").respond( + aioclient_mock.post( + _url_pattern("/axis-cgi/apidiscovery.cgi"), json=API_DISCOVERY_RESPONSE, ) - respx.post("/axis-cgi/basicdeviceinfo.cgi").respond( + aioclient_mock.post( + _url_pattern("/axis-cgi/basicdeviceinfo.cgi"), json=BASIC_DEVICE_INFO_RESPONSE, ) - respx.post("/axis-cgi/io/portmanagement.cgi").respond( + aioclient_mock.post( + _url_pattern("/axis-cgi/io/portmanagement.cgi"), json=port_management_payload, ) - respx.post("/axis-cgi/mqtt/client.cgi").respond( - json=MQTT_CLIENT_RESPONSE, status_code=mqtt_status_code + aioclient_mock.post( + _url_pattern("/axis-cgi/mqtt/client.cgi"), + json=MQTT_CLIENT_RESPONSE, + status=mqtt_status_code, ) - respx.post("/axis-cgi/streamprofile.cgi").respond( + aioclient_mock.post( + _url_pattern("/axis-cgi/streamprofile.cgi"), json=STREAM_PROFILES_RESPONSE, ) - respx.post("/axis-cgi/viewarea/info.cgi").respond(json=VIEW_AREAS_RESPONSE) - respx.post( - "/axis-cgi/param.cgi", - data={"action": "list", "group": "root.Brand"}, - ).respond( - text=BRAND_RESPONSE, - headers={"Content-Type": "text/plain"}, + aioclient_mock.post( + _url_pattern("/axis-cgi/viewarea/info.cgi"), + json=VIEW_AREAS_RESPONSE, ) - respx.post( - "/axis-cgi/param.cgi", - data={"action": "list", "group": "root.Image"}, - ).respond( - text=IMAGE_RESPONSE, - headers={"Content-Type": "text/plain"}, + aioclient_mock.post( + _url_pattern("/axis-cgi/param.cgi"), + side_effect=_param_cgi_response, ) - respx.post( - "/axis-cgi/param.cgi", - data={"action": "list", "group": "root.Input"}, - ).respond( - text=PORTS_RESPONSE, - headers={"Content-Type": "text/plain"}, - ) - respx.post( - "/axis-cgi/param.cgi", - data={"action": "list", "group": "root.IOPort"}, - ).respond( - text=param_ports_payload, - headers={"Content-Type": "text/plain"}, - ) - respx.post( - "/axis-cgi/param.cgi", - data={"action": "list", "group": "root.Output"}, - ).respond( - text=PORTS_RESPONSE, - headers={"Content-Type": "text/plain"}, - ) - respx.post( - "/axis-cgi/param.cgi", - data={"action": "list", "group": "root.Properties"}, - ).respond( - text=param_properties_payload, - headers={"Content-Type": "text/plain"}, - ) - respx.post( - "/axis-cgi/param.cgi", - data={"action": "list", "group": "root.PTZ"}, - ).respond( - text=PTZ_RESPONSE, - headers={"Content-Type": "text/plain"}, - ) - respx.post( - "/axis-cgi/param.cgi", - data={"action": "list", "group": "root.StreamProfile"}, - ).respond( - text=STREAM_PROFILES_RESPONSE, - headers={"Content-Type": "text/plain"}, - ) - respx.post("/axis-cgi/applications/list.cgi").respond( + aioclient_mock.post( + _url_pattern("/axis-cgi/applications/list.cgi"), text=APPLICATIONS_LIST_RESPONSE, headers={"Content-Type": "text/xml"}, ) - respx.post("/local/fenceguard/control.cgi").respond(json=APP_VMD4_RESPONSE) - respx.post("/local/loiteringguard/control.cgi").respond(json=APP_VMD4_RESPONSE) - respx.post("/local/motionguard/control.cgi").respond(json=APP_VMD4_RESPONSE) - respx.post("/local/vmd/control.cgi").respond(json=APP_VMD4_RESPONSE) - respx.post("/local/objectanalytics/control.cgi").respond(json=APP_AOA_RESPONSE) + aioclient_mock.post( + _url_pattern("/local/fenceguard/control.cgi"), json=APP_VMD4_RESPONSE + ) + aioclient_mock.post( + _url_pattern("/local/loiteringguard/control.cgi"), + json=APP_VMD4_RESPONSE, + ) + aioclient_mock.post( + _url_pattern("/local/motionguard/control.cgi"), json=APP_VMD4_RESPONSE + ) + aioclient_mock.post( + _url_pattern("/local/vmd/control.cgi"), json=APP_VMD4_RESPONSE + ) + aioclient_mock.post( + _url_pattern("/local/objectanalytics/control.cgi"), + json=APP_AOA_RESPONSE, + ) return __mock_default_requests @@ -241,12 +244,20 @@ def api_discovery_items() -> dict[str, Any]: @pytest.fixture(autouse=True) -def fixture_api_discovery(api_discovery_items: dict[str, Any]) -> None: +def fixture_api_discovery( + aioclient_mock: AiohttpClientMocker, + api_discovery_items: dict[str, Any], +) -> None: """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) + aioclient_mock.post( + re.compile( + rf"^https?://{re.escape(DEFAULT_HOST)}(?::\d+)?/axis-cgi/apidiscovery.cgi$" + ), + json=data, + ) @pytest.fixture(name="port_management_payload") diff --git a/tests/components/axis/test_light.py b/tests/components/axis/test_light.py index ccff3d06e2d..b5b0121268b 100644 --- a/tests/components/axis/test_light.py +++ b/tests/components/axis/test_light.py @@ -1,12 +1,14 @@ """Axis light platform tests.""" +import json +import re from typing import Any from unittest.mock import patch from axis.models.api import CONTEXT import pytest -import respx from syrupy.assertion import SnapshotAssertion +from yarl import URL from homeassistant.components.light import ATTR_BRIGHTNESS, DOMAIN as LIGHT_DOMAIN from homeassistant.const import ( @@ -23,6 +25,7 @@ from .conftest import ConfigEntryFactoryType, RtspEventMock from .const import DEFAULT_HOST, NAME from tests.common import snapshot_platform +from tests.test_util.aiohttp import AiohttpClientMocker, AiohttpClientMockResponse API_DISCOVERY_LIGHT_CONTROL = { "id": "light-control", @@ -51,23 +54,54 @@ def light_control_items() -> list[dict[str, Any]]: @pytest.fixture(autouse=True) -def light_control_fixture(light_control_items: list[dict[str, Any]]) -> None: +def light_control_fixture( + aioclient_mock: AiohttpClientMocker, + light_control_items: list[dict[str, Any]], +) -> None: """Light control mock response.""" - data = { - "apiVersion": "1.1", - "context": CONTEXT, - "method": "getLightInformation", - "data": {"items": light_control_items}, - } - respx.post( - f"http://{DEFAULT_HOST}:80/axis-cgi/lightcontrol.cgi", - json={ - "apiVersion": "1.1", - "context": CONTEXT, - "method": "getLightInformation", - }, - ).respond( - json=data, + + async def _light_control_response( + _method: str, url: URL, data: bytes | dict[str, Any] | None + ) -> AiohttpClientMockResponse: + payload: dict[str, Any] + if isinstance(data, bytes): + payload = json.loads(data) + elif isinstance(data, dict): + payload = data + else: + payload = {} + + request_method = payload.get("method") + + if request_method == "getCurrentIntensity": + response_data = { + "apiVersion": "1.1", + "context": "Axis library", + "method": "getCurrentIntensity", + "data": {"intensity": 100}, + } + elif request_method == "getValidIntensity": + response_data = { + "apiVersion": "1.1", + "context": "Axis library", + "method": "getValidIntensity", + "data": {"ranges": [{"low": 0, "high": 150}]}, + } + else: + response_data = { + "apiVersion": "1.1", + "context": CONTEXT, + "method": "getLightInformation", + "data": {"items": light_control_items}, + } + + return AiohttpClientMockResponse("post", url, json=response_data) + + aioclient_mock.post( + re.compile( + rf"^https?://{re.escape(DEFAULT_HOST)}(?::\d+)?/axis-cgi/lightcontrol.cgi$" + ), + side_effect=_light_control_response, ) @@ -100,40 +134,6 @@ async def test_lights( snapshot: SnapshotAssertion, ) -> None: """Test that lights are loaded properly.""" - # Add light - respx.post( - f"http://{DEFAULT_HOST}:80/axis-cgi/lightcontrol.cgi", - json={ - "apiVersion": "1.1", - "context": CONTEXT, - "method": "getCurrentIntensity", - "params": {"lightID": "led0"}, - }, - ).respond( - json={ - "apiVersion": "1.1", - "context": "Axis library", - "method": "getCurrentIntensity", - "data": {"intensity": 100}, - }, - ) - respx.post( - f"http://{DEFAULT_HOST}:80/axis-cgi/lightcontrol.cgi", - json={ - "apiVersion": "1.1", - "context": CONTEXT, - "method": "getValidIntensity", - "params": {"lightID": "led0"}, - }, - ).respond( - json={ - "apiVersion": "1.1", - "context": "Axis library", - "method": "getValidIntensity", - "data": {"ranges": [{"low": 0, "high": 150}]}, - }, - ) - with patch("homeassistant.components.axis.PLATFORMS", [Platform.LIGHT]): config_entry = await config_entry_factory()