1"""Test for price extractor of OANDA. 2""" 3__copyright__ = "Copyright (C) 2018 Martin Blais" 4__license__ = "GNU GPLv2" 5 6import os 7import time 8import datetime 9import unittest 10from unittest import mock 11 12from dateutil import tz 13 14from beancount.prices.sources import oanda 15from beancount.prices import source 16from beancount.core.number import D 17from beancount.utils import net_utils 18from beancount.utils import date_utils 19 20 21def response(code, contents=None): 22 urlopen = mock.MagicMock(return_value=None) 23 if isinstance(contents, str): 24 response = mock.MagicMock() 25 response.read = mock.MagicMock(return_value=contents.encode('utf-8')) 26 response.getcode = mock.MagicMock(return_value=200) 27 urlopen.return_value = response 28 return mock.patch.object(net_utils, 'retrying_urlopen', urlopen) 29 30 31class TestOandaMisc(unittest.TestCase): 32 33 def test_get_currencies(self): 34 self.assertEqual(('USD', 'CAD'), oanda._get_currencies('USD_CAD')) 35 36 def test_get_currencies_invalid(self): 37 self.assertEqual((None, None), oanda._get_currencies('USDCAD')) 38 39 40class TimezoneTestBase: 41 42 def setUp(self): 43 tz_value = 'Europe/Berlin' 44 self.tz_old = os.environ.get('TZ', None) 45 os.environ['TZ'] = tz_value 46 time.tzset() 47 48 def tearDown(self): 49 if self.tz_old is None: 50 del os.environ['TZ'] 51 else: 52 os.environ['TZ'] = self.tz_old 53 time.tzset() 54 55 56class TestOandaFetchCandles(TimezoneTestBase, unittest.TestCase): 57 58 @response(404) 59 def test_null_response(self): 60 self.assertIs(None, oanda._fetch_candles({})) 61 62 @response(200, ''' 63 { 64 "instrument" : "USD_CAD", 65 "granularity" : "S5" 66 } 67 ''') 68 def test_key_error(self): 69 self.assertIs(None, oanda._fetch_candles({})) 70 71 @response(200, ''' 72 { 73 "instrument" : "USD_CAD", 74 "granularity" : "S5", 75 "candles" : [ 76 { 77 "time" : "2017-01-23T00:45:15.000000Z", 78 "openMid" : 1.330115, 79 "highMid" : 1.33012, 80 "lowMid" : 1.33009, 81 "closeMid" : 1.33009, 82 "volume" : 9, 83 "complete" : true 84 }, 85 { 86 "time" : "2017-01-23T00:45:20.000000Z", 87 "openMid" : 1.330065, 88 "highMid" : 1.330065, 89 "lowMid" : 1.330065, 90 "closeMid" : 1.330065, 91 "volume" : 1, 92 "complete" : true 93 } 94 ] 95 } 96 ''') 97 def test_valid(self): 98 self.assertEqual([ 99 (datetime.datetime(2017, 1, 23, 0, 45, 15, tzinfo=tz.tzutc()), D('1.330115')), 100 (datetime.datetime(2017, 1, 23, 0, 45, 20, tzinfo=tz.tzutc()), D('1.330065')) 101 ], oanda._fetch_candles({})) 102 103 104class TestOandaGetLatest(unittest.TestCase): 105 106 def setUp(self): 107 self.fetcher = oanda.Source() 108 109 def test_invalid_ticker(self): 110 srcprice = self.fetcher.get_latest_price('NOTATICKER') 111 self.assertIsNone(srcprice) 112 113 def test_no_candles(self): 114 with mock.patch.object(oanda, '_fetch_candles', return_value=None): 115 self.assertEqual(None, self.fetcher.get_latest_price('USD_CAD')) 116 117 def _test_valid(self): 118 candles = [ 119 (datetime.datetime(2017, 1, 21, 0, 45, 15, tzinfo=tz.tzutc()), D('1.330115')), 120 (datetime.datetime(2017, 1, 21, 0, 45, 20, tzinfo=tz.tzutc()), D('1.330065')), 121 ] 122 with mock.patch.object(oanda, '_fetch_candles', return_value=candles): 123 srcprice = self.fetcher.get_latest_price('USD_CAD') 124 # Latest price, with current time as time. 125 self.assertEqual(source.SourcePrice( 126 D('1.330065'), 127 datetime.datetime(2017, 1, 21, 0, 45, 20, tzinfo=tz.tzutc()), 128 'CAD'), srcprice) 129 130 def test_valid(self): 131 for tzname in "America/New_York", "Europe/Berlin", "Asia/Tokyo": 132 with date_utils.intimezone(tzname): 133 self._test_valid() 134 135 136class TestOandaGetHistorical(TimezoneTestBase, unittest.TestCase): 137 138 def setUp(self): 139 self.fetcher = oanda.Source() 140 super().setUp() 141 142 def test_invalid_ticker(self): 143 srcprice = self.fetcher.get_latest_price('NOTATICKER') 144 self.assertIsNone(srcprice) 145 146 def test_no_candles(self): 147 with mock.patch.object(oanda, '_fetch_candles', return_value=None): 148 self.assertEqual(None, self.fetcher.get_latest_price('USD_CAD')) 149 150 def _check_valid(self, query_date, out_time, out_price): 151 candles = [ 152 (datetime.datetime(2017, 1, 21, 0, 0, 0, tzinfo=tz.tzutc()), D('1.3100')), 153 (datetime.datetime(2017, 1, 21, 8, 0, 0, tzinfo=tz.tzutc()), D('1.3300')), 154 (datetime.datetime(2017, 1, 21, 16, 0, 0, tzinfo=tz.tzutc()), D('1.3500')), 155 (datetime.datetime(2017, 1, 22, 0, 0, 0, tzinfo=tz.tzutc()), D('1.3700')), 156 (datetime.datetime(2017, 1, 22, 8, 0, 0, tzinfo=tz.tzutc()), D('1.3900')), 157 (datetime.datetime(2017, 1, 22, 16, 0, 0, tzinfo=tz.tzutc()), D('1.4100')), 158 (datetime.datetime(2017, 1, 23, 0, 0, 0, tzinfo=tz.tzutc()), D('1.4300')), 159 (datetime.datetime(2017, 1, 23, 8, 0, 0, tzinfo=tz.tzutc()), D('1.4500')), 160 (datetime.datetime(2017, 1, 23, 16, 0, 0, tzinfo=tz.tzutc()), D('1.4700')), 161 ] 162 with mock.patch.object(oanda, '_fetch_candles', return_value=candles): 163 query_time = datetime.datetime.combine( 164 query_date, time=datetime.time(16, 0, 0), tzinfo=tz.tzutc()) 165 srcprice = self.fetcher.get_historical_price('USD_CAD', query_time) 166 if out_time is not None: 167 self.assertEqual(source.SourcePrice(out_price, out_time, 'CAD'), srcprice) 168 else: 169 self.assertEqual(None, srcprice) 170 171 def test_valid_same_date(self): 172 for tzname in "America/New_York", "Europe/Berlin", "Asia/Tokyo": 173 with date_utils.intimezone(tzname): 174 self._check_valid( 175 datetime.date(2017, 1, 22), 176 datetime.datetime(2017, 1, 22, 16, 0, tzinfo=tz.tzutc()), 177 D('1.4100')) 178 179 def test_valid_before(self): 180 for tzname in "America/New_York", "Europe/Berlin", "Asia/Tokyo": 181 with date_utils.intimezone(tzname): 182 self._check_valid( 183 datetime.date(2017, 1, 23), 184 datetime.datetime(2017, 1, 23, 16, 0, tzinfo=tz.tzutc()), 185 D('1.4700')) 186 187 def test_valid_after(self): 188 for tzname in "America/New_York", "Europe/Berlin", "Asia/Tokyo": 189 with date_utils.intimezone(tzname): 190 self._check_valid(datetime.date(2017, 1, 20), None, None) 191 192 193if __name__ == '__main__': 194 unittest.main() 195