diff --git a/tests/components/openalpr_cloud/test_image_processing.py b/tests/components/openalpr_cloud/test_image_processing.py index e1e9192e1de..c1b88026f4d 100644 --- a/tests/components/openalpr_cloud/test_image_processing.py +++ b/tests/components/openalpr_cloud/test_image_processing.py @@ -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 diff --git a/tests/components/openalpr_local/test_image_processing.py b/tests/components/openalpr_local/test_image_processing.py index 0cde6e0454b..fefc42fe2ab 100644 --- a/tests/components/openalpr_local/test_image_processing.py +++ b/tests/components/openalpr_local/test_image_processing.py @@ -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"