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) -> None:
block_height: int = random.randint(1000, 10000)
block_hash: bytes = os.urandom(32)
prev_block_hash: bytes = os.urandom(32)
timestamp_us: int = int(time.time() * 1000_000)
cumulative_fee = random.randint(0, 10000)
last_block = Block(
block_height=block_height,
block_hash=block_hash,
timestamp=timestamp_us,
prev_hash=prev_block_hash,
cumulative_fee=cumulative_fee,
)
block_height = last_block.height - 10
block_hash: bytes = os.urandom(32)
term_start_block_height = block_height = block_height - 20
last_block: "Block" = last_block
metadata = Metadata(
block_height, block_hash, term_start_block_height, last_block
)
assert metadata.block_height == block_height
def _create_dummy_block_by_height(self, height: int):
block_hash = create_block_hash()
prev_block_hash = create_block_hash()
timestamp_us = create_timestamp()
return Block(height, block_hash, timestamp_us, prev_block_hash, 0)
def make_context(self):
self._tx_index += 1
self._context = IconScoreContext(IconScoreContextType.DIRECT)
self._context.msg = Message(self.from_address, 0)
tx_hash = create_tx_hash()
self._context.new_icon_score_mapper = IconScoreMapper()
self._context.tx = Transaction(tx_hash,
origin=self.from_address,
timestamp=randrange(1, 1000),
nonce=randrange(1, 1000))
self._context.block = Block(1, create_block_hash(), 0, None, 0)
self._context.icon_score_mapper = self._icon_score_mapper
self._context.icx = IcxEngine()
self._context.icx.open(self._icx_storage)
ContextContainer._push_context(self._context)
self._context.validate_score_blacklist = Mock()
self._context.is_service_flag_on = Mock(return_value=False)
'to': to,
'value': value,
'stepLimit': step_limit,
'timestamp': 1234567890,
'dataType': 'call',
'data': {
'method': 'getScoreStatus',
'params': {
'txHash': tx_hash
}
},
'signature': 'VAia7YZ2Ji6igKWzjR2YsGa2m53nKPrfK7uXYW78QLE+ATehAVZPC40szvAiA6NEU5gCYB4c4qaQzqDh2ugcHgA='
}
}
block = Block(block_height, block_hash, block_timestamp, self.genesis_block.hash, 0)
tx_results, state_root_hash, _, _ = self.icon_service_engine.invoke(block, [tx_v3])
self.assertIsInstance(state_root_hash, bytes)
self.assertEqual(len(state_root_hash), 32)
self.assertEqual(len(tx_results), 1)
tx_result: 'TransactionResult' = tx_results[0]
self.assertIsNotNone(tx_result.failure)
self.assertIsNone(tx_result.score_address)
self.assertEqual(tx_result.status, 0)
self.assertEqual(tx_result.block_height, block_height)
self.assertEqual(tx_result.block_hash, block_hash)
self.assertEqual(tx_result.tx_index, 0)
self.assertEqual(tx_result.tx_hash, tx_hash)
tx_v3 = {
'method': 'icx_sendTransaction',
'params': {
'nid': 3,
'version': 3,
'from': self._admin.address,
'to': self._to,
'value': value,
'stepLimit': 1000000,
'timestamp': 1234567890,
'txHash': tx_hash
}
}
block = Block(block_height,
block_hash,
block_timestamp,
self.genesis_block.hash,
0)
tx_results, state_root_hash, _, _ = self.icon_service_engine.invoke(block, [tx_v3])
self.assertIsInstance(state_root_hash, bytes)
self.assertEqual(len(state_root_hash), 32)
self.assertEqual(len(tx_results), 1)
tx_result: 'TransactionResult' = tx_results[0]
self.assertIsNone(tx_result.failure)
self.assertIsNone(tx_result.score_address)
self.assertEqual(tx_result.status, 1)
self.assertEqual(tx_result.block_height, block_height)
def _test_block_from_bytes_to_bytes_struct(
block_hash: bytes, prev_block_hash: Optional[bytes]
):
revision = 0
cumulative_fee = 10
block1 = Block(1, block_hash, 100, prev_block_hash, cumulative_fee)
data = Block.to_bytes(block1, revision)
assert isinstance(data, bytes)
assert (
1
+ DEFAULT_BYTE_SIZE
+ DEFAULT_BYTE_SIZE
+ DEFAULT_BYTE_SIZE
+ DEFAULT_BYTE_SIZE
== len(data)
)
block2 = Block.from_bytes(data)
assert block2.height == 1
assert block2.hash == block_hash
assert block2.timestamp == 100
assert block2.prev_hash == prev_block_hash
def from_dict(params: dict):
block_height = params.get("blockHeight")
block_hash = params.get("blockHash")
timestamp = params.get("timestamp", 0)
prev_hash = params.get("prevBlockHash", b"\x00" * 32)
return Block(
block_height=block_height,
block_hash=block_hash,
timestamp=timestamp,
prev_hash=prev_hash,
cumulative_fee=0,
)
def from_bytes(buf: bytes) -> "Block":
if (
len(buf) == Block._STRUCT_PACKED_BYTES_SIZE
and buf[0] == BlockVersion.STRUCT
):
return Block._from_struct_packed_bytes(buf)
else:
return Block._from_msg_packed_bytes(buf)
block_height_bytes,
block_hash_bytes,
timestamp_bytes,
block_prev_hash_bytes,
) = Block._struct.unpack(buf)
block_height = int.from_bytes(block_height_bytes, byteorder)
block_hash = block_hash_bytes
timestamp = int.from_bytes(timestamp_bytes, byteorder)
byte_prev_hash = block_prev_hash_bytes
if int(bytes.hex(byte_prev_hash), 16) == 0:
byte_prev_hash = None
prev_block_hash = byte_prev_hash
block = Block(block_height, block_hash, timestamp, prev_block_hash, 0)
return block