Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@pytest.mark.parametrize(
'types, values, expected',
(
(
['address'],
['one.eth'],
HexBytes("0x2ff37b5607484cd4eecf6d13292e22bd6e5401eaffcc07e279583bc742c68882"),
),
(
['address[]'],
[['one.eth', 'two.eth']],
HexBytes("0xb98565c0c26a962fd54d93b0ed6fb9296e03e9da29d2281ed3e3473109ef7dde"),
),
),
)
def test_solidityKeccak_ens(
self, web3: "Web3", types: Sequence[TypeStr], values: Sequence[str], expected: HexBytes
) -> None:
with ens_addresses(web3, {
'one.eth': cast(ChecksumAddress, "0x49EdDD3769c0712032808D86597B84ac5c2F5614"),
'two.eth': cast(ChecksumAddress, "0xA6b759bBbf4B59D24acf7E06e79f3a5D104fdCE5"),
}):
# when called as class method, any name lookup attempt will fail
with pytest.raises(InvalidAddress):
Web3.solidityKeccak(types, values)
# when called as instance method, ens lookups can succeed
actual = web3.solidityKeccak(types, values)
HexBytes("0x30f95d210785601eb33ae4d53d405b26f920e765dff87cca8e9a4aec99f82671"),
),
(
['bool', 'uint16'],
[True, 299],
HexBytes("0xed18599ccd80ee9fae9a28b0e34a5573c3233d7468f808fd659bc171cf0b43bd"),
),
(
['int256'],
[-10],
HexBytes("0xd6fb717f7e270a360f5093ce6a7a3752183e89c9a9afe5c0cb54b458a304d3d5"),
),
(
['int256'],
[10],
HexBytes("0xc65a7bb8d6351c1cf70c95a316cc6a92839c986682d98bc35f958f4883f9d2a8"),
),
HexBytes("0xf3f7a9fe364faab93b216da50a3214154f22a0a2b415b23a84c8169e8b636ee3"),
),
(
['address[]'],
[[
"0x49EdDD3769c0712032808D86597B84ac5c2F5614",
"0xA6b759bBbf4B59D24acf7E06e79f3a5D104fdCE5",
]],
HexBytes("0xb98565c0c26a962fd54d93b0ed6fb9296e03e9da29d2281ed3e3473109ef7dde"),
),
(
['address[]'],
[[
"0x49EdDD3769c0712032808D86597B84ac5c2F5614",
"0xa6b759bbbf4b59d24acf7e06e79f3a5d104fdce5",
]],
InvalidAddress,
'logIndex': 0,
'topics': [HexBytes('0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef'),
HexBytes('0x0000000000000000000000001111111111111111111111111111111111111111'),
HexBytes('0x0000000000000000000000000000000000000000000000000000000000000000')],
'transactionHash': HexBytes('0xf45f335a0bb17d112e870f98218ebd5159e5d7ab9f1739677d7c0b3df4879446'),
'transactionIndex': 0,
'transactionLogIndex': '0x0',
'type': 'mined'}],
[{'address': '0x1383B25f9ba231e3a1a1E45c0b5689d778D44AD5',
'blockHash': HexBytes('0xf32e073349e4ca49c5e193b161ea1b9ba7dc9c2a9bf1271725c99bb5c690bba8'),
'blockNumber': 836877, 'data': '0x0000000000000000000000000000000000000000000000000de0b6b3a7640000',
'logIndex': 0,
'topics': [HexBytes('0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef'),
HexBytes('0x0000000000000000000000000000000000000000000000000000000000000000'),
HexBytes('0x0000000000000000000000001111111111111111111111111111111111111111')],
'transactionHash': HexBytes('0xf45f335a0bb17d112e870f98218ebd5159e5d7ab9f1739677d7c0b3df4879457'),
'transactionIndex': 0,
'transactionLogIndex': '0x0',
'type': 'mined'},
{'address': '0x1383B25f9ba231e3a1a1E45c0b5689d778D44AD5',
'blockHash': HexBytes('0xf32e073349e4ca49c5e193b161ea1b9ba7dc9c2a9bf1271725c99bb5c690bbc8'),
'blockNumber': 836877, 'data': '0x0000000000000000000000000000000000000000000000000de0b6b3a7640000',
'logIndex': 0,
'topics': [HexBytes('0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef'),
HexBytes('0x0000000000000000000000000000000000000000000000000000000000000000'),
HexBytes('0x0000000000000000000000001111111111111111111111111111111111111111')],
'transactionHash': HexBytes('0xf45f335a0bb17d112e870f98218ebd5159e5d7ab9f1739677d7c0b3df4873457'),
'transactionIndex': 0,
'transactionLogIndex': '0x0',
'type': 'mined'}],
[{'address': '0x1383B25f9ba231e3a1a1E45c0b5689d778D44AD5',
'blockHash': HexBytes('0xf32e073349e4ca49c5e193b161ea1b9ba7dc9c2a9bf1271725c99bb5c690bba9'),
web3.shh.post({
'sig': sender,
'powTarget': 2.5,
'powTime': 2,
'payload': payloads[1],
'topic': topic,
'pubKey': receiver_pub
})
time.sleep(1)
received_messages = shh_filter.get_new_entries()
assert len(received_messages) == 1
message = received_messages[0]
assert message["payload"] == HexBytes(payloads[1])
assert message["topic"] == HexBytes(topic)
def get_transactions():
body = request.get_json()
try:
_get_transactions_schema_validator(body)
except fastjsonschema.JsonSchemaException as e:
return failure('Invalid JSON: ' + e.message, 400)
ret: Dict[str, List[Any]] = defaultdict(list)
for transaction in body['transactions']:
event = events_from_transaction(HexBytes(transaction), g.chain.name)
for k, v in event.items():
ret[k].extend(v)
if ret['errors']:
logging.error('Got transaction errors: %s', ret['errors'])
return failure(ret, 400)
return success(ret)
def to_address(value: str) -> str:
"""Convert a value to an address"""
if isinstance(value, bytes):
value = HexBytes(value).hex()
value = eth_utils.add_0x_prefix(str(value))
try:
return eth_utils.to_checksum_address(value)
except ValueError:
raise ValueError(f"'{value}' is not a valid ETH address.") from None
def signed_tx(cls, sutx, v, r, s):
enctx = encode_transaction(sutx, (v, r, s))
transaction_hash = keccak(enctx)
attr_dict = AttributeDict({
'rawTransaction': HexBytes(enctx),
'hash': HexBytes(transaction_hash),
'r': r,
's': s,
'v': v,
})
return attr_dict
def _get_memory(step: Dict, idx: int) -> HexBytes:
offset = int(step["stack"][idx], 16) * 2
length = int(step["stack"][idx - 1], 16) * 2
return HexBytes("".join(step["memory"])[offset : offset + length])
def get_serializable_dict(input_dict):
serializable_dict = {}
for key, value in input_dict.items():
if type(value) == list and len(value) and type(value[0]) == HexBytes:
value = [tx.hex() for tx in value]
if key == 'logs':
value = [Filler.get_serializable_dict(item) for item in value]
if type(value) == HexBytes:
value = value.hex()
serializable_dict[key] = value
return serializable_dict