hacore/tests/test_core.py at dev · dominicankev/hacore

Repository navigation

8000

"""Test to verify that Home Assistant core works."""

from __future__ import annotations

# pylint: disable=protected-access

import array

import asyncio

from datetime import datetime, timedelta

import functools

import logging

import os

from tempfile import TemporaryDirectory

from unittest.mock import MagicMock, Mock, PropertyMock, patch

import pytest

import voluptuous as vol

from homeassistant.const import (

ATTR_FRIENDLY_NAME,

CONF_UNIT_SYSTEM,

EVENT_CALL_SERVICE,

EVENT_CORE_CONFIG_UPDATE,

EVENT_HOMEASSISTANT_CLOSE,

EVENT_HOMEASSISTANT_FINAL_WRITE,

EVENT_HOMEASSISTANT_START,

EVENT_HOMEASSISTANT_STARTED,

EVENT_HOMEASSISTANT_STOP,

EVENT_SERVICE_REGISTERED,

EVENT_SERVICE_REMOVED,

EVENT_STATE_CHANGED,

MATCH_ALL,

__version__,

)

import homeassistant.core as ha

from homeassistant.core import State

from homeassistant.exceptions import (

InvalidEntityFormatError,

InvalidStateError,

MaxLengthExceeded,

ServiceNotFound,

)

import homeassistant.util.dt as dt_util

from homeassistant.util.read_only_dict import ReadOnlyDict

from homeassistant.util.unit_system import METRIC_SYSTEM

from tests.common import async_capture_events, async_mock_service

PST = dt_util.get_time_zone("America/Los_Angeles")

def test_split_entity_id():

"""Test split_entity_id."""

assert ha.split_entity_id("domain.object_id") == ("domain", "object_id")

with pytest.raises(ValueError):

ha.split_entity_id("")

with pytest.raises(ValueError):

ha.split_entity_id(".")

with pytest.raises(ValueError):

ha.split_entity_id("just_domain")

with pytest.raises(ValueError):

ha.split_entity_id("empty_object_id.")

with pytest.raises(ValueError):

ha.split_entity_id(".empty_domain")

def test_async_add_hass_job_schedule_callback():

"""Test that we schedule coroutines and add jobs to the job pool."""

hass = MagicMock()

job = MagicMock()

ha.HomeAssistant.async_add_hass_job(hass, ha.HassJob(ha.callback(job)))

assert len(hass.loop.call_soon.mock_calls) == 1

assert len(hass.loop.create_task.mock_calls) == 0

assert len(hass.add_job.mock_calls) == 0

def test_async_add_hass_job_schedule_partial_callback():

"""Test that we schedule partial coros and add jobs to the job pool."""

hass = MagicMock()

job = MagicMock()

partial = functools.partial(ha.callback(job))

ha.HomeAssistant.async_add_hass_job(hass, ha.HassJob(partial))

assert len(hass.loop.call_soon.mock_calls) == 1

assert len(hass.loop.create_task.mock_calls) == 0

assert len(hass.add_job.mock_calls) == 0

def test_async_add_hass_job_schedule_coroutinefunction(loop):

"""Test that we schedule coroutines and add jobs to the job pool."""

hass = MagicMock(loop=MagicMock(wraps=loop))

async def job():

pass

ha.HomeAssistant.async_add_hass_job(hass, ha.HassJob(job))

assert len(hass.loop.call_soon.mock_calls) == 0

assert len(hass.loop.create_task.mock_calls) == 1

assert len(hass.add_job.mock_calls) == 0

def test_async_add_hass_job_schedule_partial_coroutinefunction(loop):

"""Test that we schedule partial coros and add jobs to the job pool."""

hass = MagicMock(loop=MagicMock(wraps=loop))

async def job():

pass

partial = functools.partial(job)

ha.HomeAssistant.async_add_hass_job(hass, ha.HassJob(partial))

assert len(hass.loop.call_soon.mock_calls) == 0

assert len(hass.loop.create_task.mock_calls) == 1

assert len(hass.add_job.mock_calls) == 0

def test_async_add_job_add_hass_threaded_job_to_pool():

