forked from quandl/quandl-python
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_datatable_data.py
172 lines (149 loc) · 7.13 KB
/
test_datatable_data.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
import re
import unittest
import httpretty
import json
import pandas
import numpy
import six
from quandl.model.data import Data
from quandl.model.datatable import Datatable
from quandl.utils.request_type_util import RequestType
from mock import patch, call
from test.factories.datatable_data import DatatableDataFactory
from test.factories.datatable_meta import DatatableMetaFactory
from parameterized import parameterized
class DatatableDataTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.expected_column_names = [six.u('per_end_date'),
six.u('ticker'),
six.u('tot_oper_exp')]
cls.expected_column_types = [six.u('Date'),
six.u('String'),
six.u('String')]
cls.data_object = Data(['2015-07-11', 'AAPL', 440.0],
meta={'columns': cls.expected_column_names,
'column_types': cls.expected_column_types})
def test_to_pandas_returns_pandas_dataframe_object(self):
data = self.data_object.to_pandas()
self.assertIsInstance(data, pandas.core.frame.DataFrame)
# don't set dataFrame for datatable
def test_pandas_dataframe_index_is_datetime(self):
df = self.data_object.to_pandas()
self.assertEqual(df.index.name, 'None')
def test_to_numpys_returns_numpy_object(self):
data = self.data_object.to_numpy()
self.assertIsInstance(data, numpy.core.records.recarray)
def test_to_csv_returns_expected_csv(self):
data = self.data_object.to_csv()
expected = "None,per_end_date,ticker,tot_oper_exp\n0,2015-07-11,AAPL,440.0\n"
self.assertEqual(data, expected)
def test_column_names(self):
actual = self.data_object.columns
self.assertEqual(actual, self.expected_column_names)
def test_meta_exists(self):
self.assertIsNotNone(self.data_object.meta)
class ListDatatableDataTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
httpretty.enable()
datatable_data = {'datatable': DatatableDataFactory.build()}
meta = {'meta': DatatableMetaFactory.build()}
datatable_data.update(meta)
httpretty.register_uri(httpretty.GET,
re.compile(
'https://data.nasdaq.com/api/v3/datatables/*'),
body=json.dumps(datatable_data))
httpretty.register_uri(httpretty.POST,
re.compile(
'https://data.nasdaq.com/api/v3/datatables/*'),
body=json.dumps(datatable_data))
cls.expected_raw_data = []
cls.expected_list_values = []
@classmethod
def tearDownClass(cls):
httpretty.disable()
httpretty.reset()
def tearDown(self):
RequestType.USE_GET_REQUEST = True
@patch('quandl.connection.Connection.request')
def test_data_calls_connection_get(self, mock):
datatable = Datatable('ZACKS/FC')
Data.page(datatable, params={'ticker': ['AAPL', 'MSFT'],
'per_end_date': {'gte': '2015-01-01'},
'qopts': {'columns': ['ticker', 'per_end_date']}})
expected = call('get', 'datatables/ZACKS/FC',
params={'ticker[]': ['AAPL', 'MSFT'],
'per_end_date.gte': '2015-01-01',
'qopts.columns[]': ['ticker', 'per_end_date']})
self.assertEqual(mock.call_args, expected)
@patch('quandl.connection.Connection.request')
def test_data_calls_connection_post(self, mock):
RequestType.USE_GET_REQUEST = False
datatable = Datatable('ZACKS/FC')
Data.page(datatable, params={'ticker': ['AAPL', 'MSFT'],
'per_end_date': {'gte': '2015-01-01'},
'qopts': {'columns': ['ticker', 'per_end_date']}})
expected = call('post', 'datatables/ZACKS/FC',
json={'ticker': ['AAPL', 'MSFT'],
'per_end_date.gte': '2015-01-01',
'qopts.columns': ['ticker', 'per_end_date']})
self.assertEqual(mock.call_args, expected)
@parameterized.expand(['GET', 'POST'])
def test_values_and_meta_exist(self, request_method):
if request_method == 'POST':
RequestType.USE_GET_REQUEST = False
datatable = Datatable('ZACKS/FC')
results = Data.page(datatable, params={})
self.assertIsNotNone(results.values)
self.assertIsNotNone(results.meta)
@parameterized.expand(['GET', 'POST'])
def test_to_pandas_returns_pandas_dataframe_object(self, request_method):
if request_method == 'POST':
RequestType.USE_GET_REQUEST = False
datatable = Datatable('ZACKS/FC')
results = Data.page(datatable, params={})
df = results.to_pandas()
self.assertIsInstance(df, pandas.core.frame.DataFrame)
# no index is set for datatable.to_pandas
@parameterized.expand(['GET', 'POST'])
def test_pandas_dataframe_index_is_none(self, request_method):
if request_method == 'POST':
RequestType.USE_GET_REQUEST = False
datatable = Datatable('ZACKS/FC')
results = Data.page(datatable, params={})
df = results.to_pandas()
self.assertEqual(df.index.name, 'None')
# if datatable has Date field then it should be convert to pandas datetime
@parameterized.expand(['GET', 'POST'])
def test_pandas_dataframe_date_field_is_datetime(self, request_method):
if request_method == 'POST':
RequestType.USE_GET_REQUEST = False
datatable = Datatable('ZACKS/FC')
results = Data.page(datatable, params={})
df = results.to_pandas()
self.assertIsInstance(df['per_end_date'][0], pandas.datetime)
self.assertIsInstance(df['per_end_date'][1], pandas.datetime)
self.assertIsInstance(df['per_end_date'][2], pandas.datetime)
self.assertIsInstance(df['per_end_date'][3], pandas.datetime)
@parameterized.expand(['GET', 'POST'])
def test_to_numpy_returns_numpy_object(self, request_method):
if request_method == 'POST':
RequestType.USE_GET_REQUEST = False
datatable = Datatable('ZACKS/FC')
results = Data.page(datatable, params={})
data = results.to_numpy()
self.assertIsInstance(data, numpy.core.records.recarray)
@parameterized.expand(['GET', 'POST'])
def test_to_csv_returns_expected_csv(self, request_method):
if request_method == 'POST':
RequestType.USE_GET_REQUEST = False
datatable = Datatable('ZACKS/FC')
results = Data.page(datatable, params={})
data = results.to_csv()
expected = "None,per_end_date,ticker,tot_oper_exp\n" + \
"0,2015-07-11,AAPL,456.9\n" + \
"1,2015-07-13,433.3,\n" + \
"2,2015-07-14,AAPL,419.1\n" + \
"3,2015-07-15,476.5,\n"
self.assertEqual(data, expected)