evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Convert some tests to async and drop usage get_test_home_assistant (#64394)

Browse source 
* Fix some tests

* Update MS tests

* Convert last logbook tests to async
This commit is contained in:
Paulus Schoutsen 2022-01-18 19:37:17 -08:00 • committed by GitHub
parent 394c6850a3
commit 08083f399e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 532 additions and 610 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
homeassistant/components/camera/__init__.py
View file

@ -951,10 +951,11 @@ async def async_handle_snapshot_service(
image = await camera.async_camera_image()
def _write_image(to_file: str, image_data: bytes | None) -> None:
if image is None:
return
def _write_image(to_file: str, image_data: bytes) -> None:
"""Executor helper to write image."""
if image_data is None:
return
os.makedirs(os.path.dirname(to_file), exist_ok=True)
with open(to_file, "wb") as img_file:
img_file.write(image_data)

24
tests/components/aprs/test_device_tracker.py
View file

@ -4,9 +4,6 @@ from unittest.mock import Mock, patch
import aprslib
import homeassistant.components.aprs.device_tracker as device_tracker
from homeassistant.const import EVENT_HOMEASSISTANT_START
from tests.common import get_test_home_assistant
DEFAULT_PORT = 14580
@ -299,14 +296,11 @@ def test_aprs_listener_rx_msg_no_position():
see.assert_not_called()
def test_setup_scanner():
async def test_setup_scanner(hass):
"""Test setup_scanner."""
with patch(
"homeassistant.components.aprs.device_tracker.AprsListenerThread"
) as listener:
hass = get_test_home_assistant()
hass.start()
config = {
"username": TEST_CALLSIGN,
"password": TEST_PASSWORD,
@ -316,9 +310,9 @@ def test_setup_scanner():
}
see = Mock()
res = device_tracker.setup_scanner(hass, config, see)
hass.bus.fire(EVENT_HOMEASSISTANT_START)
hass.stop()
res = await hass.async_add_executor_job(
device_tracker.setup_scanner, hass, config, see
)
assert res
listener.assert_called_with(
@ -326,12 +320,9 @@ def test_setup_scanner():
)
def test_setup_scanner_timeout():
async def test_setup_scanner_timeout(hass):
"""Test setup_scanner failure from timeout."""
with patch("aprslib.IS.connect", side_effect=TimeoutError):
hass = get_test_home_assistant()
hass.start()
config = {
"username": TEST_CALLSIGN,
"password": TEST_PASSWORD,
@ -341,5 +332,6 @@ def test_setup_scanner_timeout():
}
see = Mock()
assert not device_tracker.setup_scanner(hass, config, see)
hass.stop()
assert not await hass.async_add_executor_job(
device_tracker.setup_scanner, hass, config, see
)

3
tests/components/camera/test_init.py
View file

@ -279,8 +279,7 @@ async def test_snapshot_service(hass, mock_camera):
mopen = mock_open()
with patch("homeassistant.components.camera.open", mopen, create=True), patch(
"homeassistant.components.camera.os.path.exists",
Mock(spec="os.path.exists", return_value=True),
"homeassistant.components.camera.os.makedirs",
), patch.object(hass.config, "is_allowed_path", return_value=True):
await hass.services.async_call(
camera.DOMAIN,

112
tests/components/logbook/test_init.py
View file

@ -35,37 +35,32 @@ from homeassistant.const import (
import homeassistant.core as ha
from homeassistant.helpers.entityfilter import CONF_ENTITY_GLOBS
from homeassistant.helpers.json import JSONEncoder
from homeassistant.setup import async_setup_component, setup_component
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import get_test_home_assistant, init_recorder_component, mock_platform
from tests.common import (
async_capture_events,
async_init_recorder_component,
mock_platform,
)
from tests.components.recorder.common import trigger_db_commit
EMPTY_CONFIG = logbook.CONFIG_SCHEMA({logbook.DOMAIN: {}})
@pytest.fixture
def hass_():
async def hass_(hass):
"""Set up things to be run when tests are started."""
hass = get_test_home_assistant()
init_recorder_component(hass) # Force an in memory DB
with patch("homeassistant.components.http.start_http_server_and_save_config"):
assert setup_component(hass, logbook.DOMAIN, EMPTY_CONFIG)
yield hass
hass.stop()
await async_init_recorder_component(hass) # Force an in memory DB
assert await async_setup_component(hass, logbook.DOMAIN, EMPTY_CONFIG)
return hass
def test_service_call_create_logbook_entry(hass_):
async def test_service_call_create_logbook_entry(hass_):
"""Test if service call create log book entry."""
calls = []
calls = async_capture_events(hass_, logbook.EVENT_LOGBOOK_ENTRY)
@ha.callback
def event_listener(event):
"""Append on event."""
calls.append(event)
hass_.bus.listen(logbook.EVENT_LOGBOOK_ENTRY, event_listener)
hass_.services.call(
await hass_.services.async_call(
logbook.DOMAIN,
"log",
{
@ -76,7 +71,7 @@ def test_service_call_create_logbook_entry(hass_):
},
True,
)
hass_.services.call(
await hass_.services.async_call(
logbook.DOMAIN,
"log",
{
@ -88,9 +83,11 @@ def test_service_call_create_logbook_entry(hass_):
# Logbook entry service call results in firing an event.
# Our service call will unblock when the event listeners have been
# scheduled. This means that they may not have been processed yet.
trigger_db_commit(hass_)
hass_.block_till_done()
hass_.data[recorder.DATA_INSTANCE].block_till_done()
await hass_.async_add_executor_job(trigger_db_commit, hass_)
await hass_.async_block_till_done()
await hass_.async_add_executor_job(
hass_.data[recorder.DATA_INSTANCE].block_till_done
)
events = list(
logbook._get_events(
@ -116,24 +113,17 @@ def test_service_call_create_logbook_entry(hass_):
assert last_call.data.get(logbook.ATTR_DOMAIN) == "logbook"
def test_service_call_create_log_book_entry_no_message(hass_):
async def test_service_call_create_log_book_entry_no_message(hass_):
"""Test if service call create log book entry without message."""
calls = []
@ha.callback
def event_listener(event):
"""Append on event."""
calls.append(event)
hass_.bus.listen(logbook.EVENT_LOGBOOK_ENTRY, event_listener)
calls = async_capture_events(hass_, logbook.EVENT_LOGBOOK_ENTRY)
with pytest.raises(vol.Invalid):
hass_.services.call(logbook.DOMAIN, "log", {}, True)
await hass_.services.async_call(logbook.DOMAIN, "log", {}, True)
# Logbook entry service call results in firing an event.
# Our service call will unblock when the event listeners have been
# scheduled. This means that they may not have been processed yet.
hass_.block_till_done()
await hass_.async_block_till_done()
assert len(calls) == 0
@ -310,7 +300,7 @@ def create_state_changed_event_from_old_new(
async def test_logbook_view(hass, hass_client):
"""Test the logbook view."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
client = await hass_client()
@ -320,7 +310,7 @@ async def test_logbook_view(hass, hass_client):
async def test_logbook_view_period_entity(hass, hass_client):
"""Test the logbook view with period and entity."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -404,7 +394,7 @@ async def test_logbook_view_period_entity(hass, hass_client):
async def test_logbook_describe_event(hass, hass_client):
"""Test teaching logbook about a new event."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
def _describe(event):
"""Describe an event."""
@ -471,7 +461,7 @@ async def test_exclude_described_event(hass, hass_client):
Mock(async_describe_events=async_describe_events),
)
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
assert await async_setup_component(
hass,
logbook.DOMAIN,
@ -515,7 +505,7 @@ async def test_exclude_described_event(hass, hass_client):
async def test_logbook_view_end_time_entity(hass, hass_client):
"""Test the logbook view with end_time and entity."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -573,7 +563,7 @@ async def test_logbook_view_end_time_entity(hass, hass_client):
async def test_logbook_entity_filter_with_automations(hass, hass_client):
"""Test the logbook view with end_time and entity with automations and scripts."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await async_setup_component(hass, "automation", {})
await async_setup_component(hass, "script", {})
@ -648,7 +638,7 @@ async def test_logbook_entity_filter_with_automations(hass, hass_client):
async def test_filter_continuous_sensor_values(hass, hass_client):
"""Test remove continuous sensor events from logbook."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -684,7 +674,7 @@ async def test_filter_continuous_sensor_values(hass, hass_client):
async def test_exclude_new_entities(hass, hass_client):
"""Test if events are excluded on first update."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -719,7 +709,7 @@ async def test_exclude_new_entities(hass, hass_client):
async def test_exclude_removed_entities(hass, hass_client):
"""Test if events are excluded on last update."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -761,7 +751,7 @@ async def test_exclude_removed_entities(hass, hass_client):
async def test_exclude_attribute_changes(hass, hass_client):
"""Test if events of attribute changes are filtered."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -799,7 +789,7 @@ async def test_exclude_attribute_changes(hass, hass_client):
async def test_logbook_entity_context_id(hass, hass_client):
"""Test the logbook view with end_time and entity with automations and scripts."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await async_setup_component(hass, "automation", {})
await async_setup_component(hass, "script", {})
@ -951,7 +941,7 @@ async def test_logbook_entity_context_id(hass, hass_client):
async def test_logbook_entity_context_parent_id(hass, hass_client):
"""Test the logbook view links events via context parent_id."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await async_setup_component(hass, "automation", {})
await async_setup_component(hass, "script", {})
@ -1132,7 +1122,7 @@ async def test_logbook_entity_context_parent_id(hass, hass_client):
async def test_logbook_context_from_template(hass, hass_client):
"""Test the logbook view with end_time and entity with automations and scripts."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
assert await async_setup_component(
hass,
@ -1218,7 +1208,7 @@ async def test_logbook_context_from_template(hass, hass_client):
async def test_logbook_entity_matches_only(hass, hass_client):
"""Test the logbook view with a single entity and entity_matches_only."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
assert await async_setup_component(
hass,
@ -1292,7 +1282,7 @@ async def test_logbook_entity_matches_only(hass, hass_client):
async def test_custom_log_entry_discoverable_via_entity_matches_only(hass, hass_client):
"""Test if a custom log entry is later discoverable via entity_matches_only."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1331,7 +1321,7 @@ async def test_custom_log_entry_discoverable_via_entity_matches_only(hass, hass_
async def test_logbook_entity_matches_only_multiple(hass, hass_client):
"""Test the logbook view with a multiple entities and entity_matches_only."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
assert await async_setup_component(
hass,
@ -1415,7 +1405,7 @@ async def test_logbook_entity_matches_only_multiple(hass, hass_client):
async def test_logbook_invalid_entity(hass, hass_client):
"""Test the logbook view with requesting an invalid entity."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await hass.async_block_till_done()
client = await hass_client()
@ -1434,7 +1424,7 @@ async def test_logbook_invalid_entity(hass, hass_client):
async def test_icon_and_state(hass, hass_client):
"""Test to ensure state and custom icons are returned."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", {})
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1481,7 +1471,7 @@ async def test_exclude_events_domain(hass, hass_client):
logbook.DOMAIN: {CONF_EXCLUDE: {CONF_DOMAINS: ["switch", "alexa"]}},
}
)
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", config)
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1521,7 +1511,7 @@ async def test_exclude_events_domain_glob(hass, hass_client):
},
}
)
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", config)
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1561,7 +1551,7 @@ async def test_include_events_entity(hass, hass_client):
},
}
)
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", config)
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1594,7 +1584,7 @@ async def test_exclude_events_entity(hass, hass_client):
logbook.DOMAIN: {CONF_EXCLUDE: {CONF_ENTITIES: [entity_id]}},
}
)
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", config)
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1628,7 +1618,7 @@ async def test_include_events_domain(hass, hass_client):
},
}
)
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", config)
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1672,7 +1662,7 @@ async def test_include_events_domain_glob(hass, hass_client):
},
}
)
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", config)
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1724,7 +1714,7 @@ async def test_include_exclude_events(hass, hass_client):
},
}
)
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", config)
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1778,7 +1768,7 @@ async def test_include_exclude_events_with_glob_filters(hass, hass_client):
},
}
)
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", config)
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1821,7 +1811,7 @@ async def test_empty_config(hass, hass_client):
logbook.DOMAIN: {},
}
)
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
await async_setup_component(hass, "logbook", config)
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
@ -1843,7 +1833,7 @@ async def test_empty_config(hass, hass_client):
async def test_context_filter(hass, hass_client):
"""Test we can filter by context."""
await hass.async_add_executor_job(init_recorder_component, hass)
await async_init_recorder_component(hass)
assert await async_setup_component(hass, "logbook", {})
await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)

479
tests/components/microsoft_face/test_init.py
View file

@ -2,6 +2,8 @@
import asyncio
from unittest.mock import patch
import pytest
from homeassistant.components import camera, microsoft_face as mf
from homeassistant.components.microsoft_face import (
ATTR_CAMERA_ENTITY,
@ -16,9 +18,9 @@ from homeassistant.components.microsoft_face import (
SERVICE_TRAIN_GROUP,
)
from homeassistant.const import ATTR_NAME
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, load_fixture
def create_group(hass, name):
@ -27,7 +29,7 @@ def create_group(hass, name):
This is a legacy helper method. Do not use it for new tests.
"""
data = {ATTR_NAME: name}
hass.services.call(DOMAIN, SERVICE_CREATE_GROUP, data)
hass.async_add_job(hass.services.async_call(DOMAIN, SERVICE_CREATE_GROUP, data))
def delete_group(hass, name):
@ -36,7 +38,7 @@ def delete_group(hass, name):
This is a legacy helper method. Do not use it for new tests.
"""
data = {ATTR_NAME: name}
hass.services.call(DOMAIN, SERVICE_DELETE_GROUP, data)
hass.async_add_job(hass.services.async_call(DOMAIN, SERVICE_DELETE_GROUP, data))
def train_group(hass, group):
@ -45,7 +47,7 @@ def train_group(hass, group):
This is a legacy helper method. Do not use it for new tests.
"""
data = {ATTR_GROUP: group}
hass.services.call(DOMAIN, SERVICE_TRAIN_GROUP, data)
hass.async_add_job(hass.services.async_call(DOMAIN, SERVICE_TRAIN_GROUP, data))
def create_person(hass, group, name):
@ -54,7 +56,7 @@ def create_person(hass, group, name):
This is a legacy helper method. Do not use it for new tests.
"""
data = {ATTR_GROUP: group, ATTR_NAME: name}
hass.services.call(DOMAIN, SERVICE_CREATE_PERSON, data)
hass.async_add_job(hass.services.async_call(DOMAIN, SERVICE_CREATE_PERSON, data))
def delete_person(hass, group, name):
@ -63,7 +65,7 @@ def delete_person(hass, group, name):
This is a legacy helper method. Do not use it for new tests.
"""
data = {ATTR_GROUP: group, ATTR_NAME: name}
hass.services.call(DOMAIN, SERVICE_DELETE_PERSON, data)
hass.async_add_job(hass.services.async_call(DOMAIN, SERVICE_DELETE_PERSON, data))
def face_person(hass, group, person, camera_entity):
@ -72,285 +74,254 @@ def face_person(hass, group, person, camera_entity):
This is a legacy helper method. Do not use it for new tests.
"""
data = {ATTR_GROUP: group, ATTR_PERSON: person, ATTR_CAMERA_ENTITY: camera_entity}
hass.services.call(DOMAIN, SERVICE_FACE_PERSON, data)
hass.async_add_job(hass.services.async_call(DOMAIN, SERVICE_FACE_PERSON, data))
class TestMicrosoftFaceSetup:
"""Test the microsoft face component."""
CONFIG = {mf.DOMAIN: {"api_key": "12345678abcdef"}}
ENDPOINT_URL = f"https://westus.{mf.FACE_API_URL}"
def setup_method(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.config = {mf.DOMAIN: {"api_key": "12345678abcdef"}}
self.endpoint_url = f"https://westus.{mf.FACE_API_URL}"
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
@patch(
@pytest.fixture
def mock_update():
"""Mock update store."""
with patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=None,
) as mock_update_store:
yield mock_update_store
async def test_setup_component(hass, mock_update):
"""Set up component."""
with assert_setup_component(3, mf.DOMAIN):
await async_setup_component(hass, mf.DOMAIN, CONFIG)
async def test_setup_component_wrong_api_key(hass, mock_update):
"""Set up component without api key."""
with assert_setup_component(0, mf.DOMAIN):
await async_setup_component(hass, mf.DOMAIN, {mf.DOMAIN: {}})
async def test_setup_component_test_service(hass, mock_update):
"""Set up component."""
with assert_setup_component(3, mf.DOMAIN):
await async_setup_component(hass, mf.DOMAIN, CONFIG)
assert hass.services.has_service(mf.DOMAIN, "create_group")
assert hass.services.has_service(mf.DOMAIN, "delete_group")
assert hass.services.has_service(mf.DOMAIN, "train_group")
assert hass.services.has_service(mf.DOMAIN, "create_person")
assert hass.services.has_service(mf.DOMAIN, "delete_person")
assert hass.services.has_service(mf.DOMAIN, "face_person")
async def test_setup_component_test_entities(hass, aioclient_mock):
"""Set up component."""
aioclient_mock.get(
ENDPOINT_URL.format("persongroups"),
text=load_fixture("microsoft_face_persongroups.json"),
)
def test_setup_component(self, mock_update):
"""Set up component."""
with assert_setup_component(3, mf.DOMAIN):
setup_component(self.hass, mf.DOMAIN, self.config)
@patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=None,
aioclient_mock.get(
ENDPOINT_URL.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
def test_setup_component_wrong_api_key(self, mock_update):
"""Set up component without api key."""
with assert_setup_component(0, mf.DOMAIN):
setup_component(self.hass, mf.DOMAIN, {mf.DOMAIN: {}})
@patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=None,
aioclient_mock.get(
ENDPOINT_URL.format("persongroups/test_group2/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
def test_setup_component_test_service(self, mock_update):
"""Set up component."""
with assert_setup_component(3, mf.DOMAIN):
setup_component(self.hass, mf.DOMAIN, self.config)
assert self.hass.services.has_service(mf.DOMAIN, "create_group")
assert self.hass.services.has_service(mf.DOMAIN, "delete_group")
assert self.hass.services.has_service(mf.DOMAIN, "train_group")
assert self.hass.services.has_service(mf.DOMAIN, "create_person")
assert self.hass.services.has_service(mf.DOMAIN, "delete_person")
assert self.hass.services.has_service(mf.DOMAIN, "face_person")
with assert_setup_component(3, mf.DOMAIN):
await async_setup_component(hass, mf.DOMAIN, CONFIG)
def test_setup_component_test_entities(self, aioclient_mock):
"""Set up component."""
aioclient_mock.get(
self.endpoint_url.format("persongroups"),
text=load_fixture("microsoft_face_persongroups.json"),
)
aioclient_mock.get(
self.endpoint_url.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
aioclient_mock.get(
self.endpoint_url.format("persongroups/test_group2/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
assert len(aioclient_mock.mock_calls) == 3
with assert_setup_component(3, mf.DOMAIN):
setup_component(self.hass, mf.DOMAIN, self.config)
entity_group1 = hass.states.get("microsoft_face.test_group1")
entity_group2 = hass.states.get("microsoft_face.test_group2")
assert len(aioclient_mock.mock_calls) == 3
assert entity_group1 is not None
assert entity_group2 is not None
entity_group1 = self.hass.states.get("microsoft_face.test_group1")
entity_group2 = self.hass.states.get("microsoft_face.test_group2")
assert entity_group1.attributes["Ryan"] == "25985303-c537-4467-b41d-bdb45cd95ca1"
assert entity_group1.attributes["David"] == "2ae4935b-9659-44c3-977f-61fac20d0538"
assert entity_group1 is not None
assert entity_group2 is not None
assert entity_group2.attributes["Ryan"] == "25985303-c537-4467-b41d-bdb45cd95ca1"
assert entity_group2.attributes["David"] == "2ae4935b-9659-44c3-977f-61fac20d0538"
assert (
entity_group1.attributes["Ryan"] == "25985303-c537-4467-b41d-bdb45cd95ca1"
)
assert (
entity_group1.attributes["David"] == "2ae4935b-9659-44c3-977f-61fac20d0538"
)
assert (
entity_group2.attributes["Ryan"] == "25985303-c537-4467-b41d-bdb45cd95ca1"
)
assert (
entity_group2.attributes["David"] == "2ae4935b-9659-44c3-977f-61fac20d0538"
)
@patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=None,
async def test_service_groups(hass, mock_update, aioclient_mock):
"""Set up component, test groups services."""
aioclient_mock.put(
ENDPOINT_URL.format("persongroups/service_group"),
status=200,
text="{}",
)
def test_service_groups(self, mock_update, aioclient_mock):
"""Set up component, test groups services."""
aioclient_mock.put(
self.endpoint_url.format("persongroups/service_group"),
status=200,
text="{}",
)
aioclient_mock.delete(
self.endpoint_url.format("persongroups/service_group"),
status=200,
text="{}",
)
with assert_setup_component(3, mf.DOMAIN):
setup_component(self.hass, mf.DOMAIN, self.config)
create_group(self.hass, "Service Group")
self.hass.block_till_done()
entity = self.hass.states.get("microsoft_face.service_group")
assert entity is not None
assert len(aioclient_mock.mock_calls) == 1
delete_group(self.hass, "Service Group")
self.hass.block_till_done()
entity = self.hass.states.get("microsoft_face.service_group")
assert entity is None
assert len(aioclient_mock.mock_calls) == 2
def test_service_person(self, aioclient_mock):
"""Set up component, test person services."""
aioclient_mock.get(
self.endpoint_url.format("persongroups"),
text=load_fixture("microsoft_face_persongroups.json"),
)
aioclient_mock.get(
self.endpoint_url.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
aioclient_mock.get(
self.endpoint_url.format("persongroups/test_group2/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
with assert_setup_component(3, mf.DOMAIN):
setup_component(self.hass, mf.DOMAIN, self.config)
assert len(aioclient_mock.mock_calls) == 3
aioclient_mock.post(
self.endpoint_url.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_create_person.json"),
)
aioclient_mock.delete(
self.endpoint_url.format(
"persongroups/test_group1/persons/"
"25985303-c537-4467-b41d-bdb45cd95ca1"
),
status=200,
text="{}",
)
create_person(self.hass, "test group1", "Hans")
self.hass.block_till_done()
entity_group1 = self.hass.states.get("microsoft_face.test_group1")
assert len(aioclient_mock.mock_calls) == 4
assert entity_group1 is not None
assert (
entity_group1.attributes["Hans"] == "25985303-c537-4467-b41d-bdb45cd95ca1"
)
delete_person(self.hass, "test group1", "Hans")
self.hass.block_till_done()
entity_group1 = self.hass.states.get("microsoft_face.test_group1")
assert len(aioclient_mock.mock_calls) == 5
assert entity_group1 is not None
assert "Hans" not in entity_group1.attributes
@patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=None,
aioclient_mock.delete(
ENDPOINT_URL.format("persongroups/service_group"),
status=200,
text="{}",
)
def test_service_train(self, mock_update, aioclient_mock):
"""Set up component, test train groups services."""
with assert_setup_component(3, mf.DOMAIN):
setup_component(self.hass, mf.DOMAIN, self.config)
aioclient_mock.post(
self.endpoint_url.format("persongroups/service_group/train"),
status=200,
text="{}",
)
with assert_setup_component(3, mf.DOMAIN):
await async_setup_component(hass, mf.DOMAIN, CONFIG)
train_group(self.hass, "Service Group")
self.hass.block_till_done()
create_group(hass, "Service Group")
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
entity = hass.states.get("microsoft_face.service_group")
assert entity is not None
assert len(aioclient_mock.mock_calls) == 1
@patch(
delete_group(hass, "Service Group")
await hass.async_block_till_done()
entity = hass.states.get("microsoft_face.service_group")
assert entity is None
assert len(aioclient_mock.mock_calls) == 2
async def test_service_person(hass, aioclient_mock):
"""Set up component, test person services."""
aioclient_mock.get(
ENDPOINT_URL.format("persongroups"),
text=load_fixture("microsoft_face_persongroups.json"),
)
aioclient_mock.get(
ENDPOINT_URL.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
aioclient_mock.get(
ENDPOINT_URL.format("persongroups/test_group2/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
with assert_setup_component(3, mf.DOMAIN):
await async_setup_component(hass, mf.DOMAIN, CONFIG)
assert len(aioclient_mock.mock_calls) == 3
aioclient_mock.post(
ENDPOINT_URL.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_create_person.json"),
)
aioclient_mock.delete(
ENDPOINT_URL.format(
"persongroups/test_group1/persons/" "25985303-c537-4467-b41d-bdb45cd95ca1"
),
status=200,
text="{}",
)
create_person(hass, "test group1", "Hans")
await hass.async_block_till_done()
entity_group1 = hass.states.get("microsoft_face.test_group1")
assert len(aioclient_mock.mock_calls) == 4
assert entity_group1 is not None
assert entity_group1.attributes["Hans"] == "25985303-c537-4467-b41d-bdb45cd95ca1"
delete_person(hass, "test group1", "Hans")
await hass.async_block_till_done()
entity_group1 = hass.states.get("microsoft_face.test_group1")
assert len(aioclient_mock.mock_calls) == 5
assert entity_group1 is not None
assert "Hans" not in entity_group1.attributes
async def test_service_train(hass, mock_update, aioclient_mock):
"""Set up component, test train groups services."""
with assert_setup_component(3, mf.DOMAIN):
await async_setup_component(hass, mf.DOMAIN, CONFIG)
aioclient_mock.post(
ENDPOINT_URL.format("persongroups/service_group/train"),
status=200,
text="{}",
)
train_group(hass, "Service Group")
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
async def test_service_face(hass, aioclient_mock):
"""Set up component, test person face services."""
aioclient_mock.get(
ENDPOINT_URL.format("persongroups"),
text=load_fixture("microsoft_face_persongroups.json"),
)
aioclient_mock.get(
ENDPOINT_URL.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
aioclient_mock.get(
ENDPOINT_URL.format("persongroups/test_group2/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
CONFIG["camera"] = {"platform": "demo"}
with assert_setup_component(3, mf.DOMAIN):
await async_setup_component(hass, mf.DOMAIN, CONFIG)
assert len(aioclient_mock.mock_calls) == 3
aioclient_mock.post(
ENDPOINT_URL.format(
"persongroups/test_group2/persons/"
"2ae4935b-9659-44c3-977f-61fac20d0538/persistedFaces"
),
status=200,
text="{}",
)
with patch(
"homeassistant.components.camera.async_get_image",
return_value=camera.Image("image/jpeg", b"Test"),
):
face_person(hass, "test_group2", "David", "camera.demo_camera")
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 4
assert aioclient_mock.mock_calls[3][2] == b"Test"
async def test_service_status_400(hass, mock_update, aioclient_mock):
"""Set up component, test groups services with error."""
aioclient_mock.put(
ENDPOINT_URL.format("persongroups/service_group"),
status=400,
text="{'error': {'message': 'Error'}}",
)
def test_service_face(self, camera_mock, aioclient_mock):
"""Set up component, test person face services."""
aioclient_mock.get(
self.endpoint_url.format("persongroups"),
text=load_fixture("microsoft_face_persongroups.json"),
)
aioclient_mock.get(
self.endpoint_url.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
aioclient_mock.get(
self.endpoint_url.format("persongroups/test_group2/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
self.config["camera"] = {"platform": "demo"}
with assert_setup_component(3, mf.DOMAIN):
setup_component(self.hass, mf.DOMAIN, self.config)
with assert_setup_component(3, mf.DOMAIN):
await async_setup_component(hass, mf.DOMAIN, CONFIG)
assert len(aioclient_mock.mock_calls) == 3
create_group(hass, "Service Group")
await hass.async_block_till_done()
aioclient_mock.post(
self.endpoint_url.format(
"persongroups/test_group2/persons/"
"2ae4935b-9659-44c3-977f-61fac20d0538/persistedFaces"
),
status=200,
text="{}",
)
entity = hass.states.get("microsoft_face.service_group")
assert entity is None
assert len(aioclient_mock.mock_calls) == 1
face_person(self.hass, "test_group2", "David", "camera.demo_camera")
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 4
assert aioclient_mock.mock_calls[3][2] == b"Test"
@patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=None,
async def test_service_status_timeout(hass, mock_update, aioclient_mock):
"""Set up component, test groups services with timeout."""
aioclient_mock.put(
ENDPOINT_URL.format("persongroups/service_group"),
status=400,
exc=asyncio.TimeoutError(),
)
def test_service_status_400(self, mock_update, aioclient_mock):
"""Set up component, test groups services with error."""
aioclient_mock.put(
self.endpoint_url.format("persongroups/service_group"),
status=400,
text="{'error': {'message': 'Error'}}",
)
with assert_setup_component(3, mf.DOMAIN):
setup_component(self.hass, mf.DOMAIN, self.config)
with assert_setup_component(3, mf.DOMAIN):
await async_setup_component(hass, mf.DOMAIN, CONFIG)
create_group(self.hass, "Service Group")
self.hass.block_till_done()
create_group(hass, "Service Group")
await hass.async_block_till_done()
entity = self.hass.states.get("microsoft_face.service_group")
assert entity is None
assert len(aioclient_mock.mock_calls) == 1
@patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=None,
)
def test_service_status_timeout(self, mock_update, aioclient_mock):
"""Set up component, test groups services with timeout."""
aioclient_mock.put(
self.endpoint_url.format("persongroups/service_group"),
status=400,
exc=asyncio.TimeoutError(),
)
with assert_setup_component(3, mf.DOMAIN):
setup_component(self.hass, mf.DOMAIN, self.config)
create_group(self.hass, "Service Group")
self.hass.block_till_done()
entity = self.hass.states.get("microsoft_face.service_group")
assert entity is None
assert len(aioclient_mock.mock_calls) == 1
entity = hass.states.get("microsoft_face.service_group")
assert entity is None
assert len(aioclient_mock.mock_calls) == 1

256
tests/components/microsoft_face_detect/test_image_processing.py
View file

@ -1,171 +1,155 @@
"""The tests for the microsoft face detect platform."""
from unittest.mock import PropertyMock, patch
import pytest
import homeassistant.components.image_processing as ip
import homeassistant.components.microsoft_face as mf
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,
mock_coro,
)
from tests.common import assert_setup_component, load_fixture
from tests.components.image_processing import common
CONFIG = {
ip.DOMAIN: {
"platform": "microsoft_face_detect",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"attributes": ["age", "gender"],
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
class TestMicrosoftFaceDetectSetup:
"""Test class for image processing."""
ENDPOINT_URL = f"https://westus.{mf.FACE_API_URL}"
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()
@patch(
@pytest.fixture
def store_mock():
"""Mock update store."""
with patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=mock_coro(),
)
def test_setup_platform(self, store_mock):
"""Set up platform with one entity."""
config = {
ip.DOMAIN: {
"platform": "microsoft_face_detect",
"source": {"entity_id": "camera.demo_camera"},
"attributes": ["age", "gender"],
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
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.microsoftface_demo_camera")
@patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=mock_coro(),
)
def test_setup_platform_name(self, store_mock):
"""Set up platform with one entity and set name."""
config = {
ip.DOMAIN: {
"platform": "microsoft_face_detect",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
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")
return_value=None,
) as mock_update_store:
yield mock_update_store
class TestMicrosoftFaceDetect:
"""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.config = {
ip.DOMAIN: {
"platform": "microsoft_face_detect",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"attributes": ["age", "gender"],
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
self.endpoint_url = f"https://westus.{mf.FACE_API_URL}"
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
@patch(
@pytest.fixture
def poll_mock():
"""Disable polling."""
with patch(
"homeassistant.components.microsoft_face_detect.image_processing."
"MicrosoftFaceDetectEntity.should_poll",
new_callable=PropertyMock(return_value=False),
):
yield
async def test_setup_platform(hass, store_mock):
"""Set up platform with one entity."""
config = {
ip.DOMAIN: {
"platform": "microsoft_face_detect",
"source": {"entity_id": "camera.demo_camera"},
"attributes": ["age", "gender"],
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
with assert_setup_component(1, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
assert hass.states.get("image_processing.microsoftface_demo_camera")
async def test_setup_platform_name(hass, store_mock):
"""Set up platform with one entity and set name."""
config = {
ip.DOMAIN: {
"platform": "microsoft_face_detect",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
with assert_setup_component(1, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
assert hass.states.get("image_processing.test_local")
async def test_ms_detect_process_image(hass, poll_mock, aioclient_mock):
"""Set up and scan a picture and test plates from event."""
aioclient_mock.get(
ENDPOINT_URL.format("persongroups"),
text=load_fixture("microsoft_face_persongroups.json"),
)
aioclient_mock.get(
ENDPOINT_URL.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
aioclient_mock.get(
ENDPOINT_URL.format("persongroups/test_group2/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
def test_ms_detect_process_image(self, poll_mock, aioclient_mock):
"""Set up and scan a picture and test plates from event."""
aioclient_mock.get(
self.endpoint_url.format("persongroups"),
text=load_fixture("microsoft_face_persongroups.json"),
)
aioclient_mock.get(
self.endpoint_url.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
aioclient_mock.get(
self.endpoint_url.format("persongroups/test_group2/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
setup_component(self.hass, ip.DOMAIN, self.config)
self.hass.block_till_done()
await async_setup_component(hass, ip.DOMAIN, CONFIG)
await hass.async_block_till_done()
state = self.hass.states.get("camera.demo_camera")
url = f"{self.hass.config.internal_url}{state.attributes.get(ATTR_ENTITY_PICTURE)}"
state = hass.states.get("camera.demo_camera")
url = f"{hass.config.internal_url}{state.attributes.get(ATTR_ENTITY_PICTURE)}"
face_events = []
face_events = []
@callback
def mock_face_event(event):
"""Mock event."""
face_events.append(event)
@callback
def mock_face_event(event):
"""Mock event."""
face_events.append(event)
self.hass.bus.listen("image_processing.detect_face", mock_face_event)
hass.bus.async_listen("image_processing.detect_face", mock_face_event)
aioclient_mock.get(url, content=b"image")
aioclient_mock.get(url, content=b"image")
aioclient_mock.post(
self.endpoint_url.format("detect"),
text=load_fixture("microsoft_face_detect.json"),
params={"returnFaceAttributes": "age,gender"},
)
aioclient_mock.post(
ENDPOINT_URL.format("detect"),
text=load_fixture("microsoft_face_detect.json"),
params={"returnFaceAttributes": "age,gender"},
)
common.scan(self.hass, entity_id="image_processing.test_local")
self.hass.block_till_done()
common.async_scan(hass, entity_id="image_processing.test_local")
await hass.async_block_till_done()
state = self.hass.states.get("image_processing.test_local")
state = hass.states.get("image_processing.test_local")
assert len(face_events) == 1
assert state.attributes.get("total_faces") == 1
assert state.state == "1"
assert len(face_events) == 1
assert state.attributes.get("total_faces") == 1
assert state.state == "1"
assert face_events[0].data["age"] == 71.0
assert face_events[0].data["gender"] == "male"
assert face_events[0].data["entity_id"] == "image_processing.test_local"
assert face_events[0].data["age"] == 71.0
assert face_events[0].data["gender"] == "male"
assert face_events[0].data["entity_id"] == "image_processing.test_local"
# Test that later, if a request is made that results in no face
# being detected, that this is reflected in the state object
aioclient_mock.clear_requests()
aioclient_mock.post(
self.endpoint_url.format("detect"),
text="[]",
params={"returnFaceAttributes": "age,gender"},
)
# Test that later, if a request is made that results in no face
# being detected, that this is reflected in the state object
aioclient_mock.clear_requests()
aioclient_mock.post(
ENDPOINT_URL.format("detect"),
text="[]",
params={"returnFaceAttributes": "age,gender"},
)
common.scan(self.hass, entity_id="image_processing.test_local")
self.hass.block_till_done()
common.async_scan(hass, entity_id="image_processing.test_local")
await hass.async_block_till_done()
state = self.hass.states.get("image_processing.test_local")
state = hass.states.get("image_processing.test_local")
# No more face events were fired
assert len(face_events) == 1
# Total faces and actual qualified number of faces reset to zero
assert state.attributes.get("total_faces") == 0
assert state.state == "0"
# No more face events were fired
assert len(face_events) == 1
# Total faces and actual qualified number of faces reset to zero
assert state.attributes.get("total_faces") == 0
assert state.state == "0"

261
tests/components/microsoft_face_identify/test_image_processing.py
View file

@ -1,171 +1,156 @@
"""The tests for the microsoft face identify platform."""
from unittest.mock import PropertyMock, patch
import pytest
import homeassistant.components.image_processing as ip
import homeassistant.components.microsoft_face as mf
from homeassistant.const import ATTR_ENTITY_PICTURE, STATE_UNKNOWN
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,
mock_coro,
)
from tests.common import assert_setup_component, load_fixture
from tests.components.image_processing import common
class TestMicrosoftFaceIdentifySetup:
"""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()
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
@patch(
@pytest.fixture
def store_mock():
"""Mock update store."""
with patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=mock_coro(),
)
def test_setup_platform(self, store_mock):
"""Set up platform with one entity."""
config = {
ip.DOMAIN: {
"platform": "microsoft_face_identify",
"source": {"entity_id": "camera.demo_camera"},
"group": "Test Group1",
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
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.microsoftface_demo_camera")
@patch(
"homeassistant.components.microsoft_face.MicrosoftFace.update_store",
return_value=mock_coro(),
)
def test_setup_platform_name(self, store_mock):
"""Set up platform with one entity and set name."""
config = {
ip.DOMAIN: {
"platform": "microsoft_face_identify",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"group": "Test Group1",
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
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")
return_value=None,
) as mock_update_store:
yield mock_update_store
class TestMicrosoftFaceIdentify:
"""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.config = {
ip.DOMAIN: {
"platform": "microsoft_face_identify",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"group": "Test Group1",
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
self.endpoint_url = f"https://westus.{mf.FACE_API_URL}"
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
@patch(
@pytest.fixture
def poll_mock():
"""Disable polling."""
with patch(
"homeassistant.components.microsoft_face_identify.image_processing."
"MicrosoftFaceIdentifyEntity.should_poll",
new_callable=PropertyMock(return_value=False),
):
yield
CONFIG = {
ip.DOMAIN: {
"platform": "microsoft_face_identify",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"group": "Test Group1",
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
ENDPOINT_URL = f"https://westus.{mf.FACE_API_URL}"
async def test_setup_platform(hass, store_mock):
"""Set up platform with one entity."""
config = {
ip.DOMAIN: {
"platform": "microsoft_face_identify",
"source": {"entity_id": "camera.demo_camera"},
"group": "Test Group1",
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
with assert_setup_component(1, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
assert hass.states.get("image_processing.microsoftface_demo_camera")
async def test_setup_platform_name(hass, store_mock):
"""Set up platform with one entity and set name."""
config = {
ip.DOMAIN: {
"platform": "microsoft_face_identify",
"source": {"entity_id": "camera.demo_camera", "name": "test local"},
"group": "Test Group1",
},
"camera": {"platform": "demo"},
mf.DOMAIN: {"api_key": "12345678abcdef6"},
}
with assert_setup_component(1, ip.DOMAIN):
await async_setup_component(hass, ip.DOMAIN, config)
await hass.async_block_till_done()
assert hass.states.get("image_processing.test_local")
async def test_ms_identify_process_image(hass, poll_mock, aioclient_mock):
"""Set up and scan a picture and test plates from event."""
aioclient_mock.get(
ENDPOINT_URL.format("persongroups"),
text=load_fixture("microsoft_face_persongroups.json"),
)
aioclient_mock.get(
ENDPOINT_URL.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
aioclient_mock.get(
ENDPOINT_URL.format("persongroups/test_group2/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
def test_ms_identify_process_image(self, poll_mock, aioclient_mock):
"""Set up and scan a picture and test plates from event."""
aioclient_mock.get(
self.endpoint_url.format("persongroups"),
text=load_fixture("microsoft_face_persongroups.json"),
)
aioclient_mock.get(
self.endpoint_url.format("persongroups/test_group1/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
aioclient_mock.get(
self.endpoint_url.format("persongroups/test_group2/persons"),
text=load_fixture("microsoft_face_persons.json"),
)
setup_component(self.hass, ip.DOMAIN, self.config)
self.hass.block_till_done()
await async_setup_component(hass, ip.DOMAIN, CONFIG)
await hass.async_block_till_done()
state = self.hass.states.get("camera.demo_camera")
url = f"{self.hass.config.internal_url}{state.attributes.get(ATTR_ENTITY_PICTURE)}"
state = hass.states.get("camera.demo_camera")
url = f"{hass.config.internal_url}{state.attributes.get(ATTR_ENTITY_PICTURE)}"
face_events = []
face_events = []
@callback
def mock_face_event(event):
"""Mock event."""
face_events.append(event)
@callback
def mock_face_event(event):
"""Mock event."""
face_events.append(event)
self.hass.bus.listen("image_processing.detect_face", mock_face_event)
hass.bus.async_listen("image_processing.detect_face", mock_face_event)
aioclient_mock.get(url, content=b"image")
aioclient_mock.get(url, content=b"image")
aioclient_mock.post(
self.endpoint_url.format("detect"),
text=load_fixture("microsoft_face_detect.json"),
)
aioclient_mock.post(
self.endpoint_url.format("identify"),
text=load_fixture("microsoft_face_identify.json"),
)
aioclient_mock.post(
ENDPOINT_URL.format("detect"),
text=load_fixture("microsoft_face_detect.json"),
)
aioclient_mock.post(
ENDPOINT_URL.format("identify"),
text=load_fixture("microsoft_face_identify.json"),
)
common.scan(self.hass, entity_id="image_processing.test_local")
self.hass.block_till_done()
common.async_scan(hass, entity_id="image_processing.test_local")
await hass.async_block_till_done()
state = self.hass.states.get("image_processing.test_local")
state = hass.states.get("image_processing.test_local")
assert len(face_events) == 1
assert state.attributes.get("total_faces") == 2
assert state.state == "David"
assert len(face_events) == 1
assert state.attributes.get("total_faces") == 2
assert state.state == "David"
assert face_events[0].data["name"] == "David"
assert face_events[0].data["confidence"] == float(92)
assert face_events[0].data["entity_id"] == "image_processing.test_local"
assert face_events[0].data["name"] == "David"
assert face_events[0].data["confidence"] == float(92)
assert face_events[0].data["entity_id"] == "image_processing.test_local"
# Test that later, if a request is made that results in no face
# being detected, that this is reflected in the state object
aioclient_mock.clear_requests()
aioclient_mock.post(self.endpoint_url.format("detect"), text="[]")
# Test that later, if a request is made that results in no face
# being detected, that this is reflected in the state object
aioclient_mock.clear_requests()
aioclient_mock.post(ENDPOINT_URL.format("detect"), text="[]")
common.scan(self.hass, entity_id="image_processing.test_local")
self.hass.block_till_done()
common.async_scan(hass, entity_id="image_processing.test_local")
await hass.async_block_till_done()
state = self.hass.states.get("image_processing.test_local")
state = hass.states.get("image_processing.test_local")
# No more face events were fired
assert len(face_events) == 1
# Total faces and actual qualified number of faces reset to zero
assert state.attributes.get("total_faces") == 0
assert state.state == STATE_UNKNOWN
# No more face events were fired
assert len(face_events) == 1
# Total faces and actual qualified number of faces reset to zero
assert state.attributes.get("total_faces") == 0
assert state.state == STATE_UNKNOWN