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