Tweak axis test fixtures (#122469)

* Don't automatically add config entry to hass

* Improve RTSP fixture typing

* Improve typing of config entry factory and remove unnecessary use of it

* Remove redundant fixture in config flow tests

* Parametrize config flow error test
This commit is contained in:
Robert Svensson 2024-07-23 19:27:38 +02:00 committed by GitHub
parent 1b7fb9ae12
commit b53800a69d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 108 additions and 123 deletions

View File

@ -2,10 +2,10 @@
from __future__ import annotations
from collections.abc import Callable, Generator
from collections.abc import Callable, Coroutine, Generator
from copy import deepcopy
from types import MappingProxyType
from typing import Any
from typing import Any, Protocol
from unittest.mock import AsyncMock, patch
from axis.rtsp import Signal, State
@ -13,7 +13,6 @@ import pytest
import respx
from homeassistant.components.axis.const import DOMAIN as AXIS_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_MODEL,
@ -47,6 +46,30 @@ from .const import (
from tests.common import MockConfigEntry
type ConfigEntryFactoryType = Callable[[], Coroutine[Any, Any, MockConfigEntry]]
type RtspStateType = Callable[[bool], None]
class RtspEventMock(Protocol):
"""Fixture to allow mocking received RTSP events."""
def __call__(
self,
topic: str,
data_type: str,
data_value: str,
operation: str = "Initialized",
source_name: str = "",
source_idx: str = "",
) -> None:
"""Send RTSP event."""
class _RtspClientMock(Protocol):
async def __call__(
self, data: dict[str, Any] | None = None, state: str = ""
) -> None: ...
@pytest.fixture(name="mock_setup_entry")
def fixture_setup_entry() -> Generator[AsyncMock]:
@ -66,9 +89,9 @@ def fixture_config_entry(
config_entry_data: MappingProxyType[str, Any],
config_entry_options: MappingProxyType[str, Any],
config_entry_version: int,
) -> ConfigEntry:
) -> MockConfigEntry:
"""Define a config entry fixture."""
config_entry = MockConfigEntry(
return MockConfigEntry(
domain=AXIS_DOMAIN,
entry_id="676abe5b73621446e6550a2e86ffe3dd",
unique_id=FORMATTED_MAC,
@ -76,8 +99,6 @@ def fixture_config_entry(
options=config_entry_options,
version=config_entry_version,
)
config_entry.add_to_hass(hass)
return config_entry
@pytest.fixture(name="config_entry_version")
@ -255,12 +276,13 @@ def fixture_default_requests(mock_requests: Callable[[str], None]) -> None:
@pytest.fixture(name="config_entry_factory")
async def fixture_config_entry_factory(
hass: HomeAssistant,
config_entry: ConfigEntry,
config_entry: MockConfigEntry,
mock_requests: Callable[[str], None],
) -> Callable[[], ConfigEntry]:
) -> ConfigEntryFactoryType:
"""Fixture factory to set up Axis network device."""
async def __mock_setup_config_entry() -> ConfigEntry:
async def __mock_setup_config_entry() -> MockConfigEntry:
config_entry.add_to_hass(hass)
mock_requests(config_entry.data[CONF_HOST])
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@ -271,8 +293,8 @@ async def fixture_config_entry_factory(
@pytest.fixture(name="config_entry_setup")
async def fixture_config_entry_setup(
hass: HomeAssistant, config_entry_factory: Callable[[], ConfigEntry]
) -> ConfigEntry:
config_entry_factory: ConfigEntryFactoryType,
) -> MockConfigEntry:
"""Define a fixture to set up Axis network device."""
return await config_entry_factory()
@ -280,8 +302,8 @@ async def fixture_config_entry_setup(
# RTSP fixtures
@pytest.fixture(autouse=True, name="mock_axis_rtspclient")
def fixture_axis_rtspclient() -> Generator[Callable[[dict | None, str], None]]:
@pytest.fixture(autouse=True, name="_mock_rtsp_client")
def fixture_axis_rtsp_client() -> Generator[_RtspClientMock]:
"""No real RTSP communication allowed."""
with patch("axis.stream_manager.RTSPClient") as rtsp_client_mock:
rtsp_client_mock.return_value.session.state = State.STOPPED
@ -298,7 +320,7 @@ def fixture_axis_rtspclient() -> Generator[Callable[[dict | None, str], None]]:
rtsp_client_mock.return_value.stop = stop_stream
def make_rtsp_call(data: dict | None = None, state: str = "") -> None:
def make_rtsp_call(data: dict[str, Any] | None = None, state: str = "") -> None:
"""Generate a RTSP call."""
axis_streammanager_session_callback = rtsp_client_mock.call_args[0][4]
@ -314,9 +336,7 @@ def fixture_axis_rtspclient() -> Generator[Callable[[dict | None, str], None]]:
@pytest.fixture(autouse=True, name="mock_rtsp_event")
def fixture_rtsp_event(
mock_axis_rtspclient: Callable[[dict | None, str], None],
) -> Callable[[str, str, str, str, str, str], None]:
def fixture_rtsp_event(_mock_rtsp_client: _RtspClientMock) -> RtspEventMock:
"""Fixture to allow mocking received RTSP events."""
def send_event(
@ -361,20 +381,18 @@ def fixture_rtsp_event(
</tt:MetadataStream>
"""
mock_axis_rtspclient(data=event.encode("utf-8"))
_mock_rtsp_client(data=event.encode("utf-8"))
return send_event
@pytest.fixture(autouse=True, name="mock_rtsp_signal_state")
def fixture_rtsp_signal_state(
mock_axis_rtspclient: Callable[[dict | None, str], None],
) -> Callable[[bool], None]:
def fixture_rtsp_signal_state(_mock_rtsp_client: _RtspClientMock) -> RtspStateType:
"""Fixture to allow mocking RTSP state signalling."""
def send_signal(connected: bool) -> None:
"""Signal state change of RTSP connection."""
signal = Signal.PLAYING if connected else Signal.FAILED
mock_axis_rtspclient(state=signal)
_mock_rtsp_client(state=signal)
return send_signal

View File

@ -1,6 +1,5 @@
"""Axis binary sensor platform tests."""
from collections.abc import Callable
from typing import Any
import pytest
@ -12,6 +11,7 @@ from homeassistant.components.binary_sensor import (
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from .conftest import RtspEventMock
from .const import NAME
@ -176,7 +176,7 @@ from .const import NAME
@pytest.mark.usefixtures("config_entry_setup")
async def test_binary_sensors(
hass: HomeAssistant,
mock_rtsp_event: Callable[[str, str, str, str, str, str], None],
mock_rtsp_event: RtspEventMock,
event: dict[str, str],
entity: dict[str, Any],
) -> None:
@ -228,7 +228,7 @@ async def test_binary_sensors(
@pytest.mark.usefixtures("config_entry_setup")
async def test_unsupported_events(
hass: HomeAssistant,
mock_rtsp_event: Callable[[str, str, str, str, str, str], None],
mock_rtsp_event: RtspEventMock,
event: dict[str, str],
) -> None:
"""Validate nothing breaks with unsupported events."""

View File

@ -1,7 +1,5 @@
"""Axis camera platform tests."""
from collections.abc import Callable
import pytest
from homeassistant.components import camera
@ -10,7 +8,6 @@ from homeassistant.components.axis.const import (
DOMAIN as AXIS_DOMAIN,
)
from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_IDLE
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@ -82,13 +79,11 @@ root.Properties.Firmware.BuildDate=Feb 15 2019 09:42
root.Properties.Firmware.BuildNumber=26
root.Properties.Firmware.Version=9.10.1
root.Properties.System.SerialNumber={MAC}
"""
""" # No image format data to signal camera support
@pytest.mark.parametrize("param_properties_payload", [PROPERTY_DATA])
async def test_camera_disabled(
hass: HomeAssistant, config_entry_factory: Callable[[], ConfigEntry]
) -> None:
@pytest.mark.usefixtures("config_entry_setup")
async def test_camera_disabled(hass: HomeAssistant) -> None:
"""Test that Axis camera platform is loaded properly but does not create camera entity."""
await config_entry_factory()
assert len(hass.states.async_entity_ids(CAMERA_DOMAIN)) == 0

View File

@ -2,7 +2,7 @@
from collections.abc import Callable
from ipaddress import ip_address
from unittest.mock import AsyncMock, patch
from unittest.mock import patch
import pytest
@ -17,13 +17,11 @@ from homeassistant.components.axis.const import (
)
from homeassistant.config_entries import (
SOURCE_DHCP,
SOURCE_IGNORE,
SOURCE_REAUTH,
SOURCE_RECONFIGURE,
SOURCE_SSDP,
SOURCE_USER,
SOURCE_ZEROCONF,
ConfigEntry,
)
from homeassistant.const import (
CONF_HOST,
@ -45,21 +43,9 @@ from tests.common import MockConfigEntry
DHCP_FORMATTED_MAC = dr.format_mac(MAC).replace(":", "")
@pytest.fixture(name="mock_config_entry")
async def mock_config_entry_fixture(
hass: HomeAssistant, config_entry: MockConfigEntry, mock_setup_entry: AsyncMock
) -> MockConfigEntry:
"""Mock config entry and setup entry."""
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
@pytest.mark.usefixtures("mock_default_requests", "mock_setup_entry")
@pytest.mark.usefixtures("mock_default_requests")
async def test_flow_manual_configuration(hass: HomeAssistant) -> None:
"""Test that config flow works."""
MockConfigEntry(domain=AXIS_DOMAIN, source=SOURCE_IGNORE).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN, context={"source": SOURCE_USER}
)
@ -93,11 +79,11 @@ async def test_flow_manual_configuration(hass: HomeAssistant) -> None:
async def test_manual_configuration_update_configuration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
config_entry_setup: MockConfigEntry,
mock_requests: Callable[[str], None],
) -> None:
"""Test that config flow fails on already configured device."""
assert mock_config_entry.data[CONF_HOST] == "1.2.3.4"
assert config_entry_setup.data[CONF_HOST] == "1.2.3.4"
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN, context={"source": SOURCE_USER}
@ -121,10 +107,19 @@ async def test_manual_configuration_update_configuration(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_HOST] == "2.3.4.5"
assert config_entry_setup.data[CONF_HOST] == "2.3.4.5"
async def test_flow_fails_faulty_credentials(hass: HomeAssistant) -> None:
@pytest.mark.parametrize(
("exc", "error"),
[
(config_flow.AuthenticationRequired, "invalid_auth"),
(config_flow.CannotConnect, "cannot_connect"),
],
)
async def test_flow_fails_on_api(
hass: HomeAssistant, exc: Exception, error: str
) -> None:
"""Test that config flow fails on faulty credentials."""
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN, context={"source": SOURCE_USER}
@ -135,7 +130,7 @@ async def test_flow_fails_faulty_credentials(hass: HomeAssistant) -> None:
with patch(
"homeassistant.components.axis.config_flow.get_axis_api",
side_effect=config_flow.AuthenticationRequired,
side_effect=exc,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@ -148,37 +143,10 @@ async def test_flow_fails_faulty_credentials(hass: HomeAssistant) -> None:
},
)
assert result["errors"] == {"base": "invalid_auth"}
assert result["errors"] == {"base": error}
async def test_flow_fails_cannot_connect(hass: HomeAssistant) -> None:
"""Test that config flow fails on cannot connect."""
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.axis.config_flow.get_axis_api",
side_effect=config_flow.CannotConnect,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_PROTOCOL: "http",
CONF_HOST: "1.2.3.4",
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 80,
},
)
assert result["errors"] == {"base": "cannot_connect"}
@pytest.mark.usefixtures("mock_default_requests", "mock_setup_entry")
@pytest.mark.usefixtures("mock_default_requests")
async def test_flow_create_entry_multiple_existing_entries_of_same_model(
hass: HomeAssistant,
) -> None:
@ -229,18 +197,18 @@ async def test_flow_create_entry_multiple_existing_entries_of_same_model(
async def test_reauth_flow_update_configuration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
config_entry_setup: MockConfigEntry,
mock_requests: Callable[[str], None],
) -> None:
"""Test that config flow fails on already configured device."""
assert mock_config_entry.data[CONF_HOST] == "1.2.3.4"
assert mock_config_entry.data[CONF_USERNAME] == "root"
assert mock_config_entry.data[CONF_PASSWORD] == "pass"
assert config_entry_setup.data[CONF_HOST] == "1.2.3.4"
assert config_entry_setup.data[CONF_USERNAME] == "root"
assert config_entry_setup.data[CONF_PASSWORD] == "pass"
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN,
context={"source": SOURCE_REAUTH},
data=mock_config_entry.data,
data=config_entry_setup.data,
)
assert result["type"] is FlowResultType.FORM
@ -261,28 +229,28 @@ async def test_reauth_flow_update_configuration(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_PROTOCOL] == "https"
assert mock_config_entry.data[CONF_HOST] == "2.3.4.5"
assert mock_config_entry.data[CONF_PORT] == 443
assert mock_config_entry.data[CONF_USERNAME] == "user2"
assert mock_config_entry.data[CONF_PASSWORD] == "pass2"
assert config_entry_setup.data[CONF_PROTOCOL] == "https"
assert config_entry_setup.data[CONF_HOST] == "2.3.4.5"
assert config_entry_setup.data[CONF_PORT] == 443
assert config_entry_setup.data[CONF_USERNAME] == "user2"
assert config_entry_setup.data[CONF_PASSWORD] == "pass2"
async def test_reconfiguration_flow_update_configuration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
config_entry_setup: MockConfigEntry,
mock_requests: Callable[[str], None],
) -> None:
"""Test that config flow reconfiguration updates configured device."""
assert mock_config_entry.data[CONF_HOST] == "1.2.3.4"
assert mock_config_entry.data[CONF_USERNAME] == "root"
assert mock_config_entry.data[CONF_PASSWORD] == "pass"
assert config_entry_setup.data[CONF_HOST] == "1.2.3.4"
assert config_entry_setup.data[CONF_USERNAME] == "root"
assert config_entry_setup.data[CONF_PASSWORD] == "pass"
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN,
context={
"source": SOURCE_RECONFIGURE,
"entry_id": mock_config_entry.entry_id,
"entry_id": config_entry_setup.entry_id,
},
)
@ -301,11 +269,11 @@ async def test_reconfiguration_flow_update_configuration(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_PROTOCOL] == "http"
assert mock_config_entry.data[CONF_HOST] == "2.3.4.5"
assert mock_config_entry.data[CONF_PORT] == 80
assert mock_config_entry.data[CONF_USERNAME] == "user"
assert mock_config_entry.data[CONF_PASSWORD] == "pass"
assert config_entry_setup.data[CONF_PROTOCOL] == "http"
assert config_entry_setup.data[CONF_HOST] == "2.3.4.5"
assert config_entry_setup.data[CONF_PORT] == 80
assert config_entry_setup.data[CONF_USERNAME] == "user"
assert config_entry_setup.data[CONF_PASSWORD] == "pass"
@pytest.mark.parametrize(
@ -372,7 +340,7 @@ async def test_reconfiguration_flow_update_configuration(
),
],
)
@pytest.mark.usefixtures("mock_default_requests", "mock_setup_entry")
@pytest.mark.usefixtures("mock_default_requests")
async def test_discovery_flow(
hass: HomeAssistant,
source: str,
@ -455,12 +423,12 @@ async def test_discovery_flow(
)
async def test_discovered_device_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
config_entry_setup: MockConfigEntry,
source: str,
discovery_info: BaseServiceInfo,
) -> None:
"""Test that discovery doesn't setup already configured devices."""
assert mock_config_entry.data[CONF_HOST] == DEFAULT_HOST
assert config_entry_setup.data[CONF_HOST] == DEFAULT_HOST
result = await hass.config_entries.flow.async_init(
AXIS_DOMAIN, data=discovery_info, context={"source": source}
@ -468,7 +436,7 @@ async def test_discovered_device_already_configured(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert mock_config_entry.data[CONF_HOST] == DEFAULT_HOST
assert config_entry_setup.data[CONF_HOST] == DEFAULT_HOST
@pytest.mark.parametrize(
@ -513,14 +481,14 @@ async def test_discovered_device_already_configured(
)
async def test_discovery_flow_updated_configuration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
config_entry_setup: MockConfigEntry,
mock_requests: Callable[[str], None],
source: str,
discovery_info: BaseServiceInfo,
expected_port: int,
) -> None:
"""Test that discovery flow update configuration with new parameters."""
assert mock_config_entry.data == {
assert config_entry_setup.data == {
CONF_HOST: DEFAULT_HOST,
CONF_PORT: 80,
CONF_USERNAME: "root",
@ -537,7 +505,7 @@ async def test_discovery_flow_updated_configuration(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert mock_config_entry.data == {
assert config_entry_setup.data == {
CONF_HOST: "2.3.4.5",
CONF_PORT: expected_port,
CONF_USERNAME: "root",
@ -646,7 +614,7 @@ async def test_discovery_flow_ignore_link_local_address(
async def test_option_flow(
hass: HomeAssistant, config_entry_setup: ConfigEntry
hass: HomeAssistant, config_entry_setup: MockConfigEntry
) -> None:
"""Test config flow options."""
assert CONF_STREAM_PROFILE not in config_entry_setup.options

View File

@ -25,6 +25,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from .conftest import RtspEventMock, RtspStateType
from .const import (
API_DISCOVERY_BASIC_DEVICE_INFO,
API_DISCOVERY_MQTT,
@ -152,8 +153,8 @@ async def test_update_address(
@pytest.mark.usefixtures("config_entry_setup")
async def test_device_unavailable(
hass: HomeAssistant,
mock_rtsp_event: Callable[[str, str, str, str, str, str], None],
mock_rtsp_signal_state: Callable[[bool], None],
mock_rtsp_event: RtspEventMock,
mock_rtsp_signal_state: RtspStateType,
) -> None:
"""Successful setup."""
# Provide an entity that can be used to verify connection state on
@ -191,6 +192,7 @@ async def test_device_not_accessible(
hass: HomeAssistant, config_entry: ConfigEntry
) -> None:
"""Failed setup schedules a retry of setup."""
config_entry.add_to_hass(hass)
with patch.object(axis, "get_axis_api", side_effect=axis.errors.CannotConnect):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@ -202,6 +204,7 @@ async def test_device_trigger_reauth_flow(
hass: HomeAssistant, config_entry: ConfigEntry
) -> None:
"""Failed authentication trigger a reauthentication flow."""
config_entry.add_to_hass(hass)
with (
patch.object(
axis, "get_axis_api", side_effect=axis.errors.AuthenticationRequired
@ -219,6 +222,7 @@ async def test_device_unknown_error(
hass: HomeAssistant, config_entry: ConfigEntry
) -> None:
"""Unknown errors are handled."""
config_entry.add_to_hass(hass)
with patch.object(axis, "get_axis_api", side_effect=Exception):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

View File

@ -18,6 +18,8 @@ async def test_setup_entry_fails(
hass: HomeAssistant, config_entry: ConfigEntry
) -> None:
"""Test successful setup of entry."""
config_entry.add_to_hass(hass)
mock_device = Mock()
mock_device.async_setup = AsyncMock(return_value=False)
@ -42,6 +44,7 @@ async def test_unload_entry(
@pytest.mark.parametrize("config_entry_version", [1])
async def test_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Test successful migration of entry data."""
config_entry.add_to_hass(hass)
assert config_entry.version == 1
mock_device = Mock()

View File

@ -1,6 +1,5 @@
"""Axis light platform tests."""
from collections.abc import Callable
from typing import Any
from unittest.mock import patch
@ -18,6 +17,7 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from .conftest import RtspEventMock
from .const import DEFAULT_HOST, NAME
API_DISCOVERY_LIGHT_CONTROL = {
@ -72,7 +72,7 @@ def light_control_fixture(light_control_items: list[dict[str, Any]]) -> None:
@pytest.mark.usefixtures("config_entry_setup")
async def test_no_light_entity_without_light_control_representation(
hass: HomeAssistant,
mock_rtsp_event: Callable[[str, str, str, str, str, str], None],
mock_rtsp_event: RtspEventMock,
) -> None:
"""Verify no lights entities get created without light control representation."""
mock_rtsp_event(
@ -89,10 +89,7 @@ async def test_no_light_entity_without_light_control_representation(
@pytest.mark.parametrize("api_discovery_items", [API_DISCOVERY_LIGHT_CONTROL])
@pytest.mark.usefixtures("config_entry_setup")
async def test_lights(
hass: HomeAssistant,
mock_rtsp_event: Callable[[str, str, str, str, str, str], None],
) -> None:
async def test_lights(hass: HomeAssistant, mock_rtsp_event: RtspEventMock) -> None:
"""Test that lights are loaded properly."""
# Add light
respx.post(

View File

@ -1,6 +1,5 @@
"""Axis switch platform tests."""
from collections.abc import Callable
from unittest.mock import patch
from axis.models.api import CONTEXT
@ -16,6 +15,7 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from .conftest import RtspEventMock
from .const import API_DISCOVERY_PORT_MANAGEMENT, NAME
PORT_DATA = """root.IOPort.I0.Configurable=yes
@ -33,7 +33,7 @@ root.IOPort.I1.Output.Active=open
@pytest.mark.usefixtures("config_entry_setup")
async def test_switches_with_port_cgi(
hass: HomeAssistant,
mock_rtsp_event: Callable[[str, str, str, str, str, str], None],
mock_rtsp_event: RtspEventMock,
) -> None:
"""Test that switches are loaded properly using port.cgi."""
mock_rtsp_event(
@ -118,7 +118,7 @@ PORT_MANAGEMENT_RESPONSE = {
@pytest.mark.usefixtures("config_entry_setup")
async def test_switches_with_port_management(
hass: HomeAssistant,
mock_rtsp_event: Callable[[str, str, str, str, str, str], None],
mock_rtsp_event: RtspEventMock,
) -> None:
"""Test that switches are loaded properly using port management."""
mock_rtsp_event(