"""Test that we schedule coroutines and add jobs to the job pool."""

hass = MagicMock()

def job():

pass

ha.HomeAssistant.async_add_hass_job(hass, ha.HassJob(job))

assert len(hass.loop.call_soon.mock_calls) == 0

assert len(hass.loop.create_task.mock_calls) == 0

assert len(hass.loop.run_in_executor.mock_calls) == 1

def test_async_create_task_schedule_coroutine(loop):

"""Test that we schedule coroutines and add jobs to the job pool."""

hass = MagicMock(loop=MagicMock(wraps=loop))

async def job():

pass

ha.HomeAssistant.async_create_task(hass, job())

assert len(hass.loop.call_soon.mock_calls) == 0

assert len(hass.loop.create_task.mock_calls) == 1

assert len(hass.add_job.mock_calls) == 0

def test_async_run_hass_job_calls_callback():

"""Test that the callback annotation is respected."""

hass = MagicMock()

calls = []

def job():

calls.append(1)

ha.HomeAssistant.async_run_hass_job(hass, ha.HassJob(ha.callback(job)))

assert len(calls) == 1

assert len(hass.async_add_job.mock_calls) == 0

def test_async_run_hass_job_delegates_non_async():

"""Test that the callback annotation is respected."""

hass = MagicMock()

calls = []

def job():

calls.append(1)

ha.HomeAssistant.async_run_hass_job(hass, ha.HassJob(job))

assert len(calls) == 0

assert len(hass.async_add_hass_job.mock_calls) == 1

async def test_stage_shutdown(hass):

"""Simulate a shutdown, test calling stuff."""

test_stop = async_capture_events(hass, EVENT_HOMEASSISTANT_STOP)

test_final_write = async_capture_events(hass, EVENT_HOMEASSISTANT_FINAL_WRITE)

test_close = async_capture_events(hass, EVENT_HOMEASSISTANT_CLOSE)

test_all = async_capture_events(hass, MATCH_ALL)

await hass.async_stop()

assert len(test_stop) == 1

assert len(test_close) == 1

assert len(test_final_write) == 1

assert len(test_all) == 2

