Drop facebox integration (#107005)
Sam Reed
GitHub
parent
c86b45b454
commit
bca629ed31
|
|
@ -1 +0,0 @@
|
|||
"""The facebox component."""
|
||||
|
|
@ -1,4 +0,0 @@
|
|||
"""Constants for the Facebox component."""
|
||||
|
||||
DOMAIN = "facebox"
|
||||
SERVICE_TEACH_FACE = "teach_face"
|
||||
|
|
@ -1,282 +0,0 @@
|
|||
"""Component for facial detection and identification via facebox."""
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
from http import HTTPStatus
|
||||
import logging
|
||||
|
||||
import requests
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.image_processing import (
|
||||
ATTR_CONFIDENCE,
|
||||
PLATFORM_SCHEMA,
|
||||
ImageProcessingFaceEntity,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
ATTR_ID,
|
||||
ATTR_NAME,
|
||||
CONF_ENTITY_ID,
|
||||
CONF_IP_ADDRESS,
|
||||
CONF_NAME,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_SOURCE,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, ServiceCall, split_entity_id
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from .const import DOMAIN, SERVICE_TEACH_FACE
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ATTR_BOUNDING_BOX = "bounding_box"
|
||||
ATTR_CLASSIFIER = "classifier"
|
||||
ATTR_IMAGE_ID = "image_id"
|
||||
ATTR_MATCHED = "matched"
|
||||
FACEBOX_NAME = "name"
|
||||
CLASSIFIER = "facebox"
|
||||
DATA_FACEBOX = "facebox_classifiers"
|
||||
FILE_PATH = "file_path"
|
||||
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_IP_ADDRESS): cv.string,
|
||||
vol.Required(CONF_PORT): cv.port,
|
||||
vol.Optional(CONF_USERNAME): cv.string,
|
||||
vol.Optional(CONF_PASSWORD): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
SERVICE_TEACH_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Optional(ATTR_ENTITY_ID): cv.entity_ids,
|
||||
vol.Required(ATTR_NAME): cv.string,
|
||||
vol.Required(FILE_PATH): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def check_box_health(url, username, password):
|
||||
"""Check the health of the classifier and return its id if healthy."""
|
||||
kwargs = {}
|
||||
if username:
|
||||
kwargs["auth"] = requests.auth.HTTPBasicAuth(username, password)
|
||||
try:
|
||||
response = requests.get(url, **kwargs, timeout=10)
|
||||
if response.status_code == HTTPStatus.UNAUTHORIZED:
|
||||
_LOGGER.error("AuthenticationError on %s", CLASSIFIER)
|
||||
return None
|
||||
if response.status_code == HTTPStatus.OK:
|
||||
return response.json()["hostname"]
|
||||
except requests.exceptions.ConnectionError:
|
||||
_LOGGER.error("ConnectionError: Is %s running?", CLASSIFIER)
|
||||
return None
|
||||
|
||||
|
||||
def encode_image(image):
|
||||
"""base64 encode an image stream."""
|
||||
base64_img = base64.b64encode(image).decode("ascii")
|
||||
return base64_img
|
||||
|
||||
|
||||
def get_matched_faces(faces):
|
||||
"""Return the name and rounded confidence of matched faces."""
|
||||
return {
|
||||
face["name"]: round(face["confidence"], 2) for face in faces if face["matched"]
|
||||
}
|
||||
|
||||
|
||||
def parse_faces(api_faces):
|
||||
"""Parse the API face data into the format required."""
|
||||
known_faces = []
|
||||
for entry in api_faces:
|
||||
face = {}
|
||||
if entry["matched"]: # This data is only in matched faces.
|
||||
face[FACEBOX_NAME] = entry["name"]
|
||||
face[ATTR_IMAGE_ID] = entry["id"]
|
||||
else: # Lets be explicit.
|
||||
face[FACEBOX_NAME] = None
|
||||
face[ATTR_IMAGE_ID] = None
|
||||
face[ATTR_CONFIDENCE] = round(100.0 * entry["confidence"], 2)
|
||||
face[ATTR_MATCHED] = entry["matched"]
|
||||
face[ATTR_BOUNDING_BOX] = entry["rect"]
|
||||
known_faces.append(face)
|
||||
return known_faces
|
||||
|
||||
|
||||
def post_image(url, image, username, password):
|
||||
"""Post an image to the classifier."""
|
||||
kwargs = {}
|
||||
if username:
|
||||
kwargs["auth"] = requests.auth.HTTPBasicAuth(username, password)
|
||||
try:
|
||||
response = requests.post(
|
||||
url, json={"base64": encode_image(image)}, timeout=10, **kwargs
|
||||
)
|
||||
if response.status_code == HTTPStatus.UNAUTHORIZED:
|
||||
_LOGGER.error("AuthenticationError on %s", CLASSIFIER)
|
||||
return None
|
||||
return response
|
||||
except requests.exceptions.ConnectionError:
|
||||
_LOGGER.error("ConnectionError: Is %s running?", CLASSIFIER)
|
||||
return None
|
||||
|
||||
|
||||
def teach_file(url, name, file_path, username, password):
|
||||
"""Teach the classifier a name associated with a file."""
|
||||
kwargs = {}
|
||||
if username:
|
||||
kwargs["auth"] = requests.auth.HTTPBasicAuth(username, password)
|
||||
try:
|
||||
with open(file_path, "rb") as open_file:
|
||||
response = requests.post(
|
||||
url,
|
||||
data={FACEBOX_NAME: name, ATTR_ID: file_path},
|
||||
files={"file": open_file},
|
||||
timeout=10,
|
||||
**kwargs,
|
||||
)
|
||||
if response.status_code == HTTPStatus.UNAUTHORIZED:
|
||||
_LOGGER.error("AuthenticationError on %s", CLASSIFIER)
|
||||
elif response.status_code == HTTPStatus.BAD_REQUEST:
|
||||
_LOGGER.error(
|
||||
"%s teaching of file %s failed with message:%s",
|
||||
CLASSIFIER,
|
||||
file_path,
|
||||
response.text,
|
||||
)
|
||||
except requests.exceptions.ConnectionError:
|
||||
_LOGGER.error("ConnectionError: Is %s running?", CLASSIFIER)
|
||||
|
||||
|
||||
def valid_file_path(file_path):
|
||||
"""Check that a file_path points to a valid file."""
|
||||
try:
|
||||
cv.isfile(file_path)
|
||||
return True
|
||||
except vol.Invalid:
|
||||
_LOGGER.error("%s error: Invalid file path: %s", CLASSIFIER, file_path)
|
||||
return False
|
||||
|
||||
|
||||
def setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the classifier."""
|
||||
if DATA_FACEBOX not in hass.data:
|
||||
hass.data[DATA_FACEBOX] = []
|
||||
|
||||
ip_address = config[CONF_IP_ADDRESS]
|
||||
port = config[CONF_PORT]
|
||||
username = config.get(CONF_USERNAME)
|
||||
password = config.get(CONF_PASSWORD)
|
||||
url_health = f"http://{ip_address}:{port}/healthz"
|
||||
hostname = check_box_health(url_health, username, password)
|
||||
if hostname is None:
|
||||
return
|
||||
|
||||
entities = []
|
||||
for camera in config[CONF_SOURCE]:
|
||||
facebox = FaceClassifyEntity(
|
||||
ip_address,
|
||||
port,
|
||||
username,
|
||||
password,
|
||||
hostname,
|
||||
camera[CONF_ENTITY_ID],
|
||||
camera.get(CONF_NAME),
|
||||
)
|
||||
entities.append(facebox)
|
||||
hass.data[DATA_FACEBOX].append(facebox)
|
||||
add_entities(entities)
|
||||
|
||||
def service_handle(service: ServiceCall) -> None:
|
||||
"""Handle for services."""
|
||||
entity_ids = service.data.get("entity_id")
|
||||
|
||||
classifiers = hass.data[DATA_FACEBOX]
|
||||
if entity_ids:
|
||||
classifiers = [c for c in classifiers if c.entity_id in entity_ids]
|
||||
|
||||
for classifier in classifiers:
|
||||
name = service.data.get(ATTR_NAME)
|
||||
file_path = service.data.get(FILE_PATH)
|
||||
classifier.teach(name, file_path)
|
||||
|
||||
hass.services.register(
|
||||
DOMAIN, SERVICE_TEACH_FACE, service_handle, schema=SERVICE_TEACH_SCHEMA
|
||||
)
|
||||
|
||||
|
||||
class FaceClassifyEntity(ImageProcessingFaceEntity):
|
||||
"""Perform a face classification."""
|
||||
|
||||
def __init__(
|
||||
self, ip_address, port, username, password, hostname, camera_entity, name=None
|
||||
):
|
||||
"""Init with the API key and model id."""
|
||||
super().__init__()
|
||||
self._url_check = f"http://{ip_address}:{port}/{CLASSIFIER}/check"
|
||||
self._url_teach = f"http://{ip_address}:{port}/{CLASSIFIER}/teach"
|
||||
self._username = username
|
||||
self._password = password
|
||||
self._hostname = hostname
|
||||
self._camera = camera_entity
|
||||
if name:
|
||||
self._name = name
|
||||
else:
|
||||
camera_name = split_entity_id(camera_entity)[1]
|
||||
self._name = f"{CLASSIFIER} {camera_name}"
|
||||
self._matched = {}
|
||||
|
||||
def process_image(self, image):
|
||||
"""Process an image."""
|
||||
response = post_image(self._url_check, image, self._username, self._password)
|
||||
if response:
|
||||
response_json = response.json()
|
||||
if response_json["success"]:
|
||||
total_faces = response_json["facesCount"]
|
||||
faces = parse_faces(response_json["faces"])
|
||||
self._matched = get_matched_faces(faces)
|
||||
self.process_faces(faces, total_faces)
|
||||
|
||||
else:
|
||||
self.total_faces = None
|
||||
self.faces = []
|
||||
self._matched = {}
|
||||
|
||||
def teach(self, name, file_path):
|
||||
"""Teach classifier a face name."""
|
||||
if not self.hass.config.is_allowed_path(file_path) or not valid_file_path(
|
||||
file_path
|
||||
):
|
||||
return
|
||||
teach_file(self._url_teach, name, file_path, self._username, self._password)
|
||||
|
||||
@property
|
||||
def camera_entity(self):
|
||||
"""Return camera entity id from process pictures."""
|
||||
return self._camera
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the classifier attributes."""
|
||||
return {
|
||||
"matched_faces": self._matched,
|
||||
"total_matched_faces": len(self._matched),
|
||||
"hostname": self._hostname,
|
||||
}
|
||||
|
|
@ -1,7 +0,0 @@
|
|||
{
|
||||
"domain": "facebox",
|
||||
"name": "Facebox",
|
||||
"codeowners": [],
|
||||
"documentation": "https://www.home-assistant.io/integrations/facebox",
|
||||
"iot_class": "local_push"
|
||||
}
|
||||
|
|
@ -1,17 +0,0 @@
|
|||
teach_face:
|
||||
fields:
|
||||
entity_id:
|
||||
selector:
|
||||
entity:
|
||||
integration: facebox
|
||||
domain: image_processing
|
||||
name:
|
||||
required: true
|
||||
example: "my_name"
|
||||
selector:
|
||||
text:
|
||||
file_path:
|
||||
required: true
|
||||
example: "/images/my_image.jpg"
|
||||
selector:
|
||||
text:
|
||||
|
|
@ -1,22 +0,0 @@
|
|||
{
|
||||
"services": {
|
||||
"teach_face": {
|
||||
"name": "Teach face",
|
||||
"description": "Teaches facebox a face using a file.",
|
||||
"fields": {
|
||||
"entity_id": {
|
||||
"name": "Entity",
|
||||
"description": "The facebox entity to teach."
|
||||
},
|
||||
"name": {
|
||||
"name": "[%key:common::config_flow::data::name%]",
|
||||
"description": "The name of the face to teach."
|
||||
},
|
||||
"file_path": {
|
||||
"name": "File path",
|
||||
"description": "The path to the image file."
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1680,12 +1680,6 @@
|
|||
"config_flow": false,
|
||||
"iot_class": "cloud_push"
|
||||
},
|
||||
"facebox": {
|
||||
"name": "Facebox",
|
||||
"integration_type": "hub",
|
||||
"config_flow": false,
|
||||
"iot_class": "local_push"
|
||||
},
|
||||
"fail2ban": {
|
||||
"name": "Fail2Ban",
|
||||
"integration_type": "hub",
|
||||
|
|
|
|||
|
|
@ -1 +0,0 @@
|
|||
"""Tests for the facebox component."""
|
||||
|
|
@ -1,341 +0,0 @@
|
|||
"""The tests for the facebox component."""
|
||||
from http import HTTPStatus
|
||||
from unittest.mock import Mock, mock_open, patch
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
import requests_mock
|
||||
|
||||
import homeassistant.components.facebox.image_processing as fb
|
||||
import homeassistant.components.image_processing as ip
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
ATTR_NAME,
|
||||
CONF_FRIENDLY_NAME,
|
||||
CONF_IP_ADDRESS,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_USERNAME,
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
MOCK_IP = "192.168.0.1"
|
||||
MOCK_PORT = "8080"
|
||||
|
||||
# Mock data returned by the facebox API.
|
||||
MOCK_BOX_ID = "b893cc4f7fd6"
|
||||
MOCK_ERROR_NO_FACE = "No face found"
|
||||
MOCK_FACE = {
|
||||
"confidence": 0.5812028911604818,
|
||||
"id": "john.jpg",
|
||||
"matched": True,
|
||||
"name": "John Lennon",
|
||||
"rect": {"height": 75, "left": 63, "top": 262, "width": 74},
|
||||
}
|
||||
|
||||
MOCK_FILE_PATH = "/images/mock.jpg"
|
||||
|
||||
MOCK_HEALTH = {
|
||||
"success": True,
|
||||
"hostname": "b893cc4f7fd6",
|
||||
"metadata": {"boxname": "facebox", "build": "development"},
|
||||
"errors": [],
|
||||
}
|
||||
|
||||
MOCK_JSON = {"facesCount": 1, "success": True, "faces": [MOCK_FACE]}
|
||||
|
||||
MOCK_NAME = "mock_name"
|
||||
MOCK_USERNAME = "mock_username"
|
||||
MOCK_PASSWORD = "mock_password"
|
||||
|
||||
# Faces data after parsing.
|
||||
PARSED_FACES = [
|
||||
{
|
||||
fb.FACEBOX_NAME: "John Lennon",
|
||||
fb.ATTR_IMAGE_ID: "john.jpg",
|
||||
fb.ATTR_CONFIDENCE: 58.12,
|
||||
fb.ATTR_MATCHED: True,
|
||||
fb.ATTR_BOUNDING_BOX: {"height": 75, "left": 63, "top": 262, "width": 74},
|
||||
}
|
||||
]
|
||||
|
||||
MATCHED_FACES = {"John Lennon": 58.12}
|
||||
|
||||
VALID_ENTITY_ID = "image_processing.facebox_demo_camera"
|
||||
VALID_CONFIG = {
|
||||
ip.DOMAIN: {
|
||||
"platform": "facebox",
|
||||
CONF_IP_ADDRESS: MOCK_IP,
|
||||
CONF_PORT: MOCK_PORT,
|
||||
ip.CONF_SOURCE: {ip.CONF_ENTITY_ID: "camera.demo_camera"},
|
||||
},
|
||||
"camera": {"platform": "demo"},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def setup_homeassistant(hass: HomeAssistant):
|
||||
"""Set up the homeassistant integration."""
|
||||
await async_setup_component(hass, "homeassistant", {})
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_healthybox():
|
||||
"""Mock fb.check_box_health."""
|
||||
check_box_health = (
|
||||
"homeassistant.components.facebox.image_processing.check_box_health"
|
||||
)
|
||||
with patch(check_box_health, return_value=MOCK_BOX_ID) as _mock_healthybox:
|
||||
yield _mock_healthybox
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_isfile():
|
||||
"""Mock os.path.isfile."""
|
||||
with patch(
|
||||
"homeassistant.components.facebox.image_processing.cv.isfile", return_value=True
|
||||
) as _mock_isfile:
|
||||
yield _mock_isfile
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_image():
|
||||
"""Return a mock camera image."""
|
||||
with patch(
|
||||
"homeassistant.components.demo.camera.DemoCamera.camera_image",
|
||||
return_value=b"Test",
|
||||
) as image:
|
||||
yield image
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_open_file():
|
||||
"""Mock open."""
|
||||
mopen = mock_open()
|
||||
with patch(
|
||||
"homeassistant.components.facebox.image_processing.open", mopen, create=True
|
||||
) as _mock_open:
|
||||
yield _mock_open
|
||||
|
||||
|
||||
def test_check_box_health(caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""Test check box health."""
|
||||
with requests_mock.Mocker() as mock_req:
|
||||
url = f"http://{MOCK_IP}:{MOCK_PORT}/healthz"
|
||||
mock_req.get(url, status_code=HTTPStatus.OK, json=MOCK_HEALTH)
|
||||
assert fb.check_box_health(url, "user", "pass") == MOCK_BOX_ID
|
||||
|
||||
mock_req.get(url, status_code=HTTPStatus.UNAUTHORIZED)
|
||||
assert fb.check_box_health(url, None, None) is None
|
||||
assert "AuthenticationError on facebox" in caplog.text
|
||||
|
||||
mock_req.get(url, exc=requests.exceptions.ConnectTimeout)
|
||||
fb.check_box_health(url, None, None)
|
||||
assert "ConnectionError: Is facebox running?" in caplog.text
|
||||
|
||||
|
||||
def test_encode_image() -> None:
|
||||
"""Test that binary data is encoded correctly."""
|
||||
assert fb.encode_image(b"test") == "dGVzdA=="
|
||||
|
||||
|
||||
def test_get_matched_faces() -> None:
|
||||
"""Test that matched_faces are parsed correctly."""
|
||||
assert fb.get_matched_faces(PARSED_FACES) == MATCHED_FACES
|
||||
|
||||
|
||||
def test_parse_faces() -> None:
|
||||
"""Test parsing of raw face data, and generation of matched_faces."""
|
||||
assert fb.parse_faces(MOCK_JSON["faces"]) == PARSED_FACES
|
||||
|
||||
|
||||
@patch("os.access", Mock(return_value=False))
|
||||
def test_valid_file_path() -> None:
|
||||
"""Test that an invalid file_path is caught."""
|
||||
assert not fb.valid_file_path("test_path")
|
||||
|
||||
|
||||
async def test_setup_platform(hass: HomeAssistant, mock_healthybox) -> None:
|
||||
"""Set up platform with one entity."""
|
||||
await async_setup_component(hass, ip.DOMAIN, VALID_CONFIG)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(VALID_ENTITY_ID)
|
||||
|
||||
|
||||
async def test_setup_platform_with_auth(hass: HomeAssistant, mock_healthybox) -> None:
|
||||
"""Set up platform with one entity and auth."""
|
||||
valid_config_auth = VALID_CONFIG.copy()
|
||||
valid_config_auth[ip.DOMAIN][CONF_USERNAME] = MOCK_USERNAME
|
||||
valid_config_auth[ip.DOMAIN][CONF_PASSWORD] = MOCK_PASSWORD
|
||||
|
||||
await async_setup_component(hass, ip.DOMAIN, valid_config_auth)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(VALID_ENTITY_ID)
|
||||
|
||||
|
||||
async def test_process_image(hass: HomeAssistant, mock_healthybox, mock_image) -> None:
|
||||
"""Test successful processing of an image."""
|
||||
await async_setup_component(hass, ip.DOMAIN, VALID_CONFIG)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(VALID_ENTITY_ID)
|
||||
|
||||
face_events = []
|
||||
|
||||
@callback
|
||||
def mock_face_event(event):
|
||||
"""Mock event."""
|
||||
face_events.append(event)
|
||||
|
||||
hass.bus.async_listen("image_processing.detect_face", mock_face_event)
|
||||
|
||||
with requests_mock.Mocker() as mock_req:
|
||||
url = f"http://{MOCK_IP}:{MOCK_PORT}/facebox/check"
|
||||
mock_req.post(url, json=MOCK_JSON)
|
||||
data = {ATTR_ENTITY_ID: VALID_ENTITY_ID}
|
||||
await hass.services.async_call(ip.DOMAIN, ip.SERVICE_SCAN, service_data=data)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(VALID_ENTITY_ID)
|
||||
assert state.state == "1"
|
||||
assert state.attributes.get("matched_faces") == MATCHED_FACES
|
||||
assert state.attributes.get("total_matched_faces") == 1
|
||||
|
||||
PARSED_FACES[0][ATTR_ENTITY_ID] = VALID_ENTITY_ID # Update.
|
||||
assert state.attributes.get("faces") == PARSED_FACES
|
||||
assert state.attributes.get(CONF_FRIENDLY_NAME) == "facebox demo_camera"
|
||||
|
||||
assert len(face_events) == 1
|
||||
assert face_events[0].data[ATTR_NAME] == PARSED_FACES[0][ATTR_NAME]
|
||||
assert (
|
||||
face_events[0].data[fb.ATTR_CONFIDENCE] == PARSED_FACES[0][fb.ATTR_CONFIDENCE]
|
||||
)
|
||||
assert face_events[0].data[ATTR_ENTITY_ID] == VALID_ENTITY_ID
|
||||
assert face_events[0].data[fb.ATTR_IMAGE_ID] == PARSED_FACES[0][fb.ATTR_IMAGE_ID]
|
||||
assert (
|
||||
face_events[0].data[fb.ATTR_BOUNDING_BOX]
|
||||
== PARSED_FACES[0][fb.ATTR_BOUNDING_BOX]
|
||||
)
|
||||
|
||||
|
||||
async def test_process_image_errors(
|
||||
hass: HomeAssistant, mock_healthybox, mock_image, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test process_image errors."""
|
||||
await async_setup_component(hass, ip.DOMAIN, VALID_CONFIG)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(VALID_ENTITY_ID)
|
||||
|
||||
# Test connection error.
|
||||
with requests_mock.Mocker() as mock_req:
|
||||
url = f"http://{MOCK_IP}:{MOCK_PORT}/facebox/check"
|
||||
mock_req.register_uri("POST", url, exc=requests.exceptions.ConnectTimeout)
|
||||
data = {ATTR_ENTITY_ID: VALID_ENTITY_ID}
|
||||
await hass.services.async_call(ip.DOMAIN, ip.SERVICE_SCAN, service_data=data)
|
||||
await hass.async_block_till_done()
|
||||
assert "ConnectionError: Is facebox running?" in caplog.text
|
||||
|
||||
state = hass.states.get(VALID_ENTITY_ID)
|
||||
assert state.state == STATE_UNKNOWN
|
||||
assert state.attributes.get("faces") == []
|
||||
assert state.attributes.get("matched_faces") == {}
|
||||
|
||||
# Now test with bad auth.
|
||||
with requests_mock.Mocker() as mock_req:
|
||||
url = f"http://{MOCK_IP}:{MOCK_PORT}/facebox/check"
|
||||
mock_req.register_uri("POST", url, status_code=HTTPStatus.UNAUTHORIZED)
|
||||
data = {ATTR_ENTITY_ID: VALID_ENTITY_ID}
|
||||
await hass.services.async_call(ip.DOMAIN, ip.SERVICE_SCAN, service_data=data)
|
||||
await hass.async_block_till_done()
|
||||
assert "AuthenticationError on facebox" in caplog.text
|
||||
|
||||
|
||||
async def test_teach_service(
|
||||
hass: HomeAssistant,
|
||||
mock_healthybox,
|
||||
mock_image,
|
||||
mock_isfile,
|
||||
mock_open_file,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test teaching of facebox."""
|
||||
await async_setup_component(hass, ip.DOMAIN, VALID_CONFIG)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(VALID_ENTITY_ID)
|
||||
|
||||
# Patch out 'is_allowed_path' as the mock files aren't allowed
|
||||
hass.config.is_allowed_path = Mock(return_value=True)
|
||||
|
||||
# Test successful teach.
|
||||
with requests_mock.Mocker() as mock_req:
|
||||
url = f"http://{MOCK_IP}:{MOCK_PORT}/facebox/teach"
|
||||
mock_req.post(url, status_code=HTTPStatus.OK)
|
||||
data = {
|
||||
ATTR_ENTITY_ID: VALID_ENTITY_ID,
|
||||
ATTR_NAME: MOCK_NAME,
|
||||
fb.FILE_PATH: MOCK_FILE_PATH,
|
||||
}
|
||||
await hass.services.async_call(
|
||||
fb.DOMAIN, fb.SERVICE_TEACH_FACE, service_data=data
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Now test with bad auth.
|
||||
with requests_mock.Mocker() as mock_req:
|
||||
url = f"http://{MOCK_IP}:{MOCK_PORT}/facebox/teach"
|
||||
mock_req.post(url, status_code=HTTPStatus.UNAUTHORIZED)
|
||||
data = {
|
||||
ATTR_ENTITY_ID: VALID_ENTITY_ID,
|
||||
ATTR_NAME: MOCK_NAME,
|
||||
fb.FILE_PATH: MOCK_FILE_PATH,
|
||||
}
|
||||
await hass.services.async_call(
|
||||
fb.DOMAIN, fb.SERVICE_TEACH_FACE, service_data=data
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert "AuthenticationError on facebox" in caplog.text
|
||||
|
||||
# Now test the failed teaching.
|
||||
with requests_mock.Mocker() as mock_req:
|
||||
url = f"http://{MOCK_IP}:{MOCK_PORT}/facebox/teach"
|
||||
mock_req.post(url, status_code=HTTPStatus.BAD_REQUEST, text=MOCK_ERROR_NO_FACE)
|
||||
data = {
|
||||
ATTR_ENTITY_ID: VALID_ENTITY_ID,
|
||||
ATTR_NAME: MOCK_NAME,
|
||||
fb.FILE_PATH: MOCK_FILE_PATH,
|
||||
}
|
||||
await hass.services.async_call(
|
||||
fb.DOMAIN, fb.SERVICE_TEACH_FACE, service_data=data
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert MOCK_ERROR_NO_FACE in caplog.text
|
||||
|
||||
# Now test connection error.
|
||||
with requests_mock.Mocker() as mock_req:
|
||||
url = f"http://{MOCK_IP}:{MOCK_PORT}/facebox/teach"
|
||||
mock_req.post(url, exc=requests.exceptions.ConnectTimeout)
|
||||
data = {
|
||||
ATTR_ENTITY_ID: VALID_ENTITY_ID,
|
||||
ATTR_NAME: MOCK_NAME,
|
||||
fb.FILE_PATH: MOCK_FILE_PATH,
|
||||
}
|
||||
await hass.services.async_call(
|
||||
fb.DOMAIN, fb.SERVICE_TEACH_FACE, service_data=data
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert "ConnectionError: Is facebox running?" in caplog.text
|
||||
|
||||
|
||||
async def test_setup_platform_with_name(hass: HomeAssistant, mock_healthybox) -> None:
|
||||
"""Set up platform with one entity and a name."""
|
||||
named_entity_id = f"image_processing.{MOCK_NAME}"
|
||||
|
||||
valid_config_named = VALID_CONFIG.copy()
|
||||
valid_config_named[ip.DOMAIN][ip.CONF_SOURCE][ip.CONF_NAME] = MOCK_NAME
|
||||
|
||||
await async_setup_component(hass, ip.DOMAIN, valid_config_named)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(named_entity_id)
|
||||
state = hass.states.get(named_entity_id)
|
||||
assert state.attributes.get(CONF_FRIENDLY_NAME) == MOCK_NAME
|
||||
Loading…
Reference in New Issue