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_dynamic_fixed_array_type_decode_u8(self):
obj = ScaleDecoder.get_decoder_class('[u8; 65]', data=ScaleBytes("0xc42b82d02bce3202f6a05d4b06d1ad46963d3be36fd0528bbe90e7f7a4e5fcd38d14234b1c9fcee920d76cfcf43b4ed5dd718e357c2bc1aae3a642975207e67f01"))
self.assertEqual('0xc42b82d02bce3202f6a05d4b06d1ad46963d3be36fd0528bbe90e7f7a4e5fcd38d14234b1c9fcee920d76cfcf43b4ed5dd718e357c2bc1aae3a642975207e67f01', obj.decode())
def test_compact_bool_invalid(self):
obj = ScaleDecoder.get_decoder_class('bool', ScaleBytes("0x02"))
self.assertRaises(InvalidScaleTypeValueException, obj.decode)
def test_all_scale_type_supported_v3(self):
metadata_decoder = MetadataDecoder(ScaleBytes(metadata_v3_hex))
metadata_decoder.decode()
self.assertEqual(metadata_decoder.version.value, "MetadataV3Decoder")
for module in metadata_decoder.metadata.modules:
if module.calls:
for call in module.calls:
for arg in call.args:
decoder_class = ScaleDecoder.get_decoder_class(arg.type, ScaleBytes('0x00'))
self.assertIsNotNone(decoder_class, msg='{} is not supported by metadata'.format(arg.type))
def test_compact_bool_false(self):
obj = ScaleDecoder.get_decoder_class('bool', ScaleBytes("0x00"))
obj.decode()
self.assertEqual(obj.value, False)
ScaleBytes(bytearray.fromhex(log.data['value']['data'].replace('0x', '')))
).decode()
if len(list(babe_predigest.values())) > 0:
babe_predigest_value = list(babe_predigest.values())[0]
log.data['value']['data'] = babe_predigest_value
self.block.authority_index = log.data['value']['data']['authorityIndex']
self.block.slot_number = log.data['value']['data']['slotNumber']
if log.data['value']['engine'] == 'aura':
aura_predigest_cls = RuntimeConfiguration().get_decoder_class('RawAuraPreDigest')
aura_predigest = aura_predigest_cls(
ScaleBytes(bytearray.fromhex(log.data['value']['data'].replace('0x', '')))
).decode()
log.data['value']['data'] = aura_predigest
self.block.slot_number = aura_predigest['slotNumber']
log.save(db_session)
block.count_events = 0
# === Extract extrinsics from block ====
extrinsics_data = json_block['block'].pop('extrinsics')
block.count_extrinsics = len(extrinsics_data)
extrinsic_idx = 0
extrinsics = []
for extrinsic in extrinsics_data:
extrinsics_decoder = ExtrinsicsDecoder(
data=ScaleBytes(extrinsic),
metadata=self.metadata_store[parent_spec_version]
)
extrinsic_data = extrinsics_decoder.decode()
# Lookup result of extrinsic
extrinsic_success = extrinsic_success_idx.get(extrinsic_idx, False)
model = Extrinsic(
block_id=block_id,
extrinsic_idx=extrinsic_idx,
extrinsic_hash=extrinsics_decoder.extrinsic_hash,
extrinsic_length=extrinsic_data.get('extrinsic_length'),
extrinsic_version=extrinsic_data.get('version_info'),
signed=extrinsics_decoder.contains_transaction,
unsigned=not extrinsics_decoder.contains_transaction,
# Look up call module from metadata
for call_index, (call_module, call_function) in self.metadata.call_index.items():
if call_module.name == value['call_module'] and call_function.name == value['call_function']:
self.call_index = call_index
self.call_module = call_module
self.call_function = call_function
break
if not self.call_index:
raise ValueError('Specified call module and function not found in metadata')
elif not self.call_module or not self.call_function:
raise ValueError('No call module and function specified')
data = ScaleBytes(bytearray.fromhex(self.call_index))
# Encode call params
if len(self.call_function.args) > 0:
for arg in self.call_function.args:
if arg.name not in value['call_args']:
raise ValueError('Parameter \'{}\' not specified'.format(arg.name))
else:
param_value = value['call_args'][arg.name]
arg_obj = self.get_decoder_class(arg.type, metadata=self.metadata)
data += arg_obj.encode(param_value)
return data
def process_encode(self, value):
if type(value) == str and value[0:2] != '0x':
# Assume SS58 encoding address
if len(value) >= 46:
from scalecodec.utils.ss58 import ss58_decode
value = '0x{}'.format(ss58_decode(value))
else:
from scalecodec.utils.ss58 import ss58_decode_account_index
index_obj = GenericAccountIndex()
value = index_obj.encode(ss58_decode_account_index(value))
if type(value) == str and value[0:2] == '0x' and len(value) == 66:
# value is AccountId
return ScaleBytes('0xff{}'.format(value[2:]))
elif type(value) == int:
# value is AccountIndex
raise NotImplementedError('Encoding of AccountIndex Adresses not supported yet')
else:
raise ValueError('Value is in unsupported format, expected 32 bytes hex-string for AccountIds or int for AccountIndex')