You've already forked homehub
536 lines
18 KiB
Python
536 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
|
from unittest import mock
|
|
import uuid
|
|
|
|
import aiojobs
|
|
import factory
|
|
import pytradfri
|
|
import pytradfri.api.aiocoap_api as pytradfri_api
|
|
|
|
from homehub_tradfri import services
|
|
|
|
|
|
class LightFactory(factory.StubFactory):
|
|
state = True
|
|
dimmer = 128
|
|
|
|
|
|
class LightControlFactory(factory.StubFactory):
|
|
lights = factory.LazyFunction(lambda: [LightFactory()])
|
|
set_state = factory.LazyFunction(
|
|
lambda: mock.Mock(return_value='fake_set_light_state_command'),
|
|
)
|
|
|
|
|
|
class DeviceFactory(factory.StubFactory):
|
|
id = factory.LazyFunction(lambda: str(uuid.uuid4()))
|
|
name = factory.Faker('domain_word')
|
|
light_control = factory.SubFactory(LightControlFactory)
|
|
reachable = True
|
|
has_light_control = True
|
|
|
|
|
|
class GroupFactory(factory.StubFactory):
|
|
id = factory.LazyFunction(lambda: str(uuid.uuid4()))
|
|
name = factory.Faker('domain_word')
|
|
member_ids = factory.LazyFunction(lambda: ['fake_light1', 'fake_light2'])
|
|
|
|
|
|
class Test_TradfriService:
|
|
def _valid_characteristics(self):
|
|
return {
|
|
'host': 'thisisntright.local',
|
|
'key': 'thisisntright',
|
|
}
|
|
|
|
def _fake_lights(self):
|
|
return [
|
|
{
|
|
'id': 'light1',
|
|
'name': 'Light 1',
|
|
'state': True,
|
|
'reachable': False,
|
|
'dimmer': 128,
|
|
},
|
|
{
|
|
'id': 'light2',
|
|
'name': 'Light 2',
|
|
'state': True,
|
|
'reachable': False,
|
|
'dimmer': 128,
|
|
},
|
|
{
|
|
'id': 'light3',
|
|
'name': 'Light 3',
|
|
'state': True,
|
|
'reachable': False,
|
|
'dimmer': 128,
|
|
},
|
|
]
|
|
|
|
def _fake_groups(self):
|
|
return [
|
|
{
|
|
'id': 'group1',
|
|
'name': 'Group 1',
|
|
'member_ids': ['light1'],
|
|
},
|
|
]
|
|
|
|
def test_init(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
|
|
# Then
|
|
assert service.api_factory is None
|
|
assert service.gateway is None
|
|
assert service.job is None
|
|
assert service.started is False
|
|
|
|
def test_state_key(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
|
|
# When
|
|
result = service.state_key()
|
|
|
|
# Then
|
|
assert result.endswith('.thisisntright.local.thisisntright')
|
|
|
|
async def test_current_data(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
|
|
with mock.patch.object(service, 'get_groups') as mock_get_groups:
|
|
with mock.patch.object(service, 'get_lights') as mock_get_lights:
|
|
fake_groups = self._fake_groups()
|
|
mock_get_groups.return_value = fake_groups
|
|
|
|
fake_lights = self._fake_lights()
|
|
mock_get_lights.return_value = fake_lights
|
|
|
|
# When
|
|
result = await service.current_data()
|
|
|
|
# Then
|
|
assert result.data == {
|
|
'groups': fake_groups,
|
|
'lights': fake_lights,
|
|
}
|
|
assert result.exception is None
|
|
|
|
async def test_current_data_get_groups_exception(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
|
|
with mock.patch.object(service, 'get_groups') as mock_get_groups:
|
|
with mock.patch.object(service, 'get_lights') as mock_get_lights:
|
|
fake_get_groups_error = RuntimeError('FIAL')
|
|
mock_get_groups.side_effect = fake_get_groups_error
|
|
|
|
fake_lights = self._fake_lights()
|
|
mock_get_lights.return_value = fake_lights
|
|
|
|
# When
|
|
result = await service.current_data()
|
|
|
|
# Then
|
|
assert result.data is None
|
|
assert result.exception == fake_get_groups_error
|
|
|
|
async def test_current_data_get_lights_exception(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
|
|
with mock.patch.object(service, 'get_groups') as mock_get_groups:
|
|
with mock.patch.object(service, 'get_lights') as mock_get_lights:
|
|
fake_groups = self._fake_groups()
|
|
mock_get_groups.return_value = fake_groups
|
|
|
|
fake_get_lights_error = RuntimeError('FIAL')
|
|
mock_get_lights.side_effect = fake_get_lights_error
|
|
|
|
# When
|
|
result = await service.current_data()
|
|
|
|
# Then
|
|
assert result.data is None
|
|
assert result.exception == fake_get_lights_error
|
|
|
|
async def test_call_api(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
|
|
service.api_factory = mock.Mock(
|
|
spec=pytradfri_api.APIFactory,
|
|
)
|
|
service.api_factory.request = mock.AsyncMock(
|
|
return_value='fake_api_result',
|
|
)
|
|
|
|
fake_commands = ['fake', 'commands']
|
|
|
|
# When
|
|
result = await service.call_api(fake_commands)
|
|
|
|
# Then
|
|
assert result == 'fake_api_result'
|
|
service.api_factory.request.assert_called_with(fake_commands)
|
|
|
|
async def test_get_lights(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
service.gateway = mock.Mock(spec=pytradfri.Gateway)
|
|
|
|
fake_get_devices_command = 'fake_get_devices_command'
|
|
service.gateway.get_devices.return_value = fake_get_devices_command
|
|
|
|
with mock.patch.object(service, 'call_api') as mock_call_api:
|
|
fake_devices = [
|
|
DeviceFactory(has_light_control=True),
|
|
DeviceFactory(has_light_control=False),
|
|
]
|
|
mock_call_api.return_value = fake_devices
|
|
|
|
# When
|
|
result = await service.get_lights()
|
|
|
|
# Then
|
|
assert len(result) == 1
|
|
|
|
light_ids = [light['id'] for light in result]
|
|
assert light_ids == [fake_devices[0].id]
|
|
|
|
light = result[0]
|
|
assert light['id'] == fake_devices[0].id
|
|
assert light['name'] == fake_devices[0].name
|
|
assert light['state'] == fake_devices[0].light_control.lights[0].state
|
|
assert light['reachable'] == fake_devices[0].reachable
|
|
assert light['dimmer'] == fake_devices[0].light_control.lights[0].dimmer
|
|
|
|
async def test_get_light(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
service.gateway = mock.Mock(spec=pytradfri.Gateway)
|
|
|
|
fake_get_device_command = 'fake_get_device_command'
|
|
service.gateway.get_device.return_value = fake_get_device_command
|
|
|
|
with mock.patch.object(service, 'call_api') as mock_call_api:
|
|
fake_device = DeviceFactory()
|
|
mock_call_api.return_value = fake_device
|
|
|
|
# When
|
|
result = await service.get_light('some_light')
|
|
|
|
# Then
|
|
assert result == fake_device
|
|
|
|
async def test_set_light_state(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
|
|
with mock.patch.object(service, 'get_light') as mock_get_light:
|
|
with mock.patch.object(service, 'call_api') as mock_call_api:
|
|
fake_light = DeviceFactory()
|
|
mock_get_light.return_value = fake_light
|
|
|
|
mock_call_api.return_value = True
|
|
|
|
# When
|
|
await service.set_light_state('some_light', True)
|
|
|
|
# Then
|
|
mock_get_light.assert_called_with('some_light')
|
|
fake_light.light_control.set_state.assert_called_with(
|
|
True, index=0,
|
|
)
|
|
mock_call_api.assert_called_with(
|
|
'fake_set_light_state_command',
|
|
)
|
|
|
|
async def test_get_groups(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
service.gateway = mock.Mock(spec=pytradfri.Gateway)
|
|
|
|
fake_get_groups_command = 'fake_get_groups_command'
|
|
service.gateway.get_groups.return_value = fake_get_groups_command
|
|
|
|
with mock.patch.object(service, 'call_api') as mock_call_api:
|
|
fake_groups = [GroupFactory(), GroupFactory()]
|
|
mock_call_api.return_value = fake_groups
|
|
|
|
# When
|
|
result = await service.get_groups()
|
|
|
|
# Then
|
|
assert len(result) == 2
|
|
|
|
group = result[0]
|
|
assert group['id'] == fake_groups[0].id
|
|
assert group['name'] == fake_groups[0].name
|
|
assert group['member_ids'] == fake_groups[0].member_ids
|
|
|
|
async def test_set_lights_state(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
|
|
with mock.patch.object(service, 'set_light_state') as mock_set_light_state:
|
|
with mock.patch.object(service, 'get_lights') as mock_get_lights:
|
|
fake_lights = self._fake_lights()
|
|
mock_get_lights.return_value = fake_lights
|
|
|
|
# When
|
|
result = await service.set_lights_state(
|
|
['light1', 'light2'], True,
|
|
)
|
|
|
|
# Then
|
|
assert result == [fake_lights[0], fake_lights[1]]
|
|
assert mock_set_light_state.call_count == 2
|
|
mock_set_light_state.assert_any_call('light1', True)
|
|
mock_set_light_state.assert_any_call('light2', True)
|
|
assert mock_get_lights.called is True
|
|
|
|
async def test_worker(self, homehub_app):
|
|
# Given
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', self._valid_characteristics(),
|
|
)
|
|
service.job = mock.Mock(spec=aiojobs._job.Job)
|
|
service.job.closed = True
|
|
service.update_weather = mock.AsyncMock()
|
|
service.notify = mock.AsyncMock()
|
|
|
|
with mock.patch('asyncio.sleep') as mock_asyncio_sleep:
|
|
# When
|
|
await service.worker()
|
|
|
|
# Then
|
|
mock_asyncio_sleep.assert_called_with(60)
|
|
assert service.notify.called is True
|
|
|
|
@mock.patch('aiojobs.aiohttp.get_scheduler_from_app')
|
|
async def test_start(self, mock_get_scheduler, homehub_app):
|
|
# Given
|
|
fake_scheduler = mock.Mock(spec=aiojobs._scheduler.Scheduler)
|
|
mock_get_scheduler.return_value = fake_scheduler
|
|
|
|
fake_job = mock.Mock(spec=aiojobs._job.Job)
|
|
fake_scheduler.spawn.return_value = fake_job
|
|
|
|
characteristics = self._valid_characteristics()
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', characteristics,
|
|
)
|
|
|
|
fake_gateway = mock.Mock(spec=pytradfri.Gateway)
|
|
|
|
fake_api_factory = mock.Mock(spec=pytradfri_api.APIFactory)
|
|
fake_api_factory.generate_psk.return_value = 'fake_psk'
|
|
|
|
fake_worker = mock.AsyncMock()
|
|
service.worker = mock.Mock()
|
|
service.worker.return_value = fake_worker
|
|
|
|
with mock.patch('pytradfri.Gateway') as mock_gateway:
|
|
with mock.patch.object(pytradfri_api.APIFactory, 'init') as mock_api_factory_init:
|
|
with mock.patch.object(service, 'set_state'):
|
|
mock_gateway.return_value = fake_gateway
|
|
mock_api_factory_init.return_value = fake_api_factory
|
|
|
|
# When
|
|
await service.start()
|
|
|
|
# Then
|
|
assert service.api_factory == fake_api_factory
|
|
pytradfri_api.APIFactory.init.assert_called_with(
|
|
host=characteristics['host'],
|
|
psk_id=mock.ANY,
|
|
psk=None,
|
|
)
|
|
psk_id = pytradfri_api.APIFactory.init.call_args.kwargs['psk_id']
|
|
assert isinstance(psk_id, str)
|
|
|
|
fake_api_factory.generate_psk.assert_called_with(
|
|
characteristics['key'],
|
|
)
|
|
|
|
assert service.gateway == fake_gateway
|
|
assert mock_gateway.called
|
|
|
|
assert service.job == fake_job
|
|
mock_get_scheduler.assert_called_with(homehub_app)
|
|
fake_scheduler.spawn.assert_called_with(fake_worker)
|
|
|
|
service.set_state.assert_called_with({
|
|
'psk_id': psk_id,
|
|
'psk': 'fake_psk',
|
|
})
|
|
|
|
assert service.started is True
|
|
|
|
@mock.patch('aiojobs.aiohttp.get_scheduler_from_app')
|
|
async def test_start_with_state(self, mock_get_scheduler, homehub_app):
|
|
# Given
|
|
fake_scheduler = mock.Mock(spec=aiojobs._scheduler.Scheduler)
|
|
mock_get_scheduler.return_value = fake_scheduler
|
|
|
|
fake_job = mock.Mock(spec=aiojobs._job.Job)
|
|
fake_scheduler.spawn.return_value = fake_job
|
|
|
|
characteristics = self._valid_characteristics()
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', characteristics,
|
|
)
|
|
service.state = {
|
|
'psk_id': 'state_psk_id',
|
|
'psk': 'state_psk',
|
|
}
|
|
|
|
fake_gateway = mock.Mock(spec=pytradfri.Gateway)
|
|
fake_api_factory = mock.Mock(spec=pytradfri_api.APIFactory)
|
|
|
|
fake_worker = mock.AsyncMock()
|
|
service.worker = mock.Mock()
|
|
service.worker.return_value = fake_worker
|
|
|
|
with mock.patch('pytradfri.Gateway') as mock_gateway:
|
|
with mock.patch.object(pytradfri_api.APIFactory, 'init') as mock_api_factory_init:
|
|
with mock.patch.object(service, 'set_state'):
|
|
mock_gateway.return_value = fake_gateway
|
|
mock_api_factory_init.return_value = fake_api_factory
|
|
|
|
# When
|
|
await service.start()
|
|
|
|
# Then
|
|
assert service.api_factory == fake_api_factory
|
|
mock_api_factory_init.assert_called_with(
|
|
host=characteristics['host'],
|
|
psk_id='state_psk_id',
|
|
psk='state_psk',
|
|
)
|
|
|
|
assert not fake_api_factory.generate_psk.called
|
|
|
|
service.set_state.assert_called_with({
|
|
'psk_id': 'state_psk_id',
|
|
'psk': 'state_psk',
|
|
})
|
|
|
|
@mock.patch('aiojobs.aiohttp.get_scheduler_from_app')
|
|
async def test_start_already_started(self, mock_get_scheduler, homehub_app):
|
|
# Given
|
|
fake_scheduler = mock.Mock(spec=aiojobs._scheduler.Scheduler)
|
|
mock_get_scheduler.return_value = fake_scheduler
|
|
|
|
characteristics = self._valid_characteristics()
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', characteristics,
|
|
)
|
|
service.started = True
|
|
|
|
fake_job = mock.Mock(spec=aiojobs._job.Job)
|
|
service.job = fake_job
|
|
|
|
fake_gateway = mock.Mock(spec=pytradfri.Gateway)
|
|
service.gateway = fake_gateway
|
|
|
|
fake_api_factory = mock.Mock(spec=pytradfri_api.APIFactory)
|
|
service.api_factory = fake_api_factory
|
|
|
|
with mock.patch('pytradfri.Gateway') as mock_gateway:
|
|
with mock.patch.object(pytradfri_api.APIFactory, 'init') as mock_api_factory_init:
|
|
with mock.patch.object(service, 'set_state'):
|
|
# When
|
|
await service.start()
|
|
|
|
# Then
|
|
assert service.api_factory == fake_api_factory
|
|
assert not mock_api_factory_init.called
|
|
|
|
assert not fake_api_factory.generate_psk.called
|
|
|
|
assert service.gateway == fake_gateway
|
|
assert not mock_gateway.called
|
|
|
|
assert service.job == fake_job
|
|
assert not mock_get_scheduler.called
|
|
assert not fake_scheduler.spawn.called
|
|
|
|
assert not service.set_state.called
|
|
|
|
assert service.started is True
|
|
|
|
async def test_stop(self, homehub_app):
|
|
# Given
|
|
characteristics = self._valid_characteristics()
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', characteristics,
|
|
)
|
|
|
|
fake_job = mock.Mock(spec=aiojobs._job.Job)
|
|
fake_job.closed = False
|
|
service.job = fake_job
|
|
|
|
fake_gateway = mock.Mock(spec=pytradfri.Gateway)
|
|
service.gateway = fake_gateway
|
|
|
|
fake_api_factory = mock.Mock(spec=pytradfri_api.APIFactory)
|
|
service.api_factory = fake_api_factory
|
|
|
|
# When
|
|
await service.stop()
|
|
|
|
# Then
|
|
assert service.api_factory is None
|
|
assert service.gateway is None
|
|
assert service.job.close.called
|
|
|
|
async def test_stop_already_stopped(self, homehub_app):
|
|
# Given
|
|
characteristics = self._valid_characteristics()
|
|
service = services.TradfriService(
|
|
homehub_app, 'testing', characteristics,
|
|
)
|
|
|
|
fake_job = mock.Mock(spec=aiojobs._job.Job)
|
|
fake_job.closed = True
|
|
service.job = fake_job
|
|
|
|
fake_gateway = mock.Mock(spec=pytradfri.Gateway)
|
|
service.gateway = fake_gateway
|
|
|
|
fake_api_factory = mock.Mock(spec=pytradfri_api.APIFactory)
|
|
service.api_factory = fake_api_factory
|
|
|
|
# When
|
|
await service.stop()
|
|
|
|
# Then
|
|
assert not service.job.close.called
|