Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
unit_test.nse_oil = {'dataset': DatasetFactory.build(
database_code='NSE', dataset_code='OIL')}
unit_test.wiki_aapl = {'dataset': DatasetFactory.build(
database_code='WIKI', dataset_code='AAPL')}
unit_test.wiki_msft = {'dataset': DatasetFactory.build(
database_code='WIKI', dataset_code='MSFT',
newest_available_date='2015-07-30', oldest_available_date='2013-01-01')}
unit_test.single_col = {'dataset': DatasetFactory.build(
database_code='SINGLE', dataset_code='COLUMN',
newest_available_date='2015-07-30', oldest_available_date='2013-01-01')}
unit_test.oil_obj = Dataset('NSE/OIL', unit_test.nse_oil['dataset'])
unit_test.aapl_obj = Dataset('WIKI/AAPL', unit_test.wiki_aapl['dataset'])
unit_test.wiki_obj = Dataset('WIKI/MSFT', unit_test.wiki_msft['dataset'])
unit_test.single_col_obj = Dataset('SINGLE/COLUMN', unit_test.single_col['dataset'])
httpretty.register_uri(httpretty.GET,
re.compile(
'https://www.quandl.com/api/v3/datasets/.*/metadata'),
responses=[httpretty.Response(body=json.dumps(dataset))
for dataset in
[unit_test.nse_oil, unit_test.wiki_aapl,
unit_test.wiki_msft]])
# mock our query param column_index request
httpretty.register_uri(httpretty.GET,
"https://www.quandl.com/api/v3/datasets/SINGLE/COLUMN/data",
body=json.dumps(unit_test.single_dataset_data))
httpretty.register_uri(httpretty.GET,
def test_dataset_calls_connection(self, mock):
d = Dataset('NSE/OIL')
d.data_fields()
expected = call('get', 'datasets/NSE/OIL/metadata', params={})
self.assertEqual(mock.call_args, expected)
database_code='NSE', dataset_code='OIL')}
unit_test.wiki_aapl = {'dataset': DatasetFactory.build(
database_code='WIKI', dataset_code='AAPL')}
unit_test.wiki_msft = {'dataset': DatasetFactory.build(
database_code='WIKI', dataset_code='MSFT',
newest_available_date='2015-07-30', oldest_available_date='2013-01-01')}
unit_test.single_col = {'dataset': DatasetFactory.build(
database_code='SINGLE', dataset_code='COLUMN',
newest_available_date='2015-07-30', oldest_available_date='2013-01-01')}
unit_test.oil_obj = Dataset('NSE/OIL', unit_test.nse_oil['dataset'])
unit_test.aapl_obj = Dataset('WIKI/AAPL', unit_test.wiki_aapl['dataset'])
unit_test.wiki_obj = Dataset('WIKI/MSFT', unit_test.wiki_msft['dataset'])
unit_test.single_col_obj = Dataset('SINGLE/COLUMN', unit_test.single_col['dataset'])
httpretty.register_uri(httpretty.GET,
re.compile(
'https://www.quandl.com/api/v3/datasets/.*/metadata'),
responses=[httpretty.Response(body=json.dumps(dataset))
for dataset in
[unit_test.nse_oil, unit_test.wiki_aapl,
unit_test.wiki_msft]])
# mock our query param column_index request
httpretty.register_uri(httpretty.GET,
"https://www.quandl.com/api/v3/datasets/SINGLE/COLUMN/data",
body=json.dumps(unit_test.single_dataset_data))
httpretty.register_uri(httpretty.GET,
"https://www.quandl.com/api/v3/datasets/WIKI/AAPL/data",
body=json.dumps(unit_test.dataset_data))
unit_test.wiki_aapl = {'dataset': DatasetFactory.build(
database_code='WIKI', dataset_code='AAPL')}
unit_test.wiki_msft = {'dataset': DatasetFactory.build(
database_code='WIKI', dataset_code='MSFT',
newest_available_date='2015-07-30', oldest_available_date='2013-01-01')}
unit_test.single_col = {'dataset': DatasetFactory.build(
database_code='SINGLE', dataset_code='COLUMN',
newest_available_date='2015-07-30', oldest_available_date='2013-01-01')}
unit_test.oil_obj = Dataset('NSE/OIL', unit_test.nse_oil['dataset'])
unit_test.aapl_obj = Dataset('WIKI/AAPL', unit_test.wiki_aapl['dataset'])
unit_test.wiki_obj = Dataset('WIKI/MSFT', unit_test.wiki_msft['dataset'])
unit_test.single_col_obj = Dataset('SINGLE/COLUMN', unit_test.single_col['dataset'])
httpretty.register_uri(httpretty.GET,
re.compile(
'https://www.quandl.com/api/v3/datasets/.*/metadata'),
responses=[httpretty.Response(body=json.dumps(dataset))
for dataset in
[unit_test.nse_oil, unit_test.wiki_aapl,
unit_test.wiki_msft]])
# mock our query param column_index request
httpretty.register_uri(httpretty.GET,
"https://www.quandl.com/api/v3/datasets/SINGLE/COLUMN/data",
body=json.dumps(unit_test.single_dataset_data))
httpretty.register_uri(httpretty.GET,
"https://www.quandl.com/api/v3/datasets/WIKI/AAPL/data",
body=json.dumps(unit_test.dataset_data))
httpretty.register_uri(httpretty.GET,
def setUpClass(cls):
httpretty.enable()
dataset = {'dataset': DatasetFactory.build(
database_code='NSE', dataset_code='OIL')}
httpretty.register_uri(httpretty.GET,
re.compile(
'https://www.quandl.com/api/v3/datasets/*'),
body=json.dumps(dataset))
dataset_code = Dataset.get_code_from_meta(dataset['dataset'])
cls.dataset_instance = Dataset(dataset_code, dataset['dataset'])
def test_to_list_returns_list(self):
results = Dataset.all()
data = results.to_list()
self.assertIsInstance(data, list)
self.assertIsInstance(data[0], list)
@patch.object(Dataset, 'data')
def test_number_becomes_column_index(self, mock_method):
get('NSE/OIL.1')
self.assertEqual(mock_method.call_count, 1)
self.assertEqual(mock_method.mock_calls[0],
call(handle_column_not_found=True, params={'column_index': 1}))
def test_dataset_column_names_match_expected(self):
dataset = Dataset('NSE/OIL')
six.assertCountEqual(self,
dataset.column_names, ['Date', 'column.1', 'column.2', 'column.3'])
def _build_dataset_object(self, dataset_code, **options):
options_copy = options.copy()
# data_codes are tuples
# e.g., ('WIKI/AAPL', {'column_index": [1,2]})
# or strings
# e.g., 'NSE/OIL'
code = self._get_request_dataset_code(dataset_code)
dataset = Dataset(code, None, **options_copy)
# save column_index param requested dynamically
# used later on to determine:
# if column_index is an array, fetch all data and use locally to filter columns
# if column_index is an empty array, fetch all data and don't filter columns
dataset.requested_column_indexes = self._get_req_dataset_col_indexes(dataset_code, code)
return dataset
def datasets(self, **options):
params = {'database_code': self.code, 'query': '', 'page': 1}
options = Util.merge_options('params', params, **options)
return quandl.model.dataset.Dataset.all(**options)