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):
_PREPS_LEN = 30
_MAIN_PREPS_LEN = 22
_AMOUNT_DELEGATE = 10000
_MINIMUM_DELEGATE_AMOUNT = 10 ** 18
# same as fee treasury address constant value
self._fee_treasury = Address.from_prefix_and_int(AddressPrefix.CONTRACT, 1)
default_icon_config[ConfigKey.IISS_CALCULATE_PERIOD] = 10
default_icon_config[ConfigKey.TERM_PERIOD] = 10
super().setUp()
self._update_governance()
self._set_revision(REV_IISS)
addr_array = [create_address() for _ in range(_PREPS_LEN)]
total_supply = 800_460_000 * ICX_IN_LOOP
# Minimum_delegate_amount is 0.02 * total_supply
# In this test delegate 0.03*total_supply because `Issue transaction` exists since REV_IISS
delegate_amount = total_supply * 2 // 1000
# generate preps
self._decentralize(addr_array, delegate_amount)
response = self._get_prep_list()
def _setUp_audit(self):
self.config.update_conf(
{
ConfigKey.SERVICE: {
ConfigKey.SERVICE_AUDIT: True,
ConfigKey.SERVICE_FEE: False,
ConfigKey.SERVICE_SCORE_PACKAGE_VALIDATOR: False,
}
}
)
self.icon_service_engine = IconServiceEngine()
self.icon_service_engine.open(self.config)
self._genesis_invoke()
self.token_initial_params = {"init_supply": hex(1000), "decimal": "0x12"}
def setUp(self):
root_clear(
self._score_root_path, self._state_db_root_path, self._iiss_db_root_path
)
self._block_height = -1
self._prev_block_hash = None
config = IconConfig("", default_icon_config)
config.load()
config.update_conf({ConfigKey.BUILTIN_SCORE_OWNER: str(self._admin.address)})
config.update_conf(
{
ConfigKey.SERVICE: {
ConfigKey.SERVICE_AUDIT: False,
ConfigKey.SERVICE_FEE: True,
ConfigKey.SERVICE_SCORE_PACKAGE_VALIDATOR: False,
}
}
)
config.update_conf(
{
ConfigKey.SCORE_ROOT_PATH: self._score_root_path,
ConfigKey.STATE_DB_ROOT_PATH: self._state_db_root_path,
}
)
config.update_conf(self._make_init_config())
def test_register_prep_burn_event_log_should_be_fixed_after_revision_9(self):
preps: list = self.create_eoa_accounts(3)
self.distribute_icx(accounts=preps,
init_balance=3000 * ICX_IN_LOOP)
config_value = self._config[ConfigKey.PREP_REGISTRATION_FEE]
# TEST: When revision is before 'DECENTRALIZATION', burn signature should not have type format
prep = preps[0]
reg_tx_result: 'TransactionResult' = self.register_prep(from_=prep, value=config_value)[0]
actual_burn_signature = reg_tx_result.event_logs[0].indexed[0]
self.assertEqual("ICXBurned", actual_burn_signature)
# TEST: When revision is between 'DECENTRALIZATION' and 'FIX_BURN_EVENT_SIGNATURE',
# burn signature should not have type format
self._init_decentralized()
prep = preps[1]
reg_tx_result: 'TransactionResult' = self.register_prep(from_=prep, value=config_value)[1]
actual_burn_signature = reg_tx_result.event_logs[0].indexed[0]
def _make_init_config(self) -> dict:
return {ConfigKey.SERVICE: {ConfigKey.SERVICE_FEE: True}}
def test_set_value_when_prep_related_set_method(self):
# failure case: except registerPRep, value can not be set when calling prep related setting method
arbitrary_value = 10
# register prep
self.register_prep(from_=self._accounts[0],
value=self._config[
ConfigKey.PREP_REGISTRATION_FEE])
# unregisterPRep
tx: dict = self.create_score_call_tx(from_=self._accounts[0],
to_=SYSTEM_SCORE_ADDRESS,
func_name="unregisterPRep",
params={},
value=arbitrary_value)
self.process_confirm_block_tx([tx],
expected_status=False)
# setPRep
tx: dict = self.create_score_call_tx(from_=self._accounts[0],
to_=SYSTEM_SCORE_ADDRESS,
func_name="setPRep",
params={"name": f"new{str(self._accounts[0].address)}"},
value=arbitrary_value)
def _add_extra_value_types(cls, table: ConfigKeyTypeDict):
# Add some value types which cannot be contained to default_conf
table[ConfigKey.LOG] = {
ConfigKey.LOGGER: str,
ConfigKey.LOG_FILE_PATH: str,
ConfigKey.LOG_LEVEL: str,
ConfigKey.LOG_OUTPUT_TYPE: str,
ConfigKey.LOG_ROTATE: {
ConfigKey.LOG_ROTATE_TYPE: str,
ConfigKey.LOG_ROTATE_PERIOD: str,
ConfigKey.LOG_ROTATE_AT_TIME: int,
ConfigKey.LOG_ROTATE_INTERVAL: int,
ConfigKey.LOG_ROTATE_MAX_BYTES: int,
ConfigKey.LOG_ROTATE_BACKUP_COUNT: int,
},
def _add_extra_value_types(cls, table: ConfigKeyTypeDict):
# Add some value types which cannot be contained to default_conf
table[ConfigKey.LOG] = {
ConfigKey.LOGGER: str,
ConfigKey.LOG_FILE_PATH: str,
ConfigKey.LOG_LEVEL: str,
ConfigKey.LOG_OUTPUT_TYPE: str,
ConfigKey.LOG_ROTATE: {
ConfigKey.LOG_ROTATE_TYPE: str,
ConfigKey.LOG_ROTATE_PERIOD: str,
ConfigKey.LOG_ROTATE_AT_TIME: int,
ConfigKey.LOG_ROTATE_INTERVAL: int,
ConfigKey.LOG_ROTATE_MAX_BYTES: int,
ConfigKey.LOG_ROTATE_BACKUP_COUNT: int,
},
def serve(self, config: "IconConfig"):
async def _serve():
await self._inner_service.connect(exclusive=True)
Logger.info(f"Start IconService Service serve!", _TAG)
channel = config[ConfigKey.CHANNEL]
amqp_key = config[ConfigKey.AMQP_KEY]
amqp_target = config[ConfigKey.AMQP_TARGET]
score_root_path = config[ConfigKey.SCORE_ROOT_PATH]
db_root_path = config[ConfigKey.STATE_DB_ROOT_PATH]
version: str = get_version()
self._set_icon_score_stub_params(channel, amqp_key, amqp_target)
Logger.info(f"==========IconService Service params==========", _TAG)
Logger.info(f"version : {version}", _TAG)
Logger.info(f"score_root_path : {score_root_path}", _TAG)
Logger.info(f"icon_score_state_db_root_path : {db_root_path}", _TAG)
Logger.info(f"amqp_target : {amqp_target}", _TAG)
Logger.info(f"amqp_key : {amqp_key}", _TAG)
Logger.info(f"icon_score_queue_name : {self._icon_score_queue_name}", _TAG)
def _start_process(conf: "IconConfig"):
Logger.info("start_server() start")
python_module_string = "iconservice.icon_service"
converted_params = {
"-sc": conf[ConfigKey.SCORE_ROOT_PATH],
"-st": conf[ConfigKey.STATE_DB_ROOT_PATH],
"-ch": conf[ConfigKey.CHANNEL],
"-ak": conf[ConfigKey.AMQP_KEY],
"-at": conf[ConfigKey.AMQP_TARGET],
"-c": conf.get(ConfigKey.CONFIG),
}
custom_argv = []
for k, v in converted_params.items():
if v is None:
continue
custom_argv.append(k)
custom_argv.append(str(v))
if conf[ConfigKey.TBEARS_MODE]:
custom_argv.append("-tbears")
if conf[ConfigKey.STEP_TRACE_FLAG]:
custom_argv.append("-steptrace")