import threading
import market, portfolio, store
-@unittest.skipUnless("unit" in limits, "Unit skipped")
class NoopLockTest(unittest.TestCase):
def test_with(self):
noop_lock = store.NoopLock()
with noop_lock:
self.assertTrue(True)
-@unittest.skipUnless("unit" in limits, "Unit skipped")
class LockedVarTest(unittest.TestCase):
def test_values(self):
thread3.join()
self.assertEqual("Bar", locked_var.get()[0:3])
-@unittest.skipUnless("unit" in limits, "Unit skipped")
class TradeStoreTest(WebMockTestCase):
def test_compute_trades(self):
self.m.balances.currencies.return_value = ["XMR", "DASH", "XVG", "BTC", "ETH"]
self.assertEqual([trade_mock1, trade_mock2], trade_store.pending)
-@unittest.skipUnless("unit" in limits, "Unit skipped")
class BalanceStoreTest(WebMockTestCase):
def setUp(self):
super().setUp()
self.assertEqual(1, as_json["BTC"])
self.assertEqual(2, as_json["ETH"])
-@unittest.skipUnless("unit" in limits, "Unit skipped")
class ReportStoreTest(WebMockTestCase):
def test_add_log(self):
- report_store = market.ReportStore(self.m)
- report_store.add_log({"foo": "bar"})
+ with self.subTest(market=self.m):
+ self.m.user_id = 1
+ self.m.market_id = 3
+ report_store = market.ReportStore(self.m)
+ result = report_store.add_log({"foo": "bar"})
+
+ self.assertEqual({"foo": "bar", "date": mock.ANY, "user_id": 1, "market_id": 3}, result)
+ self.assertEqual(result, report_store.logs[0])
+
+ with self.subTest(market=None):
+ report_store = market.ReportStore(None)
+ result = report_store.add_log({"foo": "bar"})
- self.assertEqual({"foo": "bar", "date": mock.ANY}, report_store.logs[0])
+ self.assertEqual({"foo": "bar", "date": mock.ANY, "user_id": None, "market_id": None}, result)
def test_set_verbose(self):
report_store = market.ReportStore(self.m)
self.assertFalse(report_store.verbose_print)
def test_merge(self):
+ self.m.user_id = 1
+ self.m.market_id = 3
report_store1 = market.ReportStore(self.m, verbose_print=False)
report_store2 = market.ReportStore(None, verbose_print=False)
with self.subTest(verbose=True),\
mock.patch.object(store, "datetime") as time_mock,\
mock.patch('sys.stdout', new_callable=StringIO) as stdout_mock:
- time_mock.now.return_value = datetime.datetime(2018, 2, 25, 2, 20, 10)
+ time_mock.datetime.now.return_value = datetime.datetime(2018, 2, 25, 2, 20, 10)
report_store.set_verbose(True)
report_store.print_log("Coucou")
report_store.print_log(portfolio.Amount("BTC", 1))
report_store = market.ReportStore(self.m)
self.assertEqual("2018-02-24T00:00:00",
- report_store.default_json_serial(portfolio.datetime(2018, 2, 24)))
+ report_store.default_json_serial(portfolio.datetime.datetime(2018, 2, 24)))
self.assertEqual("1.00000000 BTC",
report_store.default_json_serial(portfolio.Amount("BTC", 1)))
report_store = market.ReportStore(self.m)
report_store.logs.append({"foo": "bar"})
self.assertEqual('[\n {\n "foo": "bar"\n }\n]', report_store.to_json())
- report_store.logs.append({"date": portfolio.datetime(2018, 2, 24)})
+ report_store.logs.append({"date": portfolio.datetime.datetime(2018, 2, 24)})
self.assertEqual('[\n {\n "foo": "bar"\n },\n {\n "date": "2018-02-24T00:00:00"\n }\n]', report_store.to_json())
report_store.logs.append({"amount": portfolio.Amount("BTC", 1)})
self.assertEqual('[\n {\n "foo": "bar"\n },\n {\n "date": "2018-02-24T00:00:00"\n },\n {\n "amount": "1.00000000 BTC"\n }\n]', report_store.to_json())
}
})
- @mock.patch.object(market.ReportStore, "print_log")
- @mock.patch.object(market.ReportStore, "add_log")
- def test_log_http_request(self, add_log, print_log):
- report_store = market.ReportStore(self.m)
- response = mock.Mock()
- response.status_code = 200
- response.text = "Hey"
+ def test_log_http_request(self):
+ with mock.patch.object(market.ReportStore, "add_log") as add_log:
+ report_store = market.ReportStore(self.m)
+ response = mock.Mock()
+ response.status_code = 200
+ response.text = "Hey"
- report_store.log_http_request("method", "url", "body",
- "headers", response)
- print_log.assert_not_called()
- add_log.assert_called_once_with({
- 'type': 'http_request',
- 'method': 'method',
- 'url': 'url',
- 'body': 'body',
- 'headers': 'headers',
- 'status': 200,
- 'response': 'Hey'
- })
+ report_store.log_http_request("method", "url", "body",
+ "headers", response)
+ add_log.assert_called_once_with({
+ 'type': 'http_request',
+ 'method': 'method',
+ 'url': 'url',
+ 'body': 'body',
+ 'headers': 'headers',
+ 'status': 200,
+ 'response': 'Hey',
+ 'response_same_as': None,
+ })
- add_log.reset_mock()
- report_store.log_http_request("method", "url", "body",
- "headers", ValueError("Foo"))
- add_log.assert_called_once_with({
- 'type': 'http_request',
- 'method': 'method',
- 'url': 'url',
- 'body': 'body',
- 'headers': 'headers',
- 'status': -1,
- 'response': None,
- 'error': 'ValueError',
- 'error_message': 'Foo',
- })
+ add_log.reset_mock()
+ report_store.log_http_request("method", "url", "body",
+ "headers", ValueError("Foo"))
+ add_log.assert_called_once_with({
+ 'type': 'http_request',
+ 'method': 'method',
+ 'url': 'url',
+ 'body': 'body',
+ 'headers': 'headers',
+ 'status': -1,
+ 'response': None,
+ 'error': 'ValueError',
+ 'error_message': 'Foo',
+ })
+
+ with self.subTest(no_http_dup=True, duplicate=True):
+ self.m.user_id = 1
+ self.m.market_id = 3
+ report_store = market.ReportStore(self.m, no_http_dup=True)
+ original_add_log = report_store.add_log
+ with mock.patch.object(report_store, "add_log", side_effect=original_add_log) as add_log:
+ report_store.log_http_request("method", "url", "body",
+ "headers", response)
+ report_store.log_http_request("method", "url", "body",
+ "headers", response)
+ self.assertEqual(2, add_log.call_count)
+ self.assertIsNone(add_log.mock_calls[0][1][0]["response_same_as"])
+ self.assertIsNone(add_log.mock_calls[1][1][0]["response"])
+ self.assertEqual(add_log.mock_calls[0][1][0]["date"], add_log.mock_calls[1][1][0]["response_same_as"])
+ with self.subTest(no_http_dup=True, duplicate=False, case="Different call"):
+ self.m.user_id = 1
+ self.m.market_id = 3
+ report_store = market.ReportStore(self.m, no_http_dup=True)
+ original_add_log = report_store.add_log
+ with mock.patch.object(report_store, "add_log", side_effect=original_add_log) as add_log:
+ report_store.log_http_request("method", "url", "body",
+ "headers", response)
+ report_store.log_http_request("method2", "url", "body",
+ "headers", response)
+ self.assertEqual(2, add_log.call_count)
+ self.assertIsNone(add_log.mock_calls[0][1][0]["response_same_as"])
+ self.assertIsNone(add_log.mock_calls[1][1][0]["response_same_as"])
+ with self.subTest(no_http_dup=True, duplicate=False, case="Call inbetween"):
+ self.m.user_id = 1
+ self.m.market_id = 3
+ report_store = market.ReportStore(self.m, no_http_dup=True)
+ original_add_log = report_store.add_log
+
+ response2 = mock.Mock()
+ response2.status_code = 200
+ response2.text = "Hey there!"
+
+ with mock.patch.object(report_store, "add_log", side_effect=original_add_log) as add_log:
+ report_store.log_http_request("method", "url", "body",
+ "headers", response)
+ report_store.log_http_request("method", "url", "body",
+ "headers", response2)
+ report_store.log_http_request("method", "url", "body",
+ "headers", response)
+ self.assertEqual(3, add_log.call_count)
+ self.assertIsNone(add_log.mock_calls[0][1][0]["response_same_as"])
+ self.assertIsNone(add_log.mock_calls[1][1][0]["response_same_as"])
+ self.assertIsNone(add_log.mock_calls[2][1][0]["response_same_as"])
@mock.patch.object(market.ReportStore, "add_log")
def test_log_market(self, add_log):
report_store = market.ReportStore(self.m)
- report_store.log_market(self.market_args(debug=True, quiet=False), 4, 1)
+ report_store.log_market(self.market_args(debug=True, quiet=False))
add_log.assert_called_once_with({
"type": "market",
"commit": "$Format:%H$",
"args": { "report_path": None, "debug": True, "quiet": False },
- "user_id": 4,
- "market_id": 1,
})
@mock.patch.object(market.ReportStore, "print_log")
'action': 'Hey'
})
-@unittest.skipUnless("unit" in limits, "Unit skipped")
class PortfolioTest(WebMockTestCase):
def setUp(self):
super().setUp()
'SC': (D("0.0623"), "long"),
'ZEC': (D("0.3701"), "long"),
}
- date = portfolio.datetime(2018, 1, 8)
+ date = portfolio.datetime.datetime(2018, 1, 8)
self.assertDictEqual(expected, liquidities["high"][date])
expected = {
'XCP': (D("0.1"), "long"),
}
self.assertDictEqual(expected, liquidities["medium"][date])
- self.assertEqual(portfolio.datetime(2018, 1, 15), market.Portfolio.last_date.get())
+ self.assertEqual(portfolio.datetime.datetime(2018, 1, 15), market.Portfolio.last_date.get())
with self.subTest(description="Missing weight"):
data = store.json.loads(self.json_response, parse_int=D, parse_float=D)
else:
self.assertFalse(refetch)
self.call_count += 1
- market.Portfolio.last_date = store.LockedVar(store.datetime.now()\
- - store.timedelta(10)\
- + store.timedelta(self.call_count))
+ market.Portfolio.last_date = store.LockedVar(store.datetime.datetime.now()\
+ - store.datetime.timedelta(10)\
+ + store.datetime.timedelta(self.call_count))
get_cryptoportfolio.side_effect = _get
market.Portfolio.wait_for_recent()
self.assertTrue(store.Portfolio.worker_started)
self.assertFalse(store.Portfolio.worker.is_alive())
+ self.assertEqual(1, threading.active_count())
+
+ def test_stop_worker(self):
+ with mock.patch.object(store.Portfolio, "get_cryptoportfolio") as get,\
+ mock.patch.object(store.Portfolio, "report") as report,\
+ mock.patch.object(store.time, "sleep") as sleep:
+ store.Portfolio.start_worker(poll=3)
+ store.Portfolio.stop_worker()
+ store.Portfolio.worker.join()
+ get.assert_not_called()
+ report.assert_not_called()
+ sleep.assert_not_called()
+ self.assertFalse(store.Portfolio.worker.is_alive())
def test_wait_for_notification(self):
with self.assertRaises(RuntimeError):
store.Portfolio.callback.wait()
- report.print_log.assert_called_once_with("Fetching cryptoportfolio")
+ report.print_log.assert_called_once_with("[Worker] Fetching cryptoportfolio")
get.assert_called_once_with(refetch=True)
sleep.assert_called_once_with(3)
self.assertFalse(store.Portfolio.worker_notify.is_set())
store.Portfolio.callback.clear()
store.Portfolio.worker_started = False
store.Portfolio.worker_notify.set()
- store.Portfolio.callback.wait()
+ store.Portfolio.worker.join()
self.assertFalse(store.Portfolio.worker.is_alive())
def test_notify_and_wait(self):
projects / perso / Immae / Projets / Cryptomonnaies / Cryptoportfolio / Trader.git / blobdiff