-
David Hoese authoredDavid Hoese authored
test_data_api.py 11.91 KiB
import json
import unittest
from unittest import mock
import metobsapi
def fake_data(interval, symbols, num_vals, single_result=False):
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)
)
# make up some Nones/nulls (but not all the time)
r_idx = int(random.random() * len(columns) * 3)
# don't include time (index 0)
if 0 < r_idx < len(columns):
vals[-1][r_idx] = None
s = {
'name': measurement_name,
'columns': columns,
'tags': tags,
'values': vals,
}
if single_result:
series.append(s)
else:
series.append(ResultSet({
'series': [s],
'statement_id': 0,
}))
if single_result:
ret = {
'series': series,
'statement_id': 0,
}
return ResultSet(ret)
else:
return series
class TestDataAPI(unittest.TestCase):
def setUp(self):
metobsapi.app.config['TESTING'] = True
metobsapi.app.config['DEBUG'] = 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'])
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'])
def test_missing_symbols(self):
res = self.app.get('/api/data.json?begin=-05:00:00')
res = json.loads(res.data.decode())
self.assertEqual(res['code'], 400)
self.assertEqual(res['status'], 'error')
self.assertIn('\'symbols\'', res['message'])
def test_too_many_points(self):
res = self.app.get('/api/data.json?symbols=aoss.tower.air_temp&begin=1970-01-01T00:00:00')
self.assertEqual(res.status_code, 413)
res = json.loads(res.data.decode())
self.assertIn('too many values', res['message'])
self.assertEqual(res['code'], 413)
self.assertEqual(res['status'], 'fail')
@mock.patch('metobsapi.data_api.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'], ['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.data_api.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('air_temp', res['results']['data'])
self.assertEqual(len(res['results']['data']['air_temp']), 9)
self.assertEqual(len(res['results']['timestamps']), 9)
@mock.patch('metobsapi.data_api.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.data_api.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.data_api.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.data_api.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
lines = res.split('\n')
self.assertEqual(len(lines), 5 + 9 + 1)
# time + 1 channel
self.assertEqual(len(lines[5].split(',')), 2)
self.assertIn("# code: 200", res)
@mock.patch('metobsapi.data_api.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)
@mock.patch('metobsapi.data_api.query')
def test_three_symbol_csv(self, query_func):
"""Test that multiple channels in a CSV file are structured properly."""
r = fake_data('1m', {('aoss', 'tower'): ['time', 'air_temp', 'rel_hum', 'wind_speed']}, 9)
query_func.return_value = r
# row should be the default
res = self.app.get('/api/data.csv?symbols=aoss.tower.air_temp:'
'aoss.tower.rel_hum:aoss.tower.wind_speed&begin=-00:10:00')
res = res.data.decode()
# header, data, newline at end
lines = res.split('\n')
self.assertEqual(len(lines), 5 + 9 + 1)
# time + 3 channels
self.assertEqual(len(lines[5].split(',')), 4)
self.assertIn("# code: 200", res)
@mock.patch('metobsapi.data_api.query')
def test_three_symbol_csv_repeat(self, query_func):
"""Test that multiple channels in a CSV file are structured properly."""
r = fake_data('1m', {('aoss', 'tower'): ['time', 'air_temp', 'rel_hum', 'wind_speed']}, 9)
query_func.return_value = r
# row should be the default
res = self.app.get('/api/data.csv?symbols=aoss.tower.air_temp:'
'aoss.tower.air_temp:aoss.tower.air_temp&begin=-00:10:00')
res = res.data.decode()
# header, data, newline at end
lines = res.split('\n')
# header, data (one empty line), newline at end
self.assertEqual(len(lines), 5 + 1 + 1)
# time + 1 channel
self.assertEqual(len(lines[5].split(',')), 1)
self.assertIn("# code: 400", res)
# @mock.patch('metobsapi.data_api.query')
# def test_jsonp_bad_symbol_400(self, query_func):
# XXX: Not currently possible with flask-json
# 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=bad&begin=-00:10:00&callback=test')
# self.assertEqual(res.status_code, 400)
# res = res.data.decode()
# self.assertEqual(res['code'], 400)