Add platform and sensors for Vultr VPS (#9928)

* Initial commit of Vultr components

Have a working Vultr hub and binary sensor which pulls down the
following attributes of your VPS:
 - Date created
 - Subscription id (server id)
 - Cost per month (in US$)
 - Operating System installed
 - IPv4 address
 - label (human readable name)
 - region
 - number of vcpus
 - which storage package chosen
 - IPV6 address (if applicable)
 - RAM amount

Working next on sensor and then testing / coverage.

* Added Vultr sensor for pending charges and current bandwidth. Refactored binary_sensor and hub too

* Corrected is_on bases

* Added basic tests for Vultr binary & platform

* Updated require files

* Changing test fixture to highlight different cases

* Written basic test for sensor.vultr

* Resolved linting errors and broken test

* Increase test coverage and corrected docs

* Resolved hound issues

* Revert back negative binary test

* Another hound resolve

* Refactoring and adding is switch, moving over to vultr branch

* Made Vultr components more resiliant to invalid configs

* Added negetive test for vultr binary sensor

* Added better testing of vultr sensor

* Resolved vultr platform test affecting subsequent vultr tests

* Moving VULTR components to single use design

* Added in sensor name config

* Added missing sensors var

* Resolved init data setting of sensors, added in name conf to switch

* Made the Vultr component more resiliant to startup failure with better alerting

* Various Vultr component changes

- Refactored sensor, binary_sensor, and switch to reference one subscription
- Renamed CURRENT_BANDWIDTH_GB monitored condition to CURRENT_BANDWIDTH_USED
- Improved test coverage

* Resolved local tox linting issue

* Added more testing for Vultr switch

* Improved test coverage for Vultr components

* Made PR comment changes to vultr binary sensor

* Made PR comment changes to Vultr sensor

* resolved PR comments for Vultr Switch

* Resolved vultr sensor name and improved tests

* Improved Vultr switch testing (default name formatting)

* Removed vultr hub failure checking
This commit is contained in:
Adam Cooper 2017-11-05 13:10:14 +00:00 committed by Fabian Affolter
parent a5d5f3f727
commit 72ce9ec321
13 changed files with 1139 additions and 1 deletions

View file

@ -0,0 +1,103 @@
"""
Support for monitoring the state of Vultr subscriptions (VPS).
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/binary_sensor.vultr/
"""
import logging
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.const import CONF_NAME
from homeassistant.components.binary_sensor import (
BinarySensorDevice, PLATFORM_SCHEMA)
from homeassistant.components.vultr import (
CONF_SUBSCRIPTION, ATTR_AUTO_BACKUPS, ATTR_ALLOWED_BANDWIDTH,
ATTR_CREATED_AT, ATTR_SUBSCRIPTION_ID, ATTR_SUBSCRIPTION_NAME,
ATTR_IPV4_ADDRESS, ATTR_IPV6_ADDRESS, ATTR_MEMORY, ATTR_DISK,
ATTR_COST_PER_MONTH, ATTR_OS, ATTR_REGION, ATTR_VCPUS, DATA_VULTR)
_LOGGER = logging.getLogger(__name__)
DEFAULT_DEVICE_CLASS = 'power'
DEFAULT_NAME = 'Vultr {}'
DEPENDENCIES = ['vultr']
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_SUBSCRIPTION): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Vultr subscription (server) sensor."""
vultr = hass.data[DATA_VULTR]
subscription = config.get(CONF_SUBSCRIPTION)
name = config.get(CONF_NAME)
if subscription not in vultr.data:
_LOGGER.error("Subscription %s not found", subscription)
return False
add_devices([VultrBinarySensor(vultr, subscription, name)], True)
class VultrBinarySensor(BinarySensorDevice):
"""Representation of a Vultr subscription sensor."""
def __init__(self, vultr, subscription, name):
"""Initialize a new Vultr sensor."""
self._vultr = vultr
self._name = name
self.subscription = subscription
self.data = None
@property
def name(self):
"""Return the name of the sensor."""
try:
return self._name.format(self.data['label'])
except (KeyError, TypeError):
return self._name
@property
def icon(self):
"""Return the icon of this server."""
return 'mdi:server' if self.is_on else 'mdi:server-off'
@property
def is_on(self):
"""Return true if the binary sensor is on."""
return self.data['power_status'] == 'running'
@property
def device_class(self):
"""Return the class of this sensor."""
return DEFAULT_DEVICE_CLASS
@property
def device_state_attributes(self):
"""Return the state attributes of the Vultr subscription."""
return {
ATTR_ALLOWED_BANDWIDTH: self.data.get('allowed_bandwidth_gb'),
ATTR_AUTO_BACKUPS: self.data.get('auto_backups'),
ATTR_COST_PER_MONTH: self.data.get('cost_per_month'),
ATTR_CREATED_AT: self.data.get('date_created'),
ATTR_DISK: self.data.get('disk'),
ATTR_IPV4_ADDRESS: self.data.get('main_ip'),
ATTR_IPV6_ADDRESS: self.data.get('v6_main_ip'),
ATTR_MEMORY: self.data.get('ram'),
ATTR_OS: self.data.get('os'),
ATTR_REGION: self.data.get('location'),
ATTR_SUBSCRIPTION_ID: self.data.get('SUBID'),
ATTR_SUBSCRIPTION_NAME: self.data.get('label'),
ATTR_VCPUS: self.data.get('vcpu_count')
}
def update(self):
"""Update state of sensor."""
self._vultr.update()
self.data = self._vultr.data[self.subscription]

View file

@ -0,0 +1,115 @@
"""
Support for monitoring the state of Vultr Subscriptions.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/sensor.vultr/
"""
import logging
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.const import (
CONF_MONITORED_CONDITIONS, CONF_NAME)
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.components.vultr import (
CONF_SUBSCRIPTION, ATTR_CURRENT_BANDWIDTH_USED, ATTR_PENDING_CHARGES,
DATA_VULTR)
# Name defaults to {subscription label} {sensor name}
DEFAULT_NAME = 'Vultr {} {}'
DEPENDENCIES = ['vultr']
_LOGGER = logging.getLogger(__name__)
# Monitored conditions: name, units, icon
MONITORED_CONDITIONS = {
ATTR_CURRENT_BANDWIDTH_USED: ['Current Bandwidth Used', 'GB',
'mdi:chart-histogram'],
ATTR_PENDING_CHARGES: ['Pending Charges', 'US$',
'mdi:currency-usd']
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_SUBSCRIPTION): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_MONITORED_CONDITIONS, default=MONITORED_CONDITIONS):
vol.All(cv.ensure_list, [vol.In(MONITORED_CONDITIONS)])
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Vultr subscription (server) sensor."""
vultr = hass.data[DATA_VULTR]
subscription = config.get(CONF_SUBSCRIPTION)
name = config.get(CONF_NAME)
monitored_conditions = config.get(CONF_MONITORED_CONDITIONS)
if subscription not in vultr.data:
_LOGGER.error("Subscription %s not found", subscription)
return False
sensors = []
for condition in monitored_conditions:
sensors.append(VultrSensor(vultr,
subscription,
condition,
name))
add_devices(sensors, True)
class VultrSensor(Entity):
"""Representation of a Vultr subscription sensor."""
def __init__(self, vultr, subscription, condition, name):
"""Initialize a new Vultr sensor."""
self._vultr = vultr
self._condition = condition
self._name = name
self.subscription = subscription
self.data = None
condition_info = MONITORED_CONDITIONS[condition]
self._condition_name = condition_info[0]
self._units = condition_info[1]
self._icon = condition_info[2]
@property
def name(self):
"""Return the name of the sensor."""
try:
return self._name.format(self._condition_name)
except IndexError: # name contains more {} than fulfilled
try:
return self._name.format(self.data['label'],
self._condition_name)
except (KeyError, TypeError): # label key missing or data is None
return self._name
@property
def icon(self):
"""Icon used in the frontend if any."""
return self._icon
@property
def unit_of_measurement(self):
"""The unit of measurement to present the value in."""
return self._units
@property
def state(self):
"""Return the value of this given sensor type."""
try:
return round(float(self.data.get(self._condition)), 2)
except (TypeError, ValueError):
return self.data.get(self._condition)
def update(self):
"""Update state of sensor."""
self._vultr.update()
self.data = self._vultr.data[self.subscription]

