-
David Hoese authoredDavid Hoese authored
test_data_api.py 8.61 KiB
import json
import unittest
from unittest import mock
import metobsapi
def fake_data(interval, symbols, num_vals):
import random
from datetime import datetime, timedelta
from influxdb.resultset import ResultSet
now = datetime(2017, 3, 5, 19, 0, 0)
t_format = "%Y-%m-%dT%H:%M:%SZ"
measurement_name = "metobs_" + interval
series = []
for (site, inst), columns in symbols.items():
tags = {'site': site, 'inst': inst}
vals = []
for i in range(num_vals):
vals.append(
[(now + timedelta(minutes=i)).strftime(t_format)] + \
[random.random()] * (len(columns) - 1)
)
s = {
'name': measurement_name,
'columns': columns,
'tags': tags,
'values': vals,
}
series.append(s)
ret = {
'series': series,
'statement_id': 0,
}
return ResultSet(ret)
class TestDataAPI(unittest.TestCase):
def setUp(self):
metobsapi.app.config['TESTING'] = True
self.app = metobsapi.app.test_client()
def test_doc(self):
res = self.app.get('/api/data')
assert b'Data Request Application' in res.data
def test_bad_format(self):
res = self.app.get('/api/data.fake')
self.assertIn(b'No data file format', res.data)
def test_bad_begin_json(self):
res = self.app.get('/api/data.json?symbols=air_temp&begin=blah')
res = json.loads(res.data.decode())
self.assertEqual(res['code'], 400)
self.assertEqual(res['status'], 'error')
self.assertIn('timestamp', res['message'])
def test_bad_order(self):
res = self.app.get('/api/data.json?order=blah&symbols=air_temp')
res = json.loads(res.data.decode())
self.assertIn('column', res['message'])
self.assertIn('row', res['message'])
def test_bad_epoch(self):
res = self.app.get('/api/data.json?epoch=blah&symbols=air_temp')
res = json.loads(res.data.decode())
self.assertIn('\'h\'', res['message'])
self.assertIn('\'m\'', res['message'])
self.assertIn('\'s\'', res['message'])
self.assertIn('\'u\'', res['message'])
def test_bad_interval(self):
res = self.app.get('/api/data.json?interval=blah&symbols=air_temp')
res = json.loads(res.data.decode())
self.assertIn('\'1m\'', res['message'])
self.assertIn('\'5m\'', res['message'])
self.assertIn('\'1h\'', res['message'])
self.assertIn('\'24h\'', res['message'])
def test_missing_inst(self):
res = self.app.get('/api/data.json?site=X&symbols=air_temp&begin=-05:00:00')
res = json.loads(res.data.decode())
self.assertEqual(res['code'], 400)
self.assertEqual(res['status'], 'error')
self.assertIn('\'site\'', res['message'])
self.assertIn('\'inst\'', res['message'])
def test_missing_site(self):
res = self.app.get('/api/data.json?inst=X&symbols=air_temp&begin=-05:00:00')
res = json.loads(res.data.decode())
self.assertEqual(res['code'], 400)
self.assertEqual(res['status'], 'error')
self.assertIn('\'site\'', res['message'])
self.assertIn('\'inst\'', res['message'])
@mock.patch('metobsapi.modifyData.query')
def test_shorthand_one_symbol_json_row(self, query_func):
r = fake_data('1m', {('aoss', 'tower'): ['time', 'air_temp']}, 9)
query_func.return_value = r
# row should be the default
res = self.app.get('/api/data.json?site=aoss&inst=tower&symbols=air_temp&begin=-00:10:00')
res = json.loads(res.data.decode())
self.assertEqual(res['code'], 200)
self.assertEqual(res['num_results'], 9)
self.assertListEqual(res['results']['symbols'], ['aoss.tower.air_temp'])
self.assertEqual(len(res['results']['timestamps']), 9)
self.assertEqual(len(res['results']['data']), 9)
self.assertEqual(len(res['results']['data'][0]), 1)
@mock.patch('metobsapi.modifyData.query')
def test_shorthand_one_symbol_json_column(self, query_func):
r = fake_data('1m', {('aoss', 'tower'): ['time', 'air_temp']}, 9)
query_func.return_value = r
res = self.app.get('/api/data.json?site=aoss&inst=tower&symbols=air_temp&begin=-00:10:00&order=column')
res = json.loads(res.data.decode())
self.assertEqual(res['code'], 200)
self.assertEqual(res['num_results'], 9)
self.assertIn('aoss.tower.air_temp', res['results']['data'])
self.assertEqual(len(res['results']['data']['aoss.tower.air_temp']), 9)
self.assertEqual(len(res['results']['timestamps']), 9)
@mock.patch('metobsapi.modifyData.query')
def test_wind_speed_direction_json(self, query_func):
r = fake_data('1m', {('aoss', 'tower'): ['time', 'wind_speed', 'wind_direction', 'wind_east', 'wind_north']}, 9)
query_func.return_value = r
res = self.app.get('/api/data.json?symbols=aoss.tower.wind_speed:aoss.tower.wind_direction&begin=-00:10:00&order=column')
res = json.loads(res.data.decode())
self.assertEqual(res['code'], 200)
self.assertEqual(res['num_results'], 9)
self.assertIn('aoss.tower.wind_direction', res['results']['data'])
self.assertIn('aoss.tower.wind_speed', res['results']['data'])
self.assertEqual(len(list(res['results']['data'].keys())), 2)
@mock.patch('metobsapi.modifyData.query')
def test_one_symbol_two_insts_json_row(self, query_func):
r = fake_data('1m', {
('aoss', 'tower'): ['time', 'air_temp'],
('mendota', 'buoy'): ['time', 'air_temp'],
}, 9)
query_func.return_value = r
# row should be the default
res = self.app.get('/api/data.json?symbols=aoss.tower.air_temp:mendota.buoy.air_temp&begin=-00:10:00')
res = json.loads(res.data.decode())
self.assertEqual(res['code'], 200)
self.assertEqual(res['num_results'], 9)
self.assertListEqual(res['results']['symbols'], ['aoss.tower.air_temp', 'mendota.buoy.air_temp'])
self.assertEqual(len(res['results']['timestamps']), 9)
self.assertEqual(len(res['results']['data']), 9)
self.assertEqual(len(res['results']['data'][0]), 2)
@mock.patch('metobsapi.modifyData.query')
def test_one_symbol_three_insts_json_row(self, query_func):
r = fake_data('1m', {
('site1', 'inst1'): ['time', 'air_temp'],
('site2', 'inst2'): ['time', 'air_temp'],
('site3', 'inst3'): ['time', 'air_temp'],
}, 9)
query_func.return_value = r
# row should be the default
from metobsapi.util.data_responses import SYMBOL_TRANSLATIONS as st
st = st.copy()
st[('site1', 'inst1')] = st[('aoss', 'tower')]
st[('site2', 'inst2')] = st[('aoss', 'tower')]
st[('site3', 'inst3')] = st[('aoss', 'tower')]
with mock.patch('metobsapi.util.data_responses.SYMBOL_TRANSLATIONS', st):
res = self.app.get('/api/data.json?symbols=site1.inst1.air_temp:site2.inst2.air_temp:site3.inst3.air_temp&begin=-00:10:00')
res = json.loads(res.data.decode())
self.assertEqual(res['code'], 200)
self.assertEqual(res['num_results'], 9)
self.assertListEqual(res['results']['symbols'], ['site1.inst1.air_temp', 'site2.inst2.air_temp', 'site3.inst3.air_temp'])
self.assertEqual(len(res['results']['timestamps']), 9)
self.assertEqual(len(res['results']['data']), 9)
self.assertEqual(len(res['results']['data'][0]), 3)
@mock.patch('metobsapi.modifyData.query')
def test_one_symbol_csv(self, query_func):
r = fake_data('1m', {('aoss', 'tower'): ['time', 'air_temp']}, 9)
query_func.return_value = r
# row should be the default
res = self.app.get('/api/data.csv?symbols=aoss.tower.air_temp&begin=-00:10:00')
res = res.data.decode()
# header, data, newline at end
self.assertEqual(len(res.split('\n')), 5 + 9 + 1)
self.assertIn("# code: 200", res)
@mock.patch('metobsapi.modifyData.query')
def test_one_symbol_xml(self, query_func):
from xml.dom.minidom import parseString
r = fake_data('1m', {('aoss', 'tower'): ['time', 'air_temp']}, 9)
query_func.return_value = r
# row should be the default
res = self.app.get('/api/data.xml?symbols=aoss.tower.air_temp&begin=-00:10:00')
res = parseString(res.data.decode())
# symbols: time and air_temp
self.assertEqual(len(res.childNodes[0].childNodes[0].childNodes), 2)
# data rows
self.assertEqual(len(res.childNodes[0].childNodes[1].childNodes), 9)