Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_update_to_closed_from_signal(hass, device_factory):
"""Test the cover updates to closed when receiving a signal."""
# Arrange
device = device_factory(
"Garage", [Capability.garage_door_control], {Attribute.door: "closing"}
)
await setup_platform(hass, COVER_DOMAIN, devices=[device])
device.status.update_attribute_value(Attribute.door, "closed")
assert hass.states.get("cover.garage").state == STATE_CLOSING
# Act
async_dispatcher_send(hass, SIGNAL_SMARTTHINGS_UPDATE, [device.device_id])
# Assert
await hass.async_block_till_done()
state = hass.states.get("cover.garage")
assert state is not None
assert state.state == STATE_CLOSED
async def test_turn_on(hass, device_factory):
"""Test the switch turns of successfully."""
# Arrange
device = device_factory(
"Switch_1",
[Capability.switch, Capability.power_meter, Capability.energy_meter],
{Attribute.switch: "off", Attribute.power: 355, Attribute.energy: 11.422},
)
await setup_platform(hass, SWITCH_DOMAIN, devices=[device])
# Act
await hass.services.async_call(
"switch", "turn_on", {"entity_id": "switch.switch_1"}, blocking=True
)
# Assert
state = hass.states.get("switch.switch_1")
assert state is not None
assert state.state == "on"
assert state.attributes[ATTR_CURRENT_POWER_W] == 355
assert state.attributes[ATTR_TODAY_ENERGY_KWH] == 11.422
async def test_entity_and_device_attributes(hass, device_factory):
"""Test the attributes of the entity are correct."""
# Arrange
device = device_factory(
"Motion Sensor 1", [Capability.motion_sensor], {Attribute.motion: "inactive"}
)
entity_registry = await hass.helpers.entity_registry.async_get_registry()
device_registry = await hass.helpers.device_registry.async_get_registry()
# Act
await setup_platform(hass, BINARY_SENSOR_DOMAIN, devices=[device])
# Assert
entry = entity_registry.async_get("binary_sensor.motion_sensor_1_motion")
assert entry
assert entry.unique_id == device.device_id + "." + Attribute.motion
entry = device_registry.async_get_device({(DOMAIN, device.device_id)}, [])
assert entry
assert entry.name == device.label
assert entry.model == device.device_type_name
assert entry.manufacturer == "Unavailable"
def light_devices_fixture(device_factory):
"""Fixture returns a set of mock light devices."""
return [
device_factory(
"Dimmer 1",
capabilities=[Capability.switch, Capability.switch_level],
status={Attribute.switch: "on", Attribute.level: 100},
),
device_factory(
"Color Dimmer 1",
capabilities=[
Capability.switch,
Capability.switch_level,
Capability.color_control,
],
status={
Attribute.switch: "off",
Attribute.level: 0,
Attribute.hue: 76.0,
Attribute.saturation: 55.0,
},
),
device_factory(
async def test_unload_config_entry(hass, device_factory):
"""Test the light is removed when the config entry is unloaded."""
# Arrange
device = device_factory(
"Color Dimmer 2",
capabilities=[
Capability.switch,
Capability.switch_level,
Capability.color_control,
Capability.color_temperature,
],
status={
Attribute.switch: "off",
Attribute.level: 100,
Attribute.hue: 76.0,
Attribute.saturation: 55.0,
Attribute.color_temperature: 4500,
},
)
config_entry = await setup_platform(hass, LIGHT_DOMAIN, devices=[device])
# Act
await hass.config_entries.async_forward_entry_unload(config_entry, "light")
# Assert
assert not hass.states.get("light.color_dimmer_2")
def temperature_unit(self):
"""Return the unit of measurement."""
return UNIT_MAP.get(self._device.status.attributes[Attribute.temperature].unit)
Map(Attribute.illuminance, "Illuminance", "lux", DEVICE_CLASS_ILLUMINANCE)
],
Capability.infrared_level: [
Map(Attribute.infrared_level, "Infrared Level", "%", None)
],
Capability.media_input_source: [
Map(Attribute.input_source, "Media Input Source", None, None)
],
Capability.media_playback_repeat: [
Map(Attribute.playback_repeat_mode, "Media Playback Repeat", None, None)
],
Capability.media_playback_shuffle: [
Map(Attribute.playback_shuffle, "Media Playback Shuffle", None, None)
],
Capability.media_playback: [
Map(Attribute.playback_status, "Media Playback Status", None, None)
],
Capability.odor_sensor: [Map(Attribute.odor_level, "Odor Sensor", None, None)],
Capability.oven_mode: [Map(Attribute.oven_mode, "Oven Mode", None, None)],
Capability.oven_operating_state: [
Map(Attribute.machine_state, "Oven Machine State", None, None),
Map(Attribute.oven_job_state, "Oven Job State", None, None),
Map(Attribute.completion_time, "Oven Completion Time", None, None),
],
Capability.oven_setpoint: [
Map(Attribute.oven_setpoint, "Oven Set Point", None, None)
],
Capability.power_meter: [Map(Attribute.power, "Power Meter", POWER_WATT, None)],
Capability.power_source: [Map(Attribute.power_source, "Power Source", None, None)],
Capability.refrigeration_setpoint: [
Map(
Attribute.refrigeration_setpoint,
continue
device = self.devices.get(evt.device_id)
if not device:
continue
device.status.apply_attribute_update(
evt.component_id,
evt.capability,
evt.attribute,
evt.value,
data=evt.data,
)
# Fire events for buttons
if (
evt.capability == Capability.button
and evt.attribute == Attribute.button
):
data = {
"component_id": evt.component_id,
"device_id": evt.device_id,
"location_id": evt.location_id,
"value": evt.value,
"name": device.label,
"data": evt.data,
}
self._hass.bus.async_fire(EVENT_BUTTON, data)
_LOGGER.debug("Fired button event: %s", data)
else:
data = {
"location_id": evt.location_id,
"device_id": evt.device_id,
"component_id": evt.component_id,
def current_power_w(self):
"""Return the current power usage in W."""
return self._device.status.attributes[Attribute.power].value
DEVICE_CLASS_TIMESTAMP,
ENERGY_KILO_WATT_HOUR,
MASS_KILOGRAMS,
POWER_WATT,
TEMP_CELSIUS,
TEMP_FAHRENHEIT,
)
from . import SmartThingsEntity
from .const import DATA_BROKERS, DOMAIN
Map = namedtuple("map", "attribute name default_unit device_class")
CAPABILITY_TO_SENSORS = {
Capability.activity_lighting_mode: [
Map(Attribute.lighting_mode, "Activity Lighting Mode", None, None)
],
Capability.air_conditioner_mode: [
Map(Attribute.air_conditioner_mode, "Air Conditioner Mode", None, None)
],
Capability.air_quality_sensor: [
Map(Attribute.air_quality, "Air Quality", "CAQI", None)
],
Capability.alarm: [Map(Attribute.alarm, "Alarm", None, None)],
Capability.audio_volume: [Map(Attribute.volume, "Volume", "%", None)],
Capability.battery: [Map(Attribute.battery, "Battery", "%", DEVICE_CLASS_BATTERY)],
Capability.body_mass_index_measurement: [
Map(Attribute.bmi_measurement, "Body Mass Index", "kg/m^2", None)
],
Capability.body_weight_measurement: [
Map(Attribute.body_weight_measurement, "Body Weight", MASS_KILOGRAMS, None)
],