View file

@ -0,0 +1,106 @@
"""
Support for interacting with Vultr subscriptions.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/switch.vultr/
"""
import logging
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.const import CONF_NAME
from homeassistant.components.switch import (SwitchDevice, PLATFORM_SCHEMA)
from homeassistant.components.vultr import (
CONF_SUBSCRIPTION, ATTR_AUTO_BACKUPS, ATTR_ALLOWED_BANDWIDTH,
ATTR_CREATED_AT, ATTR_SUBSCRIPTION_ID, ATTR_SUBSCRIPTION_NAME,
ATTR_IPV4_ADDRESS, ATTR_IPV6_ADDRESS, ATTR_MEMORY, ATTR_DISK,
ATTR_COST_PER_MONTH, ATTR_OS, ATTR_REGION, ATTR_VCPUS, DATA_VULTR)
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = 'Vultr {}'
DEPENDENCIES = ['vultr']
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_SUBSCRIPTION): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Vultr subscription switch."""
vultr = hass.data[DATA_VULTR]
subscription = config.get(CONF_SUBSCRIPTION)
name = config.get(CONF_NAME)
if subscription not in vultr.data:
_LOGGER.error("Subscription %s not found", subscription)
return False
add_devices([VultrSwitch(vultr, subscription, name)], True)
class VultrSwitch(SwitchDevice):
"""Representation of a Vultr subscription switch."""
def __init__(self, vultr, subscription, name):
"""Initialize a new Vultr switch."""
self._vultr = vultr
self._name = name
self.subscription = subscription
self.data = None
@property
def name(self):
"""Return the name of the switch."""
try:
return self._name.format(self.data['label'])
except (TypeError, KeyError):
return self._name
@property
def is_on(self):
"""Return true if switch is on."""
return self.data['power_status'] == 'running'
@property
def icon(self):
"""Return the icon of this server."""
return 'mdi:server' if self.is_on else 'mdi:server-off'
@property
def device_state_attributes(self):
"""Return the state attributes of the Vultr subscription."""
return {
ATTR_ALLOWED_BANDWIDTH: self.data.get('allowed_bandwidth_gb'),
ATTR_AUTO_BACKUPS: self.data.get('auto_backups'),
ATTR_COST_PER_MONTH: self.data.get('cost_per_month'),
ATTR_CREATED_AT: self.data.get('date_created'),
ATTR_DISK: self.data.get('disk'),
ATTR_IPV4_ADDRESS: self.data.get('main_ip'),
ATTR_IPV6_ADDRESS: self.data.get('v6_main_ip'),
ATTR_MEMORY: self.data.get('ram'),
ATTR_OS: self.data.get('os'),
ATTR_REGION: self.data.get('location'),
ATTR_SUBSCRIPTION_ID: self.data.get('SUBID'),
ATTR_SUBSCRIPTION_NAME: self.data.get('label'),
ATTR_VCPUS: self.data.get('vcpu_count'),
}
def turn_on(self):
"""Boot-up the subscription."""
if self.data['power_status'] != 'running':
self._vultr.start(self.subscription)
def turn_off(self):
"""Halt the subscription."""
if self.data['power_status'] == 'running':
self._vultr.halt(self.subscription)
def update(self):
"""Get the latest data from the device and update the data."""
self._vultr.update()
self.data = self._vultr.data[self.subscription]

View file

@ -0,0 +1,105 @@
"""
Support for Vultr.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/vultr/
"""
import logging
from datetime import timedelta
import voluptuous as vol
from homeassistant.const import CONF_API_KEY
from homeassistant.util import Throttle
import homeassistant.helpers.config_validation as cv
REQUIREMENTS = ['vultr==0.1.2']
_LOGGER = logging.getLogger(__name__)
ATTR_AUTO_BACKUPS = 'auto_backups'
ATTR_ALLOWED_BANDWIDTH = 'allowed_bandwidth_gb'
ATTR_COST_PER_MONTH = 'cost_per_month'
ATTR_CURRENT_BANDWIDTH_USED = 'current_bandwidth_gb'
ATTR_CREATED_AT = 'created_at'
ATTR_DISK = 'disk'
ATTR_SUBSCRIPTION_ID = 'subid'
ATTR_SUBSCRIPTION_NAME = 'label'
ATTR_IPV4_ADDRESS = 'ipv4_address'
ATTR_IPV6_ADDRESS = 'ipv6_address'
ATTR_MEMORY = 'memory'
ATTR_OS = 'os'
ATTR_PENDING_CHARGES = 'pending_charges'
ATTR_REGION = 'region'
ATTR_VCPUS = 'vcpus'
CONF_SUBSCRIPTION = 'subscription'
DATA_VULTR = 'data_vultr'
DOMAIN = 'vultr'
NOTIFICATION_ID = 'vultr_notification'
NOTIFICATION_TITLE = 'Vultr Setup'
VULTR_PLATFORMS = ['binary_sensor', 'sensor', 'switch']
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60)
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Required(CONF_API_KEY): cv.string,
}),
}, extra=vol.ALLOW_EXTRA)
def setup(hass, config):
"""Set up the Vultr component."""
api_key = config[DOMAIN].get(CONF_API_KEY)
vultr = Vultr(api_key)
try:
vultr.update()
except RuntimeError as ex:
_LOGGER.error("Failed to make update API request because: %s",
ex)
hass.components.persistent_notification.create(
'Error: {}'
''.format(ex),
title=NOTIFICATION_TITLE,
notification_id=NOTIFICATION_ID)
return False
hass.data[DATA_VULTR] = vultr
return True
class Vultr(object):
"""Handle all communication with the Vultr API."""
def __init__(self, api_key):
"""Initialize the Vultr connection."""
from vultr import Vultr as VultrAPI
self._api_key = api_key
self.data = None
self.api = VultrAPI(self._api_key)
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Use the data from Vultr API."""
self.data = self.api.server_list()
def _force_update(self):
"""Use the data from Vultr API."""
self.data = self.api.server_list()
def halt(self, subscription):
"""Halt a subscription (hard power off)."""
self.api.server_halt(subscription)
self._force_update()
def start(self, subscription):
"""Start a subscription."""
self.api.server_start(subscription)
self._force_update()

