method, arguments = processor.method_arguments("print_tickers")
self.assertEqual(m.print_tickers, method)
+ method, arguments = processor.method_arguments("fetch_balances")
+ self.assertEqual(m.balances.fetch_balances, method)
+
def test_process_step(self):
processor = market.Processor(self.m)
mock.call("process_foo__2_sell_begin"),
mock.call("process_foo__2_sell_end"),
])
- self.m.balances.fetch_balances.assert_has_calls([
- mock.call(tag="process_foo__2_sell_begin"),
- mock.call(tag="process_foo__2_sell_end"),
- ])
- self.assertEqual(5, run_action.call_count)
+ self.assertEqual(7, run_action.call_count)
run_action.assert_has_calls([
+ mock.call('fetch_balances', {}, {'foo': 'bar', 'tag': 'process_foo__2_sell_begin'}),
mock.call('prepare_trades', {}, {'foo': 'bar'}),
mock.call('prepare_orders', {'only': 'dispose', 'compute_value': 'average'}, {'foo': 'bar'}),
mock.call('run_orders', {}, {'foo': 'bar'}),
mock.call('follow_orders', {}, {'foo': 'bar'}),
mock.call('close_trades', {}, {'foo': 'bar'}),
+ mock.call('fetch_balances', {}, {'foo': 'bar', 'tag': 'process_foo__2_sell_end'}),
])
self.m.reset_mock()
mock.call("process_foo__0_print_balances_begin"),
mock.call("process_foo__0_print_balances_end"),
])
- self.m.balances.fetch_balances.assert_has_calls([
- mock.call(add_portfolio=True, checkpoint='end',
- log_tickers=True,
- add_usdt=True,
- tag='process_foo__0_print_balances_begin')
- ])
- self.assertEqual(0, run_action.call_count)
+ self.assertEqual(1, run_action.call_count)
+ run_action.assert_has_calls([
+ mock.call('fetch_balances',
+ {'checkpoint': 'end', 'log_tickers': True, 'add_usdt': True, 'add_portfolio': True},
+ {'foo': 'bar', 'tag': 'process_foo__0_print_balances_begin'}),
+ ])
self.m.reset_mock()
with mock.patch.object(processor, "run_action") as run_action:
step = processor.scenarios["sell_needed"][1]
processor.process_step("foo", step, {"foo":"bar"})
- self.m.balances.fetch_balances.assert_not_called()
+ self.assertEqual(1, run_action.call_count)
self.m.reset_mock()
with mock.patch.object(processor, "run_action") as run_action:
step = processor.scenarios["print_balances"][0]
processor.process_step("foo", step, {"foo":"bar"})
- self.m.balances.fetch_balances.assert_called_once_with(
- add_portfolio=True, add_usdt=True, log_tickers=True,
- tag='process_foo__1_print_balances_begin')
+ run_action.assert_has_calls([
+ mock.call('fetch_balances',
+ {'log_tickers': True, 'add_usdt': True, 'add_portfolio': True},
+ {'foo': 'bar', 'tag': 'process_foo__1_print_balances_begin'}),
+ ])
def test_parse_args(self):
processor = market.Processor(self.m)