diff --git a/test/WebService_test.py b/test/WebService_test.py
index f72f7f590a32d04c7d20d292edfd67dfff0fa440..8641f809b72760340e8ef3e4c0a9fe230445557d 100644
--- a/test/WebService_test.py
+++ b/test/WebService_test.py
@@ -1,76 +1,154 @@
-"""Tests for WebService.py"""
-
+"""Unit Tests for WebService"""
+from cgi import parse_qs
+from datetime import datetime
from nose.tools import assert_equals
+from nose.tools import assert_raises
+from nose.tools import assert_is_instance
import sys
from webtest import TestApp
from geomagio.edge import EdgeFactory
+from geomagio.WebService import _get_param
from geomagio.WebService import WebService
+from obspy.core.stream import Stream
+from obspy.core.utcdatetime import UTCDateTime
+
+
+APP = TestApp(WebService(EdgeFactory()))
+
+
+def test__get_param():
+ """WebService_test.test__get_param()
+
+ Call function _get_param to make certain it gets back
+ the appropriate values and raises exceptions for invalid values.
+ """
+ params = {
+ 'id': None,
+ 'elements': 'H,E,Z,F',
+ 'sampling_period': ['1', '60'],
+ }
+ assert_raises(Exception, _get_param, params, 'id', required=True)
+ elements = _get_param(params, 'elements')
+ assert_equals(elements, 'H,E,Z,F')
+ assert_raises(Exception, _get_param, params, 'sampling_period')
+
+def test_fetch():
+ """WebService_test.test_fetch())
+
+ Call function WebService.fetch to confirm tht it returns an
+ obspy.core.stream object.
+ """
+ service = WebService(EdgeFactory())
+ query = service.parse(parse_qs('id=BOU&starttime=2016-06-06'
+ '&endtime=2016-06-07&elements=H,E,Z,F&sampling_period=60'
+ '&format=iaga2002&type=variation'))
+ timeseries = service.fetch(query)
+ assert_is_instance(timeseries, Stream)
+
+
+def test_parse():
+ """WebService_test.test_parse()
+
+ Create WebService instance and call parse to confirm that query
+ string values are applied to the correct class attribute. Also
+ confirm that default values are applied correctly.
+ """
+ service = WebService(EdgeFactory())
+ query = service.parse(parse_qs('id=BOU&starttime=2016-06-06'
+ '&endtime=2016-06-07&elements=H,E,Z,F&sampling_period=60'
+ '&format=iaga2002&type=variation'))
+ assert_equals(query.observatory_id, 'BOU')
+ assert_equals(query.starttime, UTCDateTime(2016, 6, 6, 0))
+ assert_equals(query.endtime, UTCDateTime(2016, 6, 7, 0))
+ assert_equals(query.elements, ['H', 'E', 'Z', 'F'])
+ assert_equals(query.sampling_period, '60')
+ assert_equals(query.output_format, 'iaga2002')
+ assert_equals(query.data_type, 'variation')
+ # Test that defaults are set for unspecified values
+ now = datetime.now()
+ today = UTCDateTime(year=now.year, month=now.month, day=now.day, hour=0)
+ tomorrow = today + (24 * 60 * 60 - 1)
+ query = service.parse(parse_qs('id=BOU'))
+ assert_equals(query.observatory_id, 'BOU')
+ assert_equals(query.starttime, today)
+ assert_equals(query.endtime, tomorrow)
+ assert_equals(query.elements, ('X', 'Y', 'Z', 'F'))
+ assert_equals(query.sampling_period, '60')
+ assert_equals(query.output_format, 'iaga2002')
+ assert_equals(query.data_type, 'variation')
-app = TestApp(WebService(EdgeFactory()))
+def test_requests():
+ """WebService_test.test_requests()
-def test_bad_requests():
- """WebService_test.test_bad_requests()
+ Use TestApp to confirm correct response status, status int,
+ and content-type.
"""
- # Check bad request (values)
- bad_values = [
+ # Check invalid request (bad values)
+ invalid_values = [
'/',
'/?id=bad',
'/?id=BOU&starttime=2017-50-50',
'/?id=BOU&starttime=2016-06-06&endtime=2017-50-50',
'/?id=BOU&starttime=2016-06-06&endtime=2016-06-05',
'/?id=BOU&elements=H,E,D,Z,F',
+ '/?id=BOU&sampling_period=20',
'/?id=BOU&type=bad',
'/?id=BOU&format=bad'
]
- for bad_val in bad_values:
+ for invalid_val in invalid_values:
try:
- response = app.get(bad_val, expect_errors=True)
+ response = APP.get(invalid_val, expect_errors=True)
assert_equals(response.status_int, 400)
+ assert_equals(response.status, '400 Bad Request')
+ assert_equals(response.content_type, 'text/plain')
except Exception:
exception = sys.exc_info()[1]
message = exception.args[0]
- print('FAIL: ' + message + '\nrequest = ' + bad_val)
- # Check bad request (duplicates)
+ print('FAIL: ' + message + '\nrequest = ' + invalid_val)
+ # Check invalid request (duplicates)
duplicate_values = [
'/?id=BOU&id=BOU',
'/?id=BOU&starttime=2016-06-06&starttime=2016-06-06',
'/?id=BOU&endtime=2017-50-50&endtime=2017-50-50',
'/?id=BOU&elements=H,E,Z&elements=H,E,Z',
+ '/?id=BOU&sampling_period=1&sampling_period=60',
'/?id=BOU&type=variation&type=variation',
'/?id=BOU&format=iaga2002&format=iaga2002'
]
for duplicate in duplicate_values:
try:
- response = app.get(duplicate, expect_errors=True)
+ response = APP.get(duplicate, expect_errors=True)
assert_equals(response.status_int, 400)
+ assert_equals(response.status, '400 Bad Request')
+ assert_equals(response.content_type, 'text/plain')
except Exception:
exception = sys.exc_info()[1]
message = exception.args[0]
print('FAIL: ' + message + '\nrequest = ' + duplicate)
-
-def test_good_requests():
- """WebService_test.test_good_requests()
- """
- # Check for upper and lower case
- good_requests = [
+ # Check valid request (upper and lower case)
+ valid_requests = [
'/?id=BOU',
'/?id=bou',
'/?id=bou&starttime=2016-06-06',
'/?id=bou&starttime=2016-06-06&endtime=2016-06-07',
'/?id=bou&elements=H,E,Z,F',
'/?id=bou&elements=h,e,z,f',
+ '/?id=BOU&sampling_period=1',
+ '/?id=BOU&sampling_period=60',
'/?id=BOU&type=variation',
'/?id=BOU&type=VARIATION',
'/?id=BOU&format=iaga2002',
'/?id=BOU&format=IAGA2002'
]
- for request in good_requests:
+ for request in valid_requests:
try:
- response = app.get(request)
+ response = APP.get(request)
assert_equals(response.status_int, 200)
+ assert_equals(response.status, '200 OK')
+ assert_equals(response.content_type, 'text/plain')
except Exception:
exception = sys.exc_info()[1]
message = exception.args[0]