View file

@ -1101,6 +1101,9 @@ vsure==1.3.7
# homeassistant.components.sensor.vasttrafik
vtjp==0.1.14
# homeassistant.components.vultr
vultr==0.1.2
# homeassistant.components.wake_on_lan
# homeassistant.components.media_player.panasonic_viera
# homeassistant.components.media_player.samsungtv

View file

@ -164,6 +164,9 @@ statsd==3.2.1
# homeassistant.components.camera.uvc
uvcclient==0.10.1
# homeassistant.components.vultr
vultr==0.1.2
# homeassistant.components.wake_on_lan
# homeassistant.components.media_player.panasonic_viera
# homeassistant.components.media_player.samsungtv

View file

@ -82,7 +82,8 @@ TEST_REQUIREMENTS = (
'warrant',
'yahoo-finance',
'pythonwhois',
'wakeonlan'
'wakeonlan',
'vultr'
)
IGNORE_PACKAGES = (

View file

@ -0,0 +1,165 @@
"""Test the Vultr binary sensor platform."""
import unittest
import requests_mock
import pytest
import voluptuous as vol
from components.binary_sensor import vultr
from components import vultr as base_vultr
from components.vultr import (
ATTR_ALLOWED_BANDWIDTH, ATTR_AUTO_BACKUPS, ATTR_IPV4_ADDRESS,
ATTR_COST_PER_MONTH, ATTR_CREATED_AT, ATTR_SUBSCRIPTION_ID,
CONF_SUBSCRIPTION)
from homeassistant.const import (
CONF_PLATFORM, CONF_NAME)
from tests.components.test_vultr import VALID_CONFIG
from tests.common import (
get_test_home_assistant, load_fixture)
class TestVultrBinarySensorSetup(unittest.TestCase):
"""Test the Vultr binary sensor platform."""
DEVICES = []
def add_devices(self, devices, action):
"""Mock add devices."""
for device in devices:
self.DEVICES.append(device)
def setUp(self):
"""Init values for this testcase class."""
self.hass = get_test_home_assistant()
self.configs = [
{
CONF_SUBSCRIPTION: '576965',
CONF_NAME: "A Server"
},
{
CONF_SUBSCRIPTION: '123456',
CONF_NAME: "Failed Server"
},
{
CONF_SUBSCRIPTION: '555555',
CONF_NAME: vultr.DEFAULT_NAME
}
]
def tearDown(self):
"""Stop our started services."""
self.hass.stop()
def test_failed_hub(self):
"""Test a hub setup failure."""
base_vultr.setup(self.hass, VALID_CONFIG)
@requests_mock.Mocker()
def test_binary_sensor(self, mock):
"""Test successful instance."""
mock.get(
'https://api.vultr.com/v1/account/info?api_key=ABCDEFG1234567',
text=load_fixture('vultr_account_info.json'))
mock.get(
'https://api.vultr.com/v1/server/list?api_key=ABCDEFG1234567',
text=load_fixture('vultr_server_list.json'))
# Setup hub
base_vultr.setup(self.hass, VALID_CONFIG)
# Setup each of our test configs
for config in self.configs:
vultr.setup_platform(self.hass,
config,
self.add_devices,
None)
self.assertEqual(len(self.DEVICES), 3)
for device in self.DEVICES:
# Test pre data retieval
if device.subscription == '555555':
self.assertEqual('Vultr {}', device.name)
device.update()
device_attrs = device.device_state_attributes
if device.subscription == '555555':
self.assertEqual('Vultr Another Server', device.name)
if device.name == 'A Server':
self.assertEqual(True, device.is_on)
self.assertEqual('power', device.device_class)
self.assertEqual('on', device.state)
self.assertEqual('mdi:server', device.icon)
self.assertEqual('1000',
device_attrs[ATTR_ALLOWED_BANDWIDTH])
self.assertEqual('yes',
device_attrs[ATTR_AUTO_BACKUPS])
self.assertEqual('123.123.123.123',
device_attrs[ATTR_IPV4_ADDRESS])
self.assertEqual('10.05',
device_attrs[ATTR_COST_PER_MONTH])
self.assertEqual('2013-12-19 14:45:41',
device_attrs[ATTR_CREATED_AT])
self.assertEqual('576965',
device_attrs[ATTR_SUBSCRIPTION_ID])
elif device.name == 'Failed Server':
self.assertEqual(False, device.is_on)
self.assertEqual('off', device.state)
self.assertEqual('mdi:server-off', device.icon)
self.assertEqual('1000',
device_attrs[ATTR_ALLOWED_BANDWIDTH])
self.assertEqual('no',
device_attrs[ATTR_AUTO_BACKUPS])
self.assertEqual('192.168.100.50',
device_attrs[ATTR_IPV4_ADDRESS])
self.assertEqual('73.25',
device_attrs[ATTR_COST_PER_MONTH])
self.assertEqual('2014-10-13 14:45:41',
device_attrs[ATTR_CREATED_AT])
self.assertEqual('123456',
device_attrs[ATTR_SUBSCRIPTION_ID])
def test_invalid_sensor_config(self):
"""Test config type failures."""
with pytest.raises(vol.Invalid): # No subs
vultr.PLATFORM_SCHEMA({
CONF_PLATFORM: base_vultr.DOMAIN,
})
@requests_mock.Mocker()
def test_invalid_sensors(self, mock):
"""Test the VultrBinarySensor fails."""
mock.get(
'https://api.vultr.com/v1/account/info?api_key=ABCDEFG1234567',
text=load_fixture('vultr_account_info.json'))
mock.get(
'https://api.vultr.com/v1/server/list?api_key=ABCDEFG1234567',
text=load_fixture('vultr_server_list.json'))
base_vultr.setup(self.hass, VALID_CONFIG)
bad_conf = {} # No subscription
no_subs_setup = vultr.setup_platform(self.hass,
bad_conf,
self.add_devices,
None)
self.assertFalse(no_subs_setup)
bad_conf = {
CONF_NAME: "Missing Server",
CONF_SUBSCRIPTION: '555555'
} # Sub not associated with API key (not in server_list)
wrong_subs_setup = vultr.setup_platform(self.hass,
bad_conf,
self.add_devices,
None)
self.assertFalse(wrong_subs_setup)

View file

@ -0,0 +1,165 @@
"""The tests for the Vultr sensor platform."""
import pytest
import unittest
import requests_mock
import voluptuous as vol
from components.sensor import vultr
from components import vultr as base_vultr
from components.vultr import CONF_SUBSCRIPTION
from homeassistant.const import (
CONF_NAME, CONF_MONITORED_CONDITIONS, CONF_PLATFORM)
from tests.components.test_vultr import VALID_CONFIG
from tests.common import (
get_test_home_assistant, load_fixture)
class TestVultrSensorSetup(unittest.TestCase):
"""Test the Vultr platform."""
DEVICES = []
def add_devices(self, devices, action):
"""Mock add devices."""
for device in devices:
self.DEVICES.append(device)
def setUp(self):
"""Initialize values for this testcase class."""
self.hass = get_test_home_assistant()
self.configs = [
{
CONF_NAME: vultr.DEFAULT_NAME,
CONF_SUBSCRIPTION: '576965',
CONF_MONITORED_CONDITIONS: vultr.MONITORED_CONDITIONS
},
{
CONF_NAME: 'Server {}',
CONF_SUBSCRIPTION: '123456',
CONF_MONITORED_CONDITIONS: vultr.MONITORED_CONDITIONS
},
{
CONF_NAME: 'VPS Charges',
CONF_SUBSCRIPTION: '555555',
CONF_MONITORED_CONDITIONS: [
'pending_charges'
]
}
]
def tearDown(self):
"""Stop everything that was started."""
self.hass.stop()
@requests_mock.Mocker()
def test_sensor(self, mock):
"""Test the Vultr sensor class and methods."""
mock.get(
'https://api.vultr.com/v1/account/info?api_key=ABCDEFG1234567',
text=load_fixture('vultr_account_info.json'))
mock.get(
'https://api.vultr.com/v1/server/list?api_key=ABCDEFG1234567',
text=load_fixture('vultr_server_list.json'))
base_vultr.setup(self.hass, VALID_CONFIG)
for config in self.configs:
setup = vultr.setup_platform(self.hass,
config,
self.add_devices,
None)
self.assertIsNone(setup)
self.assertEqual(5, len(self.DEVICES))
tested = 0
for device in self.DEVICES:
# Test pre update
if device.subscription == '576965':
self.assertEqual(vultr.DEFAULT_NAME, device.name)
device.update()
if device.unit_of_measurement == 'GB': # Test Bandwidth Used
if device.subscription == '576965':
self.assertEqual(
'Vultr my new server Current Bandwidth Used',
device.name)
self.assertEqual('mdi:chart-histogram', device.icon)
self.assertEqual(131.51, device.state)
self.assertEqual('mdi:chart-histogram', device.icon)
tested += 1
elif device.subscription == '123456':
self.assertEqual('Server Current Bandwidth Used',
device.name)
self.assertEqual(957.46, device.state)
tested += 1
elif device.unit_of_measurement == 'US$': # Test Pending Charges
if device.subscription == '576965': # Default 'Vultr {} {}'
self.assertEqual('Vultr my new server Pending Charges',
device.name)
self.assertEqual('mdi:currency-usd', device.icon)
self.assertEqual(46.67, device.state)
self.assertEqual('mdi:currency-usd', device.icon)
tested += 1
elif device.subscription == '123456': # Custom name with 1 {}
self.assertEqual('Server Pending Charges', device.name)
self.assertEqual('not a number', device.state)
tested += 1
elif device.subscription == '555555': # No {} in name
self.assertEqual('VPS Charges', device.name)
self.assertEqual(5.45, device.state)
tested += 1
self.assertEqual(tested, 5)
def test_invalid_sensor_config(self):
"""Test config type failures."""
with pytest.raises(vol.Invalid): # No subscription
vultr.PLATFORM_SCHEMA({
CONF_PLATFORM: base_vultr.DOMAIN,
CONF_MONITORED_CONDITIONS: vultr.MONITORED_CONDITIONS
})
with pytest.raises(vol.Invalid): # Bad monitored_conditions
vultr.PLATFORM_SCHEMA({
CONF_PLATFORM: base_vultr.DOMAIN,
CONF_SUBSCRIPTION: '123456',
CONF_MONITORED_CONDITIONS: [
'non-existent-condition',
]
})
@requests_mock.Mocker()
def test_invalid_sensors(self, mock):
"""Test the VultrSensor fails."""
mock.get(
'https://api.vultr.com/v1/account/info?api_key=ABCDEFG1234567',
text=load_fixture('vultr_account_info.json'))
mock.get(
'https://api.vultr.com/v1/server/list?api_key=ABCDEFG1234567',
text=load_fixture('vultr_server_list.json'))
base_vultr.setup(self.hass, VALID_CONFIG)
bad_conf = {
CONF_MONITORED_CONDITIONS: vultr.MONITORED_CONDITIONS,
} # No subs at all
no_sub_setup = vultr.setup_platform(self.hass,
bad_conf,
self.add_devices,
None)
self.assertIsNotNone(no_sub_setup)
self.assertEqual(0, len(self.DEVICES))

View file

@ -0,0 +1,201 @@
"""Test the Vultr switch platform."""
import unittest
import requests_mock
import pytest
import voluptuous as vol
from components.switch import vultr
from components import vultr as base_vultr
from components.vultr import (
ATTR_ALLOWED_BANDWIDTH, ATTR_AUTO_BACKUPS, ATTR_IPV4_ADDRESS,
ATTR_COST_PER_MONTH, ATTR_CREATED_AT, ATTR_SUBSCRIPTION_ID,
CONF_SUBSCRIPTION)
from homeassistant.const import (
CONF_PLATFORM, CONF_NAME)
from tests.components.test_vultr import VALID_CONFIG
from tests.common import (
get_test_home_assistant, load_fixture)
class TestVultrSwitchSetup(unittest.TestCase):
"""Test the Vultr switch platform."""
DEVICES = []
def add_devices(self, devices, action):
"""Mock add devices."""
for device in devices:
self.DEVICES.append(device)
def setUp(self):
"""Init values for this testcase class."""
self.hass = get_test_home_assistant()
self.configs = [
{
CONF_SUBSCRIPTION: '576965',
CONF_NAME: "A Server"
},
{
CONF_SUBSCRIPTION: '123456',
CONF_NAME: "Failed Server"
},
{
CONF_SUBSCRIPTION: '555555',
CONF_NAME: vultr.DEFAULT_NAME
}
]
def tearDown(self):
"""Stop our started services."""
self.hass.stop()
@requests_mock.Mocker()
def test_switch(self, mock):
"""Test successful instance."""
mock.get(
'https://api.vultr.com/v1/account/info?api_key=ABCDEFG1234567',
text=load_fixture('vultr_account_info.json'))
mock.get(
'https://api.vultr.com/v1/server/list?api_key=ABCDEFG1234567',
text=load_fixture('vultr_server_list.json'))
# Setup hub
base_vultr.setup(self.hass, VALID_CONFIG)
# Setup each of our test configs
for config in self.configs:
vultr.setup_platform(self.hass,
config,
self.add_devices,
None)
self.assertEqual(len(self.DEVICES), 3)
tested = 0
for device in self.DEVICES:
if device.subscription == '555555':
self.assertEqual('Vultr {}', device.name)
tested += 1
device.update()
device_attrs = device.device_state_attributes
if device.subscription == '555555':
self.assertEqual('Vultr Another Server', device.name)
tested += 1
if device.name == 'A Server':
self.assertEqual(True, device.is_on)
self.assertEqual('on', device.state)
self.assertEqual('mdi:server', device.icon)
self.assertEqual('1000',
device_attrs[ATTR_ALLOWED_BANDWIDTH])
self.assertEqual('yes',
device_attrs[ATTR_AUTO_BACKUPS])
self.assertEqual('123.123.123.123',
device_attrs[ATTR_IPV4_ADDRESS])
self.assertEqual('10.05',
device_attrs[ATTR_COST_PER_MONTH])
self.assertEqual('2013-12-19 14:45:41',
device_attrs[ATTR_CREATED_AT])
self.assertEqual('576965',
device_attrs[ATTR_SUBSCRIPTION_ID])
tested += 1
elif device.name == 'Failed Server':
self.assertEqual(False, device.is_on)
self.assertEqual('off', device.state)
self.assertEqual('mdi:server-off', device.icon)
self.assertEqual('1000',
device_attrs[ATTR_ALLOWED_BANDWIDTH])
self.assertEqual('no',
device_attrs[ATTR_AUTO_BACKUPS])
self.assertEqual('192.168.100.50',
device_attrs[ATTR_IPV4_ADDRESS])
self.assertEqual('73.25',
device_attrs[ATTR_COST_PER_MONTH])
self.assertEqual('2014-10-13 14:45:41',
device_attrs[ATTR_CREATED_AT])
self.assertEqual('123456',
device_attrs[ATTR_SUBSCRIPTION_ID])
tested += 1
self.assertEqual(4, tested)
@requests_mock.Mocker()
def test_turn_on(self, mock):
"""Test turning a subscription on."""
mock.get(
'https://api.vultr.com/v1/server/list?api_key=ABCDEFG1234567',
text=load_fixture('vultr_server_list.json'))
mock.post(
'https://api.vultr.com/v1/server/start?api_key=ABCDEFG1234567')
for device in self.DEVICES:
if device.name == 'Failed Server':
device.turn_on()
# Turn on, force date update
self.assertEqual(2, mock.call_count)
@requests_mock.Mocker()
def test_turn_off(self, mock):
"""Test turning a subscription off."""
mock.get(
'https://api.vultr.com/v1/server/list?api_key=ABCDEFG1234567',
text=load_fixture('vultr_server_list.json'))
mock.post(
'https://api.vultr.com/v1/server/halt?api_key=ABCDEFG1234567')
for device in self.DEVICES:
if device.name == 'A Server':
device.turn_off()
# Turn off, force update
self.assertEqual(2, mock.call_count)
def test_invalid_switch_config(self):
"""Test config type failures."""
with pytest.raises(vol.Invalid): # No subscription
vultr.PLATFORM_SCHEMA({
CONF_PLATFORM: base_vultr.DOMAIN,
})
@requests_mock.Mocker()
def test_invalid_switches(self, mock):
"""Test the VultrSwitch fails."""
mock.get(
'https://api.vultr.com/v1/account/info?api_key=ABCDEFG1234567',
text=load_fixture('vultr_account_info.json'))
mock.get(
'https://api.vultr.com/v1/server/list?api_key=ABCDEFG1234567',
text=load_fixture('vultr_server_list.json'))
base_vultr.setup(self.hass, VALID_CONFIG)
bad_conf = {} # No subscription
no_subs_setup = vultr.setup_platform(self.hass,
bad_conf,
self.add_devices,
None)
self.assertIsNotNone(no_subs_setup)
bad_conf = {
CONF_NAME: "Missing Server",
CONF_SUBSCRIPTION: '665544'
} # Sub not associated with API key (not in server_list)
wrong_subs_setup = vultr.setup_platform(self.hass,
bad_conf,
self.add_devices,
None)
self.assertIsNotNone(wrong_subs_setup)

View file

@ -0,0 +1,48 @@
"""The tests for the Vultr component."""
import unittest
import requests_mock
from copy import deepcopy
from homeassistant import setup
import components.vultr as vultr
from tests.common import (
get_test_home_assistant, load_fixture)
VALID_CONFIG = {
'vultr': {
'api_key': 'ABCDEFG1234567'
}
}
class TestVultr(unittest.TestCase):
"""Tests the Vultr component."""
def setUp(self):
"""Initialize values for this test case class."""
self.hass = get_test_home_assistant()
self.config = VALID_CONFIG
def tearDown(self): # pylint: disable=invalid-name
"""Stop everything that we started."""
self.hass.stop()
@requests_mock.Mocker()
def test_setup(self, mock):
"""Test successful setup."""
mock.get(
'https://api.vultr.com/v1/account/info?api_key=ABCDEFG1234567',
text=load_fixture('vultr_account_info.json'))
mock.get(
'https://api.vultr.com/v1/server/list?api_key=ABCDEFG1234567',
text=load_fixture('vultr_server_list.json'))
response = vultr.setup(self.hass, self.config)
self.assertTrue(response)
def test_setup_no_api_key(self):
"""Test failed setup with missing API Key."""
conf = deepcopy(self.config)
del conf['vultr']['api_key']
assert not setup.setup_component(self.hass, vultr.DOMAIN, conf)

View file

@ -0,0 +1 @@
{"balance":"-123.00","pending_charges":"3.38","last_payment_date":"2017-08-11 15:04:04","last_payment_amount":"-10.00"}

122
tests/fixtures/vultr_server_list.json vendored Normal file
View file

@ -0,0 +1,122 @@
{
"576965": {
"SUBID": "576965",
"os": "CentOS 6 x64",
"ram": "4096 MB",
"disk": "Virtual 60 GB",
"main_ip": "123.123.123.123",
"vcpu_count": "2",
"location": "New Jersey",
"DCID": "1",
"default_password": "nreqnusibni",
"date_created": "2013-12-19 14:45:41",
"pending_charges": "46.67",
"status": "active",
"cost_per_month": "10.05",
"current_bandwidth_gb": 131.512,
"allowed_bandwidth_gb": "1000",
"netmask_v4": "255.255.255.248",
"gateway_v4": "123.123.123.1",
"power_status": "running",
"server_state": "ok",
"VPSPLANID": "28",
"v6_network": "2001:DB8:1000::",
"v6_main_ip": "2001:DB8:1000::100",
"v6_network_size": "64",
"v6_networks": [
{
"v6_network": "2001:DB8:1000::",
"v6_main_ip": "2001:DB8:1000::100",
"v6_network_size": "64"
}
],
"label": "my new server",
"internal_ip": "10.99.0.10",
"kvm_url": "https://my.vultr.com/subs/novnc/api.php?data=eawxFVZw2mXnhGUV",
"auto_backups": "yes",
"tag": "mytag",
"OSID": "127",
"APPID": "0",
"FIREWALLGROUPID": "0"
},
"123456": {
"SUBID": "123456",
"os": "CentOS 6 x64",
"ram": "4096 MB",
"disk": "Virtual 60 GB",
"main_ip": "192.168.100.50",
"vcpu_count": "2",
"location": "New Jersey",
"DCID": "1",
"default_password": "nreqnusibni",
"date_created": "2014-10-13 14:45:41",
"pending_charges": "not a number",
"status": "active",
"cost_per_month": "73.25",
"current_bandwidth_gb": 957.457,
"allowed_bandwidth_gb": "1000",
"netmask_v4": "255.255.255.248",
"gateway_v4": "123.123.123.1",
"power_status": "halted",
"server_state": "ok",
"VPSPLANID": "28",
"v6_network": "2001:DB8:1000::",
"v6_main_ip": "2001:DB8:1000::100",
"v6_network_size": "64",
"v6_networks": [
{
"v6_network": "2001:DB8:1000::",
"v6_main_ip": "2001:DB8:1000::100",
"v6_network_size": "64"
}
],
"label": "my failed server",
"internal_ip": "10.99.0.10",
"kvm_url": "https://my.vultr.com/subs/novnc/api.php?data=eawxFVZw2mXnhGUV",
"auto_backups": "no",
"tag": "mytag",
"OSID": "127",
"APPID": "0",
"FIREWALLGROUPID": "0"
},
"555555": {
"SUBID": "555555",
"os": "CentOS 7 x64",
"ram": "1024 MB",
"disk": "Virtual 30 GB",
"main_ip": "192.168.250.50",
"vcpu_count": "1",
"location": "London",
"DCID": "7",
"default_password": "password",
"date_created": "2014-10-15 14:45:41",
"pending_charges": "5.45",
"status": "active",
"cost_per_month": "73.25",
"current_bandwidth_gb": 57.457,
"allowed_bandwidth_gb": "100",
"netmask_v4": "255.255.255.248",
"gateway_v4": "123.123.123.1",
"power_status": "halted",
"server_state": "ok",
"VPSPLANID": "28",
"v6_network": "2001:DB8:1000::",
"v6_main_ip": "2001:DB8:1000::100",
"v6_network_size": "64",
"v6_networks": [
{
"v6_network": "2001:DB8:1000::",
"v6_main_ip": "2001:DB8:1000::100",
"v6_network_size": "64"
}
],
"label": "Another Server",
"internal_ip": "10.99.0.10",
"kvm_url": "https://my.vultr.com/subs/novnc/api.php?data=eawxFVZw2mXnhGUV",
"auto_backups": "no",
"tag": "mytag",
"OSID": "127",
"APPID": "0",
"FIREWALLGROUPID": "0"
}
}