Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
self.mem_db = {}
self.key_ = b"key1"
self.last_value = None
def get(caller, key):
return self.last_value
self.score_address = Address.from_data(AddressPrefix.CONTRACT, b'score')
context_db = Mock(spec=ContextDatabase)
context_db.get = get
self._observer = Mock(spec=DatabaseObserver)
self._icon_score_database = IconScoreDatabase(self.score_address, context_db)
self._icon_score_database.set_observer(self._observer)
def test_negative_balance(self):
step_price = 0
self.icx_engine.get_balance = Mock(return_value=0)
params = {
'version': 3,
'txHash': create_tx_hash(b'tx'),
'from': Address.from_data(AddressPrefix.EOA, b'from'),
'to': Address.from_data(AddressPrefix.CONTRACT, b'to'),
'value': -10,
'stepLimit': 100,
'timestamp': int(time.time() * 10 ** 6),
'nonce': 1
}
# too small balance
with self.assertRaises(InvalidParamsException) as cm:
self.validator.execute(params, step_price, 100)
self.assertEqual(ExceptionCode.INVALID_PARAMS, cm.exception.code)
self.assertEqual('value < 0', cm.exception.message)
def test_convert_score_parameters():
def func(
_bool: bool,
_bytes: bytes,
_int: int,
_str: str,
_address: Address,
_list: List[int],
_dict: Dict[str, int],
_struct: User,
_list_of_struct: List[User],
_dict_of_str_and_struct: Dict[str, User],
):
pass
address = Address.from_data(AddressPrefix.EOA, os.urandom(20))
params = {
"_bool": True,
"_bytes": b"hello",
"_int": 100,
"_str": "world",
"_address": address,
"_list": [0, 1, 2, 3, 4, 5],
"_dict": {"0": 0, "1": 1, "2": 2},
"_struct": {"name": "hello", "age": 30, "single": True, "wallet": address},
"_list_of_struct": [
{"name": "hello", "age": 30, "single": True, "wallet": address},
{"name": "world", "age": 40, "single": False, "wallet": address},
],
"_dict_of_str_and_struct": {
"a": {"name": "hello", "age": 30, "single": True, "wallet": address},
def test_call_event_kwarg(self):
context = ContextContainer._get_context()
name = "name"
address = Address.from_data(AddressPrefix.EOA, os.urandom(20))
age = 10
# Call with ordered arguments
self._mock_score.OneIndexEvent(name, address, age)
self.assertEqual(len(context.event_logs), 1)
event_log_ordered_args = context.event_logs[0]
# Call with ordered arguments and keyword arguments
self._mock_score.OneIndexEvent(
name, age=age, address=address)
self.assertEqual(len(context.event_logs), 2)
event_log_keyword_args = context.event_logs[1]
self.assertEqual(event_log_ordered_args.score_address,
event_log_keyword_args.score_address)
self.assertEqual(event_log_ordered_args.indexed,
def test_revert(self, mocker):
mocker.patch.object(IconServiceEngine, "_charge_transaction_fee")
mocker.patch.object(IconScoreEngine, "invoke")
context = ContextContainer._get_context()
icon_service_engine = IconServiceEngine()
icon_service_engine._icx_engine = Mock(spec=IcxEngine)
icon_service_engine._icon_score_deploy_engine = Mock(spec=DeployEngine)
icon_service_engine._icon_pre_validator = Mock(spec=IconPreValidator)
context.tx_batch = TransactionBatch()
context.clear_batch = Mock()
context.update_batch = Mock()
from_ = Address.from_data(AddressPrefix.EOA, os.urandom(20))
to_ = Address.from_data(AddressPrefix.CONTRACT, os.urandom(20))
tx_index = randrange(0, 100)
context.tx = Transaction(os.urandom(32), tx_index, from_, 0)
context.msg = Message(from_)
def intercept_charge_transaction_fee(*args, **kwargs):
return {}, Mock(spec=int)
IconServiceEngine._charge_transaction_fee.side_effect = (
intercept_charge_transaction_fee
)
icon_service_engine._icon_score_deploy_engine.attach_mock(
Mock(return_value=False), "is_data_type_supported"
)
def test_tx_success(self):
from_ = Address.from_data(AddressPrefix.EOA, os.urandom(20))
to_ = Address.from_data(AddressPrefix.EOA, os.urandom(20))
tx_index = randrange(0, 100)
self._mock_context.tx = Transaction(os.urandom(32), tx_index, from_, to_, 0)
self._mock_context.msg = Message(from_)
params = {
'version': 3,
'from': from_,
'to': to_,
'value': 0,
'timestamp': 1234567890,
'nonce': 1
}
self._icon_service_engine._process_transaction = Mock(return_value=None)
tx_result = self._icon_service_engine._handle_icx_send_transaction(
def test_sample_result(self):
from_ = Address.from_data(AddressPrefix.EOA, b'from')
to_ = Address.from_data(AddressPrefix.CONTRACT, b'to')
self._mock_context.tx = Transaction(os.urandom(32), 1234, from_, to_, 0)
self._mock_context.msg = Message(from_)
IconScoreContext.engine.deploy.invoke = Mock()
self._icon_service_engine._process_transaction = Mock(return_value=to_)
tx_result = self._icon_service_engine._handle_icx_send_transaction(
self._mock_context, {'from': from_, 'to': to_})
tx_result.score_address = to_
tx_result.event_logs = [
EventLog(
Address.from_data(AddressPrefix.CONTRACT, b'addr_to'),
[b'indexed', Address.from_data(AddressPrefix.EOA, b'index')],
[True, 1234, 'str', None, b'test']
)
mocker.patch.object(IconServiceEngine, "_charge_transaction_fee")
mocker.patch.object(IconScoreEngine, "invoke")
context = ContextContainer._get_context()
icon_service_engine = IconServiceEngine()
icon_service_engine._icx_engine = Mock(spec=IcxEngine)
icon_service_engine._icon_score_deploy_engine = Mock(spec=DeployEngine)
icon_service_engine._icon_pre_validator = Mock(spec=IconPreValidator)
context.tx_batch = TransactionBatch()
context.clear_batch = Mock()
context.update_batch = Mock()
from_ = Address.from_data(AddressPrefix.EOA, os.urandom(20))
to_ = Address.from_data(AddressPrefix.CONTRACT, os.urandom(20))
tx_index = randrange(0, 100)
context.tx = Transaction(os.urandom(32), tx_index, from_, 0)
context.msg = Message(from_)
def intercept_charge_transaction_fee(*args, **kwargs):
return {}, Mock(spec=int)
IconServiceEngine._charge_transaction_fee.side_effect = (
intercept_charge_transaction_fee
)
icon_service_engine._icon_score_deploy_engine.attach_mock(
Mock(return_value=False), "is_data_type_supported"
)
reason = Mock(spec=str)
'from': Address.from_data(AddressPrefix.EOA, b'from'),
'to': Address.from_data(AddressPrefix.CONTRACT, b'to'),
'value': 1,
'stepLimit': 5000,
'timestamp': int(time.time() * 10 ** 6),
'nonce': 1
}
self.icx_engine.get_balance = Mock(return_value=100)
self.validator.execute(params, step_price, 4000)
params = {
'version': 3,
'txHash': create_tx_hash(b'tx'),
'from': Address.from_data(AddressPrefix.EOA, b'from'),
'to': Address.from_data(AddressPrefix.CONTRACT, b'to'),
'value': 10,
'stepLimit': 100,
'timestamp': int(time.time() * 10 ** 6),
'nonce': 1
}
# too small balance
if step_price > 0:
with self.assertRaises(InvalidRequestException) as cm:
self.validator.execute(params, step_price, 4000)
self.assertEqual(ExceptionCode.INVALID_REQUEST, cm.exception.code)
self.assertEqual('Step limit too low', cm.exception.message)
def generate_score_address_for_tbears(score_path: str) -> 'Address':
"""
:param score_path:
The path of a SCORE which is under development with tbears
:return:
"""
project_name = score_path.split('/')[-1]
return Address.from_data(AddressPrefix.CONTRACT, project_name.encode())