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)