Convert remaining image processing tests to async (#64506)

This commit is contained in:
Paulus Schoutsen 2022-01-20 09:00:58 -08:00 committed by GitHub
parent ddf548cd27
commit 6803219133
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 264 additions and 283 deletions

View File

@ -2,200 +2,188 @@
import asyncio
from unittest.mock import PropertyMock, patch
import pytest
from homeassistant.components import camera, image_processing as ip
from homeassistant.components.openalpr_cloud.image_processing import OPENALPR_API_URL
from homeassistant.core import callback
from homeassistant.setup import setup_component
from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component, get_test_home_assistant, load_fixture
from tests.common import assert_setup_component, async_capture_events, load_fixture
from tests.components.image_processing import common
class TestOpenAlprCloudSetup:
"""Test class for image processing."""
@pytest.fixture
async def setup_openalpr_cloud(hass):
"""Set up openalpr cloud."""
config = {
ip.DOMAIN: {
"platform": "openalpr_cloud",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"region": "eu",
"api_key": "sk_abcxyz123456",
},
"camera": {"platform": "demo"},
}
def setup_method(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_setup_platform(self):
"""Set up platform with one entity."""
config = {
ip.DOMAIN: {
"platform": "openalpr_cloud",
"source": {"entity_id": "camera.demo_camera"},
"region": "eu",
"api_key": "sk_abcxyz123456",
},
"camera": {"platform": "demo"},
}
with assert_setup_component(1, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
self.hass.block_till_done()
assert self.hass.states.get("image_processing.openalpr_demo_camera")
def test_setup_platform_name(self):
"""Set up platform with one entity and set name."""
config = {
ip.DOMAIN: {
"platform": "openalpr_cloud",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"region": "eu",
"api_key": "sk_abcxyz123456",
},
"camera": {"platform": "demo"},
}
with assert_setup_component(1, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
self.hass.block_till_done()
assert self.hass.states.get("image_processing.test_local")
def test_setup_platform_without_api_key(self):
"""Set up platform with one entity without api_key."""
config = {
ip.DOMAIN: {
"platform": "openalpr_cloud",
"source": {"entity_id": "camera.demo_camera"},
"region": "eu",
},
"camera": {"platform": "demo"},
}
with assert_setup_component(0, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
def test_setup_platform_without_region(self):
"""Set up platform with one entity without region."""
config = {
ip.DOMAIN: {
"platform": "openalpr_cloud",
"source": {"entity_id": "camera.demo_camera"},
"api_key": "sk_abcxyz123456",
},
"camera": {"platform": "demo"},
}
with assert_setup_component(0, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
with patch(
"homeassistant.components.openalpr_cloud.image_processing."
"OpenAlprCloudEntity.should_poll",
new_callable=PropertyMock(return_value=False),
):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
class TestOpenAlprCloud:
"""Test class for image processing."""
@pytest.fixture
async def alpr_events(hass):
"""Listen for events."""
return async_capture_events(hass, "image_processing.found_plate")
def setup_method(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
config = {
ip.DOMAIN: {
"platform": "openalpr_cloud",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"region": "eu",
"api_key": "sk_abcxyz123456",
},
"camera": {"platform": "demo"},
}
PARAMS = {
"secret_key": "sk_abcxyz123456",
"tasks": "plate",
"return_image": 0,
"country": "eu",
}
with patch(
"homeassistant.components.openalpr_cloud.image_processing."
"OpenAlprCloudEntity.should_poll",
new_callable=PropertyMock(return_value=False),
):
setup_component(self.hass, ip.DOMAIN, config)
self.hass.block_till_done()
self.alpr_events = []
async def test_setup_platform(hass):
"""Set up platform with one entity."""
config = {
ip.DOMAIN: {
"platform": "openalpr_cloud",
"source": {"entity_id": "camera.demo_camera"},
"region": "eu",
"api_key": "sk_abcxyz123456",
},
"camera": {"platform": "demo"},
}
@callback
def mock_alpr_event(event):
"""Mock event."""
self.alpr_events.append(event)
with assert_setup_component(1, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
self.hass.bus.listen("image_processing.found_plate", mock_alpr_event)
assert hass.states.get("image_processing.openalpr_demo_camera")
self.params = {
"secret_key": "sk_abcxyz123456",
"tasks": "plate",
"return_image": 0,
"country": "eu",
}
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
async def test_setup_platform_name(hass):
"""Set up platform with one entity and set name."""
config = {
ip.DOMAIN: {
"platform": "openalpr_cloud",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"region": "eu",
"api_key": "sk_abcxyz123456",
},
"camera": {"platform": "demo"},
}
def test_openalpr_process_image(self, aioclient_mock):
"""Set up and scan a picture and test plates from event."""
aioclient_mock.post(
OPENALPR_API_URL,
params=self.params,
text=load_fixture("alpr_cloud.json"),
status=200,
)
with assert_setup_component(1, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
with patch(
"homeassistant.components.camera.async_get_image",
return_value=camera.Image("image/jpeg", b"image"),
):
common.scan(self.hass, entity_id="image_processing.test_local")
self.hass.block_till_done()
assert hass.states.get("image_processing.test_local")
state = self.hass.states.get("image_processing.test_local")
assert len(aioclient_mock.mock_calls) == 1
assert len(self.alpr_events) == 5
assert state.attributes.get("vehicles") == 1
assert state.state == "H786P0J"
async def test_setup_platform_without_api_key(hass):
"""Set up platform with one entity without api_key."""
config = {
ip.DOMAIN: {
"platform": "openalpr_cloud",
"source": {"entity_id": "camera.demo_camera"},
"region": "eu",
},
"camera": {"platform": "demo"},
}
event_data = [
event.data
for event in self.alpr_events
if event.data.get("plate") == "H786P0J"
]
assert len(event_data) == 1
assert event_data[0]["plate"] == "H786P0J"
assert event_data[0]["confidence"] == float(90.436699)
assert event_data[0]["entity_id"] == "image_processing.test_local"
with assert_setup_component(0, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
def test_openalpr_process_image_api_error(self, aioclient_mock):
"""Set up and scan a picture and test api error."""
aioclient_mock.post(
OPENALPR_API_URL,
params=self.params,
text="{'error': 'error message'}",
status=400,
)
with patch(
"homeassistant.components.camera.async_get_image",
return_value=camera.Image("image/jpeg", b"image"),
):
common.scan(self.hass, entity_id="image_processing.test_local")
self.hass.block_till_done()
async def test_setup_platform_without_region(hass):
"""Set up platform with one entity without region."""
config = {
ip.DOMAIN: {
"platform": "openalpr_cloud",
"source": {"entity_id": "camera.demo_camera"},
"api_key": "sk_abcxyz123456",
},
"camera": {"platform": "demo"},
}
assert len(aioclient_mock.mock_calls) == 1
assert len(self.alpr_events) == 0
with assert_setup_component(0, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
def test_openalpr_process_image_api_timeout(self, aioclient_mock):
"""Set up and scan a picture and test api error."""
aioclient_mock.post(
OPENALPR_API_URL, params=self.params, exc=asyncio.TimeoutError()
)
with patch(
"homeassistant.components.camera.async_get_image",
return_value=camera.Image("image/jpeg", b"image"),
):
common.scan(self.hass, entity_id="image_processing.test_local")
self.hass.block_till_done()
async def test_openalpr_process_image(
alpr_events, setup_openalpr_cloud, hass, aioclient_mock
):
"""Set up and scan a picture and test plates from event."""
aioclient_mock.post(
OPENALPR_API_URL,
params=PARAMS,
text=load_fixture("alpr_cloud.json"),
status=200,
)
assert len(aioclient_mock.mock_calls) == 1
assert len(self.alpr_events) == 0
with patch(
"homeassistant.components.camera.async_get_image",
return_value=camera.Image("image/jpeg", b"image"),
):
common.async_scan(hass, entity_id="image_processing.test_local")
await hass.async_block_till_done()
state = hass.states.get("image_processing.test_local")
assert len(aioclient_mock.mock_calls) == 1
assert len(alpr_events) == 5
assert state.attributes.get("vehicles") == 1
assert state.state == "H786P0J"
event_data = [
event.data for event in alpr_events if event.data.get("plate") == "H786P0J"
]
assert len(event_data) == 1
assert event_data[0]["plate"] == "H786P0J"
assert event_data[0]["confidence"] == float(90.436699)
assert event_data[0]["entity_id"] == "image_processing.test_local"
async def test_openalpr_process_image_api_error(
alpr_events, setup_openalpr_cloud, hass, aioclient_mock
):
"""Set up and scan a picture and test api error."""
aioclient_mock.post(
OPENALPR_API_URL,
params=PARAMS,
text="{'error': 'error message'}",
status=400,
)
with patch(
"homeassistant.components.camera.async_get_image",
return_value=camera.Image("image/jpeg", b"image"),
):
common.async_scan(hass, entity_id="image_processing.test_local")
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(alpr_events) == 0
async def test_openalpr_process_image_api_timeout(
alpr_events, setup_openalpr_cloud, hass, aioclient_mock
):
"""Set up and scan a picture and test api error."""
aioclient_mock.post(OPENALPR_API_URL, params=PARAMS, exc=asyncio.TimeoutError())
with patch(
"homeassistant.components.camera.async_get_image",
return_value=camera.Image("image/jpeg", b"image"),
):
common.async_scan(hass, entity_id="image_processing.test_local")
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(alpr_events) == 0

View File

@ -1,16 +1,52 @@
"""The tests for the openalpr local platform."""
from unittest.mock import MagicMock, PropertyMock, patch
import pytest
import homeassistant.components.image_processing as ip
from homeassistant.const import ATTR_ENTITY_PICTURE
from homeassistant.core import callback
from homeassistant.setup import setup_component
from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component, get_test_home_assistant, load_fixture
from tests.common import assert_setup_component, async_capture_events, load_fixture
from tests.components.image_processing import common
def mock_async_subprocess():
@pytest.fixture
async def setup_openalpr_local(hass):
"""Set up openalpr local."""
config = {
ip.DOMAIN: {
"platform": "openalpr_local",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"region": "eu",
},
"camera": {"platform": "demo"},
}
with patch(
"homeassistant.components.openalpr_local.image_processing."
"OpenAlprLocalEntity.should_poll",
new_callable=PropertyMock(return_value=False),
):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
@pytest.fixture
def url(hass, setup_openalpr_local):
"""Return the camera URL."""
state = hass.states.get("camera.demo_camera")
return f"{hass.config.internal_url}{state.attributes.get(ATTR_ENTITY_PICTURE)}"
@pytest.fixture
async def alpr_events(hass):
"""Listen for events."""
return async_capture_events(hass, "image_processing.found_plate")
@pytest.fixture
def popen_mock():
"""Get a Popen mock back."""
async_popen = MagicMock()
@ -20,130 +56,87 @@ def mock_async_subprocess():
return (fixture, None)
async_popen.communicate = communicate
return async_popen
with patch("asyncio.create_subprocess_exec", return_value=async_popen) as mock:
yield mock
class TestOpenAlprLocalSetup:
"""Test class for image processing."""
async def test_setup_platform(hass):
"""Set up platform with one entity."""
config = {
ip.DOMAIN: {
"platform": "openalpr_local",
"source": {"entity_id": "camera.demo_camera"},
"region": "eu",
},
"camera": {"platform": "demo"},
}
def setup_method(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
with assert_setup_component(1, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_setup_platform(self):
"""Set up platform with one entity."""
config = {
ip.DOMAIN: {
"platform": "openalpr_local",
"source": {"entity_id": "camera.demo_camera"},
"region": "eu",
},
"camera": {"platform": "demo"},
}
with assert_setup_component(1, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
self.hass.block_till_done()
assert self.hass.states.get("image_processing.openalpr_demo_camera")
def test_setup_platform_name(self):
"""Set up platform with one entity and set name."""
config = {
ip.DOMAIN: {
"platform": "openalpr_local",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"region": "eu",
},
"camera": {"platform": "demo"},
}
with assert_setup_component(1, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
self.hass.block_till_done()
assert self.hass.states.get("image_processing.test_local")
def test_setup_platform_without_region(self):
"""Set up platform with one entity without region."""
config = {
ip.DOMAIN: {
"platform": "openalpr_local",
"source": {"entity_id": "camera.demo_camera"},
},
"camera": {"platform": "demo"},
}
with assert_setup_component(0, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
self.hass.block_till_done()
assert hass.states.get("image_processing.openalpr_demo_camera")
class TestOpenAlprLocal:
"""Test class for image processing."""
async def test_setup_platform_name(hass):
"""Set up platform with one entity and set name."""
config = {
ip.DOMAIN: {
"platform": "openalpr_local",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"region": "eu",
},
"camera": {"platform": "demo"},
}
def setup_method(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
with assert_setup_component(1, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
config = {
ip.DOMAIN: {
"platform": "openalpr_local",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"region": "eu",
},
"camera": {"platform": "demo"},
}
assert hass.states.get("image_processing.test_local")
with patch(
"homeassistant.components.openalpr_local.image_processing."
"OpenAlprLocalEntity.should_poll",
new_callable=PropertyMock(return_value=False),
):
setup_component(self.hass, ip.DOMAIN, config)
self.hass.block_till_done()
state = self.hass.states.get("camera.demo_camera")
self.url = f"{self.hass.config.internal_url}{state.attributes.get(ATTR_ENTITY_PICTURE)}"
async def test_setup_platform_without_region(hass):
"""Set up platform with one entity without region."""
config = {
ip.DOMAIN: {
"platform": "openalpr_local",
"source": {"entity_id": "camera.demo_camera"},
},
"camera": {"platform": "demo"},
}
self.alpr_events = []
with assert_setup_component(0, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
@callback
def mock_alpr_event(event):
"""Mock event."""
self.alpr_events.append(event)
self.hass.bus.listen("image_processing.found_plate", mock_alpr_event)
async def test_openalpr_process_image(
setup_openalpr_local,
url,
hass,
alpr_events,
popen_mock,
aioclient_mock,
):
"""Set up and scan a picture and test plates from event."""
aioclient_mock.get(url, content=b"image")
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
common.async_scan(hass, entity_id="image_processing.test_local")
await hass.async_block_till_done()
@patch("asyncio.create_subprocess_exec", return_value=mock_async_subprocess())
def test_openalpr_process_image(self, popen_mock, aioclient_mock):
"""Set up and scan a picture and test plates from event."""
aioclient_mock.get(self.url, content=b"image")
state = hass.states.get("image_processing.test_local")
common.scan(self.hass, entity_id="image_processing.test_local")
self.hass.block_till_done()
assert popen_mock.called
assert len(alpr_events) == 5
assert state.attributes.get("vehicles") == 1
assert state.state == "PE3R2X"
state = self.hass.states.get("image_processing.test_local")
assert popen_mock.called
assert len(self.alpr_events) == 5
assert state.attributes.get("vehicles") == 1
assert state.state == "PE3R2X"
event_data = [
event.data
for event in self.alpr_events
if event.data.get("plate") == "PE3R2X"
]
assert len(event_data) == 1
assert event_data[0]["plate"] == "PE3R2X"
assert event_data[0]["confidence"] == float(98.9371)
assert event_data[0]["entity_id"] == "image_processing.test_local"
event_data = [
event.data for event in alpr_events if event.data.get("plate") == "PE3R2X"
]
assert len(event_data) == 1
assert event_data[0]["plate"] == "PE3R2X"
assert event_data[0]["confidence"] == float(98.9371)
assert event_data[0]["entity_id"] == "image_processing.test_local"