我正在使用mock来测试一个ccxt.binance类对象,该模拟对象作为参数传递给我正在测试的函数。最初它是可以工作的,因为我能够测试到一定程度。然而,当我改变了该对象的值时,函数没有遵守这个值,而是返回一个StopIteration错误。我在mock设置方面漏掉了什么吗?以下是代码示例。
def exchange_history_load(args, ccxt_exchange):
# skip exchange if has no true fetchOHLCV or if exchange has no timeframe specific data
if ((not ccxt_exchange.has['fetchOHLCV']) or (ccxt_exchange.has['fetchOHLCV'] == 'emulated') or
(not ccxt_exchange.timeframes.get(args.timeframe))):
logging.warning(f'Skipping {ccxt_exchange.name}.')
return
# check if exchange is in database. Write to database if not.
exchange_db_data = get_exchange(ccxt_exchange.id)
if not exchange_db_data:
insert_exchange_to_db(ccxt_exchange.id, ccxt_exchange.name)
logging.info(f'Inserted {ccxt_exchange.name} to database.')
exchange_db_data = get_exchange(ccxt_exchange.id)
exchange_db_id = exchange_db_data[0]
logging.info(f'Fetched {ccxt_exchange.name} data.')
# load exchange markets
markets = ccxt_exchange.load_markets()
logging.info(f'Loaded {ccxt_exchange.name} markets.')
return exchange_db_id, markets
这是测试函数
@patch('src.cron.historical_price.hp_load_data.insert_exchange_to_db')
@patch('src.cron.historical_price.hp_load_data.get_exchange')
@patch('ccxt.binance')
def test_exchange_history_load(mock_binance, mock_get_exchange, mock_insert_exchange_to_db):
# create sample arguments for function
args = argparse.Namespace(timeframe='1d')
# mock up sample return values for necessary functions
mock_get_exchange.side_effect = [None, (1, 'binance', 'Binance'), (1, 'binance', 'Binance')]
mock_binance.return_value.has.return_value = {'fetchOHLCV': True}
mock_binance.return_value.load_markets.return_value = {
'BTC/USDT': {
'id': 'btcusdt',
'symbol': 'BTC/USDT',
'base': 'BTC',
'quote': 'USDT',
'active': True
}
}
mock_binance.return_value.id.return_value = 'binance'
mock_binance.return_value.name.return_value = 'Binance'
ccxt_exchange = ccxt.binance()
# case 1: exchange has fetchOHLCV, timeframe, but is not in database
# test assertions with case 1
exchange_db_id, markets = exchange_history_load(args, ccxt_exchange)
assert type(exchange_db_id) == int
assert type(markets) == dict
assert mock_get_exchange.call_count == 2
calls = [call(mock_binance.return_value.id), call(mock_binance.return_value.id)]
mock_get_exchange.assert_has_calls(calls)
mock_insert_exchange_to_db.assert_called_once()
mock_binance.return_value.load_markets.assert_called_once()
# case 2: exchange has fetchOHLCV, timeframe, and is in database
# reset mock calls for case 2
mock_insert_exchange_to_db.reset_mock()
mock_get_exchange.reset_mock()
mock_binance.return_value.load_markets.reset_mock()
# test assertions with case 2
exchange_db_id, markets = exchange_history_load(args, ccxt_exchange)
assert type(exchange_db_id) == int
assert type(markets) == dict
mock_get_exchange.assert_called_once()
assert not mock_insert_exchange_to_db.called
mock_binance.return_value.load_markets.assert_called_once()
calls = [call(mock_binance.return_value.id)]
mock_get_exchange.assert_has_calls(calls)
# case 3: exchange doesn't have true fetchOHLCV
# reset_mock calls for case 3
mock_get_exchange.reset_mock()
mock_insert_exchange_to_db.reset_mock()
mock_binance.reset_mock()
mock_binance.return_value.has.return_value = {'fetchOHLCV': True}
# mock_binance.return_value.load_markets.reset_mock()
# test assertions with case 3
exchange_db_id, markets = exchange_history_load(args, ccxt_exchange)
# assert not exchange_db_id
# assert not markets
assert not mock_get_exchange.called
assert not mock_insert_exchange_to_db.called
assert not mock_binance.return_value.load_markets.called
第一种和第二种情况都能正常工作。但是在第三种情况下,会抛出一个错误。
mock_binance = <MagicMock name='binance' id='140285596974272'>
mock_get_exchange = <MagicMock name='get_exchange' id='140285596584312'>
mock_insert_exchange_to_db = <MagicMock name='insert_exchange_to_db' id='140285596584480'>
@patch('src.cron.historical_price.hp_load_data.insert_exchange_to_db')
@patch('src.cron.historical_price.hp_load_data.get_exchange')
@patch('ccxt.binance')
def test_exchange_history_load(mock_binance, mock_get_exchange, mock_insert_exchange_to_db):
# create sample arguments for function
args = argparse.Namespace(timeframe='1d')
# mock up sample return values for necessary functions
mock_get_exchange.side_effect = [None, (1, 'binance', 'Binance'), (1, 'binance', 'Binance')]
mock_binance.return_value.has.return_value = {'fetchOHLCV': True}
mock_binance.return_value.load_markets.return_value = {
'BTC/USDT': {
'id': 'btcusdt',
'symbol': 'BTC/USDT',
'base': 'BTC',
'quote': 'USDT',
'active': True
}
}
mock_binance.return_value.id.return_value = 'binance'
mock_binance.return_value.name.return_value = 'Binance'
ccxt_exchange = ccxt.binance()
# case 1: exchange is not in database
# test assertions with case 1
exchange_db_id, markets = exchange_history_load(args, ccxt_exchange)
assert type(exchange_db_id) == int
assert type(markets) == dict
assert mock_get_exchange.call_count == 2
calls = [call(mock_binance.return_value.id), call(mock_binance.return_value.id)]
mock_get_exchange.assert_has_calls(calls)
mock_insert_exchange_to_db.assert_called_once()
mock_binance.return_value.load_markets.assert_called_once()
# case 2: exchange is in database
# reset mock calls for case 2
mock_insert_exchange_to_db.reset_mock()
mock_get_exchange.reset_mock()
mock_binance.return_value.load_markets.reset_mock()
# test assertions with case 2
exchange_db_id, markets = exchange_history_load(args, ccxt_exchange)
assert type(exchange_db_id) == int
assert type(markets) == dict
mock_get_exchange.assert_called_once()
assert not mock_insert_exchange_to_db.called
mock_binance.return_value.load_markets.assert_called_once()
calls = [call(mock_binance.return_value.id)]
mock_get_exchange.assert_has_calls(calls)
# case 3: exchange doesn't have true fetchOHLCV
# reset_mock calls for case 3
mock_get_exchange.reset_mock()
mock_insert_exchange_to_db.reset_mock()
mock_binance.reset_mock()
mock_binance.return_value.has.return_value = {'fetchOHLCV': True}
# mock_binance.return_value.load_markets.reset_mock()
# test assertions with case 3
> exchange_db_id, markets = exchange_history_load(args, ccxt_exchange)
test_cron/test_historical_price/test_hp_load_data.py:89:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../src/cron/historical_price/hp_load_data.py:75: in exchange_history_load
exchange_db_data = get_exchange(ccxt_exchange.id)
/usr/lib/python3.6/unittest/mock.py:939: in __call__
return _mock_self._mock_call(*args, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
_mock_self = <MagicMock name='get_exchange' id='140285596584312'>
args = (<MagicMock name='binance().id' id='140285596659328'>,), kwargs = {}
self = <MagicMock name='get_exchange' id='140285596584312'>, _new_name = ''
_new_parent = None
_call = call(<MagicMock name='binance().id' id='140285596659328'>), seen = set()
skip_next_dot = False, do_method_calls = False, name = 'get_exchange'
def _mock_call(_mock_self, *args, **kwargs):
self = _mock_self
self.called = True
self.call_count += 1
_new_name = self._mock_new_name
_new_parent = self._mock_new_parent
_call = _Call((args, kwargs), two=True)
self.call_args = _call
self.call_args_list.append(_call)
self.mock_calls.append(_Call(('', args, kwargs)))
seen = set()
skip_next_dot = _new_name == '()'
do_method_calls = self._mock_parent is not None
name = self._mock_name
while _new_parent is not None:
this_mock_call = _Call((_new_name, args, kwargs))
if _new_parent._mock_new_name:
dot = '.'
if skip_next_dot:
dot = ''
skip_next_dot = False
if _new_parent._mock_new_name == '()':
skip_next_dot = True
_new_name = _new_parent._mock_new_name + dot + _new_name
if do_method_calls:
if _new_name == name:
this_method_call = this_mock_call
else:
this_method_call = _Call((name, args, kwargs))
_new_parent.method_calls.append(this_method_call)
do_method_calls = _new_parent._mock_parent is not None
if do_method_calls:
name = _new_parent._mock_name + '.' + name
_new_parent.mock_calls.append(this_mock_call)
_new_parent = _new_parent._mock_new_parent
# use ids here so as not to call __hash__ on the mocks
_new_parent_id = id(_new_parent)
if _new_parent_id in seen:
break
seen.add(_new_parent_id)
ret_val = DEFAULT
effect = self.side_effect
if effect is not None:
if _is_exception(effect):
raise effect
if not _callable(effect):
> result = next(effect)
E StopIteration
/usr/lib/python3.6/unittest/mock.py:998: StopIteration