Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_putting_i_score_data_on_current_db_and_get_from_the_db(self, rc_data_storage):
# TEST: Put i-score response and get it from the db
expected_i_score = 10_000
expected_version = 1
expected_block_height = 0
expected_state_hash: bytes = sha3_256(b"state_hash")
rc_data_storage.put_calc_response_from_rc(expected_i_score, expected_block_height, expected_state_hash)
actual_i_score_db_data = MsgPackForDB.loads(
rc_data_storage._db.get(rc_data_storage.KEY_FOR_CALC_RESPONSE_FROM_RC))
assert actual_i_score_db_data[0] == expected_version
assert actual_i_score_db_data[1] == expected_i_score
assert actual_i_score_db_data[2] == expected_block_height
assert actual_i_score_db_data[3] == expected_state_hash
# TEST: Put i-score response and get using get_calc_response_from_rc method
actual_i_score, _, _ = rc_data_storage.get_calc_response_from_rc()
assert actual_i_score == expected_i_score
def to_bytes(self) -> bytes:
version: int = 0
return MsgPackForDB.dumps([version, self._prev_node_address_mapper])
def to_bytes(self) -> bytes:
return MsgPackForDB.dumps([
self._VERSION,
self.name,
self.email,
self.website,
self.details,
self.p2p_end_point,
self.public_key,
self.incentive_rep,
self.block_height,
self.tx_index,
])
def get_calc_response_from_rc(self) -> Tuple[int, int, Optional[bytes]]:
response_from_rc: Optional[bytes] = self._db.get(self.KEY_FOR_CALC_RESPONSE_FROM_RC)
if response_from_rc is None:
return -1, -1, None
response_from_rc: list = MsgPackForDB.loads(response_from_rc)
version = response_from_rc[0]
if version == 0:
iscore = response_from_rc[1]
block_height = response_from_rc[2]
state_hash = None
elif version == 1:
iscore = response_from_rc[1]
block_height = response_from_rc[2]
state_hash = response_from_rc[3]
else:
raise DatabaseException(f"get_calc_response_from_rc invalid version: {version}")
return iscore, block_height, state_hash
def from_bytes(cls, buf: bytes) -> 'RegulatorVariable':
data: list = MsgPackForDB.loads(buf)
version = data[0]
return cls(*data[1:])
def get_calc_period(self, context: 'IconScoreContext') -> Optional[int]:
value: bytes = self._db.get(context, self.CALC_PERIOD_KEY)
if value:
data: list = MsgPackForDB.loads(value)
version: int = data[0]
assert version == 0
calc_period: int = data[1]
return calc_period
return None
def from_bytes(cls, buf: bytes) -> 'RewardRate':
data: list = MsgPackForDB.loads(buf)
version = data[0]
assert version == cls._VERSION
return cls(*data[1:])
def to_bytes(self) -> bytes:
"""Converts DepositMeta object into bytes.
:return: DepositMeta in bytes
"""
data: list = [self.version, self.head_id, self.tail_id,
self.available_head_id_of_virtual_step, self.available_head_id_of_deposit,
self.expires_of_virtual_step, self.expires_of_deposit]
return MsgPackForDB.dumps(data)
def put_last_calc_info(self,
context: 'IconScoreContext',
start: int,
end: int):
version = 0
value: bytes = MsgPackForDB.dumps([version, start, end])
self._db.put(context, self._KEY_LAST_CALC_INFO, value)
def migrate(self, context: 'IconScoreContext', data: List['Value']):
for value in data:
self.put_value(context, value)
self._db.put(context, self.MIGRATION_FLAG, MsgPackForDB.dumps(True))