Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

TekSavvy Sensor unlimited bandwidth support (#12325) 

* Support TekSavvy Unlimited Plans

Support TekSavvy account usage for unlimited plans.
Seeing cap limit to 0 will now provide unlimited behaviour on usage calculations.

* Add unit tests to sensor.teksavvy

Add coverage unit tests to TekSavvy Sensor component, none existing previously.
pull/12801/head
Mike O'Driscoll 2018-02-28 19:21:10 -05:00 committed by Paulus Schoutsen
parent b1cc9bf452
commit 3416d3f5f1
3 changed files with 214 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -647,7 +647,6 @@ omit =
homeassistant/components/sensor/sytadin.py
homeassistant/components/sensor/tank_utility.py
homeassistant/components/sensor/ted5000.py
homeassistant/components/sensor/teksavvy.py
homeassistant/components/sensor/temper.py
homeassistant/components/sensor/tibber.py
homeassistant/components/sensor/time_date.py

46
homeassistant/components/sensor/teksavvy.py
View File

@ -31,11 +31,11 @@ MIN_TIME_BETWEEN_UPDATES = timedelta(hours=1)
REQUEST_TIMEOUT = 5 # seconds
SENSOR_TYPES = {
'usage': ['Usage', PERCENT, 'mdi:percent'],
'usage': ['Usage Ratio', PERCENT, 'mdi:percent'],
'usage_gb': ['Usage', GIGABYTES, 'mdi:download'],
'limit': ['Data limit', GIGABYTES, 'mdi:download'],
'onpeak_download': ['On Peak Download', GIGABYTES, 'mdi:download'],
'onpeak_upload': ['On Peak Upload ', GIGABYTES, 'mdi:upload'],
'onpeak_upload': ['On Peak Upload', GIGABYTES, 'mdi:upload'],
'onpeak_total': ['On Peak Total', GIGABYTES, 'mdi:download'],
'offpeak_download': ['Off Peak download', GIGABYTES, 'mdi:download'],
'offpeak_upload': ['Off Peak Upload', GIGABYTES, 'mdi:upload'],
@ -128,7 +128,9 @@ class TekSavvyData(object):
self.websession = websession
self.api_key = api_key
self.bandwidth_cap = bandwidth_cap
self.data = {"limit": self.bandwidth_cap}
# Set unlimited users to infinite, otherwise the cap.
self.data = {"limit": self.bandwidth_cap} if self.bandwidth_cap > 0 \
else {"limit": float('inf')}
@asyncio.coroutine
@Throttle(MIN_TIME_BETWEEN_UPDATES)
@ -143,17 +145,27 @@ class TekSavvyData(object):
if req.status != 200:
_LOGGER.error("Request failed with status: %u", req.status)
return False
data = yield from req.json()
for (api, ha_name) in API_HA_MAP:
self.data[ha_name] = float(data["value"][0][api])
on_peak_download = self.data["onpeak_download"]
on_peak_upload = self.data["onpeak_upload"]
off_peak_download = self.data["offpeak_download"]
off_peak_upload = self.data["offpeak_upload"]
limit = self.data["limit"]
self.data["usage"] = 100*on_peak_download/self.bandwidth_cap
self.data["usage_gb"] = on_peak_download
self.data["onpeak_total"] = on_peak_download + on_peak_upload
self.data["offpeak_total"] = off_peak_download + off_peak_upload
self.data["onpeak_remaining"] = limit - on_peak_download
return True
try:
data = yield from req.json()
for (api, ha_name) in API_HA_MAP:
self.data[ha_name] = float(data["value"][0][api])
on_peak_download = self.data["onpeak_download"]
on_peak_upload = self.data["onpeak_upload"]
off_peak_download = self.data["offpeak_download"]
off_peak_upload = self.data["offpeak_upload"]
limit = self.data["limit"]
# Support "unlimited" users
if self.bandwidth_cap > 0:
self.data["usage"] = 100*on_peak_download/self.bandwidth_cap
else:
self.data["usage"] = 0
self.data["usage_gb"] = on_peak_download
self.data["onpeak_total"] = on_peak_download + on_peak_upload
self.data["offpeak_total"] =\
off_peak_download + off_peak_upload
self.data["onpeak_remaining"] = limit - on_peak_download
return True
except ValueError:
_LOGGER.error("JSON Decode Failed")
return False

185
tests/components/sensor/test_teksavvy.py Normal file
View File

@ -0,0 +1,185 @@
"""Tests for the TekSavvy sensor platform."""
import asyncio
from homeassistant.bootstrap import async_setup_component
from homeassistant.components.sensor.teksavvy import TekSavvyData
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@asyncio.coroutine
def test_capped_setup(hass, aioclient_mock):
"""Test the default setup."""
config = {'platform': 'teksavvy',
'api_key': 'NOTAKEY',
'total_bandwidth': 400,
'monitored_variables': [
'usage',
'usage_gb',
'limit',
'onpeak_download',
'onpeak_upload',
'onpeak_total',
'offpeak_download',
'offpeak_upload',
'offpeak_total',
'onpeak_remaining']}
result = '{"odata.metadata":"http://api.teksavvy.com/web/Usage/$metadata'\
'#UsageSummaryRecords","value":[{'\
'"StartDate":"2018-01-01T00:00:00",'\
'"EndDate":"2018-01-31T00:00:00",'\
'"OID":"999999","IsCurrent":true,'\
'"OnPeakDownload":226.75,'\
'"OnPeakUpload":8.82,'\
'"OffPeakDownload":36.24,"OffPeakUpload":1.58'\
'}]}'
aioclient_mock.get("https://api.teksavvy.com/"
"web/Usage/UsageSummaryRecords?"
"$filter=IsCurrent%20eq%20true",
text=result)
yield from async_setup_component(hass, 'sensor', {'sensor': config})
state = hass.states.get('sensor.teksavvy_data_limit')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '400'
state = hass.states.get('sensor.teksavvy_off_peak_download')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '36.24'
state = hass.states.get('sensor.teksavvy_off_peak_upload')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '1.58'
state = hass.states.get('sensor.teksavvy_off_peak_total')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '37.82'
state = hass.states.get('sensor.teksavvy_on_peak_download')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '226.75'
state = hass.states.get('sensor.teksavvy_on_peak_upload')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '8.82'
state = hass.states.get('sensor.teksavvy_on_peak_total')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '235.57'
state = hass.states.get('sensor.teksavvy_usage_ratio')
assert state.attributes.get('unit_of_measurement') == '%'
assert state.state == '56.69'
state = hass.states.get('sensor.teksavvy_usage')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '226.75'
state = hass.states.get('sensor.teksavvy_remaining')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '173.25'
@asyncio.coroutine
def test_unlimited_setup(hass, aioclient_mock):
"""Test the default setup."""
config = {'platform': 'teksavvy',
'api_key': 'NOTAKEY',
'total_bandwidth': 0,
'monitored_variables': [
'usage',
'usage_gb',
'limit',
'onpeak_download',
'onpeak_upload',
'onpeak_total',
'offpeak_download',
'offpeak_upload',
'offpeak_total',
'onpeak_remaining']}
result = '{"odata.metadata":"http://api.teksavvy.com/web/Usage/$metadata'\
'#UsageSummaryRecords","value":[{'\
'"StartDate":"2018-01-01T00:00:00",'\
'"EndDate":"2018-01-31T00:00:00",'\
'"OID":"999999","IsCurrent":true,'\
'"OnPeakDownload":226.75,'\
'"OnPeakUpload":8.82,'\
'"OffPeakDownload":36.24,"OffPeakUpload":1.58'\
'}]}'
aioclient_mock.get("https://api.teksavvy.com/"
"web/Usage/UsageSummaryRecords?"
"$filter=IsCurrent%20eq%20true",
text=result)
yield from async_setup_component(hass, 'sensor', {'sensor': config})
state = hass.states.get('sensor.teksavvy_data_limit')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == 'inf'
state = hass.states.get('sensor.teksavvy_off_peak_download')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '36.24'
state = hass.states.get('sensor.teksavvy_off_peak_upload')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '1.58'
state = hass.states.get('sensor.teksavvy_off_peak_total')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '37.82'
state = hass.states.get('sensor.teksavvy_on_peak_download')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '226.75'
state = hass.states.get('sensor.teksavvy_on_peak_upload')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '8.82'
state = hass.states.get('sensor.teksavvy_on_peak_total')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '235.57'
state = hass.states.get('sensor.teksavvy_usage')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == '226.75'
state = hass.states.get('sensor.teksavvy_usage_ratio')
assert state.attributes.get('unit_of_measurement') == '%'
assert state.state == '0'
state = hass.states.get('sensor.teksavvy_remaining')
assert state.attributes.get('unit_of_measurement') == 'GB'
assert state.state == 'inf'
@asyncio.coroutine
def test_bad_return_code(hass, aioclient_mock):
"""Test handling a return code that isn't HTTP OK."""
aioclient_mock.get("https://api.teksavvy.com/"
"web/Usage/UsageSummaryRecords?"
"$filter=IsCurrent%20eq%20true",
status=404)
tsd = TekSavvyData(hass.loop, async_get_clientsession(hass),
'notakey', 400)
result = yield from tsd.async_update()
assert result is False
@asyncio.coroutine
def test_bad_json_decode(hass, aioclient_mock):
"""Test decoding invalid json result."""
aioclient_mock.get("https://api.teksavvy.com/"
"web/Usage/UsageSummaryRecords?"
"$filter=IsCurrent%20eq%20true",
text='this is not json')
tsd = TekSavvyData(hass.loop, async_get_clientsession(hass),
'notakey', 400)
result = yield from tsd.async_update()
assert result is False