Merge branch 'master' into get-pages

# Conflicts:
#	tests/test_client.py
This commit is contained in:
Luke McNinch 2021-11-30 16:31:50 -05:00
commit c152480ad9
8 changed files with 412 additions and 150 deletions

View file

@ -3,9 +3,20 @@ name: Build-CI
on: [pull_request, push] on: [pull_request, push]
jobs: jobs:
approve: # First step
runs-on: ubuntu-latest
steps:
- name: Approve
run: echo For security reasons, all pull requests to this repository need to be approved first before running any automated CI.
build: build:
runs-on: ubuntu-latest runs-on: ubuntu-latest
needs: [approve] # Require the first step to finish
environment:
name: IO
steps: steps:
- uses: actions/checkout@v2 - uses: actions/checkout@v2

View file

@ -21,5 +21,5 @@
from .client import Client from .client import Client
from .mqtt_client import MQTTClient from .mqtt_client import MQTTClient
from .errors import AdafruitIOError, RequestError, ThrottlingError, MQTTError from .errors import AdafruitIOError, RequestError, ThrottlingError, MQTTError
from .model import Data, Feed, Group from .model import Data, Feed, Group, Dashboard, Block, Layout
from ._version import __version__ from ._version import __version__

View file

@ -18,6 +18,7 @@
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE. # SOFTWARE.
import time
from time import struct_time from time import struct_time
import json import json
import platform import platform
@ -30,7 +31,7 @@ from urllib.parse import parse_qs
import requests import requests
from .errors import RequestError, ThrottlingError from .errors import RequestError, ThrottlingError
from .model import Data, Feed, Group from .model import Data, Feed, Group, Dashboard, Block, Layout
API_PAGE_LIMIT = 1000 API_PAGE_LIMIT = 1000
@ -194,9 +195,19 @@ class Client(object):
https://docs.python.org/3.7/library/time.html#time.struct_time https://docs.python.org/3.7/library/time.html#time.struct_time
""" """
path = 'integrations/time/struct.json' path = 'integrations/time/struct.json'
time = self._get(path) return self._parse_time_struct(self._get(path))
return struct_time((time['year'], time['mon'], time['mday'], time['hour'],
time['min'], time['sec'], time['wday'], time['yday'], time['isdst'])) @staticmethod
def _parse_time_struct(time_dict: dict) -> time.struct_time:
"""Parse the time data returned by the server and return a time_struct
Corrects for the weekday returned by the server in Sunday=0 format
(Python expects Monday=0)
"""
wday = (time_dict['wday'] - 1) % 7
return struct_time((time_dict['year'], time_dict['mon'], time_dict['mday'],
time_dict['hour'], time_dict['min'], time_dict['sec'],
wday, time_dict['yday'], time_dict['isdst']))
def receive_weather(self, weather_id=None): def receive_weather(self, weather_id=None):
"""Adafruit IO Weather Service, Powered by Dark Sky """Adafruit IO Weather Service, Powered by Dark Sky
@ -326,11 +337,13 @@ class Client(object):
:param string feed: Key of Adafruit IO feed. :param string feed: Key of Adafruit IO feed.
:param group_key group: Group to place new feed in. :param group_key group: Group to place new feed in.
""" """
f = feed._asdict()
del f['id'] # Don't pass id on create call
path = "feeds/" path = "feeds/"
if group_key is not None: # create feed in a group if group_key is not None: # create feed in a group
path="/groups/%s/feeds"%group_key path="/groups/%s/feeds"%group_key
return Feed.from_dict(self._post(path, {"feed": feed._asdict()})) return Feed.from_dict(self._post(path, {"feed": f}))
return Feed.from_dict(self._post(path, {"feed": feed._asdict()})) return Feed.from_dict(self._post(path, {"feed": f}))
def delete_feed(self, feed): def delete_feed(self, feed):
"""Delete the specified feed. """Delete the specified feed.
@ -363,3 +376,73 @@ class Client(object):
""" """
path = "groups/{0}".format(group) path = "groups/{0}".format(group)
self._delete(path) self._delete(path)
# Dashboard functionality.
def dashboards(self, dashboard=None):
"""Retrieve a list of all dashboards, or the specified dashboard.
:param string dashboard: Key of Adafruit IO Dashboard. Defaults to None.
"""
if dashboard is None:
path = "dashboards/"
return list(map(Dashboard.from_dict, self._get(path)))
path = "dashboards/{0}".format(dashboard)
return Dashboard.from_dict(self._get(path))
def create_dashboard(self, dashboard):
"""Create the specified dashboard.
:param Dashboard dashboard: Dashboard object to create
"""
path = "dashboards/"
return Dashboard.from_dict(self._post(path, dashboard._asdict()))
def delete_dashboard(self, dashboard):
"""Delete the specified dashboard.
:param string dashboard: Key of Adafruit IO Dashboard.
"""
path = "dashboards/{0}".format(dashboard)
self._delete(path)
# Block functionality.
def blocks(self, dashboard, block=None):
"""Retrieve a list of all blocks from a dashboard, or the specified block.
:param string dashboard: Key of Adafruit IO Dashboard.
:param string block: id of Adafruit IO Block. Defaults to None.
"""
if block is None:
path = "dashboards/{0}/blocks".format(dashboard)
return list(map(Block.from_dict, self._get(path)))
path = "dashboards/{0}/blocks/{1}".format(dashboard, block)
return Block.from_dict(self._get(path))
def create_block(self, dashboard, block):
"""Create the specified block under the specified dashboard.
:param string dashboard: Key of Adafruit IO Dashboard.
:param Block block: Block object to create under dashboard
"""
path = "dashboards/{0}/blocks".format(dashboard)
return Block.from_dict(self._post(path, block._asdict()))
def delete_block(self, dashboard, block):
"""Delete the specified block.
:param string dashboard: Key of Adafruit IO Dashboard.
:param string block: id of Adafruit IO Block.
"""
path = "dashboards/{0}/blocks/{1}".format(dashboard, block)
self._delete(path)
# Layout functionality.
def layouts(self, dashboard):
"""Retrieve the layouts array from a dashboard
:param string dashboard: key of Adafruit IO Dashboard.
"""
path = "dashboards/{0}".format(dashboard)
dashboard = self._get(path)
return Layout.from_dict(dashboard['layouts'])
def update_layout(self, dashboard, layout):
"""Update the layout of the specified dashboard.
:param string dashboard: Key of Adafruit IO Dashboard.
:param Layout layout: Layout object to update under dashboard
"""
path = "dashboards/{0}/update_layouts".format(dashboard)
return Layout.from_dict(self._post(path, {'layouts': layout._asdict()}))

