Fix asyncio DeprecationWarning [3.12] (#98989)
* Fix asyncio DeprecationWarning [3.12] * Use AsyncMock * Rewrite ffmpeg tests * Remove test classes * Rename test filepull/99014/head
parent
bab7d289a6
commit
3ebd7d2fd1
|
@ -0,0 +1,127 @@
|
|||
"""The tests for Home Assistant ffmpeg binary sensor."""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from homeassistant.const import EVENT_HOMEASSISTANT_START
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import assert_setup_component
|
||||
|
||||
CONFIG_NOISE = {
|
||||
"binary_sensor": {"platform": "ffmpeg_noise", "input": "testinputvideo"}
|
||||
}
|
||||
CONFIG_MOTION = {
|
||||
"binary_sensor": {"platform": "ffmpeg_motion", "input": "testinputvideo"}
|
||||
}
|
||||
|
||||
|
||||
# -- ffmpeg noise binary_sensor --
|
||||
|
||||
|
||||
async def test_noise_setup_component(hass: HomeAssistant) -> None:
|
||||
"""Set up ffmpeg component."""
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
await async_setup_component(hass, "binary_sensor", CONFIG_NOISE)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert hass.states.get("binary_sensor.ffmpeg_noise") is not None
|
||||
|
||||
|
||||
@patch("haffmpeg.sensor.SensorNoise.open_sensor", side_effect=AsyncMock())
|
||||
async def test_noise_setup_component_start(mock_start, hass: HomeAssistant):
|
||||
"""Set up ffmpeg component."""
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
await async_setup_component(hass, "binary_sensor", CONFIG_NOISE)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert hass.states.get("binary_sensor.ffmpeg_noise") is not None
|
||||
|
||||
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_start.called
|
||||
|
||||
entity = hass.states.get("binary_sensor.ffmpeg_noise")
|
||||
assert entity.state == "unavailable"
|
||||
|
||||
|
||||
@patch("haffmpeg.sensor.SensorNoise")
|
||||
async def test_noise_setup_component_start_callback(mock_ffmpeg, hass: HomeAssistant):
|
||||
"""Set up ffmpeg component."""
|
||||
mock_ffmpeg().open_sensor.side_effect = AsyncMock()
|
||||
mock_ffmpeg().close = AsyncMock()
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
await async_setup_component(hass, "binary_sensor", CONFIG_NOISE)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert hass.states.get("binary_sensor.ffmpeg_noise") is not None
|
||||
|
||||
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity = hass.states.get("binary_sensor.ffmpeg_noise")
|
||||
assert entity.state == "off"
|
||||
|
||||
hass.async_add_job(mock_ffmpeg.call_args[0][1], True)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity = hass.states.get("binary_sensor.ffmpeg_noise")
|
||||
assert entity.state == "on"
|
||||
|
||||
|
||||
# -- ffmpeg motion binary_sensor --
|
||||
|
||||
|
||||
async def test_motion_setup_component(hass: HomeAssistant) -> None:
|
||||
"""Set up ffmpeg component."""
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
await async_setup_component(hass, "binary_sensor", CONFIG_MOTION)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert hass.states.get("binary_sensor.ffmpeg_motion") is not None
|
||||
|
||||
|
||||
@patch("haffmpeg.sensor.SensorMotion.open_sensor", side_effect=AsyncMock())
|
||||
async def test_motion_setup_component_start(mock_start, hass: HomeAssistant):
|
||||
"""Set up ffmpeg component."""
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
await async_setup_component(hass, "binary_sensor", CONFIG_MOTION)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert hass.states.get("binary_sensor.ffmpeg_motion") is not None
|
||||
|
||||
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_start.called
|
||||
|
||||
entity = hass.states.get("binary_sensor.ffmpeg_motion")
|
||||
assert entity.state == "unavailable"
|
||||
|
||||
|
||||
@patch("haffmpeg.sensor.SensorMotion")
|
||||
async def test_motion_setup_component_start_callback(mock_ffmpeg, hass: HomeAssistant):
|
||||
"""Set up ffmpeg component."""
|
||||
mock_ffmpeg().open_sensor.side_effect = AsyncMock()
|
||||
mock_ffmpeg().close = AsyncMock()
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
await async_setup_component(hass, "binary_sensor", CONFIG_MOTION)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert hass.states.get("binary_sensor.ffmpeg_motion") is not None
|
||||
|
||||
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity = hass.states.get("binary_sensor.ffmpeg_motion")
|
||||
assert entity.state == "off"
|
||||
|
||||
hass.async_add_job(mock_ffmpeg.call_args[0][1], True)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity = hass.states.get("binary_sensor.ffmpeg_motion")
|
||||
assert entity.state == "on"
|
|
@ -1,130 +0,0 @@
|
|||
"""The tests for Home Assistant ffmpeg binary sensor."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.setup import setup_component
|
||||
|
||||
from tests.common import assert_setup_component, get_test_home_assistant, mock_coro
|
||||
|
||||
|
||||
class TestFFmpegNoiseSetup:
|
||||
"""Test class for ffmpeg."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
|
||||
self.config = {
|
||||
"binary_sensor": {"platform": "ffmpeg_noise", "input": "testinputvideo"}
|
||||
}
|
||||
|
||||
def teardown_method(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
def test_setup_component(self):
|
||||
"""Set up ffmpeg component."""
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
setup_component(self.hass, "binary_sensor", self.config)
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert self.hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert self.hass.states.get("binary_sensor.ffmpeg_noise") is not None
|
||||
|
||||
@patch("haffmpeg.sensor.SensorNoise.open_sensor", return_value=mock_coro())
|
||||
def test_setup_component_start(self, mock_start):
|
||||
"""Set up ffmpeg component."""
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
setup_component(self.hass, "binary_sensor", self.config)
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert self.hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert self.hass.states.get("binary_sensor.ffmpeg_noise") is not None
|
||||
|
||||
self.hass.start()
|
||||
assert mock_start.called
|
||||
|
||||
entity = self.hass.states.get("binary_sensor.ffmpeg_noise")
|
||||
assert entity.state == "unavailable"
|
||||
|
||||
@patch("haffmpeg.sensor.SensorNoise")
|
||||
def test_setup_component_start_callback(self, mock_ffmpeg):
|
||||
"""Set up ffmpeg component."""
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
setup_component(self.hass, "binary_sensor", self.config)
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert self.hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert self.hass.states.get("binary_sensor.ffmpeg_noise") is not None
|
||||
|
||||
self.hass.start()
|
||||
|
||||
entity = self.hass.states.get("binary_sensor.ffmpeg_noise")
|
||||
assert entity.state == "off"
|
||||
|
||||
self.hass.add_job(mock_ffmpeg.call_args[0][1], True)
|
||||
self.hass.block_till_done()
|
||||
|
||||
entity = self.hass.states.get("binary_sensor.ffmpeg_noise")
|
||||
assert entity.state == "on"
|
||||
|
||||
|
||||
class TestFFmpegMotionSetup:
|
||||
"""Test class for ffmpeg."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
|
||||
self.config = {
|
||||
"binary_sensor": {"platform": "ffmpeg_motion", "input": "testinputvideo"}
|
||||
}
|
||||
|
||||
def teardown_method(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
def test_setup_component(self):
|
||||
"""Set up ffmpeg component."""
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
setup_component(self.hass, "binary_sensor", self.config)
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert self.hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert self.hass.states.get("binary_sensor.ffmpeg_motion") is not None
|
||||
|
||||
@patch("haffmpeg.sensor.SensorMotion.open_sensor", return_value=mock_coro())
|
||||
def test_setup_component_start(self, mock_start):
|
||||
"""Set up ffmpeg component."""
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
setup_component(self.hass, "binary_sensor", self.config)
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert self.hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert self.hass.states.get("binary_sensor.ffmpeg_motion") is not None
|
||||
|
||||
self.hass.start()
|
||||
assert mock_start.called
|
||||
|
||||
entity = self.hass.states.get("binary_sensor.ffmpeg_motion")
|
||||
assert entity.state == "unavailable"
|
||||
|
||||
@patch("haffmpeg.sensor.SensorMotion")
|
||||
def test_setup_component_start_callback(self, mock_ffmpeg):
|
||||
"""Set up ffmpeg component."""
|
||||
with assert_setup_component(1, "binary_sensor"):
|
||||
setup_component(self.hass, "binary_sensor", self.config)
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert self.hass.data["ffmpeg"].binary == "ffmpeg"
|
||||
assert self.hass.states.get("binary_sensor.ffmpeg_motion") is not None
|
||||
|
||||
self.hass.start()
|
||||
|
||||
entity = self.hass.states.get("binary_sensor.ffmpeg_motion")
|
||||
assert entity.state == "off"
|
||||
|
||||
self.hass.add_job(mock_ffmpeg.call_args[0][1], True)
|
||||
self.hass.block_till_done()
|
||||
|
||||
entity = self.hass.states.get("binary_sensor.ffmpeg_motion")
|
||||
assert entity.state == "on"
|
|
@ -1,7 +1,6 @@
|
|||
"""Test the Minecraft Server config flow."""
|
||||
|
||||
import asyncio
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import aiodns
|
||||
from mcstatus.status_response import JavaStatusResponse
|
||||
|
@ -72,9 +71,6 @@ USER_INPUT_PORT_TOO_LARGE = {
|
|||
CONF_HOST: "mc.dummyserver.com:65536",
|
||||
}
|
||||
|
||||
SRV_RECORDS = asyncio.Future()
|
||||
SRV_RECORDS.set_result([QueryMock()])
|
||||
|
||||
|
||||
async def test_show_config_form(hass: HomeAssistant) -> None:
|
||||
"""Test if initial configuration form is shown."""
|
||||
|
@ -173,7 +169,7 @@ async def test_connection_succeeded_with_srv_record(hass: HomeAssistant) -> None
|
|||
"""Test config entry in case of a successful connection with a SRV record."""
|
||||
with patch(
|
||||
"aiodns.DNSResolver.query",
|
||||
return_value=SRV_RECORDS,
|
||||
side_effect=AsyncMock(return_value=[QueryMock()]),
|
||||
), patch(
|
||||
"mcstatus.server.JavaServer.async_status",
|
||||
return_value=JavaStatusResponse(
|
||||
|
|
Loading…
Reference in New Issue