async def test_shutdown_calls_block_till_done_after_shutdown_run_callback_threadsafe(

hass,

"""Ensure shutdown_run_callback_threadsafe is called before the final async_block_till_done."""

stop_calls = []

async def _record_block_till_done():

nonlocal stop_calls

stop_calls.append("async_block_till_done")

def _record_shutdown_run_callback_threadsafe(loop):

nonlocal stop_calls

stop_calls.append(("shutdown_run_callback_threadsafe", loop))

with patch.object(hass, "async_block_till_done", _record_block_till_done), patch(

"homeassistant.core.shutdown_run_callback_threadsafe",

_record_shutdown_run_callback_threadsafe,

await hass.async_stop()

assert stop_calls[-2] == ("shutdown_run_callback_threadsafe", hass.loop)

assert stop_calls[-1] == "async_block_till_done"

async def test_pending_sheduler(hass):

"""Add a coro to pending tasks."""

call_count = []

async def test_coro():

"""Test Coro."""

call_count.append("call")

for _ in range(3):

hass.async_add_job(test_coro())

await asyncio.wait(hass._pending_tasks)

assert len(hass._pending_tasks) == 3

assert len(call_count) == 3

async def test_async_add_job_pending_tasks_coro(hass):

"""Add a coro to pending tasks."""

call_count = []

async def test_coro():

"""Test Coro."""

call_count.append("call")

for _ in range(2):

hass.add_job(test_coro())

async def wait_finish_callback():

"""Wait until all stuff is scheduled."""

await asyncio.sleep(0)

await wait_finish_callback()

assert len(hass._pending_tasks) == 2

await hass.async_block_till_done()

assert len(call_count) == 2

async def test_async_create_task_pending_tasks_coro(hass):

"""Add a coro to pending tasks."""

call_count = []

async def test_coro():

"""Test Coro."""

call_count.append("call")

for _ in range(2):

hass.create_task(test_coro())

async def wait_finish_callback():

"""Wait until all stuff is scheduled."""

await asyncio.sleep(0)

await wait_finish_callback()

assert len(hass._pending_tasks) == 2

await hass.async_block_till_done()

assert len(call_count) == 2

async def test_async_add_job_pending_tasks_executor(hass):

"""Run an executor in pending tasks."""

call_count = []

def test_executor():

"""Test executor."""

call_count.append("call")

async def wait_finish_callback():

"""Wait until all stuff is scheduled."""

await asyncio.sleep(0)

for _ in range(2):

hass.async_add_job(test_executor)

await wait_finish_callback()

assert len(hass._pending_tasks) == 2

await hass.async_block_till_done()

assert len(call_count) == 2

async def test_async_add_job_pending_tasks_callback(hass):

"""Run a callback in pending tasks."""

call_count = []

@ha.callback

def test_callback():

"""Test callback."""

call_count.append("call")

async def wait_finish_callback():

"""Wait until all stuff is scheduled."""

await asyncio.sleep(0)

for _ in range(2):

hass.async_add_job(test_callback)

await wait_finish_callback()

await hass.async_block_till_done()

assert len(hass._pending_tasks) == 0

assert len(call_count) == 2

async def test_add_job_with_none(hass):

"""Try to add a job with None as function."""

with pytest.raises(ValueError):

hass.async_add_job(None, "test_arg")

def test_event_eq():

"""Test events."""

now = dt_util.utcnow()

data = {"some": "attr"}

context = ha.Context()

event1, event2 = (

ha.Event("some_type", data, time_fired=now, context=context) for _ in range(2)

)

assert event1 == event2

def test_event_repr():

"""Test that Event repr method works."""

assert str(ha.Event("TestEvent")) == "<Event TestEvent[L]>"

assert (

str(ha.Event("TestEvent", {"beer": "nice"}, ha.EventOrigin.remote))

== "<Event TestEvent[R]: beer=nice>"

)

def test_event_as_dict():

"""Test an Event as dictionary."""

event_type = "some_type"

now = dt_util.utcnow()

data = {"some": "attr"}

event = ha.Event(event_type, data, ha.EventOrigin.local, now)

expected = {

"event_type": event_type,

"data": data,

"origin": "LOCAL",

"time_fired": now.isoformat(),

"context": {

"id": event.context.id,

"parent_id": None,

"user_id": event.context.user_id,

}

assert event.as_dict() == expected

# 2nd time to verify cache

assert event.as_dict() == expected

def test_state_as_dict():

"""Test a State as dictionary."""

last_time = datetime(1984, 12, 8, 12, 0, 0)

state = ha.State(

"happy.happy",

"on",

{"pig": "dog"},

last_updated=last_time,

last_changed=last_time,

)

expected = {

"context": {

"id": state.context.id,

"parent_id": None,

"user_id": state.context.user_id,

"entity_id": "happy.happy",

"attributes": {"pig": "dog"},

"last_changed": last_time.isoformat(),

"last_updated": last_time.isoformat(),

"state": "on",

}

as_dict_1 = state.as_dict()

assert isinstance(as_dict_1, ReadOnlyDict)

assert isinstance(as_dict_1["attributes"], ReadOnlyDict)

assert isinstance(as_dict_1["context"], ReadOnlyDict)

assert as_dict_1 == expected

# 2nd time to verify cache

assert state.as_dict() == expected

assert state.as_dict() is as_dict_1

async def test_eventbus_add_remove_listener(hass):

"""Test remove_listener method."""

old_count = len(hass.bus.async_listeners())

def listener(_):

pass

unsub = hass.bus.async_listen("test", listener)

assert old_count + 1 == len(hass.bus.async_listeners())

# Remove listener

unsub()

assert old_count == len(hass.bus.async_listeners())

# Should do nothing now

unsub()

async def test_eventbus_filtered_listener(hass):

"""Test we can prefilter events."""

calls = []

@ha.callback

def listener(event):

"""Mock listener."""

calls.append(event)

@ha.callback

def filter(event):

"""Mock filter."""

return not event.data["filtered"]

unsub = hass.bus.async_listen("test", listener, event_filter=filter)

hass.bus.async_fire("test", {"filtered": True})

await hass.async_block_till_done()

assert len(calls) == 0

hass.bus.async_fire("test", {"filtered": False})

await hass.async_block_till_done()

assert len(calls) == 1

unsub()

async def test_eventbus_run_immediately(hass):

"""Test we can call events immediately."""

calls = []

@ha.callback

def listener(event):

"""Mock listener."""

calls.append(event)

unsub = hass.bus.async_listen("test", listener, run_immediately=True)

hass.bus.async_fire("test", {"event": True})

# No async_block_till_done here

assert len(calls) == 1

unsub()

async def test_eventbus_unsubscribe_listener(hass):

"""Test unsubscribe listener from returned function."""

calls = []

@ha.callback

def listener(event):

"""Mock listener."""

calls.append(event)

unsub = hass.bus.async_listen("test", listener)

hass.bus.async_fire("test")

await hass.async_block_till_done()

assert len(calls) == 1

unsub()

hass.bus.async_fire("event")

await hass.async_block_till_done()

assert len(calls) == 1

async def test_eventbus_listen_once_event_with_callback(hass):

"""Test listen_once_event method."""

runs = []

@ha.callback

def event_handler(event):

runs.append(event)

hass.bus.async_listen_once("test_event", event_handler)

hass.bus.async_fire("test_event")

# Second time it should not increase runs

hass.bus.async_fire("test_event")

await hass.async_block_till_done()

assert len(runs) == 1

async def test_eventbus_listen_once_event_with_coroutine(hass):

"""Test listen_once_event method."""

runs = []

async def event_handler(event):

runs.append(event)

hass.bus.async_listen_once("test_event", event_handler)

hass.bus.async_fire("test_event")

# Second time it should not increase runs

hass.bus.async_fire("test_event")

await hass.async_block_till_done()

assert len(runs) == 1

async def test_eventbus_listen_once_event_with_thread(hass):

"""Test listen_once_event method."""

runs = []

def event_handler(event):

runs.append(event)

hass.bus.async_listen_once("test_event", event_handler)

hass.bus.async_fire("test_event")

# Second time it should not increase runs

hass.bus.async_fire("test_event")

await hass.async_block_till_done()

assert len(runs) == 1

async def test_eventbus_thread_event_listener(hass):

"""Test thread event listener."""

thread_calls = []

def thread_listener(event):

thread_calls.append(event)

hass.bus.async_listen("test_thread", thread_listener)

hass.bus.async_fire("test_thread")

await hass.async_block_till_done()

assert len(thread_calls) == 1

async def test_eventbus_callback_event_listener(hass):

"""Test callback event listener."""

callback_calls = []

@ha.callback

def callback_listener(event):

callback_calls.append(event)

hass.bus.async_listen("test_callback", callback_listener)

hass.bus.async_fire("test_callback")

await hass.async_block_till_done()

assert len(callback_calls) == 1

async def test_eventbus_coroutine_event_listener(hass):

"""Test coroutine event listener."""

coroutine_calls = []

async def coroutine_listener(event):

coroutine_calls.append(event)

hass.bus.async_listen("test_coroutine", coroutine_listener)

hass.bus.async_fire("test_coroutine")

await hass.async_block_till_done()

assert len(coroutine_calls) == 1

async def test_eventbus_max_length_exceeded(hass):

"""Test that an exception is raised when the max character length is exceeded."""

long_evt_name = (

"this_event_exceeds_the_max_character_length_even_with_the_new_limit"

)

with pytest.raises(MaxLengthExceeded) as exc_info:

hass.bus.async_fire(long_evt_name)

assert exc_info.value.property_name == "event_type"

assert exc_info.value.max_length == 64

assert exc_info.value.value == long_evt_name

def test_state_init():

"""Test state.init."""

with pytest.raises(InvalidEntityFormatError):

ha.State("invalid_entity_format", "test_state")

with pytest.raises(InvalidStateError):

ha.State("domain.long_state", "t" * 256)

def test_state_domain():

"""Test domain."""

state = ha.State("some_domain.hello", "world")

assert state.domain == "some_domain"

def test_state_object_id():

"""Test object ID."""

state = ha.State("domain.hello", "world")

assert state.object_id == "hello"

def test_state_name_if_no_friendly_name_attr():

"""Test if there is no friendly name."""

state = ha.State("domain.hello_world", "world")

assert state.name == "hello world"

def test_state_name_if_friendly_name_attr():

"""Test if there is a friendly name."""

name = "Some Unique Name"

state = ha.State("domain.hello_world", "world", {ATTR_FRIENDLY_NAME: name})

assert state.name == name

def test_state_dict_conversion():

"""Test conversion of dict."""

state = ha.State("domain.hello", "world", {"some": "attr"})

assert state == ha.State.from_dict(state.as_dict())

def test_state_dict_conversion_with_wrong_data():

"""Test conversion with wrong data."""

assert ha.State.from_dict(None) is None

assert ha.State.from_dict({"state": "yes"}) is None

assert ha.State.from_dict({"entity_id": "yes"}) is None

# Make sure invalid context data doesn't crash

wrong_context = ha.State.from_dict(

{

"entity_id": "light.kitchen",

"state": "on",

"context": {"id": "123", "non-existing": "crash"},

}

)

assert wrong_context is not None

assert wrong_context.context.id == "123"

def test_state_repr():

"""Test state.repr."""

assert (

str(ha.State("happy.happy", "on", last_changed=datetime(1984, 12, 8, 12, 0, 0)))

== "<state happy.happy=on @ 1984-12-08T12:00:00+00:00>"

)

assert (

str(

ha.State(

"happy.happy",

"on",

{"brightness": 144},

datetime(1984, 12, 8, 12, 0, 0),

)

== "<state happy.happy=on; brightness=144 @ "

"1984-12-08T12:00:00+00:00>"

)

async def test_statemachine_is_state(hass):

"""Test is_state method."""

hass.states.async_set("light.bowl", "on", {})

assert hass.states.is_state("light.Bowl", "on")

assert not hass.states.is_state("light.Bowl", "off")

assert not hass.states.is_state("light.Non_existing", "on")

async def test_statemachine_entity_ids(hass):

"""Test get_entity_ids method."""

hass.states.async_set("light.bowl", "on", {})

hass.states.async_set("SWITCH.AC", "off", {})

ent_ids = hass.states.async_entity_ids()

assert len(ent_ids) == 2

assert "light.bowl" in ent_ids

assert "switch.ac" in ent_ids

ent_ids = hass.states.async_entity_ids("light")

assert len(ent_ids) == 1

assert "light.bowl" in ent_ids

states = sorted(state.entity_id for state in hass.states.async_all())

assert states == ["light.bowl", "switch.ac"]

async def test_statemachine_remove(hass):

"""Test remove method."""

hass.states.async_set("light.bowl", "on", {})

events = async_capture_events(hass, EVENT_STATE_CHANGED)

assert "light.bowl" in hass.states.async_entity_ids()

assert hass.states.async_remove("light.bowl")

await hass.async_block_till_done()

assert "light.bowl" not in hass.states.async_entity_ids()

assert len(events) == 1

assert events[0].data.get("entity_id") == "light.bowl"

assert events[0].data.get("old_state") is not None

assert events[0].data["old_state"].entity_id == "light.bowl"

assert events[0].data.get("new_state") is None

# If it does not exist, we should get False

assert not hass.states.async_remove("light.Bowl")

await hass.async_block_till_done()

assert len(events) == 1

async def test_statemachine_case_insensitivty(hass):

"""Test insensitivty."""

events = async_capture_events(hass, EVENT_STATE_CHANGED)

hass.states.async_set("light.BOWL", "off")

await hass.async_block_till_done()

assert hass.states.is_state("light.bowl", "off")

assert len(events) == 1

async def test_statemachine_last_changed_not_updated_on_same_state(hass):

"""Test to not update the existing, same state."""

hass.states.async_set("light.bowl", "on", {})

state = hass.states.get("light.Bowl")

future = dt_util.utcnow() + timedelta(hours=10)

with patch("homeassistant.util.dt.utcnow", return_value=future):

hass.states.async_set("light.Bowl", "on", {"attr": "triggers_change"})

await hass.async_block_till_done()

state2 = hass.states.get("light.Bowl")

assert state2 is not None

assert state.last_changed == state2.last_changed

async def test_statemachine_force_update(hass):

"""Test force update option."""

hass.states.async_set("light.bowl", "on", {})

events = async_capture_events(hass, EVENT_STATE_CHANGED)

hass.states.async_set("light.bowl", "on")

await hass.async_block_till_done()

assert len(events) == 0

hass.states.async_set("light.bowl", "on", None, True)

await hass.async_block_till_done()

assert len(events) == 1

def test_service_call_repr():

"""Test ServiceCall repr."""

call = ha.ServiceCall("homeassistant", "start")

assert str(call) == f"<ServiceCall homeassistant.start (c:{call.context.id})>"

call2 = ha.ServiceCall("homeassistant", "start", {"fast": "yes"})

assert (

str(call2)

== f"<ServiceCall homeassistant.start (c:{call2.context.id}): fast=yes>"

)

async def test_serviceregistry_has_service(hass):

"""Test has_service method."""

hass.services.async_register("test_domain", "test_service", lambda call: None)

assert len(hass.services.async_services()) == 1

assert hass.services.has_service("tesT_domaiN", "tesT_servicE")

assert not hass.services.has_service("test_domain", "non_existing")

assert not hass.services.has_service("non_existing", "test_service")

async def test_serviceregistry_call_with_blocking_done_in_time(hass):

"""Test call with blocking."""

registered_events = async_capture_events(hass, EVENT_SERVICE_REGISTERED)

calls = async_mock_service(hass, "test_domain", "register_calls")

await hass.async_block_till_done()

assert len(registered_events) == 1

assert registered_events[0].data["domain"] == "test_domain"

assert registered_events[0].data["service"] == "register_calls"

assert await hass.services.async_call(

"test_domain", "REGISTER_CALLS", blocking=True

)

assert len(calls) == 1

async def test_serviceregistry_call_non_existing_with_blocking(hass):

"""Test non-existing with blocking."""

with pytest.raises(ha.ServiceNotFound):

await hass.services.async_call("test_domain", "i_do_not_exist", blocking=True)

async def test_serviceregistry_async_service(hass):

"""Test registering and calling an async service."""

calls = []

async def service_handler(call):

"""Service handler coroutine."""

calls.append(call)

hass.services.async_register("test_domain", "register_calls", service_handler)

assert await hass.services.async_call(

"test_domain", "REGISTER_CALLS", blocking=True

)

assert len(calls) == 1

async def test_serviceregistry_async_service_partial(hass):

"""Test registering and calling an wrapped async service."""

calls = []

async def service_handler(call):

"""Service handler coroutine."""

calls.append(call)

hass.services.async_register(

"test_domain", "register_calls", functools.partial(service_handler)

)

await hass.async_block_till_done()

assert await hass.services.async_call(

"test_domain", "REGISTER_CALLS", blocking=True

)

assert len(calls) == 1

async def test_serviceregistry_callback_service(hass):

"""Test registering and calling an async service."""

calls = []

@ha.callback

def service_handler(call):

"""Service handler coroutine."""

calls.append(call)

hass.services.async_register("test_domain", "register_calls", service_handler)

assert await hass.services.async_call(

"test_domain", "REGISTER_CALLS", blocking=True

)

assert len(calls) == 1

async def test_serviceregistry_remove_service(hass):

"""Test remove service."""

calls_remove = async_capture_events(hass, EVENT_SERVICE_REMOVED)

hass.services.async_register("test_domain", "test_service", lambda call: None)

assert hass.services.has_service("test_Domain", "test_Service")

hass.services.async_remove("test_Domain", "test_Service")

await hass.async_block_till_done()

assert not hass.services.has_service("test_Domain", "test_Service")

assert len(calls_remove) == 1

assert calls_remove[-1].data["domain"] == "test_domain"

assert calls_remove[-1].data["service"] == "test_service"

async def test_serviceregistry_service_that_not_exists(hass):

"""Test remove service that not exists."""

calls_remove = async_capture_events(hass, EVENT_SERVICE_REMOVED)

assert not hass.services.has_service("test_xxx", "test_yyy")

hass.services.async_remove("test_xxx", "test_yyy")

await hass.async_block_till_done()

assert len(calls_remove) == 0

with pytest.raises(ServiceNotFound):

await hass.services.async_call("test_do_not", "exist", {})

async def test_serviceregistry_async_service_raise_exception(hass):

"""Test registering and calling an async service raise exception."""

async def service_handler(_):

"""Service handler coroutine."""

raise ValueError

hass.services.async_register("test_domain", "register_calls", service_handler)

with pytest.raises(ValueError):

assert await hass.services.async_call(

"test_domain", "REGISTER_CALLS", blocking=True

)

# Non-blocking service call never throw exception

await hass.services.async_call("test_domain", "REGISTER_CALLS", blocking=False)

await hass.async_block_till_done()

async def test_serviceregistry_callback_service_raise_exception(hass):

"""Test registering and calling an callback service raise exception."""

@ha.callback

def service_handler(_):

"""Service handler coroutine."""

raise ValueError

hass.services.async_register("test_domain", "register_calls", service_handler)

with pytest.raises(ValueError):

assert await hass.services.async_call(

"test_domain", "REGISTER_CALLS", blocking=True

)

# Non-blocking service call never throw exception

await hass.services.async_call("test_domain", "REGISTER_CALLS", blocking=False)

await hass.async_block_till_done()

def test_config_defaults():

"""Test config defaults."""

hass = Mock()

config = ha.Config(hass)

assert config.hass is hass

assert config.latitude == 0

assert config.longitude == 0

assert config.elevation == 0

assert config.location_name == "Home"

assert config.time_zone == "UTC"

assert config.internal_url is None

assert config.external_url is None

assert config.config_source is ha.ConfigSource.DEFAULT

assert config.skip_pip is False

assert config.components == set()

assert config.api is None

assert config.config_dir is None

assert config.allowlist_external_dirs == set()

assert config.allowlist_external_urls == set()

assert config.media_dirs == {}

assert config.safe_mode is False

assert config.legacy_templates is False

assert config.currency == "EUR"

def test_config_path_with_file():

"""Test get_config_path method."""

config = ha.Config(None)

config.config_dir = "/test/ha-config"

assert config.path("test.conf") == "/test/ha-config/test.conf"

def test_config_path_with_dir_and_file():

"""Test get_config_path method."""

config = ha.Config(None)

config.config_dir = "/test/ha-config"

assert config.path("dir", "test.conf") == "/test/ha-config/dir/test.conf"

def test_config_as_dict():

"""Test as dict."""

config = ha.Config(None)

config.config_dir = "/test/ha-config"

config.hass = MagicMock()

type(config.hass.state).value = PropertyMock(return_value="RUNNING")

expected = {

"latitude": 0,

"longitude": 0,

"elevation": 0,

CONF_UNIT_SYSTEM: METRIC_SYSTEM.as_dict(),

"location_name": "Home",

"time_zone": "UTC",

"components": set(),

"config_dir": "/test/ha-config",

"whitelist_external_dirs": set(),

"allowlist_external_dirs": set(),

"allowlist_external_urls": set(),

"version": __version__,

"config_source": ha.ConfigSource.DEFAULT,

"safe_mode": False,

"state": "RUNNING",

"external_url": None,

"internal_url": None,

"currency": "EUR",

}

assert expected == config.as_dict()

def test_config_is_allowed_path():

"""Test is_allowed_path method."""

config = ha.Config(None)

with TemporaryDirectory() as tmp_dir:

# The created dir is in /tmp. This is a symlink on OS X

# causing this test to fail unless we resolve path first.

View remainder of file in raw view

FilesExpand file tree

test_core.py

Latest commit

History

test_core.py

File metadata and controls