View file

@ -43,6 +43,7 @@ DATA_FIELDS = [ 'created_epoch',
FEED_FIELDS = [ 'name', FEED_FIELDS = [ 'name',
'key', 'key',
'id',
'description', 'description',
'unit_type', 'unit_type',
'unit_symbol', 'unit_symbol',
@ -61,6 +62,26 @@ GROUP_FIELDS = [ 'description',
'properties', 'properties',
'name' ] 'name' ]
DASHBOARD_FIELDS = [ 'name',
'key',
'description',
'show_header',
'color_mode',
'block_borders',
'header_image_url',
'blocks' ]
BLOCK_FIELDS = [ 'name',
'id',
'visual_type',
'properties',
'block_feeds' ]
LAYOUT_FIELDS = ['xl',
'lg',
'md',
'sm',
'xs' ]
# These are very simple data model classes that are based on namedtuple. This is # These are very simple data model classes that are based on namedtuple. This is
# to keep the classes simple and prevent any confusion around updating data # to keep the classes simple and prevent any confusion around updating data
@ -71,15 +92,24 @@ GROUP_FIELDS = [ 'description',
Data = namedtuple('Data', DATA_FIELDS) Data = namedtuple('Data', DATA_FIELDS)
Feed = namedtuple('Feed', FEED_FIELDS) Feed = namedtuple('Feed', FEED_FIELDS)
Group = namedtuple('Group', GROUP_FIELDS) Group = namedtuple('Group', GROUP_FIELDS)
Dashboard = namedtuple('Dashboard', DASHBOARD_FIELDS)
Block = namedtuple('Block', BLOCK_FIELDS)
Layout = namedtuple('Layout', LAYOUT_FIELDS)
# Magic incantation to make all parameters to the initializers optional with a # Magic incantation to make all parameters to the initializers optional with a
# default value of None. # default value of None.
Group.__new__.__defaults__ = tuple(None for x in GROUP_FIELDS) Group.__new__.__defaults__ = tuple(None for x in GROUP_FIELDS)
Data.__new__.__defaults__ = tuple(None for x in DATA_FIELDS) Data.__new__.__defaults__ = tuple(None for x in DATA_FIELDS)
Layout.__new__.__defaults__ = tuple(None for x in LAYOUT_FIELDS)
# explicitly set dashboard values so that 'color_mode' is 'dark'
Dashboard.__new__.__defaults__ = (None, None, None, False, "dark", True, None, None)
# explicitly set block values so 'properties' is a dictionary
Block.__new__.__defaults__ = (None, None, None, {}, None)
# explicitly set feed values # explicitly set feed values
Feed.__new__.__defaults__ = (None, None, None, None, None, 'ON', 'Private', None, None, None) Feed.__new__.__defaults__ = (None, None, None, None, None, None, 'ON', 'Private', None, None, None)
# Define methods to convert from dicts to the data types. # Define methods to convert from dicts to the data types.
def _from_dict(cls, data): def _from_dict(cls, data):
@ -103,7 +133,17 @@ def _group_from_dict(cls, data):
return cls(**params) return cls(**params)
def _dashboard_from_dict(cls, data):
params = {x: data.get(x, None) for x in cls._fields}
# Parse the blocks if they're provided and generate block instances.
params['blocks'] = tuple(map(Block.from_dict, data.get('blocks', [])))
return cls(**params)
# Now add the from_dict class methods defined above to the data types. # Now add the from_dict class methods defined above to the data types.
Data.from_dict = classmethod(_from_dict) Data.from_dict = classmethod(_from_dict)
Feed.from_dict = classmethod(_feed_from_dict) Feed.from_dict = classmethod(_feed_from_dict)
Group.from_dict = classmethod(_group_from_dict) Group.from_dict = classmethod(_group_from_dict)
Dashboard.from_dict = classmethod(_dashboard_from_dict)
Block.from_dict = classmethod(_from_dict)
Layout.from_dict = classmethod(_from_dict)

View file

@ -0,0 +1,88 @@
"""
'dashboard.py'
=========================================
Creates a dashboard with 3 blocks and feed it data
Author(s): Doug Zobel
"""
from time import sleep
from random import randrange
from Adafruit_IO import Client, Feed, Block, Dashboard, Layout
# Set to your Adafruit IO key.
# Remember, your key is a secret,
# so make sure not to publish it when you publish this code!
ADAFRUIT_IO_USERNAME = ''
# Set to your Adafruit IO username.
# (go to https://accounts.adafruit.com to find your username)
ADAFRUIT_IO_KEY = ''
# Create an instance of the REST client.
aio = Client(ADAFRUIT_IO_USERNAME, ADAFRUIT_IO_KEY)
# Create a new feed named 'Dashboard Data' under the default group
feed = aio.create_feed(Feed(name="Dashboard Data"), "default")
# Fetch group info (group.id needed when adding feeds to blocks)
group = aio.groups("default")
# Create a new dasbhoard named 'Example Dashboard'
dashboard = aio.create_dashboard(Dashboard(name="Example Dashboard"))
# Create a line_chart
linechart = Block(name="Linechart Data",
visual_type = 'line_chart',
properties = {
"gridLines": True,
"historyHours": "2"},
block_feeds = [{
"group_id": group.id,
"feed_id": feed.id
}])
linechart = aio.create_block(dashboard.key, linechart)
# Create a gauge
gauge = Block(name="Gauge Data",
visual_type = 'gauge',
block_feeds = [{
"group_id": group.id,
"feed_id": feed.id
}])
gauge = aio.create_block(dashboard.key, gauge)
# Create a text stream
stream = Block(name="Stream Data",
visual_type = 'stream',
properties = {
"fontSize": "12",
"fontColor": "#63de00",
"showGroupName": "no"},
block_feeds = [{
"group_id": group.id,
"feed_id": feed.id
}])
stream = aio.create_block(dashboard.key, stream)
# Update the large layout to:
# |----------------|
# | Line Chart |
# |----------------|
# | Gauge | Stream |
# |----------------|
layout = Layout(lg = [
{'x': 0, 'y': 0, 'w': 16, 'h': 4, 'i': str(linechart.id)},
{'x': 0, 'y': 4, 'w': 8, 'h': 4, 'i': str(gauge.id)},
{'x': 8, 'y': 4, 'w': 8, 'h': 4, 'i': str(stream.id)}])
aio.update_layout(dashboard.key, layout)
print("Dashboard created at: " +
"https://io.adafruit.com/{0}/dashboards/{1}".format(ADAFRUIT_IO_USERNAME,
dashboard.key))
# Now send some data
value = 0
while True:
value = (value + randrange(0, 10)) % 100
print('sending data: ', value)
aio.send_data(feed.key, value)
sleep(3)

View file

@ -1,133 +0,0 @@
"""
'environmental_monitor.py'
===============================================================================
Example of sending I2C sensor data
from multiple sensors to Adafruit IO.
Tutorial Link: https://learn.adafruit.com/adafruit-io-air-quality-monitor
Adafruit invests time and resources providing this open source code.
Please support Adafruit and open source hardware by purchasing
products from Adafruit!
Author(s): Brent Rubell for Adafruit Industries
Copyright (c) 2018 Adafruit Industries
Licensed under the MIT license.
All text above must be included in any redistribution.
Dependencies:
- Adafruit_Blinka (CircuitPython, on Pi.)
(https://github.com/adafruit/Adafruit_Blinka)
- Adafruit_CircuitPython_SGP30.
(https://github.com/adafruit/Adafruit_CircuitPython_SGP30)
- Adafruit_CircuitPython_VEML6070.
(https://github.com/adafruit/Adafruit_CircuitPython_VEML6070)
- Adafruit_CircuitPython_BME280.
(https://github.com/adafruit/Adafruit_CircuitPython_BME280)
"""
# Import standard python modules
import time
# import Adafruit Blinka
import board
import busio
# import CircuitPython sensor libraries
import adafruit_sgp30
import adafruit_veml6070
import adafruit_bme280
# import Adafruit IO REST client
from Adafruit_IO import Client, Feed, RequestError
# loop timeout, in seconds.
LOOP_DELAY = 10
# Set to your Adafruit IO key.
# Remember, your key is a secret,
# so make sure not to publish it when you publish this code!
ADAFRUIT_IO_KEY = 'YOUR_AIO_KEY'
# Set to your Adafruit IO username.
# (go to https://accounts.adafruit.com to find your username)
ADAFRUIT_IO_USERNAME = 'YOUR_AIO_USERNAME'
# Create an instance of the REST client
aio = Client(ADAFRUIT_IO_USERNAME, ADAFRUIT_IO_KEY)
try: # if we already have the feeds, assign them.
tvoc_feed = aio.feeds('tvoc')
eCO2_feed = aio.feeds('eco2')
uv_feed = aio.feeds('uv')
temperature_feed = aio.feeds('temperature')
humidity_feed = aio.feeds('humidity')
pressure_feed = aio.feeds('pressure')
altitude_feed = aio.feeds('altitude')
except RequestError: # if we don't, create and assign them.
tvoc_feed = aio.create_feed(Feed(name='tvoc'))
eCO2_feed = aio.create_feed(Feed(name='eco2'))
uv_feed = aio.create_feed(Feed(name='uv'))
temperature_feed = aio.create_feed(Feed(name='temperature'))
humidity_feed = aio.create_feed(Feed(name='humidity'))
pressure_feed = aio.create_feed(Feed(name='pressure'))
altitude_feed = aio.create_feed(Feed(name='altitude'))
# Create busio I2C
i2c = busio.I2C(board.SCL, board.SDA, frequency=100000)
# Create VEML6070 object.
uv = adafruit_veml6070.VEML6070(i2c)
# Create BME280 object.
bme280 = adafruit_bme280.Adafruit_BME280_I2C(i2c)
bme280.sea_level_pressure = 1013.25
# Create SGP30 object using I2C.
sgp30 = adafruit_sgp30.Adafruit_SGP30(i2c)
sgp30.iaq_init()
sgp30.set_iaq_baseline(0x8973, 0x8aae)
# Sample VEML6070
def sample_VEML():
for _ in range(10):
uv_raw = uv.uv_raw
return uv_raw
while True:
print('Reading sensors...')
# Read SGP30.
eCO2_data = sgp30.eCO2
tvoc_data = sgp30.TVOC
# Read VEML6070.
uv_data = sample_VEML()
# Read BME280.
temp_data = bme280.temperature
# convert temperature (C->F)
temp_data = int(temp_data) * 1.8 + 32
humid_data = bme280.humidity
pressure_data = bme280.pressure
alt_data = bme280.altitude
print('sending data to adafruit io...')
# Send SGP30 Data to Adafruit IO.
print('eCO2:', eCO2_data)
aio.send(eCO2_feed.key, eCO2_data)
print('tvoc:', tvoc_data)
aio.send(tvoc_feed.key, tvoc_data)
time.sleep(2)
# Send VEML6070 Data to Adafruit IO.
print('UV Level: ', uv_data)
aio.send(uv_feed.key, uv_data)
time.sleep(2)
# Send BME280 Data to Adafruit IO.
print('Temperature: %0.1f C' % temp_data)
aio.send(temperature_feed.key, temp_data)
print("Humidity: %0.1f %%" % humid_data)
aio.send(humidity_feed.key, int(humid_data))
time.sleep(2)
print("Pressure: %0.1f hPa" % pressure_data)
aio.send(pressure_feed.key, int(pressure_data))
print("Altitude = %0.2f meters" % alt_data)
aio.send(altitude_feed.key, int(alt_data))
# avoid timeout from adafruit io
time.sleep(LOOP_DELAY * 60)

View file

@ -3,7 +3,7 @@
import time import time
import unittest import unittest
from Adafruit_IO import Client, Data, Feed, Group, RequestError from Adafruit_IO import Client, Data, Feed, Group, Dashboard, Block, Layout, RequestError
import base import base
@ -22,8 +22,8 @@ class TestClient(base.IOTestCase):
# If your IP isn't put on the list of non-throttled IPs, uncomment the # If your IP isn't put on the list of non-throttled IPs, uncomment the
# function below to waste time between tests to prevent throttling. # function below to waste time between tests to prevent throttling.
def tearDown(self): #def tearDown(self):
time.sleep(30.0) # time.sleep(30.0)
# Helper Methods # Helper Methods
def get_client(self): def get_client(self):
@ -46,6 +46,22 @@ class TestClient(base.IOTestCase):
# Swallow the error if the group doesn't exist. # Swallow the error if the group doesn't exist.
pass pass
def ensure_dashboard_deleted(self, client, dashboard):
# Delete the specified dashboard if it exists.
try:
client.delete_dashboard(dashboard)
except RequestError:
# Swallow the error if the dashboard doesn't exist.
pass
def ensure_block_deleted(self, client, dashboard, block):
# Delete the specified block if it exists.
try:
client.delete_block(dashboard, block)
except RequestError:
# Swallow the error if the block doesn't exist.
pass
def empty_feed(self, client, feed): def empty_feed(self, client, feed):
# Remove all the data from a specified feed (but don't delete the feed). # Remove all the data from a specified feed (but don't delete the feed).
data = client.data(feed, max_results=None) data = client.data(feed, max_results=None)
@ -160,11 +176,41 @@ class TestClient(base.IOTestCase):
"""receive_time """receive_time
""" """
aio = self.get_client() aio = self.get_client()
time = aio.receive_time() server_time = aio.receive_time()
# Check that each value is rx'd properly # Check that each value is rx'd properly
# (should never be None type) # (should never be None type)
for time_data in time: for time_data in server_time:
self.assertIsNotNone(time_data) self.assertIsNotNone(time_data)
# Check that the week day was interpreted properly
adjusted_time = time.localtime(time.mktime(server_time))
self.assertEqual(server_time.tm_wday, adjusted_time.tm_wday)
def test_parse_time_struct(self):
"""Ensure the _parse_time_struct method properly handles all 7
week days. Particularly important to make sure Sunday is 6,
not -1"""
# Zero time is a dictionary as would be provided by server
# (wday is one higher than it should be)
zero_time = {'year': 1970,
'mon': 1,
'mday': 1,
'hour': 0,
'min': 0,
'sec': 0,
'wday': 4,
'yday': 1,
'isdst': 0}
# Create a good struct for each day of the week and make sure
# the server-style dictionary is parsed correctly
for k in range(7):
real_struct = time.gmtime(k * 86400)
d = zero_time.copy()
d['mday'] += k
d['wday'] += k
d['yday'] += k
newd = Client._parse_time_struct(d)
self.assertEqual(newd.tm_wday, real_struct.tm_wday)
# Test Feed Functionality # Test Feed Functionality
def test_append_by_feed_name(self): def test_append_by_feed_name(self):
@ -269,3 +315,95 @@ class TestClient(base.IOTestCase):
group = io.create_group(Group(name='grouprx')) group = io.create_group(Group(name='grouprx'))
response = io.groups(group.key) response = io.groups(group.key)
self.assertEqual(response.key, 'grouprx') self.assertEqual(response.key, 'grouprx')
# Test Dashboard Functionality
def test_dashboard_create_dashboard(self):
io = self.get_client()
self.ensure_dashboard_deleted(io, 'dashtest')
response = io.create_dashboard(Dashboard(name='dashtest'))
self.assertEqual(response.name, 'dashtest')
def test_dashboard_returns_all_dashboards(self):
io = self.get_client()
self.ensure_dashboard_deleted(io, 'dashtest')
dashboard = io.create_dashboard(Dashboard(name='dashtest'))
response = io.dashboards()
self.assertGreaterEqual(len(response), 1)
def test_dashboard_returns_requested_feed(self):
io = self.get_client()
self.ensure_dashboard_deleted(io, 'dashtest')
dashboard = io.create_dashboard(Dashboard(name='dashtest'))
response = io.dashboards('dashtest')
self.assertEqual(response.name, 'dashtest')
# Test Block Functionality
def test_block_create_block(self):
io = self.get_client()
self.ensure_block_deleted(io, 'dashtest', 'blocktest')
self.ensure_dashboard_deleted(io, 'dashtest')
dash = io.create_dashboard(Dashboard(name='dashtest'))
block = io.create_block(dash.key, Block(name='blocktest',
visual_type = 'line_chart'))
self.assertEqual(block.name, 'blocktest')
io.delete_block(dash.key, block.id)
io.delete_dashboard(dash.key)
def test_dashboard_returns_all_blocks(self):
io = self.get_client()
self.ensure_block_deleted(io, 'dashtest', 'blocktest')
self.ensure_dashboard_deleted(io, 'dashtest')
dash = io.create_dashboard(Dashboard(name='dashtest'))
block = io.create_block(dash.key, Block(name='blocktest',
visual_type = 'line_chart'))
response = io.blocks(dash.key)
self.assertEqual(len(response), 1)
io.delete_block(dash.key, block.id)
io.delete_dashboard(dash.key)
def test_dashboard_returns_requested_block(self):
io = self.get_client()
self.ensure_block_deleted(io, 'dashtest', 'blocktest')
self.ensure_dashboard_deleted(io, 'dashtest')
dash = io.create_dashboard(Dashboard(name='dashtest'))
block = io.create_block(dash.key, Block(name='blocktest',
visual_type = 'line_chart'))
response = io.blocks(dash.key, block.id)
self.assertEqual(response.name, 'blocktest')
io.delete_block(dash.key, block.id)
io.delete_dashboard(dash.key)
# Test Layout Functionality
def test_layout_returns_all_layouts(self):
io = self.get_client()
self.ensure_block_deleted(io, 'dashtest', 'blocktest')
self.ensure_dashboard_deleted(io, 'dashtest')
dash = io.create_dashboard(Dashboard(name='dashtest'))
block = io.create_block(dash.key, Block(name='blocktest',
visual_type = 'line_chart'))
response = io.layouts(dash.key)
self.assertEqual(len(response), 5) # 5 layouts: xs, sm, md, lg, xl
self.assertEqual(len(response.lg), 1)
io.delete_block(dash.key, block.id)
io.delete_dashboard(dash.key)
def test_layout_update_layout(self):
io = self.get_client()
self.ensure_block_deleted(io, 'dashtest', 'blocktest')
self.ensure_dashboard_deleted(io, 'dashtest')
dash = io.create_dashboard(Dashboard(name='dashtest'))
block = io.create_block(dash.key, Block(name='blocktest',
visual_type = 'line_chart'))
layout = Layout(lg = [
{'x': 0, 'y': 0, 'w': 16, 'h': 4, 'i': str(block.id)}])
io.update_layout(dash.key, layout)
response = io.layouts(dash.key)
self.assertEqual(len(response.lg), 1)
self.assertEqual(response.lg[0]['w'], 16)
io.delete_block(dash.key, block.id)
io.delete_dashboard(dash.key)
if __name__ == "__main__":
unittest.main()

View file

@ -18,7 +18,7 @@
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE. # SOFTWARE.
from Adafruit_IO import Data, Feed, Group from Adafruit_IO import Data, Feed, Group, Dashboard, Block, Layout
import base import base
@ -45,11 +45,12 @@ class TestData(base.IOTestCase):
def test_feeds_have_explicitly_set_values(self): def test_feeds_have_explicitly_set_values(self):
""" Let's make sure feeds are explicitly set from within the model: """ Let's make sure feeds are explicitly set from within the model:
Feed.__new__.__defaults__ = (None, None, None, None, None, 'ON', 'Private', None, None, None) Feed.__new__.__defaults__ = (None, None, None, None, None, None, 'ON', 'Private', None, None, None)
""" """
feed = Feed(name='foo') feed = Feed(name='foo')
self.assertEqual(feed.name, 'foo') self.assertEqual(feed.name, 'foo')
self.assertIsNone(feed.key) self.assertIsNone(feed.key)
self.assertIsNone(feed.id)
self.assertIsNone(feed.description) self.assertIsNone(feed.description)
self.assertIsNone(feed.unit_type) self.assertIsNone(feed.unit_type)
self.assertIsNone(feed.unit_symbol) self.assertIsNone(feed.unit_symbol)
@ -69,6 +70,40 @@ class TestData(base.IOTestCase):
self.assertIsNone(group.feeds) self.assertIsNone(group.feeds)
self.assertIsNone(group.properties) self.assertIsNone(group.properties)
""" Let's make sure feeds are explicitly set from within the model:
Dashboard.__new__.__defaults__ = (None, None, None, False, "dark", True, None, None)
"""
def test_dashboard_have_explicitly_set_values(self):
dashboard = Dashboard(name="foo")
self.assertEqual(dashboard.name, 'foo')
self.assertIsNone(dashboard.key)
self.assertIsNone(dashboard.description)
self.assertFalse(dashboard.show_header)
self.assertEqual(dashboard.color_mode, 'dark')
self.assertTrue(dashboard.block_borders)
self.assertIsNone(dashboard.header_image_url)
self.assertIsNone(dashboard.blocks)
""" Let's make sure feeds are explicitly set from within the model:
Block.__new__.__defaults__ = (None, None, None {}, None)
"""
def test_block_have_explicitly_set_values(self):
block = Block(name="foo")
self.assertEqual(block.name, 'foo')
self.assertIsNone(block.id)
self.assertIsNone(block.visual_type)
self.assertEqual(type(block.properties), dict)
self.assertEqual(len(block.properties), 0)
self.assertIsNone(block.block_feeds)
def test_layout_properties_are_optional(self):
layout = Layout()
self.assertIsNone(layout.xl)
self.assertIsNone(layout.lg)
self.assertIsNone(layout.md)
self.assertIsNone(layout.sm)
self.assertIsNone(layout.xs)
def test_from_dict_ignores_unknown_items(self): def test_from_dict_ignores_unknown_items(self):
data = Data.from_dict({'value': 'foo', 'feed_id': 10, 'unknown_param': 42}) data = Data.from_dict({'value': 'foo', 'feed_id': 10, 'unknown_param': 42})