Convert image_processing tests to pytest (#56451)
* Convert image_processing tests to pytest * Use existing fixtures and test helpers * Remove useless code
This commit is contained in:
Erik Montnemery
GitHub
parent
26d310fc8a
commit
ac053388b4
1 changed files with 194 additions and 252 deletions
|
|
@ -4,289 +4,231 @@ from unittest.mock import PropertyMock, patch
|
|||
import homeassistant.components.http as http
|
||||
import homeassistant.components.image_processing as ip
|
||||
from homeassistant.const import ATTR_ENTITY_PICTURE
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.loader import DATA_CUSTOM_COMPONENTS
|
||||
from homeassistant.setup import setup_component
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import (
|
||||
assert_setup_component,
|
||||
get_test_home_assistant,
|
||||
get_test_instance_port,
|
||||
)
|
||||
from tests.common import assert_setup_component, async_capture_events
|
||||
from tests.components.image_processing import common
|
||||
|
||||
|
||||
class TestSetupImageProcessing:
|
||||
"""Test class for setup image processing."""
|
||||
|
||||
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_component(self):
|
||||
"""Set up demo platform on image_process component."""
|
||||
config = {ip.DOMAIN: {"platform": "demo"}}
|
||||
|
||||
with assert_setup_component(1, ip.DOMAIN):
|
||||
setup_component(self.hass, ip.DOMAIN, config)
|
||||
|
||||
def test_setup_component_with_service(self):
|
||||
"""Set up demo platform on image_process component test service."""
|
||||
config = {ip.DOMAIN: {"platform": "demo"}}
|
||||
|
||||
with assert_setup_component(1, ip.DOMAIN):
|
||||
setup_component(self.hass, ip.DOMAIN, config)
|
||||
|
||||
assert self.hass.services.has_service(ip.DOMAIN, "scan")
|
||||
def get_url(hass):
|
||||
"""Return camera url."""
|
||||
state = hass.states.get("camera.demo_camera")
|
||||
return f"{hass.config.internal_url}{state.attributes.get(ATTR_ENTITY_PICTURE)}"
|
||||
|
||||
|
||||
class TestImageProcessing:
|
||||
"""Test class for image processing."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self.hass.data.pop(DATA_CUSTOM_COMPONENTS)
|
||||
|
||||
setup_component(
|
||||
self.hass,
|
||||
http.DOMAIN,
|
||||
{http.DOMAIN: {http.CONF_SERVER_PORT: get_test_instance_port()}},
|
||||
)
|
||||
|
||||
config = {ip.DOMAIN: {"platform": "test"}, "camera": {"platform": "demo"}}
|
||||
|
||||
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)}"
|
||||
|
||||
def teardown_method(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
@patch(
|
||||
"homeassistant.components.demo.camera.Path.read_bytes",
|
||||
return_value=b"Test",
|
||||
async def setup_image_processing(hass, aiohttp_unused_port):
|
||||
"""Set up things to be run when tests are started."""
|
||||
await async_setup_component(
|
||||
hass,
|
||||
http.DOMAIN,
|
||||
{http.DOMAIN: {http.CONF_SERVER_PORT: aiohttp_unused_port()}},
|
||||
)
|
||||
def test_get_image_from_camera(self, mock_camera_read):
|
||||
"""Grab an image from camera entity."""
|
||||
common.scan(self.hass, entity_id="image_processing.test")
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get("image_processing.test")
|
||||
config = {ip.DOMAIN: {"platform": "test"}, "camera": {"platform": "demo"}}
|
||||
|
||||
assert mock_camera_read.called
|
||||
assert state.state == "1"
|
||||
assert state.attributes["image"] == b"Test"
|
||||
|
||||
@patch(
|
||||
"homeassistant.components.camera.async_get_image",
|
||||
side_effect=HomeAssistantError(),
|
||||
)
|
||||
def test_get_image_without_exists_camera(self, mock_image):
|
||||
"""Try to get image without exists camera."""
|
||||
self.hass.states.remove("camera.demo_camera")
|
||||
|
||||
common.scan(self.hass, entity_id="image_processing.test")
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get("image_processing.test")
|
||||
|
||||
assert mock_image.called
|
||||
assert state.state == "0"
|
||||
await async_setup_component(hass, ip.DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
class TestImageProcessingAlpr:
|
||||
"""Test class for alpr image processing."""
|
||||
async def setup_image_processing_alpr(hass):
|
||||
"""Set up things to be run when tests are started."""
|
||||
config = {ip.DOMAIN: {"platform": "demo"}, "camera": {"platform": "demo"}}
|
||||
|
||||
def setup_method(self):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
await async_setup_component(hass, ip.DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
config = {ip.DOMAIN: {"platform": "demo"}, "camera": {"platform": "demo"}}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.demo.image_processing."
|
||||
"DemoImageProcessingAlpr.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)}"
|
||||
|
||||
self.alpr_events = []
|
||||
|
||||
@callback
|
||||
def mock_alpr_event(event):
|
||||
"""Mock event."""
|
||||
self.alpr_events.append(event)
|
||||
|
||||
self.hass.bus.listen("image_processing.found_plate", mock_alpr_event)
|
||||
|
||||
def teardown_method(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
def test_alpr_event_single_call(self, aioclient_mock):
|
||||
"""Set up and scan a picture and test plates from event."""
|
||||
aioclient_mock.get(self.url, content=b"image")
|
||||
|
||||
common.scan(self.hass, entity_id="image_processing.demo_alpr")
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get("image_processing.demo_alpr")
|
||||
|
||||
assert len(self.alpr_events) == 4
|
||||
assert state.state == "AC3829"
|
||||
|
||||
event_data = [
|
||||
event.data
|
||||
for event in self.alpr_events
|
||||
if event.data.get("plate") == "AC3829"
|
||||
]
|
||||
assert len(event_data) == 1
|
||||
assert event_data[0]["plate"] == "AC3829"
|
||||
assert event_data[0]["confidence"] == 98.3
|
||||
assert event_data[0]["entity_id"] == "image_processing.demo_alpr"
|
||||
|
||||
def test_alpr_event_double_call(self, aioclient_mock):
|
||||
"""Set up and scan a picture and test plates from event."""
|
||||
aioclient_mock.get(self.url, content=b"image")
|
||||
|
||||
common.scan(self.hass, entity_id="image_processing.demo_alpr")
|
||||
common.scan(self.hass, entity_id="image_processing.demo_alpr")
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get("image_processing.demo_alpr")
|
||||
|
||||
assert len(self.alpr_events) == 4
|
||||
assert state.state == "AC3829"
|
||||
|
||||
event_data = [
|
||||
event.data
|
||||
for event in self.alpr_events
|
||||
if event.data.get("plate") == "AC3829"
|
||||
]
|
||||
assert len(event_data) == 1
|
||||
assert event_data[0]["plate"] == "AC3829"
|
||||
assert event_data[0]["confidence"] == 98.3
|
||||
assert event_data[0]["entity_id"] == "image_processing.demo_alpr"
|
||||
|
||||
@patch(
|
||||
"homeassistant.components.demo.image_processing."
|
||||
"DemoImageProcessingAlpr.confidence",
|
||||
new_callable=PropertyMock(return_value=95),
|
||||
)
|
||||
def test_alpr_event_single_call_confidence(self, confidence_mock, aioclient_mock):
|
||||
"""Set up and scan a picture and test plates from event."""
|
||||
aioclient_mock.get(self.url, content=b"image")
|
||||
|
||||
common.scan(self.hass, entity_id="image_processing.demo_alpr")
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get("image_processing.demo_alpr")
|
||||
|
||||
assert len(self.alpr_events) == 2
|
||||
assert state.state == "AC3829"
|
||||
|
||||
event_data = [
|
||||
event.data
|
||||
for event in self.alpr_events
|
||||
if event.data.get("plate") == "AC3829"
|
||||
]
|
||||
assert len(event_data) == 1
|
||||
assert event_data[0]["plate"] == "AC3829"
|
||||
assert event_data[0]["confidence"] == 98.3
|
||||
assert event_data[0]["entity_id"] == "image_processing.demo_alpr"
|
||||
return async_capture_events(hass, "image_processing.found_plate")
|
||||
|
||||
|
||||
class TestImageProcessingFace:
|
||||
"""Test class for face image processing."""
|
||||
async def setup_image_processing_face(hass):
|
||||
"""Set up things to be run when tests are started."""
|
||||
config = {ip.DOMAIN: {"platform": "demo"}, "camera": {"platform": "demo"}}
|
||||
|
||||
def setup_method(self):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
await async_setup_component(hass, ip.DOMAIN, config)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
config = {ip.DOMAIN: {"platform": "demo"}, "camera": {"platform": "demo"}}
|
||||
return async_capture_events(hass, "image_processing.detect_face")
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.demo.image_processing."
|
||||
"DemoImageProcessingFace.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_component(hass):
|
||||
"""Set up demo platform on image_process component."""
|
||||
config = {ip.DOMAIN: {"platform": "demo"}}
|
||||
|
||||
self.face_events = []
|
||||
with assert_setup_component(1, ip.DOMAIN):
|
||||
assert await async_setup_component(hass, ip.DOMAIN, config)
|
||||
|
||||
@callback
|
||||
def mock_face_event(event):
|
||||
"""Mock event."""
|
||||
self.face_events.append(event)
|
||||
|
||||
self.hass.bus.listen("image_processing.detect_face", mock_face_event)
|
||||
async def test_setup_component_with_service(hass):
|
||||
"""Set up demo platform on image_process component test service."""
|
||||
config = {ip.DOMAIN: {"platform": "demo"}}
|
||||
|
||||
def teardown_method(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
with assert_setup_component(1, ip.DOMAIN):
|
||||
assert await async_setup_component(hass, ip.DOMAIN, config)
|
||||
|
||||
def test_face_event_call(self, aioclient_mock):
|
||||
"""Set up and scan a picture and test faces from event."""
|
||||
aioclient_mock.get(self.url, content=b"image")
|
||||
assert hass.services.has_service(ip.DOMAIN, "scan")
|
||||
|
||||
common.scan(self.hass, entity_id="image_processing.demo_face")
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get("image_processing.demo_face")
|
||||
@patch(
|
||||
"homeassistant.components.demo.camera.Path.read_bytes",
|
||||
return_value=b"Test",
|
||||
)
|
||||
async def test_get_image_from_camera(
|
||||
mock_camera_read, hass, aiohttp_unused_port, enable_custom_integrations
|
||||
):
|
||||
"""Grab an image from camera entity."""
|
||||
await setup_image_processing(hass, aiohttp_unused_port)
|
||||
|
||||
assert len(self.face_events) == 2
|
||||
assert state.state == "Hans"
|
||||
assert state.attributes["total_faces"] == 4
|
||||
common.async_scan(hass, entity_id="image_processing.test")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
event_data = [
|
||||
event.data for event in self.face_events if event.data.get("name") == "Hans"
|
||||
]
|
||||
assert len(event_data) == 1
|
||||
assert event_data[0]["name"] == "Hans"
|
||||
assert event_data[0]["confidence"] == 98.34
|
||||
assert event_data[0]["gender"] == "male"
|
||||
assert event_data[0]["entity_id"] == "image_processing.demo_face"
|
||||
state = hass.states.get("image_processing.test")
|
||||
|
||||
@patch(
|
||||
"homeassistant.components.demo.image_processing."
|
||||
"DemoImageProcessingFace.confidence",
|
||||
new_callable=PropertyMock(return_value=None),
|
||||
)
|
||||
def test_face_event_call_no_confidence(self, mock_config, aioclient_mock):
|
||||
"""Set up and scan a picture and test faces from event."""
|
||||
aioclient_mock.get(self.url, content=b"image")
|
||||
assert mock_camera_read.called
|
||||
assert state.state == "1"
|
||||
assert state.attributes["image"] == b"Test"
|
||||
|
||||
common.scan(self.hass, entity_id="image_processing.demo_face")
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get("image_processing.demo_face")
|
||||
@patch(
|
||||
"homeassistant.components.camera.async_get_image",
|
||||
side_effect=HomeAssistantError(),
|
||||
)
|
||||
async def test_get_image_without_exists_camera(
|
||||
mock_image, hass, aiohttp_unused_port, enable_custom_integrations
|
||||
):
|
||||
"""Try to get image without exists camera."""
|
||||
await setup_image_processing(hass, aiohttp_unused_port)
|
||||
|
||||
assert len(self.face_events) == 3
|
||||
assert state.state == "4"
|
||||
assert state.attributes["total_faces"] == 4
|
||||
hass.states.async_remove("camera.demo_camera")
|
||||
|
||||
event_data = [
|
||||
event.data for event in self.face_events if event.data.get("name") == "Hans"
|
||||
]
|
||||
assert len(event_data) == 1
|
||||
assert event_data[0]["name"] == "Hans"
|
||||
assert event_data[0]["confidence"] == 98.34
|
||||
assert event_data[0]["gender"] == "male"
|
||||
assert event_data[0]["entity_id"] == "image_processing.demo_face"
|
||||
common.async_scan(hass, entity_id="image_processing.test")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("image_processing.test")
|
||||
|
||||
assert mock_image.called
|
||||
assert state.state == "0"
|
||||
|
||||
|
||||
async def test_alpr_event_single_call(hass, aioclient_mock):
|
||||
"""Set up and scan a picture and test plates from event."""
|
||||
alpr_events = await setup_image_processing_alpr(hass)
|
||||
aioclient_mock.get(get_url(hass), content=b"image")
|
||||
|
||||
common.async_scan(hass, entity_id="image_processing.demo_alpr")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("image_processing.demo_alpr")
|
||||
|
||||
assert len(alpr_events) == 4
|
||||
assert state.state == "AC3829"
|
||||
|
||||
event_data = [
|
||||
event.data for event in alpr_events if event.data.get("plate") == "AC3829"
|
||||
]
|
||||
assert len(event_data) == 1
|
||||
assert event_data[0]["plate"] == "AC3829"
|
||||
assert event_data[0]["confidence"] == 98.3
|
||||
assert event_data[0]["entity_id"] == "image_processing.demo_alpr"
|
||||
|
||||
|
||||
async def test_alpr_event_double_call(hass, aioclient_mock):
|
||||
"""Set up and scan a picture and test plates from event."""
|
||||
alpr_events = await setup_image_processing_alpr(hass)
|
||||
aioclient_mock.get(get_url(hass), content=b"image")
|
||||
|
||||
common.async_scan(hass, entity_id="image_processing.demo_alpr")
|
||||
common.async_scan(hass, entity_id="image_processing.demo_alpr")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("image_processing.demo_alpr")
|
||||
|
||||
assert len(alpr_events) == 4
|
||||
assert state.state == "AC3829"
|
||||
|
||||
event_data = [
|
||||
event.data for event in alpr_events if event.data.get("plate") == "AC3829"
|
||||
]
|
||||
assert len(event_data) == 1
|
||||
assert event_data[0]["plate"] == "AC3829"
|
||||
assert event_data[0]["confidence"] == 98.3
|
||||
assert event_data[0]["entity_id"] == "image_processing.demo_alpr"
|
||||
|
||||
|
||||
@patch(
|
||||
"homeassistant.components.demo.image_processing.DemoImageProcessingAlpr.confidence",
|
||||
new_callable=PropertyMock(return_value=95),
|
||||
)
|
||||
async def test_alpr_event_single_call_confidence(confidence_mock, hass, aioclient_mock):
|
||||
"""Set up and scan a picture and test plates from event."""
|
||||
alpr_events = await setup_image_processing_alpr(hass)
|
||||
aioclient_mock.get(get_url(hass), content=b"image")
|
||||
|
||||
common.async_scan(hass, entity_id="image_processing.demo_alpr")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("image_processing.demo_alpr")
|
||||
|
||||
assert len(alpr_events) == 2
|
||||
assert state.state == "AC3829"
|
||||
|
||||
event_data = [
|
||||
event.data for event in alpr_events if event.data.get("plate") == "AC3829"
|
||||
]
|
||||
assert len(event_data) == 1
|
||||
assert event_data[0]["plate"] == "AC3829"
|
||||
assert event_data[0]["confidence"] == 98.3
|
||||
assert event_data[0]["entity_id"] == "image_processing.demo_alpr"
|
||||
|
||||
|
||||
async def test_face_event_call(hass, aioclient_mock):
|
||||
"""Set up and scan a picture and test faces from event."""
|
||||
face_events = await setup_image_processing_face(hass)
|
||||
aioclient_mock.get(get_url(hass), content=b"image")
|
||||
|
||||
common.async_scan(hass, entity_id="image_processing.demo_face")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("image_processing.demo_face")
|
||||
|
||||
assert len(face_events) == 2
|
||||
assert state.state == "Hans"
|
||||
assert state.attributes["total_faces"] == 4
|
||||
|
||||
event_data = [
|
||||
event.data for event in face_events if event.data.get("name") == "Hans"
|
||||
]
|
||||
assert len(event_data) == 1
|
||||
assert event_data[0]["name"] == "Hans"
|
||||
assert event_data[0]["confidence"] == 98.34
|
||||
assert event_data[0]["gender"] == "male"
|
||||
assert event_data[0]["entity_id"] == "image_processing.demo_face"
|
||||
|
||||
|
||||
@patch(
|
||||
"homeassistant.components.demo.image_processing."
|
||||
"DemoImageProcessingFace.confidence",
|
||||
new_callable=PropertyMock(return_value=None),
|
||||
)
|
||||
async def test_face_event_call_no_confidence(mock_config, hass, aioclient_mock):
|
||||
"""Set up and scan a picture and test faces from event."""
|
||||
face_events = await setup_image_processing_face(hass)
|
||||
aioclient_mock.get(get_url(hass), content=b"image")
|
||||
|
||||
common.async_scan(hass, entity_id="image_processing.demo_face")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("image_processing.demo_face")
|
||||
|
||||
assert len(face_events) == 3
|
||||
assert state.state == "4"
|
||||
assert state.attributes["total_faces"] == 4
|
||||
|
||||
event_data = [
|
||||
event.data for event in face_events if event.data.get("name") == "Hans"
|
||||
]
|
||||
assert len(event_data) == 1
|
||||
assert event_data[0]["name"] == "Hans"
|
||||
assert event_data[0]["confidence"] == 98.34
|
||||
assert event_data[0]["gender"] == "male"
|
||||
assert event_data[0]["entity_id"] == "image_processing.demo_face"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue