Skip to content
Snippets Groups Projects
Verified Commit 913e60d0 authored by David Hoese's avatar David Hoese
Browse files

Switch data api tests to pytest

parent c5789b77
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
metobsapi/tests/test_data_api.py
+ 123
− 118
View file @ 913e60d0
import json
import unittest
from unittest import mock
import metobsapi
import pytest
def fake_data(interval, symbols, num_vals, single_result=False):
......@@ -53,121 +52,129 @@ def fake_data(interval, symbols, num_vals, single_result=False):
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()
@pytest.fixture
def app():
from metobsapi import app as real_app
def test_doc(self):
res = self.app.get("/api/data")
real_app.config.update({"TESTING": True, "DEBUG": True})
yield real_app
@pytest.fixture
def client(app):
return app.test_client()
class TestDataAPI:
def test_doc(self, client):
res = client.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_format(self, client):
res = client.get("/api/data.fake")
assert b"No data file format" in res.data
def test_bad_begin_json(self):
res = self.app.get("/api/data.json?symbols=air_temp&begin=blah")
def test_bad_begin_json(self, client):
res = client.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"])
assert res["code"] == 400
assert res["status"] == "error"
assert "timestamp" in res["message"]
def test_bad_order(self):
res = self.app.get("/api/data.json?order=blah&symbols=air_temp")
def test_bad_order(self, client):
res = client.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"])
assert "column" in res["message"]
assert "row" in res["message"]
def test_bad_epoch(self):
res = self.app.get("/api/data.json?epoch=blah&symbols=air_temp")
def test_bad_epoch(self, client):
res = client.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"])
assert "'h'" in res["message"]
assert "'m'" in res["message"]
assert "'s'" in res["message"]
assert "'u'" in res["message"]
def test_bad_interval(self):
res = self.app.get("/api/data.json?interval=blah&symbols=air_temp")
def test_bad_interval(self, client):
res = client.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"])
assert "'1m'" in res["message"]
assert "'5m'" in res["message"]
assert "'1h'" in res["message"]
def test_missing_inst(self):
res = self.app.get("/api/data.json?site=X&symbols=air_temp&begin=-05:00:00")
def test_missing_inst(self, client):
res = client.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"])
assert res["code"] == 400
assert res["status"] == "error"
assert "'site'" in res["message"]
assert "'inst'" in res["message"]
def test_missing_site(self):
res = self.app.get("/api/data.json?inst=X&symbols=air_temp&begin=-05:00:00")
def test_missing_site(self, client):
res = client.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"])
assert res["code"] == 400
assert res["status"] == "error"
assert "'site'" in res["message"]
assert "'inst'" in res["message"]
def test_missing_symbols(self):
res = self.app.get("/api/data.json?begin=-05:00:00")
def test_missing_symbols(self, client):
res = client.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"])
assert res["code"] == 400
assert res["status"] == "error"
assert "'symbols'" in 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)
def test_too_many_points(self, client):
res = client.get("/api/data.json?symbols=aoss.tower.air_temp&begin=1970-01-01T00:00:00")
assert 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")
assert "too many values" in res["message"]
assert res["code"] == 413
assert res["status"] == "fail"
@mock.patch("metobsapi.data_api.query")
def test_shorthand_one_symbol_json_row(self, query_func):
def test_shorthand_one_symbol_json_row(self, query_func, client):
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 = client.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)
assert res["code"] == 200
assert res["num_results"] == 9
assert res["results"]["symbols"] == ["air_temp"]
assert len(res["results"]["timestamps"]) == 9
assert len(res["results"]["data"]) == 9
assert len(res["results"]["data"][0]) == 1
@mock.patch("metobsapi.data_api.query")
def test_shorthand_one_symbol_json_column(self, query_func):
def test_shorthand_one_symbol_json_column(self, query_func, client):
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 = client.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)
assert res["code"] == 200
assert res["num_results"] == 9
assert "air_temp" in res["results"]["data"]
assert len(res["results"]["data"]["air_temp"]) == 9
assert len(res["results"]["timestamps"]) == 9
@mock.patch("metobsapi.data_api.query")
def test_wind_speed_direction_json(self, query_func):
def test_wind_speed_direction_json(self, query_func, client):
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(
res = client.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)
assert res["code"] == 200
assert res["num_results"] == 9
assert "aoss.tower.wind_direction" in res["results"]["data"]
assert "aoss.tower.wind_speed" in res["results"]["data"]
assert len(list(res["results"]["data"].keys())) == 2
@mock.patch("metobsapi.data_api.query")
def test_one_symbol_two_insts_json_row(self, query_func):
def test_one_symbol_two_insts_json_row(self, query_func, client):
r = fake_data(
"1m",
{
......@@ -178,17 +185,17 @@ class TestDataAPI(unittest.TestCase):
)
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 = client.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)
assert res["code"] == 200
assert res["num_results"] == 9
assert res["results"]["symbols"] == ["aoss.tower.air_temp", "mendota.buoy.air_temp"]
assert len(res["results"]["timestamps"]) == 9
assert len(res["results"]["data"]) == 9
assert len(res["results"]["data"][0]) == 2
@mock.patch("metobsapi.data_api.query")
def test_one_symbol_three_insts_json_row(self, query_func):
def test_one_symbol_three_insts_json_row(self, query_func, client):
r = fake_data(
"1m",
{
......@@ -207,89 +214,87 @@ class TestDataAPI(unittest.TestCase):
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(
res = client.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)
assert res["code"] == 200
assert res["num_results"] == 9
assert res["results"]["symbols"] == ["site1.inst1.air_temp", "site2.inst2.air_temp", "site3.inst3.air_temp"]
assert len(res["results"]["timestamps"]) == 9
assert len(res["results"]["data"]) == 9
assert len(res["results"]["data"][0]) == 3
@mock.patch("metobsapi.data_api.query")
def test_one_symbol_csv(self, query_func):
def test_one_symbol_csv(self, query_func, client):
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 = client.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)
assert len(lines) == 5 + 9 + 1
# time + 1 channel
self.assertEqual(len(lines[5].split(",")), 2)
self.assertIn("# code: 200", res)
assert len(lines[5].split(",")) == 2
assert "# code: 200" in res
@mock.patch("metobsapi.data_api.query")
def test_one_symbol_xml(self, query_func):
def test_one_symbol_xml(self, query_func, client):
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 = client.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)
assert len(res.childNodes[0].childNodes[0].childNodes) == 2
# data rows
self.assertEqual(len(res.childNodes[0].childNodes[1].childNodes), 9)
assert len(res.childNodes[0].childNodes[1].childNodes) == 9
@mock.patch("metobsapi.data_api.query")
def test_three_symbol_csv(self, query_func):
def test_three_symbol_csv(self, query_func, client):
"""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(
res = client.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)
assert len(lines) == 5 + 9 + 1
# time + 3 channels
self.assertEqual(len(lines[5].split(",")), 4)
self.assertIn("# code: 200", res)
assert len(lines[5].split(",")) == 4
assert "# code: 200" in res
@mock.patch("metobsapi.data_api.query")
def test_three_symbol_csv_repeat(self, query_func):
def test_three_symbol_csv_repeat(self, query_func, client):
"""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(
res = client.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)
assert len(lines) == 5 + 1 + 1
# time + 1 channel
self.assertEqual(len(lines[5].split(",")), 1)
self.assertIn("# code: 400", res)
assert len(lines[5].split(",")) == 1
assert "# code: 400" in res
# @mock.patch('metobsapi.data_api.query')
# def test_jsonp_bad_symbol_400(self, query_func):
# def test_jsonp_bad_symbol_400(self, query_func, client):
# 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 = client.get('/api/data.json?site=aoss&inst=tower&symbols=bad&begin=-00:10:00&callback=test')
# assert res.status_code == 400
# res = res.data.decode()
# self.assertEqual(res['code'], 400)
# assert res['code'] == 400
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment