mirror of
https://github.com/home-assistant/core.git
synced 2025-04-24 09:17:53 +00:00
Fix source code using Windows newline (#12248)
* 🚜 Fix usage of carriage return * 🤝 Rebase and repeat * 🚜 Fix file permissions
This commit is contained in:
parent
e4874fd7c7
commit
2ae0c5653e
188
homeassistant/components/binary_sensor/mercedesme.py
Executable file → Normal file
188
homeassistant/components/binary_sensor/mercedesme.py
Executable file → Normal file
@ -1,94 +1,94 @@
|
||||
"""
|
||||
Support for Mercedes cars with Mercedes ME.
|
||||
|
||||
For more details about this component, please refer to the documentation at
|
||||
https://home-assistant.io/components/binary_sensor.mercedesme/
|
||||
"""
|
||||
import logging
|
||||
import datetime
|
||||
|
||||
from homeassistant.components.binary_sensor import (BinarySensorDevice)
|
||||
from homeassistant.components.mercedesme import (
|
||||
DATA_MME, MercedesMeEntity, BINARY_SENSORS)
|
||||
|
||||
DEPENDENCIES = ['mercedesme']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Setup the sensor platform."""
|
||||
data = hass.data[DATA_MME].data
|
||||
|
||||
if not data.cars:
|
||||
_LOGGER.error("No cars found. Check component log.")
|
||||
return
|
||||
|
||||
devices = []
|
||||
for car in data.cars:
|
||||
for key, value in sorted(BINARY_SENSORS.items()):
|
||||
devices.append(MercedesMEBinarySensor(
|
||||
data, key, value[0], car["vin"], None))
|
||||
|
||||
add_devices(devices, True)
|
||||
|
||||
|
||||
class MercedesMEBinarySensor(MercedesMeEntity, BinarySensorDevice):
|
||||
"""Representation of a Sensor."""
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return the state of the binary sensor."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
if self._internal_name == "windowsClosed":
|
||||
return {
|
||||
"window_front_left": self._car["windowStatusFrontLeft"],
|
||||
"window_front_right": self._car["windowStatusFrontRight"],
|
||||
"window_rear_left": self._car["windowStatusRearLeft"],
|
||||
"window_rear_right": self._car["windowStatusRearRight"],
|
||||
"original_value": self._car[self._internal_name],
|
||||
"last_update": datetime.datetime.fromtimestamp(
|
||||
self._car["lastUpdate"]).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"car": self._car["license"]
|
||||
}
|
||||
elif self._internal_name == "tireWarningLight":
|
||||
return {
|
||||
"front_right_tire_pressure_kpa":
|
||||
self._car["frontRightTirePressureKpa"],
|
||||
"front_left_tire_pressure_kpa":
|
||||
self._car["frontLeftTirePressureKpa"],
|
||||
"rear_right_tire_pressure_kpa":
|
||||
self._car["rearRightTirePressureKpa"],
|
||||
"rear_left_tire_pressure_kpa":
|
||||
self._car["rearLeftTirePressureKpa"],
|
||||
"original_value": self._car[self._internal_name],
|
||||
"last_update": datetime.datetime.fromtimestamp(
|
||||
self._car["lastUpdate"]
|
||||
).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"car": self._car["license"],
|
||||
}
|
||||
return {
|
||||
"original_value": self._car[self._internal_name],
|
||||
"last_update": datetime.datetime.fromtimestamp(
|
||||
self._car["lastUpdate"]).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"car": self._car["license"]
|
||||
}
|
||||
|
||||
def update(self):
|
||||
"""Fetch new state data for the sensor."""
|
||||
self._car = next(
|
||||
car for car in self._data.cars if car["vin"] == self._vin)
|
||||
|
||||
if self._internal_name == "windowsClosed":
|
||||
self._state = bool(self._car[self._internal_name] == "CLOSED")
|
||||
elif self._internal_name == "tireWarningLight":
|
||||
self._state = bool(self._car[self._internal_name] != "INACTIVE")
|
||||
else:
|
||||
self._state = self._car[self._internal_name] is True
|
||||
|
||||
_LOGGER.debug("Updated %s Value: %s IsOn: %s",
|
||||
self._internal_name, self._state, self.is_on)
|
||||
"""
|
||||
Support for Mercedes cars with Mercedes ME.
|
||||
|
||||
For more details about this component, please refer to the documentation at
|
||||
https://home-assistant.io/components/binary_sensor.mercedesme/
|
||||
"""
|
||||
import logging
|
||||
import datetime
|
||||
|
||||
from homeassistant.components.binary_sensor import (BinarySensorDevice)
|
||||
from homeassistant.components.mercedesme import (
|
||||
DATA_MME, MercedesMeEntity, BINARY_SENSORS)
|
||||
|
||||
DEPENDENCIES = ['mercedesme']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Setup the sensor platform."""
|
||||
data = hass.data[DATA_MME].data
|
||||
|
||||
if not data.cars:
|
||||
_LOGGER.error("No cars found. Check component log.")
|
||||
return
|
||||
|
||||
devices = []
|
||||
for car in data.cars:
|
||||
for key, value in sorted(BINARY_SENSORS.items()):
|
||||
devices.append(MercedesMEBinarySensor(
|
||||
data, key, value[0], car["vin"], None))
|
||||
|
||||
add_devices(devices, True)
|
||||
|
||||
|
||||
class MercedesMEBinarySensor(MercedesMeEntity, BinarySensorDevice):
|
||||
"""Representation of a Sensor."""
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
"""Return the state of the binary sensor."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
if self._internal_name == "windowsClosed":
|
||||
return {
|
||||
"window_front_left": self._car["windowStatusFrontLeft"],
|
||||
"window_front_right": self._car["windowStatusFrontRight"],
|
||||
"window_rear_left": self._car["windowStatusRearLeft"],
|
||||
"window_rear_right": self._car["windowStatusRearRight"],
|
||||
"original_value": self._car[self._internal_name],
|
||||
"last_update": datetime.datetime.fromtimestamp(
|
||||
self._car["lastUpdate"]).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"car": self._car["license"]
|
||||
}
|
||||
elif self._internal_name == "tireWarningLight":
|
||||
return {
|
||||
"front_right_tire_pressure_kpa":
|
||||
self._car["frontRightTirePressureKpa"],
|
||||
"front_left_tire_pressure_kpa":
|
||||
self._car["frontLeftTirePressureKpa"],
|
||||
"rear_right_tire_pressure_kpa":
|
||||
self._car["rearRightTirePressureKpa"],
|
||||
"rear_left_tire_pressure_kpa":
|
||||
self._car["rearLeftTirePressureKpa"],
|
||||
"original_value": self._car[self._internal_name],
|
||||
"last_update": datetime.datetime.fromtimestamp(
|
||||
self._car["lastUpdate"]
|
||||
).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"car": self._car["license"],
|
||||
}
|
||||
return {
|
||||
"original_value": self._car[self._internal_name],
|
||||
"last_update": datetime.datetime.fromtimestamp(
|
||||
self._car["lastUpdate"]).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"car": self._car["license"]
|
||||
}
|
||||
|
||||
def update(self):
|
||||
"""Fetch new state data for the sensor."""
|
||||
self._car = next(
|
||||
car for car in self._data.cars if car["vin"] == self._vin)
|
||||
|
||||
if self._internal_name == "windowsClosed":
|
||||
self._state = bool(self._car[self._internal_name] == "CLOSED")
|
||||
elif self._internal_name == "tireWarningLight":
|
||||
self._state = bool(self._car[self._internal_name] != "INACTIVE")
|
||||
else:
|
||||
self._state = self._car[self._internal_name] is True
|
||||
|
||||
_LOGGER.debug("Updated %s Value: %s IsOn: %s",
|
||||
self._internal_name, self._state, self.is_on)
|
||||
|
0
homeassistant/components/camera/xeoma.py
Executable file → Normal file
0
homeassistant/components/camera/xeoma.py
Executable file → Normal file
142
homeassistant/components/device_tracker/mercedesme.py
Executable file → Normal file
142
homeassistant/components/device_tracker/mercedesme.py
Executable file → Normal file
@ -1,71 +1,71 @@
|
||||
"""
|
||||
Support for Mercedes cars with Mercedes ME.
|
||||
|
||||
For more details about this component, please refer to the documentation at
|
||||
https://home-assistant.io/components/device_tracker.mercedesme/
|
||||
"""
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
from homeassistant.components.mercedesme import DATA_MME
|
||||
from homeassistant.helpers.event import track_time_interval
|
||||
from homeassistant.util import Throttle
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEPENDENCIES = ['mercedesme']
|
||||
|
||||
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=30)
|
||||
|
||||
|
||||
def setup_scanner(hass, config, see, discovery_info=None):
|
||||
"""Set up the Mercedes ME tracker."""
|
||||
if discovery_info is None:
|
||||
return False
|
||||
|
||||
data = hass.data[DATA_MME].data
|
||||
|
||||
if not data.cars:
|
||||
return False
|
||||
|
||||
MercedesMEDeviceTracker(hass, config, see, data)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class MercedesMEDeviceTracker(object):
|
||||
"""A class representing a Mercedes ME device tracker."""
|
||||
|
||||
def __init__(self, hass, config, see, data):
|
||||
"""Initialize the Mercedes ME device tracker."""
|
||||
self.see = see
|
||||
self.data = data
|
||||
self.update_info()
|
||||
|
||||
track_time_interval(
|
||||
hass, self.update_info, MIN_TIME_BETWEEN_SCANS)
|
||||
|
||||
@Throttle(MIN_TIME_BETWEEN_SCANS)
|
||||
def update_info(self, now=None):
|
||||
"""Update the device info."""
|
||||
for device in self.data.cars:
|
||||
_LOGGER.debug("Updating %s", device["vin"])
|
||||
location = self.data.get_location(device["vin"])
|
||||
if location is None:
|
||||
return False
|
||||
dev_id = device["vin"]
|
||||
name = device["license"]
|
||||
|
||||
lat = location['positionLat']['value']
|
||||
lon = location['positionLong']['value']
|
||||
attrs = {
|
||||
'trackr_id': dev_id,
|
||||
'id': dev_id,
|
||||
'name': name
|
||||
}
|
||||
self.see(
|
||||
dev_id=dev_id, host_name=name,
|
||||
gps=(lat, lon), attributes=attrs
|
||||
)
|
||||
|
||||
return True
|
||||
"""
|
||||
Support for Mercedes cars with Mercedes ME.
|
||||
|
||||
For more details about this component, please refer to the documentation at
|
||||
https://home-assistant.io/components/device_tracker.mercedesme/
|
||||
"""
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
from homeassistant.components.mercedesme import DATA_MME
|
||||
from homeassistant.helpers.event import track_time_interval
|
||||
from homeassistant.util import Throttle
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEPENDENCIES = ['mercedesme']
|
||||
|
||||
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=30)
|
||||
|
||||
|
||||
def setup_scanner(hass, config, see, discovery_info=None):
|
||||
"""Set up the Mercedes ME tracker."""
|
||||
if discovery_info is None:
|
||||
return False
|
||||
|
||||
data = hass.data[DATA_MME].data
|
||||
|
||||
if not data.cars:
|
||||
return False
|
||||
|
||||
MercedesMEDeviceTracker(hass, config, see, data)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class MercedesMEDeviceTracker(object):
|
||||
"""A class representing a Mercedes ME device tracker."""
|
||||
|
||||
def __init__(self, hass, config, see, data):
|
||||
"""Initialize the Mercedes ME device tracker."""
|
||||
self.see = see
|
||||
self.data = data
|
||||
self.update_info()
|
||||
|
||||
track_time_interval(
|
||||
hass, self.update_info, MIN_TIME_BETWEEN_SCANS)
|
||||
|
||||
@Throttle(MIN_TIME_BETWEEN_SCANS)
|
||||
def update_info(self, now=None):
|
||||
"""Update the device info."""
|
||||
for device in self.data.cars:
|
||||
_LOGGER.debug("Updating %s", device["vin"])
|
||||
location = self.data.get_location(device["vin"])
|
||||
if location is None:
|
||||
return False
|
||||
dev_id = device["vin"]
|
||||
name = device["license"]
|
||||
|
||||
lat = location['positionLat']['value']
|
||||
lon = location['positionLong']['value']
|
||||
attrs = {
|
||||
'trackr_id': dev_id,
|
||||
'id': dev_id,
|
||||
'name': name
|
||||
}
|
||||
self.see(
|
||||
dev_id=dev_id, host_name=name,
|
||||
gps=(lat, lon), attributes=attrs
|
||||
)
|
||||
|
||||
return True
|
||||
|
308
homeassistant/components/mercedesme.py
Executable file → Normal file
308
homeassistant/components/mercedesme.py
Executable file → Normal file
@ -1,154 +1,154 @@
|
||||
"""
|
||||
Support for MercedesME System.
|
||||
|
||||
For more details about this component, please refer to the documentation at
|
||||
https://home-assistant.io/components/mercedesme/
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
import voluptuous as vol
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
from homeassistant.const import (
|
||||
CONF_USERNAME, CONF_PASSWORD, CONF_SCAN_INTERVAL, LENGTH_KILOMETERS)
|
||||
from homeassistant.helpers import discovery
|
||||
from homeassistant.helpers.dispatcher import (
|
||||
async_dispatcher_connect, dispatcher_send)
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.event import track_time_interval
|
||||
|
||||
REQUIREMENTS = ['mercedesmejsonpy==0.1.2']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
BINARY_SENSORS = {
|
||||
'doorsClosed': ['Doors closed'],
|
||||
'windowsClosed': ['Windows closed'],
|
||||
'locked': ['Doors locked'],
|
||||
'tireWarningLight': ['Tire Warning']
|
||||
}
|
||||
|
||||
SENSORS = {
|
||||
'fuelLevelPercent': ['Fuel Level', '%'],
|
||||
'fuelRangeKm': ['Fuel Range', LENGTH_KILOMETERS],
|
||||
'latestTrip': ['Latest Trip', None],
|
||||
'odometerKm': ['Odometer', LENGTH_KILOMETERS],
|
||||
'serviceIntervalDays': ['Next Service', 'days']
|
||||
}
|
||||
|
||||
DATA_MME = 'mercedesme'
|
||||
DOMAIN = 'mercedesme'
|
||||
|
||||
NOTIFICATION_ID = 'mercedesme_integration_notification'
|
||||
NOTIFICATION_TITLE = 'Mercedes me integration setup'
|
||||
|
||||
SIGNAL_UPDATE_MERCEDESME = "mercedesme_update"
|
||||
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema({
|
||||
DOMAIN: vol.Schema({
|
||||
vol.Required(CONF_USERNAME): cv.string,
|
||||
vol.Required(CONF_PASSWORD): cv.string,
|
||||
vol.Optional(CONF_SCAN_INTERVAL, default=30):
|
||||
vol.All(cv.positive_int, vol.Clamp(min=10))
|
||||
})
|
||||
}, extra=vol.ALLOW_EXTRA)
|
||||
|
||||
|
||||
def setup(hass, config):
|
||||
"""Set up MercedesMe System."""
|
||||
from mercedesmejsonpy.controller import Controller
|
||||
from mercedesmejsonpy import Exceptions
|
||||
|
||||
conf = config[DOMAIN]
|
||||
username = conf.get(CONF_USERNAME)
|
||||
password = conf.get(CONF_PASSWORD)
|
||||
scan_interval = conf.get(CONF_SCAN_INTERVAL)
|
||||
|
||||
try:
|
||||
mercedesme_api = Controller(username, password, scan_interval)
|
||||
if not mercedesme_api.is_valid_session:
|
||||
raise Exceptions.MercedesMeException(500)
|
||||
hass.data[DATA_MME] = MercedesMeHub(mercedesme_api)
|
||||
except Exceptions.MercedesMeException as ex:
|
||||
if ex.code == 401:
|
||||
hass.components.persistent_notification.create(
|
||||
"Error:<br />Please check username and password."
|
||||
"You will need to restart Home Assistant after fixing.",
|
||||
title=NOTIFICATION_TITLE,
|
||||
notification_id=NOTIFICATION_ID)
|
||||
else:
|
||||
hass.components.persistent_notification.create(
|
||||
"Error:<br />Can't communicate with Mercedes me API.<br />"
|
||||
"Error code: {} Reason: {}"
|
||||
"You will need to restart Home Assistant after fixing."
|
||||
"".format(ex.code, ex.message),
|
||||
title=NOTIFICATION_TITLE,
|
||||
notification_id=NOTIFICATION_ID)
|
||||
|
||||
_LOGGER.error("Unable to communicate with Mercedes me API: %s",
|
||||
ex.message)
|
||||
return False
|
||||
|
||||
discovery.load_platform(hass, 'sensor', DOMAIN, {}, config)
|
||||
discovery.load_platform(hass, 'device_tracker', DOMAIN, {}, config)
|
||||
discovery.load_platform(hass, 'binary_sensor', DOMAIN, {}, config)
|
||||
|
||||
def hub_refresh(event_time):
|
||||
"""Call Mercedes me API to refresh information."""
|
||||
_LOGGER.info("Updating Mercedes me component.")
|
||||
hass.data[DATA_MME].data.update()
|
||||
dispatcher_send(hass, SIGNAL_UPDATE_MERCEDESME)
|
||||
|
||||
track_time_interval(
|
||||
hass,
|
||||
hub_refresh,
|
||||
timedelta(seconds=scan_interval))
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class MercedesMeHub(object):
|
||||
"""Representation of a base MercedesMe device."""
|
||||
|
||||
def __init__(self, data):
|
||||
"""Initialize the entity."""
|
||||
self.data = data
|
||||
|
||||
|
||||
class MercedesMeEntity(Entity):
|
||||
"""Entity class for MercedesMe devices."""
|
||||
|
||||
def __init__(self, data, internal_name, sensor_name, vin, unit):
|
||||
"""Initialize the MercedesMe entity."""
|
||||
self._car = None
|
||||
self._data = data
|
||||
self._state = False
|
||||
self._name = sensor_name
|
||||
self._internal_name = internal_name
|
||||
self._unit = unit
|
||||
self._vin = vin
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_added_to_hass(self):
|
||||
"""Register callbacks."""
|
||||
async_dispatcher_connect(
|
||||
self.hass, SIGNAL_UPDATE_MERCEDESME, self._update_callback)
|
||||
|
||||
def _update_callback(self):
|
||||
"""Callback update method."""
|
||||
# If the method is made a callback this should be changed
|
||||
# to the async version. Check core.callback
|
||||
self.schedule_update_ha_state(True)
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
"""Return the unit of measurement."""
|
||||
return self._unit
|
||||
"""
|
||||
Support for MercedesME System.
|
||||
|
||||
For more details about this component, please refer to the documentation at
|
||||
https://home-assistant.io/components/mercedesme/
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
import voluptuous as vol
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
from homeassistant.const import (
|
||||
CONF_USERNAME, CONF_PASSWORD, CONF_SCAN_INTERVAL, LENGTH_KILOMETERS)
|
||||
from homeassistant.helpers import discovery
|
||||
from homeassistant.helpers.dispatcher import (
|
||||
async_dispatcher_connect, dispatcher_send)
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.event import track_time_interval
|
||||
|
||||
REQUIREMENTS = ['mercedesmejsonpy==0.1.2']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
BINARY_SENSORS = {
|
||||
'doorsClosed': ['Doors closed'],
|
||||
'windowsClosed': ['Windows closed'],
|
||||
'locked': ['Doors locked'],
|
||||
'tireWarningLight': ['Tire Warning']
|
||||
}
|
||||
|
||||
SENSORS = {
|
||||
'fuelLevelPercent': ['Fuel Level', '%'],
|
||||
'fuelRangeKm': ['Fuel Range', LENGTH_KILOMETERS],
|
||||
'latestTrip': ['Latest Trip', None],
|
||||
'odometerKm': ['Odometer', LENGTH_KILOMETERS],
|
||||
'serviceIntervalDays': ['Next Service', 'days']
|
||||
}
|
||||
|
||||
DATA_MME = 'mercedesme'
|
||||
DOMAIN = 'mercedesme'
|
||||
|
||||
NOTIFICATION_ID = 'mercedesme_integration_notification'
|
||||
NOTIFICATION_TITLE = 'Mercedes me integration setup'
|
||||
|
||||
SIGNAL_UPDATE_MERCEDESME = "mercedesme_update"
|
||||
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema({
|
||||
DOMAIN: vol.Schema({
|
||||
vol.Required(CONF_USERNAME): cv.string,
|
||||
vol.Required(CONF_PASSWORD): cv.string,
|
||||
vol.Optional(CONF_SCAN_INTERVAL, default=30):
|
||||
vol.All(cv.positive_int, vol.Clamp(min=10))
|
||||
})
|
||||
}, extra=vol.ALLOW_EXTRA)
|
||||
|
||||
|
||||
def setup(hass, config):
|
||||
"""Set up MercedesMe System."""
|
||||
from mercedesmejsonpy.controller import Controller
|
||||
from mercedesmejsonpy import Exceptions
|
||||
|
||||
conf = config[DOMAIN]
|
||||
username = conf.get(CONF_USERNAME)
|
||||
password = conf.get(CONF_PASSWORD)
|
||||
scan_interval = conf.get(CONF_SCAN_INTERVAL)
|
||||
|
||||
try:
|
||||
mercedesme_api = Controller(username, password, scan_interval)
|
||||
if not mercedesme_api.is_valid_session:
|
||||
raise Exceptions.MercedesMeException(500)
|
||||
hass.data[DATA_MME] = MercedesMeHub(mercedesme_api)
|
||||
except Exceptions.MercedesMeException as ex:
|
||||
if ex.code == 401:
|
||||
hass.components.persistent_notification.create(
|
||||
"Error:<br />Please check username and password."
|
||||
"You will need to restart Home Assistant after fixing.",
|
||||
title=NOTIFICATION_TITLE,
|
||||
notification_id=NOTIFICATION_ID)
|
||||
else:
|
||||
hass.components.persistent_notification.create(
|
||||
"Error:<br />Can't communicate with Mercedes me API.<br />"
|
||||
"Error code: {} Reason: {}"
|
||||
"You will need to restart Home Assistant after fixing."
|
||||
"".format(ex.code, ex.message),
|
||||
title=NOTIFICATION_TITLE,
|
||||
notification_id=NOTIFICATION_ID)
|
||||
|
||||
_LOGGER.error("Unable to communicate with Mercedes me API: %s",
|
||||
ex.message)
|
||||
return False
|
||||
|
||||
discovery.load_platform(hass, 'sensor', DOMAIN, {}, config)
|
||||
discovery.load_platform(hass, 'device_tracker', DOMAIN, {}, config)
|
||||
discovery.load_platform(hass, 'binary_sensor', DOMAIN, {}, config)
|
||||
|
||||
def hub_refresh(event_time):
|
||||
"""Call Mercedes me API to refresh information."""
|
||||
_LOGGER.info("Updating Mercedes me component.")
|
||||
hass.data[DATA_MME].data.update()
|
||||
dispatcher_send(hass, SIGNAL_UPDATE_MERCEDESME)
|
||||
|
||||
track_time_interval(
|
||||
hass,
|
||||
hub_refresh,
|
||||
timedelta(seconds=scan_interval))
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class MercedesMeHub(object):
|
||||
"""Representation of a base MercedesMe device."""
|
||||
|
||||
def __init__(self, data):
|
||||
"""Initialize the entity."""
|
||||
self.data = data
|
||||
|
||||
|
||||
class MercedesMeEntity(Entity):
|
||||
"""Entity class for MercedesMe devices."""
|
||||
|
||||
def __init__(self, data, internal_name, sensor_name, vin, unit):
|
||||
"""Initialize the MercedesMe entity."""
|
||||
self._car = None
|
||||
self._data = data
|
||||
self._state = False
|
||||
self._name = sensor_name
|
||||
self._internal_name = internal_name
|
||||
self._unit = unit
|
||||
self._vin = vin
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_added_to_hass(self):
|
||||
"""Register callbacks."""
|
||||
async_dispatcher_connect(
|
||||
self.hass, SIGNAL_UPDATE_MERCEDESME, self._update_callback)
|
||||
|
||||
def _update_callback(self):
|
||||
"""Callback update method."""
|
||||
# If the method is made a callback this should be changed
|
||||
# to the async version. Check core.callback
|
||||
self.schedule_update_ha_state(True)
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
"""Return the unit of measurement."""
|
||||
return self._unit
|
||||
|
0
homeassistant/components/remote/xiaomi_miio.py
Executable file → Normal file
0
homeassistant/components/remote/xiaomi_miio.py
Executable file → Normal file
166
homeassistant/components/sensor/mercedesme.py
Executable file → Normal file
166
homeassistant/components/sensor/mercedesme.py
Executable file → Normal file
@ -1,83 +1,83 @@
|
||||
"""
|
||||
Support for Mercedes cars with Mercedes ME.
|
||||
|
||||
For more details about this component, please refer to the documentation at
|
||||
https://home-assistant.io/components/sensor.mercedesme/
|
||||
"""
|
||||
import logging
|
||||
import datetime
|
||||
|
||||
from homeassistant.components.mercedesme import (
|
||||
DATA_MME, MercedesMeEntity, SENSORS)
|
||||
|
||||
|
||||
DEPENDENCIES = ['mercedesme']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Setup the sensor platform."""
|
||||
if discovery_info is None:
|
||||
return
|
||||
|
||||
data = hass.data[DATA_MME].data
|
||||
|
||||
if not data.cars:
|
||||
return
|
||||
|
||||
devices = []
|
||||
for car in data.cars:
|
||||
for key, value in sorted(SENSORS.items()):
|
||||
devices.append(
|
||||
MercedesMESensor(data, key, value[0], car["vin"], value[1]))
|
||||
|
||||
add_devices(devices, True)
|
||||
|
||||
|
||||
class MercedesMESensor(MercedesMeEntity):
|
||||
"""Representation of a Sensor."""
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the sensor."""
|
||||
return self._state
|
||||
|
||||
def update(self):
|
||||
"""Get the latest data and updates the states."""
|
||||
_LOGGER.debug("Updating %s", self._internal_name)
|
||||
|
||||
self._car = next(
|
||||
car for car in self._data.cars if car["vin"] == self._vin)
|
||||
|
||||
if self._internal_name == "latestTrip":
|
||||
self._state = self._car["latestTrip"]["id"]
|
||||
else:
|
||||
self._state = self._car[self._internal_name]
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
if self._internal_name == "latestTrip":
|
||||
return {
|
||||
"duration_seconds":
|
||||
self._car["latestTrip"]["durationSeconds"],
|
||||
"distance_traveled_km":
|
||||
self._car["latestTrip"]["distanceTraveledKm"],
|
||||
"started_at": datetime.datetime.fromtimestamp(
|
||||
self._car["latestTrip"]["startedAt"]
|
||||
).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"average_speed_km_per_hr":
|
||||
self._car["latestTrip"]["averageSpeedKmPerHr"],
|
||||
"finished": self._car["latestTrip"]["finished"],
|
||||
"last_update": datetime.datetime.fromtimestamp(
|
||||
self._car["lastUpdate"]
|
||||
).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"car": self._car["license"]
|
||||
}
|
||||
|
||||
return {
|
||||
"last_update": datetime.datetime.fromtimestamp(
|
||||
self._car["lastUpdate"]).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"car": self._car["license"]
|
||||
}
|
||||
"""
|
||||
Support for Mercedes cars with Mercedes ME.
|
||||
|
||||
For more details about this component, please refer to the documentation at
|
||||
https://home-assistant.io/components/sensor.mercedesme/
|
||||
"""
|
||||
import logging
|
||||
import datetime
|
||||
|
||||
from homeassistant.components.mercedesme import (
|
||||
DATA_MME, MercedesMeEntity, SENSORS)
|
||||
|
||||
|
||||
DEPENDENCIES = ['mercedesme']
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_devices, discovery_info=None):
|
||||
"""Setup the sensor platform."""
|
||||
if discovery_info is None:
|
||||
return
|
||||
|
||||
data = hass.data[DATA_MME].data
|
||||
|
||||
if not data.cars:
|
||||
return
|
||||
|
||||
devices = []
|
||||
for car in data.cars:
|
||||
for key, value in sorted(SENSORS.items()):
|
||||
devices.append(
|
||||
MercedesMESensor(data, key, value[0], car["vin"], value[1]))
|
||||
|
||||
add_devices(devices, True)
|
||||
|
||||
|
||||
class MercedesMESensor(MercedesMeEntity):
|
||||
"""Representation of a Sensor."""
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state of the sensor."""
|
||||
return self._state
|
||||
|
||||
def update(self):
|
||||
"""Get the latest data and updates the states."""
|
||||
_LOGGER.debug("Updating %s", self._internal_name)
|
||||
|
||||
self._car = next(
|
||||
car for car in self._data.cars if car["vin"] == self._vin)
|
||||
|
||||
if self._internal_name == "latestTrip":
|
||||
self._state = self._car["latestTrip"]["id"]
|
||||
else:
|
||||
self._state = self._car[self._internal_name]
|
||||
|
||||
@property
|
||||
def device_state_attributes(self):
|
||||
"""Return the state attributes."""
|
||||
if self._internal_name == "latestTrip":
|
||||
return {
|
||||
"duration_seconds":
|
||||
self._car["latestTrip"]["durationSeconds"],
|
||||
"distance_traveled_km":
|
||||
self._car["latestTrip"]["distanceTraveledKm"],
|
||||
"started_at": datetime.datetime.fromtimestamp(
|
||||
self._car["latestTrip"]["startedAt"]
|
||||
).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"average_speed_km_per_hr":
|
||||
self._car["latestTrip"]["averageSpeedKmPerHr"],
|
||||
"finished": self._car["latestTrip"]["finished"],
|
||||
"last_update": datetime.datetime.fromtimestamp(
|
||||
self._car["lastUpdate"]
|
||||
).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"car": self._car["license"]
|
||||
}
|
||||
|
||||
return {
|
||||
"last_update": datetime.datetime.fromtimestamp(
|
||||
self._car["lastUpdate"]).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"car": self._car["license"]
|
||||
}
|
||||
|
@ -1,423 +1,423 @@
|
||||
"""The tests for the google calendar component."""
|
||||
# pylint: disable=protected-access
|
||||
import logging
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
import homeassistant.components.calendar as calendar_base
|
||||
import homeassistant.components.calendar.google as calendar
|
||||
import homeassistant.util.dt as dt_util
|
||||
from homeassistant.const import CONF_PLATFORM, STATE_OFF, STATE_ON
|
||||
from homeassistant.helpers.template import DATE_STR_FORMAT
|
||||
from tests.common import get_test_home_assistant
|
||||
|
||||
TEST_PLATFORM = {calendar_base.DOMAIN: {CONF_PLATFORM: 'test'}}
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TestComponentsGoogleCalendar(unittest.TestCase):
|
||||
"""Test the Google calendar."""
|
||||
|
||||
hass = None # HomeAssistant
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def setUp(self):
|
||||
"""Setup things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
|
||||
# Set our timezone to CST/Regina so we can check calculations
|
||||
# This keeps UTC-6 all year round
|
||||
dt_util.set_default_time_zone(dt_util.get_time_zone('America/Regina'))
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
dt_util.set_default_time_zone(dt_util.get_time_zone('UTC'))
|
||||
|
||||
self.hass.stop()
|
||||
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_all_day_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
week_from_today = dt_util.dt.date.today() \
|
||||
+ dt_util.dt.timedelta(days=7)
|
||||
event = {
|
||||
'summary': 'Test All Day Event',
|
||||
'start': {
|
||||
'date': week_from_today.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'date': (week_from_today + dt_util.dt.timedelta(days=1))
|
||||
.isoformat()
|
||||
},
|
||||
'location': 'Test Cases',
|
||||
'description': 'We\'re just testing that all day events get setup '
|
||||
'correctly',
|
||||
'kind': 'calendar#event',
|
||||
'created': '2016-06-23T16:37:57.000Z',
|
||||
'transparency': 'transparent',
|
||||
'updated': '2016-06-24T01:57:21.045Z',
|
||||
'reminders': {'useDefault': True},
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'id': '_c8rinwq863h45qnucyoi43ny8',
|
||||
'etag': '"2933466882090000"',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
'iCalUID': 'cydrevtfuybguinhomj@google.com',
|
||||
'status': 'confirmed'
|
||||
}
|
||||
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test All Day'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None,
|
||||
'', {'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_OFF)
|
||||
|
||||
self.assertFalse(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event['summary'],
|
||||
'all_day': True,
|
||||
'offset_reached': False,
|
||||
'start_time': '{} 00:00:00'.format(event['start']['date']),
|
||||
'end_time': '{} 00:00:00'.format(event['end']['date']),
|
||||
'location': event['location'],
|
||||
'description': event['description']
|
||||
})
|
||||
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_future_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
one_hour_from_now = dt_util.now() \
|
||||
+ dt_util.dt.timedelta(minutes=30)
|
||||
event = {
|
||||
'start': {
|
||||
'dateTime': one_hour_from_now.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'dateTime': (one_hour_from_now
|
||||
+ dt_util.dt.timedelta(minutes=60))
|
||||
.isoformat()
|
||||
},
|
||||
'summary': 'Test Event in 30 minutes',
|
||||
'reminders': {'useDefault': True},
|
||||
'id': 'aioehgni435lihje',
|
||||
'status': 'confirmed',
|
||||
'updated': '2016-11-05T15:52:07.329Z',
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True,
|
||||
},
|
||||
'created': '2016-11-05T15:52:07.000Z',
|
||||
'iCalUID': 'dsfohuygtfvgbhnuju@google.com',
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
},
|
||||
'etag': '"2956722254658000"',
|
||||
'kind': 'calendar#event',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
}
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test Future Event'
|
||||
device_id = 'test_future_event'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None, device_id,
|
||||
{'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_OFF)
|
||||
|
||||
self.assertFalse(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event['summary'],
|
||||
'all_day': False,
|
||||
'offset_reached': False,
|
||||
'start_time': one_hour_from_now.strftime(DATE_STR_FORMAT),
|
||||
'end_time':
|
||||
(one_hour_from_now + dt_util.dt.timedelta(minutes=60))
|
||||
.strftime(DATE_STR_FORMAT),
|
||||
'location': '',
|
||||
'description': ''
|
||||
})
|
||||
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_in_progress_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
middle_of_event = dt_util.now() \
|
||||
- dt_util.dt.timedelta(minutes=30)
|
||||
event = {
|
||||
'start': {
|
||||
'dateTime': middle_of_event.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'dateTime': (middle_of_event + dt_util.dt
|
||||
.timedelta(minutes=60))
|
||||
.isoformat()
|
||||
},
|
||||
'summary': 'Test Event in Progress',
|
||||
'reminders': {'useDefault': True},
|
||||
'id': 'aioehgni435lihje',
|
||||
'status': 'confirmed',
|
||||
'updated': '2016-11-05T15:52:07.329Z',
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True,
|
||||
},
|
||||
'created': '2016-11-05T15:52:07.000Z',
|
||||
'iCalUID': 'dsfohuygtfvgbhnuju@google.com',
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
},
|
||||
'etag': '"2956722254658000"',
|
||||
'kind': 'calendar#event',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
}
|
||||
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test Event in Progress'
|
||||
device_id = 'test_event_in_progress'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None, device_id,
|
||||
{'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_ON)
|
||||
|
||||
self.assertFalse(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event['summary'],
|
||||
'all_day': False,
|
||||
'offset_reached': False,
|
||||
'start_time': middle_of_event.strftime(DATE_STR_FORMAT),
|
||||
'end_time':
|
||||
(middle_of_event + dt_util.dt.timedelta(minutes=60))
|
||||
.strftime(DATE_STR_FORMAT),
|
||||
'location': '',
|
||||
'description': ''
|
||||
})
|
||||
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_offset_in_progress_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
middle_of_event = dt_util.now() \
|
||||
+ dt_util.dt.timedelta(minutes=14)
|
||||
event_summary = 'Test Event in Progress'
|
||||
event = {
|
||||
'start': {
|
||||
'dateTime': middle_of_event.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'dateTime': (middle_of_event + dt_util.dt
|
||||
.timedelta(minutes=60))
|
||||
.isoformat()
|
||||
},
|
||||
'summary': '{} !!-15'.format(event_summary),
|
||||
'reminders': {'useDefault': True},
|
||||
'id': 'aioehgni435lihje',
|
||||
'status': 'confirmed',
|
||||
'updated': '2016-11-05T15:52:07.329Z',
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True,
|
||||
},
|
||||
'created': '2016-11-05T15:52:07.000Z',
|
||||
'iCalUID': 'dsfohuygtfvgbhnuju@google.com',
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
},
|
||||
'etag': '"2956722254658000"',
|
||||
'kind': 'calendar#event',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
}
|
||||
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test Event in Progress'
|
||||
device_id = 'test_event_in_progress'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None, device_id,
|
||||
{'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_OFF)
|
||||
|
||||
self.assertTrue(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event_summary,
|
||||
'all_day': False,
|
||||
'offset_reached': True,
|
||||
'start_time': middle_of_event.strftime(DATE_STR_FORMAT),
|
||||
'end_time':
|
||||
(middle_of_event + dt_util.dt.timedelta(minutes=60))
|
||||
.strftime(DATE_STR_FORMAT),
|
||||
'location': '',
|
||||
'description': ''
|
||||
})
|
||||
|
||||
@pytest.mark.skip
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_all_day_offset_in_progress_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
tomorrow = dt_util.dt.date.today() \
|
||||
+ dt_util.dt.timedelta(days=1)
|
||||
|
||||
event_summary = 'Test All Day Event Offset In Progress'
|
||||
event = {
|
||||
'summary': '{} !!-25:0'.format(event_summary),
|
||||
'start': {
|
||||
'date': tomorrow.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'date': (tomorrow + dt_util.dt.timedelta(days=1))
|
||||
.isoformat()
|
||||
},
|
||||
'location': 'Test Cases',
|
||||
'description': 'We\'re just testing that all day events get setup '
|
||||
'correctly',
|
||||
'kind': 'calendar#event',
|
||||
'created': '2016-06-23T16:37:57.000Z',
|
||||
'transparency': 'transparent',
|
||||
'updated': '2016-06-24T01:57:21.045Z',
|
||||
'reminders': {'useDefault': True},
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'id': '_c8rinwq863h45qnucyoi43ny8',
|
||||
'etag': '"2933466882090000"',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
'iCalUID': 'cydrevtfuybguinhomj@google.com',
|
||||
'status': 'confirmed'
|
||||
}
|
||||
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test All Day Offset In Progress'
|
||||
device_id = 'test_all_day_offset_in_progress'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None, device_id,
|
||||
{'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_OFF)
|
||||
|
||||
self.assertTrue(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event_summary,
|
||||
'all_day': True,
|
||||
'offset_reached': True,
|
||||
'start_time': '{} 06:00:00'.format(event['start']['date']),
|
||||
'end_time': '{} 06:00:00'.format(event['end']['date']),
|
||||
'location': event['location'],
|
||||
'description': event['description']
|
||||
})
|
||||
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_all_day_offset_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
tomorrow = dt_util.dt.date.today() \
|
||||
+ dt_util.dt.timedelta(days=2)
|
||||
|
||||
offset_hours = (1 + dt_util.now().hour)
|
||||
event_summary = 'Test All Day Event Offset'
|
||||
event = {
|
||||
'summary': '{} !!-{}:0'.format(event_summary, offset_hours),
|
||||
'start': {
|
||||
'date': tomorrow.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'date': (tomorrow + dt_util.dt.timedelta(days=1))
|
||||
.isoformat()
|
||||
},
|
||||
'location': 'Test Cases',
|
||||
'description': 'We\'re just testing that all day events get setup '
|
||||
'correctly',
|
||||
'kind': 'calendar#event',
|
||||
'created': '2016-06-23T16:37:57.000Z',
|
||||
'transparency': 'transparent',
|
||||
'updated': '2016-06-24T01:57:21.045Z',
|
||||
'reminders': {'useDefault': True},
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'id': '_c8rinwq863h45qnucyoi43ny8',
|
||||
'etag': '"2933466882090000"',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
'iCalUID': 'cydrevtfuybguinhomj@google.com',
|
||||
'status': 'confirmed'
|
||||
}
|
||||
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test All Day Offset'
|
||||
device_id = 'test_all_day_offset'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None, device_id,
|
||||
{'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_OFF)
|
||||
|
||||
self.assertFalse(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event_summary,
|
||||
'all_day': True,
|
||||
'offset_reached': False,
|
||||
'start_time': '{} 00:00:00'.format(event['start']['date']),
|
||||
'end_time': '{} 00:00:00'.format(event['end']['date']),
|
||||
'location': event['location'],
|
||||
'description': event['description']
|
||||
})
|
||||
"""The tests for the google calendar component."""
|
||||
# pylint: disable=protected-access
|
||||
import logging
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
import homeassistant.components.calendar as calendar_base
|
||||
import homeassistant.components.calendar.google as calendar
|
||||
import homeassistant.util.dt as dt_util
|
||||
from homeassistant.const import CONF_PLATFORM, STATE_OFF, STATE_ON
|
||||
from homeassistant.helpers.template import DATE_STR_FORMAT
|
||||
from tests.common import get_test_home_assistant
|
||||
|
||||
TEST_PLATFORM = {calendar_base.DOMAIN: {CONF_PLATFORM: 'test'}}
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TestComponentsGoogleCalendar(unittest.TestCase):
|
||||
"""Test the Google calendar."""
|
||||
|
||||
hass = None # HomeAssistant
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def setUp(self):
|
||||
"""Setup things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
|
||||
# Set our timezone to CST/Regina so we can check calculations
|
||||
# This keeps UTC-6 all year round
|
||||
dt_util.set_default_time_zone(dt_util.get_time_zone('America/Regina'))
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
dt_util.set_default_time_zone(dt_util.get_time_zone('UTC'))
|
||||
|
||||
self.hass.stop()
|
||||
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_all_day_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
week_from_today = dt_util.dt.date.today() \
|
||||
+ dt_util.dt.timedelta(days=7)
|
||||
event = {
|
||||
'summary': 'Test All Day Event',
|
||||
'start': {
|
||||
'date': week_from_today.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'date': (week_from_today + dt_util.dt.timedelta(days=1))
|
||||
.isoformat()
|
||||
},
|
||||
'location': 'Test Cases',
|
||||
'description': 'We\'re just testing that all day events get setup '
|
||||
'correctly',
|
||||
'kind': 'calendar#event',
|
||||
'created': '2016-06-23T16:37:57.000Z',
|
||||
'transparency': 'transparent',
|
||||
'updated': '2016-06-24T01:57:21.045Z',
|
||||
'reminders': {'useDefault': True},
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'id': '_c8rinwq863h45qnucyoi43ny8',
|
||||
'etag': '"2933466882090000"',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
'iCalUID': 'cydrevtfuybguinhomj@google.com',
|
||||
'status': 'confirmed'
|
||||
}
|
||||
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test All Day'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None,
|
||||
'', {'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_OFF)
|
||||
|
||||
self.assertFalse(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event['summary'],
|
||||
'all_day': True,
|
||||
'offset_reached': False,
|
||||
'start_time': '{} 00:00:00'.format(event['start']['date']),
|
||||
'end_time': '{} 00:00:00'.format(event['end']['date']),
|
||||
'location': event['location'],
|
||||
'description': event['description']
|
||||
})
|
||||
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_future_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
one_hour_from_now = dt_util.now() \
|
||||
+ dt_util.dt.timedelta(minutes=30)
|
||||
event = {
|
||||
'start': {
|
||||
'dateTime': one_hour_from_now.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'dateTime': (one_hour_from_now
|
||||
+ dt_util.dt.timedelta(minutes=60))
|
||||
.isoformat()
|
||||
},
|
||||
'summary': 'Test Event in 30 minutes',
|
||||
'reminders': {'useDefault': True},
|
||||
'id': 'aioehgni435lihje',
|
||||
'status': 'confirmed',
|
||||
'updated': '2016-11-05T15:52:07.329Z',
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True,
|
||||
},
|
||||
'created': '2016-11-05T15:52:07.000Z',
|
||||
'iCalUID': 'dsfohuygtfvgbhnuju@google.com',
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
},
|
||||
'etag': '"2956722254658000"',
|
||||
'kind': 'calendar#event',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
}
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test Future Event'
|
||||
device_id = 'test_future_event'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None, device_id,
|
||||
{'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_OFF)
|
||||
|
||||
self.assertFalse(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event['summary'],
|
||||
'all_day': False,
|
||||
'offset_reached': False,
|
||||
'start_time': one_hour_from_now.strftime(DATE_STR_FORMAT),
|
||||
'end_time':
|
||||
(one_hour_from_now + dt_util.dt.timedelta(minutes=60))
|
||||
.strftime(DATE_STR_FORMAT),
|
||||
'location': '',
|
||||
'description': ''
|
||||
})
|
||||
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_in_progress_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
middle_of_event = dt_util.now() \
|
||||
- dt_util.dt.timedelta(minutes=30)
|
||||
event = {
|
||||
'start': {
|
||||
'dateTime': middle_of_event.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'dateTime': (middle_of_event + dt_util.dt
|
||||
.timedelta(minutes=60))
|
||||
.isoformat()
|
||||
},
|
||||
'summary': 'Test Event in Progress',
|
||||
'reminders': {'useDefault': True},
|
||||
'id': 'aioehgni435lihje',
|
||||
'status': 'confirmed',
|
||||
'updated': '2016-11-05T15:52:07.329Z',
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True,
|
||||
},
|
||||
'created': '2016-11-05T15:52:07.000Z',
|
||||
'iCalUID': 'dsfohuygtfvgbhnuju@google.com',
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
},
|
||||
'etag': '"2956722254658000"',
|
||||
'kind': 'calendar#event',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
}
|
||||
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test Event in Progress'
|
||||
device_id = 'test_event_in_progress'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None, device_id,
|
||||
{'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_ON)
|
||||
|
||||
self.assertFalse(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event['summary'],
|
||||
'all_day': False,
|
||||
'offset_reached': False,
|
||||
'start_time': middle_of_event.strftime(DATE_STR_FORMAT),
|
||||
'end_time':
|
||||
(middle_of_event + dt_util.dt.timedelta(minutes=60))
|
||||
.strftime(DATE_STR_FORMAT),
|
||||
'location': '',
|
||||
'description': ''
|
||||
})
|
||||
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_offset_in_progress_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
middle_of_event = dt_util.now() \
|
||||
+ dt_util.dt.timedelta(minutes=14)
|
||||
event_summary = 'Test Event in Progress'
|
||||
event = {
|
||||
'start': {
|
||||
'dateTime': middle_of_event.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'dateTime': (middle_of_event + dt_util.dt
|
||||
.timedelta(minutes=60))
|
||||
.isoformat()
|
||||
},
|
||||
'summary': '{} !!-15'.format(event_summary),
|
||||
'reminders': {'useDefault': True},
|
||||
'id': 'aioehgni435lihje',
|
||||
'status': 'confirmed',
|
||||
'updated': '2016-11-05T15:52:07.329Z',
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True,
|
||||
},
|
||||
'created': '2016-11-05T15:52:07.000Z',
|
||||
'iCalUID': 'dsfohuygtfvgbhnuju@google.com',
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
},
|
||||
'etag': '"2956722254658000"',
|
||||
'kind': 'calendar#event',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
}
|
||||
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test Event in Progress'
|
||||
device_id = 'test_event_in_progress'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None, device_id,
|
||||
{'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_OFF)
|
||||
|
||||
self.assertTrue(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event_summary,
|
||||
'all_day': False,
|
||||
'offset_reached': True,
|
||||
'start_time': middle_of_event.strftime(DATE_STR_FORMAT),
|
||||
'end_time':
|
||||
(middle_of_event + dt_util.dt.timedelta(minutes=60))
|
||||
.strftime(DATE_STR_FORMAT),
|
||||
'location': '',
|
||||
'description': ''
|
||||
})
|
||||
|
||||
@pytest.mark.skip
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_all_day_offset_in_progress_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
tomorrow = dt_util.dt.date.today() \
|
||||
+ dt_util.dt.timedelta(days=1)
|
||||
|
||||
event_summary = 'Test All Day Event Offset In Progress'
|
||||
event = {
|
||||
'summary': '{} !!-25:0'.format(event_summary),
|
||||
'start': {
|
||||
'date': tomorrow.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'date': (tomorrow + dt_util.dt.timedelta(days=1))
|
||||
.isoformat()
|
||||
},
|
||||
'location': 'Test Cases',
|
||||
'description': 'We\'re just testing that all day events get setup '
|
||||
'correctly',
|
||||
'kind': 'calendar#event',
|
||||
'created': '2016-06-23T16:37:57.000Z',
|
||||
'transparency': 'transparent',
|
||||
'updated': '2016-06-24T01:57:21.045Z',
|
||||
'reminders': {'useDefault': True},
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'id': '_c8rinwq863h45qnucyoi43ny8',
|
||||
'etag': '"2933466882090000"',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
'iCalUID': 'cydrevtfuybguinhomj@google.com',
|
||||
'status': 'confirmed'
|
||||
}
|
||||
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test All Day Offset In Progress'
|
||||
device_id = 'test_all_day_offset_in_progress'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None, device_id,
|
||||
{'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_OFF)
|
||||
|
||||
self.assertTrue(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event_summary,
|
||||
'all_day': True,
|
||||
'offset_reached': True,
|
||||
'start_time': '{} 06:00:00'.format(event['start']['date']),
|
||||
'end_time': '{} 06:00:00'.format(event['end']['date']),
|
||||
'location': event['location'],
|
||||
'description': event['description']
|
||||
})
|
||||
|
||||
@patch('homeassistant.components.calendar.google.GoogleCalendarData')
|
||||
def test_all_day_offset_event(self, mock_next_event):
|
||||
"""Test that we can create an event trigger on device."""
|
||||
tomorrow = dt_util.dt.date.today() \
|
||||
+ dt_util.dt.timedelta(days=2)
|
||||
|
||||
offset_hours = (1 + dt_util.now().hour)
|
||||
event_summary = 'Test All Day Event Offset'
|
||||
event = {
|
||||
'summary': '{} !!-{}:0'.format(event_summary, offset_hours),
|
||||
'start': {
|
||||
'date': tomorrow.isoformat()
|
||||
},
|
||||
'end': {
|
||||
'date': (tomorrow + dt_util.dt.timedelta(days=1))
|
||||
.isoformat()
|
||||
},
|
||||
'location': 'Test Cases',
|
||||
'description': 'We\'re just testing that all day events get setup '
|
||||
'correctly',
|
||||
'kind': 'calendar#event',
|
||||
'created': '2016-06-23T16:37:57.000Z',
|
||||
'transparency': 'transparent',
|
||||
'updated': '2016-06-24T01:57:21.045Z',
|
||||
'reminders': {'useDefault': True},
|
||||
'organizer': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'sequence': 0,
|
||||
'creator': {
|
||||
'email': 'uvrttabwegnui4gtia3vyqb@import.calendar.google.com',
|
||||
'displayName': 'Organizer Name',
|
||||
'self': True
|
||||
},
|
||||
'id': '_c8rinwq863h45qnucyoi43ny8',
|
||||
'etag': '"2933466882090000"',
|
||||
'htmlLink': 'https://www.google.com/calendar/event?eid=*******',
|
||||
'iCalUID': 'cydrevtfuybguinhomj@google.com',
|
||||
'status': 'confirmed'
|
||||
}
|
||||
|
||||
mock_next_event.return_value.event = event
|
||||
|
||||
device_name = 'Test All Day Offset'
|
||||
device_id = 'test_all_day_offset'
|
||||
|
||||
cal = calendar.GoogleCalendarEventDevice(self.hass, None, device_id,
|
||||
{'name': device_name})
|
||||
|
||||
self.assertEqual(cal.name, device_name)
|
||||
|
||||
self.assertEqual(cal.state, STATE_OFF)
|
||||
|
||||
self.assertFalse(cal.offset_reached())
|
||||
|
||||
self.assertEqual(cal.device_state_attributes, {
|
||||
'message': event_summary,
|
||||
'all_day': True,
|
||||
'offset_reached': False,
|
||||
'start_time': '{} 00:00:00'.format(event['start']['date']),
|
||||
'end_time': '{} 00:00:00'.format(event['end']['date']),
|
||||
'location': event['location'],
|
||||
'description': event['description']
|
||||
})
|
||||
|
@ -1 +1 @@
|
||||
"""Tests for the cloud component."""
|
||||
"""Tests for the cloud component."""
|
||||
|
@ -1,265 +1,265 @@
|
||||
"""The tests for the Xiaomi router device tracker platform."""
|
||||
import logging
|
||||
import unittest
|
||||
from unittest import mock
|
||||
from unittest.mock import patch
|
||||
|
||||
import requests
|
||||
|
||||
from homeassistant.components.device_tracker import DOMAIN, xiaomi as xiaomi
|
||||
from homeassistant.components.device_tracker.xiaomi import get_scanner
|
||||
from homeassistant.const import (CONF_HOST, CONF_USERNAME, CONF_PASSWORD,
|
||||
CONF_PLATFORM)
|
||||
from tests.common import get_test_home_assistant
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
INVALID_USERNAME = 'bob'
|
||||
TOKEN_TIMEOUT_USERNAME = 'tok'
|
||||
URL_AUTHORIZE = 'http://192.168.0.1/cgi-bin/luci/api/xqsystem/login'
|
||||
URL_LIST_END = 'api/misystem/devicelist'
|
||||
|
||||
FIRST_CALL = True
|
||||
|
||||
|
||||
def mocked_requests(*args, **kwargs):
|
||||
"""Mock requests.get invocations."""
|
||||
class MockResponse:
|
||||
"""Class to represent a mocked response."""
|
||||
|
||||
def __init__(self, json_data, status_code):
|
||||
"""Initialize the mock response class."""
|
||||
self.json_data = json_data
|
||||
self.status_code = status_code
|
||||
|
||||
def json(self):
|
||||
"""Return the json of the response."""
|
||||
return self.json_data
|
||||
|
||||
@property
|
||||
def content(self):
|
||||
"""Return the content of the response."""
|
||||
return self.json()
|
||||
|
||||
def raise_for_status(self):
|
||||
"""Raise an HTTPError if status is not 200."""
|
||||
if self.status_code != 200:
|
||||
raise requests.HTTPError(self.status_code)
|
||||
|
||||
data = kwargs.get('data')
|
||||
global FIRST_CALL
|
||||
|
||||
if data and data.get('username', None) == INVALID_USERNAME:
|
||||
# deliver an invalid token
|
||||
return MockResponse({
|
||||
"code": "401",
|
||||
"msg": "Invalid token"
|
||||
}, 200)
|
||||
elif data and data.get('username', None) == TOKEN_TIMEOUT_USERNAME:
|
||||
# deliver an expired token
|
||||
return MockResponse({
|
||||
"url": "/cgi-bin/luci/;stok=ef5860/web/home",
|
||||
"token": "timedOut",
|
||||
"code": "0"
|
||||
}, 200)
|
||||
elif str(args[0]).startswith(URL_AUTHORIZE):
|
||||
# deliver an authorized token
|
||||
return MockResponse({
|
||||
"url": "/cgi-bin/luci/;stok=ef5860/web/home",
|
||||
"token": "ef5860",
|
||||
"code": "0"
|
||||
}, 200)
|
||||
elif str(args[0]).endswith("timedOut/" + URL_LIST_END) \
|
||||
and FIRST_CALL is True:
|
||||
FIRST_CALL = False
|
||||
# deliver an error when called with expired token
|
||||
return MockResponse({
|
||||
"code": "401",
|
||||
"msg": "Invalid token"
|
||||
}, 200)
|
||||
elif str(args[0]).endswith(URL_LIST_END):
|
||||
# deliver the device list
|
||||
return MockResponse({
|
||||
"mac": "1C:98:EC:0E:D5:A4",
|
||||
"list": [
|
||||
{
|
||||
"mac": "23:83:BF:F6:38:A0",
|
||||
"oname": "12255ff",
|
||||
"isap": 0,
|
||||
"parent": "",
|
||||
"authority": {
|
||||
"wan": 1,
|
||||
"pridisk": 0,
|
||||
"admin": 1,
|
||||
"lan": 0
|
||||
},
|
||||
"push": 0,
|
||||
"online": 1,
|
||||
"name": "Device1",
|
||||
"times": 0,
|
||||
"ip": [
|
||||
{
|
||||
"downspeed": "0",
|
||||
"online": "496957",
|
||||
"active": 1,
|
||||
"upspeed": "0",
|
||||
"ip": "192.168.0.25"
|
||||
}
|
||||
],
|
||||
"statistics": {
|
||||
"downspeed": "0",
|
||||
"online": "496957",
|
||||
"upspeed": "0"
|
||||
},
|
||||
"icon": "",
|
||||
"type": 1
|
||||
},
|
||||
{
|
||||
"mac": "1D:98:EC:5E:D5:A6",
|
||||
"oname": "CdddFG58",
|
||||
"isap": 0,
|
||||
"parent": "",
|
||||
"authority": {
|
||||
"wan": 1,
|
||||
"pridisk": 0,
|
||||
"admin": 1,
|
||||
"lan": 0
|
||||
},
|
||||
"push": 0,
|
||||
"online": 1,
|
||||
"name": "Device2",
|
||||
"times": 0,
|
||||
"ip": [
|
||||
{
|
||||
"downspeed": "0",
|
||||
"online": "347325",
|
||||
"active": 1,
|
||||
"upspeed": "0",
|
||||
"ip": "192.168.0.3"
|
||||
}
|
||||
],
|
||||
"statistics": {
|
||||
"downspeed": "0",
|
||||
"online": "347325",
|
||||
"upspeed": "0"
|
||||
},
|
||||
"icon": "",
|
||||
"type": 0
|
||||
},
|
||||
],
|
||||
"code": 0
|
||||
}, 200)
|
||||
else:
|
||||
_LOGGER.debug('UNKNOWN ROUTE')
|
||||
|
||||
|
||||
class TestXiaomiDeviceScanner(unittest.TestCase):
|
||||
"""Xiaomi device scanner test class."""
|
||||
|
||||
def setUp(self):
|
||||
"""Initialize values for this testcase class."""
|
||||
self.hass = get_test_home_assistant()
|
||||
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
@mock.patch(
|
||||
'homeassistant.components.device_tracker.xiaomi.XiaomiDeviceScanner',
|
||||
return_value=mock.MagicMock())
|
||||
def test_config(self, xiaomi_mock):
|
||||
"""Testing minimal configuration."""
|
||||
config = {
|
||||
DOMAIN: xiaomi.PLATFORM_SCHEMA({
|
||||
CONF_PLATFORM: xiaomi.DOMAIN,
|
||||
CONF_HOST: '192.168.0.1',
|
||||
CONF_PASSWORD: 'passwordTest'
|
||||
})
|
||||
}
|
||||
xiaomi.get_scanner(self.hass, config)
|
||||
self.assertEqual(xiaomi_mock.call_count, 1)
|
||||
self.assertEqual(xiaomi_mock.call_args, mock.call(config[DOMAIN]))
|
||||
call_arg = xiaomi_mock.call_args[0][0]
|
||||
self.assertEqual(call_arg['username'], 'admin')
|
||||
self.assertEqual(call_arg['password'], 'passwordTest')
|
||||
self.assertEqual(call_arg['host'], '192.168.0.1')
|
||||
self.assertEqual(call_arg['platform'], 'device_tracker')
|
||||
|
||||
@mock.patch(
|
||||
'homeassistant.components.device_tracker.xiaomi.XiaomiDeviceScanner',
|
||||
return_value=mock.MagicMock())
|
||||
def test_config_full(self, xiaomi_mock):
|
||||
"""Testing full configuration."""
|
||||
config = {
|
||||
DOMAIN: xiaomi.PLATFORM_SCHEMA({
|
||||
CONF_PLATFORM: xiaomi.DOMAIN,
|
||||
CONF_HOST: '192.168.0.1',
|
||||
CONF_USERNAME: 'alternativeAdminName',
|
||||
CONF_PASSWORD: 'passwordTest'
|
||||
})
|
||||
}
|
||||
xiaomi.get_scanner(self.hass, config)
|
||||
self.assertEqual(xiaomi_mock.call_count, 1)
|
||||
self.assertEqual(xiaomi_mock.call_args, mock.call(config[DOMAIN]))
|
||||
call_arg = xiaomi_mock.call_args[0][0]
|
||||
self.assertEqual(call_arg['username'], 'alternativeAdminName')
|
||||
self.assertEqual(call_arg['password'], 'passwordTest')
|
||||
self.assertEqual(call_arg['host'], '192.168.0.1')
|
||||
self.assertEqual(call_arg['platform'], 'device_tracker')
|
||||
|
||||
@patch('requests.get', side_effect=mocked_requests)
|
||||
@patch('requests.post', side_effect=mocked_requests)
|
||||
def test_invalid_credential(self, mock_get, mock_post):
|
||||
""""Testing invalid credential handling."""
|
||||
config = {
|
||||
DOMAIN: xiaomi.PLATFORM_SCHEMA({
|
||||
CONF_PLATFORM: xiaomi.DOMAIN,
|
||||
CONF_HOST: '192.168.0.1',
|
||||
CONF_USERNAME: INVALID_USERNAME,
|
||||
CONF_PASSWORD: 'passwordTest'
|
||||
})
|
||||
}
|
||||
self.assertIsNone(get_scanner(self.hass, config))
|
||||
|
||||
@patch('requests.get', side_effect=mocked_requests)
|
||||
@patch('requests.post', side_effect=mocked_requests)
|
||||
def test_valid_credential(self, mock_get, mock_post):
|
||||
""""Testing valid refresh."""
|
||||
config = {
|
||||
DOMAIN: xiaomi.PLATFORM_SCHEMA({
|
||||
CONF_PLATFORM: xiaomi.DOMAIN,
|
||||
CONF_HOST: '192.168.0.1',
|
||||
CONF_USERNAME: 'admin',
|
||||
CONF_PASSWORD: 'passwordTest'
|
||||
})
|
||||
}
|
||||
scanner = get_scanner(self.hass, config)
|
||||
self.assertIsNotNone(scanner)
|
||||
self.assertEqual(2, len(scanner.scan_devices()))
|
||||
self.assertEqual("Device1",
|
||||
scanner.get_device_name("23:83:BF:F6:38:A0"))
|
||||
self.assertEqual("Device2",
|
||||
scanner.get_device_name("1D:98:EC:5E:D5:A6"))
|
||||
|
||||
@patch('requests.get', side_effect=mocked_requests)
|
||||
@patch('requests.post', side_effect=mocked_requests)
|
||||
def test_token_timed_out(self, mock_get, mock_post):
|
||||
""""Testing refresh with a timed out token.
|
||||
|
||||
New token is requested and list is downloaded a second time.
|
||||
"""
|
||||
config = {
|
||||
DOMAIN: xiaomi.PLATFORM_SCHEMA({
|
||||
CONF_PLATFORM: xiaomi.DOMAIN,
|
||||
CONF_HOST: '192.168.0.1',
|
||||
CONF_USERNAME: TOKEN_TIMEOUT_USERNAME,
|
||||
CONF_PASSWORD: 'passwordTest'
|
||||
})
|
||||
}
|
||||
scanner = get_scanner(self.hass, config)
|
||||
self.assertIsNotNone(scanner)
|
||||
self.assertEqual(2, len(scanner.scan_devices()))
|
||||
self.assertEqual("Device1",
|
||||
scanner.get_device_name("23:83:BF:F6:38:A0"))
|
||||
self.assertEqual("Device2",
|
||||
scanner.get_device_name("1D:98:EC:5E:D5:A6"))
|
||||
"""The tests for the Xiaomi router device tracker platform."""
|
||||
import logging
|
||||
import unittest
|
||||
from unittest import mock
|
||||
from unittest.mock import patch
|
||||
|
||||
import requests
|
||||
|
||||
from homeassistant.components.device_tracker import DOMAIN, xiaomi as xiaomi
|
||||
from homeassistant.components.device_tracker.xiaomi import get_scanner
|
||||
from homeassistant.const import (CONF_HOST, CONF_USERNAME, CONF_PASSWORD,
|
||||
CONF_PLATFORM)
|
||||
from tests.common import get_test_home_assistant
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
INVALID_USERNAME = 'bob'
|
||||
TOKEN_TIMEOUT_USERNAME = 'tok'
|
||||
URL_AUTHORIZE = 'http://192.168.0.1/cgi-bin/luci/api/xqsystem/login'
|
||||
URL_LIST_END = 'api/misystem/devicelist'
|
||||
|
||||
FIRST_CALL = True
|
||||
|
||||
|
||||
def mocked_requests(*args, **kwargs):
|
||||
"""Mock requests.get invocations."""
|
||||
class MockResponse:
|
||||
"""Class to represent a mocked response."""
|
||||
|
||||
def __init__(self, json_data, status_code):
|
||||
"""Initialize the mock response class."""
|
||||
self.json_data = json_data
|
||||
self.status_code = status_code
|
||||
|
||||
def json(self):
|
||||
"""Return the json of the response."""
|
||||
return self.json_data
|
||||
|
||||
@property
|
||||
def content(self):
|
||||
"""Return the content of the response."""
|
||||
return self.json()
|
||||
|
||||
def raise_for_status(self):
|
||||
"""Raise an HTTPError if status is not 200."""
|
||||
if self.status_code != 200:
|
||||
raise requests.HTTPError(self.status_code)
|
||||
|
||||
data = kwargs.get('data')
|
||||
global FIRST_CALL
|
||||
|
||||
if data and data.get('username', None) == INVALID_USERNAME:
|
||||
# deliver an invalid token
|
||||
return MockResponse({
|
||||
"code": "401",
|
||||
"msg": "Invalid token"
|
||||
}, 200)
|
||||
elif data and data.get('username', None) == TOKEN_TIMEOUT_USERNAME:
|
||||
# deliver an expired token
|
||||
return MockResponse({
|
||||
"url": "/cgi-bin/luci/;stok=ef5860/web/home",
|
||||
"token": "timedOut",
|
||||
"code": "0"
|
||||
}, 200)
|
||||
elif str(args[0]).startswith(URL_AUTHORIZE):
|
||||
# deliver an authorized token
|
||||
return MockResponse({
|
||||
"url": "/cgi-bin/luci/;stok=ef5860/web/home",
|
||||
"token": "ef5860",
|
||||
"code": "0"
|
||||
}, 200)
|
||||
elif str(args[0]).endswith("timedOut/" + URL_LIST_END) \
|
||||
and FIRST_CALL is True:
|
||||
FIRST_CALL = False
|
||||
# deliver an error when called with expired token
|
||||
return MockResponse({
|
||||
"code": "401",
|
||||
"msg": "Invalid token"
|
||||
}, 200)
|
||||
elif str(args[0]).endswith(URL_LIST_END):
|
||||
# deliver the device list
|
||||
return MockResponse({
|
||||
"mac": "1C:98:EC:0E:D5:A4",
|
||||
"list": [
|
||||
{
|
||||
"mac": "23:83:BF:F6:38:A0",
|
||||
"oname": "12255ff",
|
||||
"isap": 0,
|
||||
"parent": "",
|
||||
"authority": {
|
||||
"wan": 1,
|
||||
"pridisk": 0,
|
||||
"admin": 1,
|
||||
"lan": 0
|
||||
},
|
||||
"push": 0,
|
||||
"online": 1,
|
||||
"name": "Device1",
|
||||
"times": 0,
|
||||
"ip": [
|
||||
{
|
||||
"downspeed": "0",
|
||||
"online": "496957",
|
||||
"active": 1,
|
||||
"upspeed": "0",
|
||||
"ip": "192.168.0.25"
|
||||
}
|
||||
],
|
||||
"statistics": {
|
||||
"downspeed": "0",
|
||||
"online": "496957",
|
||||
"upspeed": "0"
|
||||
},
|
||||
"icon": "",
|
||||
"type": 1
|
||||
},
|
||||
{
|
||||
"mac": "1D:98:EC:5E:D5:A6",
|
||||
"oname": "CdddFG58",
|
||||
"isap": 0,
|
||||
"parent": "",
|
||||
"authority": {
|
||||
"wan": 1,
|
||||
"pridisk": 0,
|
||||
"admin": 1,
|
||||
"lan": 0
|
||||
},
|
||||
"push": 0,
|
||||
"online": 1,
|
||||
"name": "Device2",
|
||||
"times": 0,
|
||||
"ip": [
|
||||
{
|
||||
"downspeed": "0",
|
||||
"online": "347325",
|
||||
"active": 1,
|
||||
"upspeed": "0",
|
||||
"ip": "192.168.0.3"
|
||||
}
|
||||
],
|
||||
"statistics": {
|
||||
"downspeed": "0",
|
||||
"online": "347325",
|
||||
"upspeed": "0"
|
||||
},
|
||||
"icon": "",
|
||||
"type": 0
|
||||
},
|
||||
],
|
||||
"code": 0
|
||||
}, 200)
|
||||
else:
|
||||
_LOGGER.debug('UNKNOWN ROUTE')
|
||||
|
||||
|
||||
class TestXiaomiDeviceScanner(unittest.TestCase):
|
||||
"""Xiaomi device scanner test class."""
|
||||
|
||||
def setUp(self):
|
||||
"""Initialize values for this testcase class."""
|
||||
self.hass = get_test_home_assistant()
|
||||
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
@mock.patch(
|
||||
'homeassistant.components.device_tracker.xiaomi.XiaomiDeviceScanner',
|
||||
return_value=mock.MagicMock())
|
||||
def test_config(self, xiaomi_mock):
|
||||
"""Testing minimal configuration."""
|
||||
config = {
|
||||
DOMAIN: xiaomi.PLATFORM_SCHEMA({
|
||||
CONF_PLATFORM: xiaomi.DOMAIN,
|
||||
CONF_HOST: '192.168.0.1',
|
||||
CONF_PASSWORD: 'passwordTest'
|
||||
})
|
||||
}
|
||||
xiaomi.get_scanner(self.hass, config)
|
||||
self.assertEqual(xiaomi_mock.call_count, 1)
|
||||
self.assertEqual(xiaomi_mock.call_args, mock.call(config[DOMAIN]))
|
||||
call_arg = xiaomi_mock.call_args[0][0]
|
||||
self.assertEqual(call_arg['username'], 'admin')
|
||||
self.assertEqual(call_arg['password'], 'passwordTest')
|
||||
self.assertEqual(call_arg['host'], '192.168.0.1')
|
||||
self.assertEqual(call_arg['platform'], 'device_tracker')
|
||||
|
||||
@mock.patch(
|
||||
'homeassistant.components.device_tracker.xiaomi.XiaomiDeviceScanner',
|
||||
return_value=mock.MagicMock())
|
||||
def test_config_full(self, xiaomi_mock):
|
||||
"""Testing full configuration."""
|
||||
config = {
|
||||
DOMAIN: xiaomi.PLATFORM_SCHEMA({
|
||||
CONF_PLATFORM: xiaomi.DOMAIN,
|
||||
CONF_HOST: '192.168.0.1',
|
||||
CONF_USERNAME: 'alternativeAdminName',
|
||||
CONF_PASSWORD: 'passwordTest'
|
||||
})
|
||||
}
|
||||
xiaomi.get_scanner(self.hass, config)
|
||||
self.assertEqual(xiaomi_mock.call_count, 1)
|
||||
self.assertEqual(xiaomi_mock.call_args, mock.call(config[DOMAIN]))
|
||||
call_arg = xiaomi_mock.call_args[0][0]
|
||||
self.assertEqual(call_arg['username'], 'alternativeAdminName')
|
||||
self.assertEqual(call_arg['password'], 'passwordTest')
|
||||
self.assertEqual(call_arg['host'], '192.168.0.1')
|
||||
self.assertEqual(call_arg['platform'], 'device_tracker')
|
||||
|
||||
@patch('requests.get', side_effect=mocked_requests)
|
||||
@patch('requests.post', side_effect=mocked_requests)
|
||||
def test_invalid_credential(self, mock_get, mock_post):
|
||||
""""Testing invalid credential handling."""
|
||||
config = {
|
||||
DOMAIN: xiaomi.PLATFORM_SCHEMA({
|
||||
CONF_PLATFORM: xiaomi.DOMAIN,
|
||||
CONF_HOST: '192.168.0.1',
|
||||
CONF_USERNAME: INVALID_USERNAME,
|
||||
CONF_PASSWORD: 'passwordTest'
|
||||
})
|
||||
}
|
||||
self.assertIsNone(get_scanner(self.hass, config))
|
||||
|
||||
@patch('requests.get', side_effect=mocked_requests)
|
||||
@patch('requests.post', side_effect=mocked_requests)
|
||||
def test_valid_credential(self, mock_get, mock_post):
|
||||
""""Testing valid refresh."""
|
||||
config = {
|
||||
DOMAIN: xiaomi.PLATFORM_SCHEMA({
|
||||
CONF_PLATFORM: xiaomi.DOMAIN,
|
||||
CONF_HOST: '192.168.0.1',
|
||||
CONF_USERNAME: 'admin',
|
||||
CONF_PASSWORD: 'passwordTest'
|
||||
})
|
||||
}
|
||||
scanner = get_scanner(self.hass, config)
|
||||
self.assertIsNotNone(scanner)
|
||||
self.assertEqual(2, len(scanner.scan_devices()))
|
||||
self.assertEqual("Device1",
|
||||
scanner.get_device_name("23:83:BF:F6:38:A0"))
|
||||
self.assertEqual("Device2",
|
||||
scanner.get_device_name("1D:98:EC:5E:D5:A6"))
|
||||
|
||||
@patch('requests.get', side_effect=mocked_requests)
|
||||
@patch('requests.post', side_effect=mocked_requests)
|
||||
def test_token_timed_out(self, mock_get, mock_post):
|
||||
""""Testing refresh with a timed out token.
|
||||
|
||||
New token is requested and list is downloaded a second time.
|
||||
"""
|
||||
config = {
|
||||
DOMAIN: xiaomi.PLATFORM_SCHEMA({
|
||||
CONF_PLATFORM: xiaomi.DOMAIN,
|
||||
CONF_HOST: '192.168.0.1',
|
||||
CONF_USERNAME: TOKEN_TIMEOUT_USERNAME,
|
||||
CONF_PASSWORD: 'passwordTest'
|
||||
})
|
||||
}
|
||||
scanner = get_scanner(self.hass, config)
|
||||
self.assertIsNotNone(scanner)
|
||||
self.assertEqual(2, len(scanner.scan_devices()))
|
||||
self.assertEqual("Device1",
|
||||
scanner.get_device_name("23:83:BF:F6:38:A0"))
|
||||
self.assertEqual("Device2",
|
||||
scanner.get_device_name("1D:98:EC:5E:D5:A6"))
|
||||
|
@ -1,128 +1,128 @@
|
||||
"""Test the Emulated Hue component."""
|
||||
import json
|
||||
|
||||
from unittest.mock import patch, Mock, mock_open
|
||||
|
||||
from homeassistant.components.emulated_hue import Config, _LOGGER
|
||||
|
||||
|
||||
def test_config_google_home_entity_id_to_number():
|
||||
"""Test config adheres to the type."""
|
||||
conf = Config(Mock(), {
|
||||
'type': 'google_home'
|
||||
})
|
||||
|
||||
mop = mock_open(read_data=json.dumps({'1': 'light.test2'}))
|
||||
handle = mop()
|
||||
|
||||
with patch('homeassistant.util.json.open', mop, create=True):
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '2'
|
||||
assert handle.write.call_count == 1
|
||||
assert json.loads(handle.write.mock_calls[0][1][0]) == {
|
||||
'1': 'light.test2',
|
||||
'2': 'light.test',
|
||||
}
|
||||
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '2'
|
||||
assert handle.write.call_count == 1
|
||||
|
||||
number = conf.entity_id_to_number('light.test2')
|
||||
assert number == '1'
|
||||
assert handle.write.call_count == 1
|
||||
|
||||
entity_id = conf.number_to_entity_id('1')
|
||||
assert entity_id == 'light.test2'
|
||||
|
||||
|
||||
def test_config_google_home_entity_id_to_number_altered():
|
||||
"""Test config adheres to the type."""
|
||||
conf = Config(Mock(), {
|
||||
'type': 'google_home'
|
||||
})
|
||||
|
||||
mop = mock_open(read_data=json.dumps({'21': 'light.test2'}))
|
||||
handle = mop()
|
||||
|
||||
with patch('homeassistant.util.json.open', mop, create=True):
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '22'
|
||||
assert handle.write.call_count == 1
|
||||
assert json.loads(handle.write.mock_calls[0][1][0]) == {
|
||||
'21': 'light.test2',
|
||||
'22': 'light.test',
|
||||
}
|
||||
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '22'
|
||||
assert handle.write.call_count == 1
|
||||
|
||||
number = conf.entity_id_to_number('light.test2')
|
||||
assert number == '21'
|
||||
assert handle.write.call_count == 1
|
||||
|
||||
entity_id = conf.number_to_entity_id('21')
|
||||
assert entity_id == 'light.test2'
|
||||
|
||||
|
||||
def test_config_google_home_entity_id_to_number_empty():
|
||||
"""Test config adheres to the type."""
|
||||
conf = Config(Mock(), {
|
||||
'type': 'google_home'
|
||||
})
|
||||
|
||||
mop = mock_open(read_data='')
|
||||
handle = mop()
|
||||
|
||||
with patch('homeassistant.util.json.open', mop, create=True):
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '1'
|
||||
assert handle.write.call_count == 1
|
||||
assert json.loads(handle.write.mock_calls[0][1][0]) == {
|
||||
'1': 'light.test',
|
||||
}
|
||||
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '1'
|
||||
assert handle.write.call_count == 1
|
||||
|
||||
number = conf.entity_id_to_number('light.test2')
|
||||
assert number == '2'
|
||||
assert handle.write.call_count == 2
|
||||
|
||||
entity_id = conf.number_to_entity_id('2')
|
||||
assert entity_id == 'light.test2'
|
||||
|
||||
|
||||
def test_config_alexa_entity_id_to_number():
|
||||
"""Test config adheres to the type."""
|
||||
conf = Config(None, {
|
||||
'type': 'alexa'
|
||||
})
|
||||
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == 'light.test'
|
||||
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == 'light.test'
|
||||
|
||||
number = conf.entity_id_to_number('light.test2')
|
||||
assert number == 'light.test2'
|
||||
|
||||
entity_id = conf.number_to_entity_id('light.test')
|
||||
assert entity_id == 'light.test'
|
||||
|
||||
|
||||
def test_warning_config_google_home_listen_port():
|
||||
"""Test we warn when non-default port is used for Google Home."""
|
||||
with patch.object(_LOGGER, 'warning') as mock_warn:
|
||||
Config(None, {
|
||||
'type': 'google_home',
|
||||
'host_ip': '123.123.123.123',
|
||||
'listen_port': 8300
|
||||
})
|
||||
|
||||
assert mock_warn.called
|
||||
assert mock_warn.mock_calls[0][1][0] == \
|
||||
"When targeting Google Home, listening port has to be port 80"
|
||||
"""Test the Emulated Hue component."""
|
||||
import json
|
||||
|
||||
from unittest.mock import patch, Mock, mock_open
|
||||
|
||||
from homeassistant.components.emulated_hue import Config, _LOGGER
|
||||
|
||||
|
||||
def test_config_google_home_entity_id_to_number():
|
||||
"""Test config adheres to the type."""
|
||||
conf = Config(Mock(), {
|
||||
'type': 'google_home'
|
||||
})
|
||||
|
||||
mop = mock_open(read_data=json.dumps({'1': 'light.test2'}))
|
||||
handle = mop()
|
||||
|
||||
with patch('homeassistant.util.json.open', mop, create=True):
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '2'
|
||||
assert handle.write.call_count == 1
|
||||
assert json.loads(handle.write.mock_calls[0][1][0]) == {
|
||||
'1': 'light.test2',
|
||||
'2': 'light.test',
|
||||
}
|
||||
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '2'
|
||||
assert handle.write.call_count == 1
|
||||
|
||||
number = conf.entity_id_to_number('light.test2')
|
||||
assert number == '1'
|
||||
assert handle.write.call_count == 1
|
||||
|
||||
entity_id = conf.number_to_entity_id('1')
|
||||
assert entity_id == 'light.test2'
|
||||
|
||||
|
||||
def test_config_google_home_entity_id_to_number_altered():
|
||||
"""Test config adheres to the type."""
|
||||
conf = Config(Mock(), {
|
||||
'type': 'google_home'
|
||||
})
|
||||
|
||||
mop = mock_open(read_data=json.dumps({'21': 'light.test2'}))
|
||||
handle = mop()
|
||||
|
||||
with patch('homeassistant.util.json.open', mop, create=True):
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '22'
|
||||
assert handle.write.call_count == 1
|
||||
assert json.loads(handle.write.mock_calls[0][1][0]) == {
|
||||
'21': 'light.test2',
|
||||
'22': 'light.test',
|
||||
}
|
||||
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '22'
|
||||
assert handle.write.call_count == 1
|
||||
|
||||
number = conf.entity_id_to_number('light.test2')
|
||||
assert number == '21'
|
||||
assert handle.write.call_count == 1
|
||||
|
||||
entity_id = conf.number_to_entity_id('21')
|
||||
assert entity_id == 'light.test2'
|
||||
|
||||
|
||||
def test_config_google_home_entity_id_to_number_empty():
|
||||
"""Test config adheres to the type."""
|
||||
conf = Config(Mock(), {
|
||||
'type': 'google_home'
|
||||
})
|
||||
|
||||
mop = mock_open(read_data='')
|
||||
handle = mop()
|
||||
|
||||
with patch('homeassistant.util.json.open', mop, create=True):
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '1'
|
||||
assert handle.write.call_count == 1
|
||||
assert json.loads(handle.write.mock_calls[0][1][0]) == {
|
||||
'1': 'light.test',
|
||||
}
|
||||
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == '1'
|
||||
assert handle.write.call_count == 1
|
||||
|
||||
number = conf.entity_id_to_number('light.test2')
|
||||
assert number == '2'
|
||||
assert handle.write.call_count == 2
|
||||
|
||||
entity_id = conf.number_to_entity_id('2')
|
||||
assert entity_id == 'light.test2'
|
||||
|
||||
|
||||
def test_config_alexa_entity_id_to_number():
|
||||
"""Test config adheres to the type."""
|
||||
conf = Config(None, {
|
||||
'type': 'alexa'
|
||||
})
|
||||
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == 'light.test'
|
||||
|
||||
number = conf.entity_id_to_number('light.test')
|
||||
assert number == 'light.test'
|
||||
|
||||
number = conf.entity_id_to_number('light.test2')
|
||||
assert number == 'light.test2'
|
||||
|
||||
entity_id = conf.number_to_entity_id('light.test')
|
||||
assert entity_id == 'light.test'
|
||||
|
||||
|
||||
def test_warning_config_google_home_listen_port():
|
||||
"""Test we warn when non-default port is used for Google Home."""
|
||||
with patch.object(_LOGGER, 'warning') as mock_warn:
|
||||
Config(None, {
|
||||
'type': 'google_home',
|
||||
'host_ip': '123.123.123.123',
|
||||
'listen_port': 8300
|
||||
})
|
||||
|
||||
assert mock_warn.called
|
||||
assert mock_warn.mock_calls[0][1][0] == \
|
||||
"When targeting Google Home, listening port has to be port 80"
|
||||
|
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -1,216 +1,216 @@
|
||||
"""The tests for the hddtemp platform."""
|
||||
import socket
|
||||
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.setup import setup_component
|
||||
|
||||
from tests.common import get_test_home_assistant
|
||||
|
||||
VALID_CONFIG_MINIMAL = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_NAME = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'name': 'FooBar',
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_ONE_DISK = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'disks': [
|
||||
'/dev/sdd1',
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_WRONG_DISK = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'disks': [
|
||||
'/dev/sdx1',
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_MULTIPLE_DISKS = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'host': 'foobar.local',
|
||||
'disks': [
|
||||
'/dev/sda1',
|
||||
'/dev/sdb1',
|
||||
'/dev/sdc1',
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_HOST = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'host': 'alice.local',
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_HOST_UNREACHABLE = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'host': 'bob.local',
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class TelnetMock():
|
||||
"""Mock class for the telnetlib.Telnet object."""
|
||||
|
||||
def __init__(self, host, port, timeout=0):
|
||||
"""Initialize Telnet object."""
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.timeout = timeout
|
||||
self.sample_data = bytes('|/dev/sda1|WDC WD30EZRX-12DC0B0|29|C|' +
|
||||
'|/dev/sdb1|WDC WD15EADS-11P7B2|32|C|' +
|
||||
'|/dev/sdc1|WDC WD20EARX-22MMMB0|29|C|' +
|
||||
'|/dev/sdd1|WDC WD15EARS-00Z5B1|89|F|',
|
||||
'ascii')
|
||||
|
||||
def read_all(self):
|
||||
"""Return sample values."""
|
||||
if self.host == 'alice.local':
|
||||
raise ConnectionRefusedError
|
||||
elif self.host == 'bob.local':
|
||||
raise socket.gaierror
|
||||
else:
|
||||
return self.sample_data
|
||||
return None
|
||||
|
||||
|
||||
class TestHDDTempSensor(unittest.TestCase):
|
||||
"""Test the hddtemp sensor."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up things to run when tests begin."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self.config = VALID_CONFIG_ONE_DISK
|
||||
self.reference = {'/dev/sda1': {'device': '/dev/sda1',
|
||||
'temperature': '29',
|
||||
'unit_of_measurement': '°C',
|
||||
'model': 'WDC WD30EZRX-12DC0B0', },
|
||||
'/dev/sdb1': {'device': '/dev/sdb1',
|
||||
'temperature': '32',
|
||||
'unit_of_measurement': '°C',
|
||||
'model': 'WDC WD15EADS-11P7B2', },
|
||||
'/dev/sdc1': {'device': '/dev/sdc1',
|
||||
'temperature': '29',
|
||||
'unit_of_measurement': '°C',
|
||||
'model': 'WDC WD20EARX-22MMMB0', },
|
||||
'/dev/sdd1': {'device': '/dev/sdd1',
|
||||
'temperature': '32',
|
||||
'unit_of_measurement': '°C',
|
||||
'model': 'WDC WD15EARS-00Z5B1', }, }
|
||||
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_min_config(self):
|
||||
"""Test minimal hddtemp configuration."""
|
||||
assert setup_component(self.hass, 'sensor', VALID_CONFIG_MINIMAL)
|
||||
|
||||
entity = self.hass.states.all()[0].entity_id
|
||||
state = self.hass.states.get(entity)
|
||||
|
||||
reference = self.reference[state.attributes.get('device')]
|
||||
|
||||
self.assertEqual(state.state, reference['temperature'])
|
||||
self.assertEqual(state.attributes.get('device'), reference['device'])
|
||||
self.assertEqual(state.attributes.get('model'), reference['model'])
|
||||
self.assertEqual(state.attributes.get('unit_of_measurement'),
|
||||
reference['unit_of_measurement'])
|
||||
self.assertEqual(state.attributes.get('friendly_name'),
|
||||
'HD Temperature ' + reference['device'])
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_rename_config(self):
|
||||
"""Test hddtemp configuration with different name."""
|
||||
assert setup_component(self.hass, 'sensor', VALID_CONFIG_NAME)
|
||||
|
||||
entity = self.hass.states.all()[0].entity_id
|
||||
state = self.hass.states.get(entity)
|
||||
|
||||
reference = self.reference[state.attributes.get('device')]
|
||||
|
||||
self.assertEqual(state.attributes.get('friendly_name'),
|
||||
'FooBar ' + reference['device'])
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_one_disk(self):
|
||||
"""Test hddtemp one disk configuration."""
|
||||
assert setup_component(self.hass, 'sensor', VALID_CONFIG_ONE_DISK)
|
||||
|
||||
state = self.hass.states.get('sensor.hd_temperature_devsdd1')
|
||||
|
||||
reference = self.reference[state.attributes.get('device')]
|
||||
|
||||
self.assertEqual(state.state, reference['temperature'])
|
||||
self.assertEqual(state.attributes.get('device'), reference['device'])
|
||||
self.assertEqual(state.attributes.get('model'), reference['model'])
|
||||
self.assertEqual(state.attributes.get('unit_of_measurement'),
|
||||
reference['unit_of_measurement'])
|
||||
self.assertEqual(state.attributes.get('friendly_name'),
|
||||
'HD Temperature ' + reference['device'])
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_wrong_disk(self):
|
||||
"""Test hddtemp wrong disk configuration."""
|
||||
assert setup_component(self.hass, 'sensor', VALID_CONFIG_WRONG_DISK)
|
||||
|
||||
self.assertEqual(len(self.hass.states.all()), 1)
|
||||
state = self.hass.states.get('sensor.hd_temperature_devsdx1')
|
||||
self.assertEqual(state.attributes.get('friendly_name'),
|
||||
'HD Temperature ' + '/dev/sdx1')
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_multiple_disks(self):
|
||||
"""Test hddtemp multiple disk configuration."""
|
||||
assert setup_component(self.hass,
|
||||
'sensor', VALID_CONFIG_MULTIPLE_DISKS)
|
||||
|
||||
for sensor in ['sensor.hd_temperature_devsda1',
|
||||
'sensor.hd_temperature_devsdb1',
|
||||
'sensor.hd_temperature_devsdc1']:
|
||||
|
||||
state = self.hass.states.get(sensor)
|
||||
|
||||
reference = self.reference[state.attributes.get('device')]
|
||||
|
||||
self.assertEqual(state.state,
|
||||
reference['temperature'])
|
||||
self.assertEqual(state.attributes.get('device'),
|
||||
reference['device'])
|
||||
self.assertEqual(state.attributes.get('model'),
|
||||
reference['model'])
|
||||
self.assertEqual(state.attributes.get('unit_of_measurement'),
|
||||
reference['unit_of_measurement'])
|
||||
self.assertEqual(state.attributes.get('friendly_name'),
|
||||
'HD Temperature ' + reference['device'])
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_host_refused(self):
|
||||
"""Test hddtemp if host unreachable."""
|
||||
assert setup_component(self.hass, 'sensor', VALID_CONFIG_HOST)
|
||||
self.assertEqual(len(self.hass.states.all()), 0)
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_host_unreachable(self):
|
||||
"""Test hddtemp if host unreachable."""
|
||||
assert setup_component(self.hass, 'sensor',
|
||||
VALID_CONFIG_HOST_UNREACHABLE)
|
||||
self.assertEqual(len(self.hass.states.all()), 0)
|
||||
"""The tests for the hddtemp platform."""
|
||||
import socket
|
||||
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.setup import setup_component
|
||||
|
||||
from tests.common import get_test_home_assistant
|
||||
|
||||
VALID_CONFIG_MINIMAL = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_NAME = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'name': 'FooBar',
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_ONE_DISK = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'disks': [
|
||||
'/dev/sdd1',
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_WRONG_DISK = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'disks': [
|
||||
'/dev/sdx1',
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_MULTIPLE_DISKS = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'host': 'foobar.local',
|
||||
'disks': [
|
||||
'/dev/sda1',
|
||||
'/dev/sdb1',
|
||||
'/dev/sdc1',
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_HOST = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'host': 'alice.local',
|
||||
}
|
||||
}
|
||||
|
||||
VALID_CONFIG_HOST_UNREACHABLE = {
|
||||
'sensor': {
|
||||
'platform': 'hddtemp',
|
||||
'host': 'bob.local',
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class TelnetMock():
|
||||
"""Mock class for the telnetlib.Telnet object."""
|
||||
|
||||
def __init__(self, host, port, timeout=0):
|
||||
"""Initialize Telnet object."""
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.timeout = timeout
|
||||
self.sample_data = bytes('|/dev/sda1|WDC WD30EZRX-12DC0B0|29|C|' +
|
||||
'|/dev/sdb1|WDC WD15EADS-11P7B2|32|C|' +
|
||||
'|/dev/sdc1|WDC WD20EARX-22MMMB0|29|C|' +
|
||||
'|/dev/sdd1|WDC WD15EARS-00Z5B1|89|F|',
|
||||
'ascii')
|
||||
|
||||
def read_all(self):
|
||||
"""Return sample values."""
|
||||
if self.host == 'alice.local':
|
||||
raise ConnectionRefusedError
|
||||
elif self.host == 'bob.local':
|
||||
raise socket.gaierror
|
||||
else:
|
||||
return self.sample_data
|
||||
return None
|
||||
|
||||
|
||||
class TestHDDTempSensor(unittest.TestCase):
|
||||
"""Test the hddtemp sensor."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up things to run when tests begin."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self.config = VALID_CONFIG_ONE_DISK
|
||||
self.reference = {'/dev/sda1': {'device': '/dev/sda1',
|
||||
'temperature': '29',
|
||||
'unit_of_measurement': '°C',
|
||||
'model': 'WDC WD30EZRX-12DC0B0', },
|
||||
'/dev/sdb1': {'device': '/dev/sdb1',
|
||||
'temperature': '32',
|
||||
'unit_of_measurement': '°C',
|
||||
'model': 'WDC WD15EADS-11P7B2', },
|
||||
'/dev/sdc1': {'device': '/dev/sdc1',
|
||||
'temperature': '29',
|
||||
'unit_of_measurement': '°C',
|
||||
'model': 'WDC WD20EARX-22MMMB0', },
|
||||
'/dev/sdd1': {'device': '/dev/sdd1',
|
||||
'temperature': '32',
|
||||
'unit_of_measurement': '°C',
|
||||
'model': 'WDC WD15EARS-00Z5B1', }, }
|
||||
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_min_config(self):
|
||||
"""Test minimal hddtemp configuration."""
|
||||
assert setup_component(self.hass, 'sensor', VALID_CONFIG_MINIMAL)
|
||||
|
||||
entity = self.hass.states.all()[0].entity_id
|
||||
state = self.hass.states.get(entity)
|
||||
|
||||
reference = self.reference[state.attributes.get('device')]
|
||||
|
||||
self.assertEqual(state.state, reference['temperature'])
|
||||
self.assertEqual(state.attributes.get('device'), reference['device'])
|
||||
self.assertEqual(state.attributes.get('model'), reference['model'])
|
||||
self.assertEqual(state.attributes.get('unit_of_measurement'),
|
||||
reference['unit_of_measurement'])
|
||||
self.assertEqual(state.attributes.get('friendly_name'),
|
||||
'HD Temperature ' + reference['device'])
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_rename_config(self):
|
||||
"""Test hddtemp configuration with different name."""
|
||||
assert setup_component(self.hass, 'sensor', VALID_CONFIG_NAME)
|
||||
|
||||
entity = self.hass.states.all()[0].entity_id
|
||||
state = self.hass.states.get(entity)
|
||||
|
||||
reference = self.reference[state.attributes.get('device')]
|
||||
|
||||
self.assertEqual(state.attributes.get('friendly_name'),
|
||||
'FooBar ' + reference['device'])
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_one_disk(self):
|
||||
"""Test hddtemp one disk configuration."""
|
||||
assert setup_component(self.hass, 'sensor', VALID_CONFIG_ONE_DISK)
|
||||
|
||||
state = self.hass.states.get('sensor.hd_temperature_devsdd1')
|
||||
|
||||
reference = self.reference[state.attributes.get('device')]
|
||||
|
||||
self.assertEqual(state.state, reference['temperature'])
|
||||
self.assertEqual(state.attributes.get('device'), reference['device'])
|
||||
self.assertEqual(state.attributes.get('model'), reference['model'])
|
||||
self.assertEqual(state.attributes.get('unit_of_measurement'),
|
||||
reference['unit_of_measurement'])
|
||||
self.assertEqual(state.attributes.get('friendly_name'),
|
||||
'HD Temperature ' + reference['device'])
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_wrong_disk(self):
|
||||
"""Test hddtemp wrong disk configuration."""
|
||||
assert setup_component(self.hass, 'sensor', VALID_CONFIG_WRONG_DISK)
|
||||
|
||||
self.assertEqual(len(self.hass.states.all()), 1)
|
||||
state = self.hass.states.get('sensor.hd_temperature_devsdx1')
|
||||
self.assertEqual(state.attributes.get('friendly_name'),
|
||||
'HD Temperature ' + '/dev/sdx1')
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_multiple_disks(self):
|
||||
"""Test hddtemp multiple disk configuration."""
|
||||
assert setup_component(self.hass,
|
||||
'sensor', VALID_CONFIG_MULTIPLE_DISKS)
|
||||
|
||||
for sensor in ['sensor.hd_temperature_devsda1',
|
||||
'sensor.hd_temperature_devsdb1',
|
||||
'sensor.hd_temperature_devsdc1']:
|
||||
|
||||
state = self.hass.states.get(sensor)
|
||||
|
||||
reference = self.reference[state.attributes.get('device')]
|
||||
|
||||
self.assertEqual(state.state,
|
||||
reference['temperature'])
|
||||
self.assertEqual(state.attributes.get('device'),
|
||||
reference['device'])
|
||||
self.assertEqual(state.attributes.get('model'),
|
||||
reference['model'])
|
||||
self.assertEqual(state.attributes.get('unit_of_measurement'),
|
||||
reference['unit_of_measurement'])
|
||||
self.assertEqual(state.attributes.get('friendly_name'),
|
||||
'HD Temperature ' + reference['device'])
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_host_refused(self):
|
||||
"""Test hddtemp if host unreachable."""
|
||||
assert setup_component(self.hass, 'sensor', VALID_CONFIG_HOST)
|
||||
self.assertEqual(len(self.hass.states.all()), 0)
|
||||
|
||||
@patch('telnetlib.Telnet', new=TelnetMock)
|
||||
def test_hddtemp_host_unreachable(self):
|
||||
"""Test hddtemp if host unreachable."""
|
||||
assert setup_component(self.hass, 'sensor',
|
||||
VALID_CONFIG_HOST_UNREACHABLE)
|
||||
self.assertEqual(len(self.hass.states.all()), 0)
|
||||
|
@ -1,192 +1,192 @@
|
||||
"""The tests for the wake on lan switch platform."""
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.setup import setup_component
|
||||
from homeassistant.const import STATE_ON, STATE_OFF
|
||||
import homeassistant.components.switch as switch
|
||||
|
||||
from tests.common import get_test_home_assistant, mock_service
|
||||
|
||||
|
||||
TEST_STATE = None
|
||||
|
||||
|
||||
def send_magic_packet(*macs, **kwargs):
|
||||
"""Fake call for sending magic packets."""
|
||||
return
|
||||
|
||||
|
||||
def call(cmd, stdout, stderr):
|
||||
"""Return fake subprocess return codes."""
|
||||
if cmd[5] == 'validhostname' and TEST_STATE:
|
||||
return 0
|
||||
return 2
|
||||
|
||||
|
||||
def system():
|
||||
"""Fake system call to test the windows platform."""
|
||||
return 'Windows'
|
||||
|
||||
|
||||
class TestWOLSwitch(unittest.TestCase):
|
||||
"""Test the wol switch."""
|
||||
|
||||
def setUp(self):
|
||||
"""Setup things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
def test_valid_hostname(self):
|
||||
"""Test with valid hostname."""
|
||||
global TEST_STATE
|
||||
TEST_STATE = False
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
'host': 'validhostname',
|
||||
}
|
||||
}))
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
TEST_STATE = True
|
||||
|
||||
switch.turn_on(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
|
||||
switch.turn_off(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
@patch('platform.system', new=system)
|
||||
def test_valid_hostname_windows(self):
|
||||
"""Test with valid hostname on windows."""
|
||||
global TEST_STATE
|
||||
TEST_STATE = False
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
'host': 'validhostname',
|
||||
}
|
||||
}))
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
TEST_STATE = True
|
||||
|
||||
switch.turn_on(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
def test_minimal_config(self):
|
||||
"""Test with minimal config."""
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
}
|
||||
}))
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
def test_broadcast_config(self):
|
||||
"""Test with broadcast address config."""
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
'broadcast_address': '255.255.255.255',
|
||||
}
|
||||
}))
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
switch.turn_on(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
def test_off_script(self):
|
||||
"""Test with turn off script."""
|
||||
global TEST_STATE
|
||||
TEST_STATE = False
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
'host': 'validhostname',
|
||||
'turn_off': {
|
||||
'service': 'shell_command.turn_off_TARGET',
|
||||
},
|
||||
}
|
||||
}))
|
||||
calls = mock_service(self.hass, 'shell_command', 'turn_off_TARGET')
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
TEST_STATE = True
|
||||
|
||||
switch.turn_on(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
assert len(calls) == 0
|
||||
|
||||
TEST_STATE = False
|
||||
|
||||
switch.turn_off(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
assert len(calls) == 1
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
@patch('platform.system', new=system)
|
||||
def test_invalid_hostname_windows(self):
|
||||
"""Test with invalid hostname on windows."""
|
||||
global TEST_STATE
|
||||
TEST_STATE = False
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
'host': 'invalidhostname',
|
||||
}
|
||||
}))
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
TEST_STATE = True
|
||||
|
||||
switch.turn_on(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
"""The tests for the wake on lan switch platform."""
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.setup import setup_component
|
||||
from homeassistant.const import STATE_ON, STATE_OFF
|
||||
import homeassistant.components.switch as switch
|
||||
|
||||
from tests.common import get_test_home_assistant, mock_service
|
||||
|
||||
|
||||
TEST_STATE = None
|
||||
|
||||
|
||||
def send_magic_packet(*macs, **kwargs):
|
||||
"""Fake call for sending magic packets."""
|
||||
return
|
||||
|
||||
|
||||
def call(cmd, stdout, stderr):
|
||||
"""Return fake subprocess return codes."""
|
||||
if cmd[5] == 'validhostname' and TEST_STATE:
|
||||
return 0
|
||||
return 2
|
||||
|
||||
|
||||
def system():
|
||||
"""Fake system call to test the windows platform."""
|
||||
return 'Windows'
|
||||
|
||||
|
||||
class TestWOLSwitch(unittest.TestCase):
|
||||
"""Test the wol switch."""
|
||||
|
||||
def setUp(self):
|
||||
"""Setup things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
def test_valid_hostname(self):
|
||||
"""Test with valid hostname."""
|
||||
global TEST_STATE
|
||||
TEST_STATE = False
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
'host': 'validhostname',
|
||||
}
|
||||
}))
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
TEST_STATE = True
|
||||
|
||||
switch.turn_on(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
|
||||
switch.turn_off(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
@patch('platform.system', new=system)
|
||||
def test_valid_hostname_windows(self):
|
||||
"""Test with valid hostname on windows."""
|
||||
global TEST_STATE
|
||||
TEST_STATE = False
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
'host': 'validhostname',
|
||||
}
|
||||
}))
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
TEST_STATE = True
|
||||
|
||||
switch.turn_on(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
def test_minimal_config(self):
|
||||
"""Test with minimal config."""
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
}
|
||||
}))
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
def test_broadcast_config(self):
|
||||
"""Test with broadcast address config."""
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
'broadcast_address': '255.255.255.255',
|
||||
}
|
||||
}))
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
switch.turn_on(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
def test_off_script(self):
|
||||
"""Test with turn off script."""
|
||||
global TEST_STATE
|
||||
TEST_STATE = False
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
'host': 'validhostname',
|
||||
'turn_off': {
|
||||
'service': 'shell_command.turn_off_TARGET',
|
||||
},
|
||||
}
|
||||
}))
|
||||
calls = mock_service(self.hass, 'shell_command', 'turn_off_TARGET')
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
TEST_STATE = True
|
||||
|
||||
switch.turn_on(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_ON, state.state)
|
||||
assert len(calls) == 0
|
||||
|
||||
TEST_STATE = False
|
||||
|
||||
switch.turn_off(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
assert len(calls) == 1
|
||||
|
||||
@patch('wakeonlan.send_magic_packet', new=send_magic_packet)
|
||||
@patch('subprocess.call', new=call)
|
||||
@patch('platform.system', new=system)
|
||||
def test_invalid_hostname_windows(self):
|
||||
"""Test with invalid hostname on windows."""
|
||||
global TEST_STATE
|
||||
TEST_STATE = False
|
||||
self.assertTrue(setup_component(self.hass, switch.DOMAIN, {
|
||||
'switch': {
|
||||
'platform': 'wake_on_lan',
|
||||
'mac_address': '00-01-02-03-04-05',
|
||||
'host': 'invalidhostname',
|
||||
}
|
||||
}))
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
||||
TEST_STATE = True
|
||||
|
||||
switch.turn_on(self.hass, 'switch.wake_on_lan')
|
||||
self.hass.block_till_done()
|
||||
|
||||
state = self.hass.states.get('switch.wake_on_lan')
|
||||
self.assertEqual(STATE_OFF, state.state)
|
||||
|
Loading…
x
Reference in New Issue
Block a user