Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_set_lock_state(self):
mock_controller = mock.create_autospec(pyvera.VeraController)
status_json = json.loads(
'{'
' "id": 33,'
' "deviceInfo": {'
# pyvera.CATEGORY_LOCK
' "category": 7,'
' "categoryName": "Doorlock",'
' "name": "MyTestDeadbolt",'
' "locked": 0'
' }'
'}'
)
lock = pyvera.VeraLock(status_json, [], mock_controller)
lock.set_lock_state(1)
self.assertTrue(lock.get_value('locked'), '1')
def test_controller_custom_subscription_registry() -> None:
"""Test function."""
class CustomSubscriptionRegistry(pyvera.AbstractSubscriptionRegistry):
"""Test registry."""
def start(self) -> None:
"""Start the polling."""
def stop(self) -> None:
"""Stop the polling."""
controller = VeraController("URL", CustomSubscriptionRegistry())
assert controller.subscription_registry.get_controller() == controller
CATEGORY_POWER_METER,
CATEGORY_SCENE_CONTROLLER,
CATEGORY_SENSOR,
CATEGORY_SWITCH,
CATEGORY_TEMPERATURE_SENSOR,
CATEGORY_THERMOSTAT,
CATEGORY_UV_SENSOR,
VeraController,
)
VeraApiData = NamedTuple(
"VeraApiData", [("sdata", dict), ("status", dict), ("lu_sdata", dict)],
)
VeraControllerData = NamedTuple(
"ControllerData", [("api_data", VeraApiData), ("controller", VeraController)]
)
def new_vera_api_data():
return VeraApiData(
sdata=deepcopy(RESPONSE_SDATA),
status=deepcopy(RESPONSE_STATUS),
lu_sdata=deepcopy(RESPONSE_LU_SDATA_EMPTY),
)
def find_device_object(device_id: int, data_list: list) -> Optional[dict]:
"""Find a vera device object in a list of devices."""
for device in data_list or []:
if device.get("id") == device_id:
return device
"""Common code for tests."""
from typing import Callable, NamedTuple, Tuple
from mock import MagicMock
from pyvera import VeraController, VeraDevice, VeraScene
from homeassistant.components.vera import CONF_CONTROLLER, DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
ComponentData = NamedTuple("ComponentData", (("controller", VeraController),))
class ComponentFactory:
"""Factory class."""
def __init__(self, init_controller_mock):
"""Initialize component factory."""
self.init_controller_mock = init_controller_mock
async def configure_component(
self,
hass: HomeAssistant,
devices: Tuple[VeraDevice] = (),
scenes: Tuple[VeraScene] = (),
setup_callback: Callable[[VeraController], None] = None,
) -> ComponentData:
if state.get("service") != service_id
or state.get("variable") != state_variable_name
]
status["states"].append(
{
"service": service_id,
"variable": state_variable_name,
"value": state_variable_value,
}
)
return ResponseStub(json={})
return ResponseStub(json={})
controller = VeraController("http://127.0.0.1:123")
controller.data_request = data_request
return VeraControllerData(api_data=api_data, controller=controller)
def test__event_device_for_vera_lock_status() -> None:
"""Test function."""
registry = SubscriptionRegistry()
registry.set_controller(MagicMock(spec=VeraController))
mock_lock = MagicMock(spec=VeraLock)
mock_lock.name = MagicMock(return_value="MyTestDeadbolt")
# Deadbolt changing but not done
device_json: dict = {"state": STATE_JOB_IN_PROGRESS}
registry._event_device(mock_lock, device_json, [])
mock_lock.update.assert_not_called()
# Deadbolt progress with reset state but not done
device_json = {
"state": STATE_NO_JOB,
"comment": "MyTestDeadbolt: Sending the Z-Wave command after 0 retries",
}
registry._event_device(mock_lock, device_json, [])
mock_lock.update.assert_not_called()
async def configure_component(
self,
hass: HomeAssistant,
devices: Tuple[VeraDevice] = (),
scenes: Tuple[VeraScene] = (),
setup_callback: Callable[[VeraController], None] = None,
) -> ComponentData:
"""Configure the component with specific mock data."""
controller_url = "http://127.0.0.1:123"
hass_config = {
DOMAIN: {CONF_CONTROLLER: controller_url},
}
controller = MagicMock(spec=VeraController) # type: VeraController
controller.base_url = controller_url
controller.register = MagicMock()
controller.get_devices = MagicMock(return_value=devices or ())
controller.get_scenes = MagicMock(return_value=scenes or ())
for vera_obj in controller.get_devices() + controller.get_scenes():
vera_obj.vera_controller = controller
controller.get_devices.reset_mock()
controller.get_scenes.reset_mock()
if setup_callback:
setup_callback(controller, hass_config)
def init_controller(base_url: str) -> list:
nonlocal controller
def test_set_lock_state(self):
mock_controller = mock.create_autospec(pyvera.VeraController)
status_json = json.loads(
'{'
' "id": 33,'
' "deviceInfo": {'
# pyvera.CATEGORY_LOCK
' "category": 7,'
' "categoryName": "Doorlock",'
' "name": "MyTestDeadbolt",'
' "locked": 0'
' }'
'}'
)
lock = pyvera.VeraLock(status_json, [], mock_controller)
lock.set_lock_state(1)
self.assertTrue(lock.get_value('locked'), '1')
def test_lock(vera_controller_data: VeraControllerData) -> None:
"""Test function."""
controller = vera_controller_data.controller
device = cast(VeraLock, controller.get_device_by_id(DEVICE_LOCK_ID))
controller.register(device, lambda device: None)
assert device.is_locked() is False
device.lock()
assert device.is_locked() is True
device.unlock()
assert device.is_locked() is False
def test__event_device_for_vera_lock_status():
"""Test function."""
registry = pyvera.SubscriptionRegistry()
mock_lock = mock.create_autospec(pyvera.VeraLock)
mock_lock.name = mock.MagicMock(return_value="MyTestDeadbolt")
# Deadbolt changing but not done
device_json = json.loads(
"{"
# subscribe.STATE_JOB_IN_PROGRESS
' "state": "1"'
"}"
)
registry._event_device(mock_lock, device_json, [])
mock_lock.update.assert_not_called()
# Deadbolt progress with reset state but not done
device_json = json.loads(
"{"
# subscribe.STATE_NO_JOB