Nucypher
/
nucypher
mirror of https://github.com/nucypher/nucypher.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Sort fixtures that compile to localized acceptance/unit conftest respectively; Faster unit test suite.

pull/3021/head
Kieran Prasch 2022-11-17 14:31:17 +00:00
parent 679a655558
commit ff3f414331
8 changed files with 379 additions and 331 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

168
tests/acceptance/blockchain/conditions/conftest.py Normal file
View File

@ -0,0 +1,168 @@
from pathlib import Path
import pytest
import nucypher
from nucypher.blockchain.eth.agents import ContractAgency, NucypherTokenAgent, SubscriptionManagerAgent
from nucypher.blockchain.eth.signers.software import Web3Signer
from nucypher.blockchain.eth.sol.compile.compile import multiversion_compile
from nucypher.blockchain.eth.sol.compile.types import SourceBundle
from nucypher.crypto.powers import TransactingPower
from nucypher.policy.conditions.context import USER_ADDRESS_CONTEXT
from nucypher.policy.conditions.evm import ContractCondition
from nucypher.policy.conditions.lingo import ReturnValueTest, ConditionLingo, OR, AND
from tests.constants import TESTERCHAIN_CHAIN_ID
@pytest.fixture()
def condition_providers(testerchain):
providers = {testerchain.client.chain_id: testerchain.provider}
return providers
@pytest.fixture(autouse=True)
def mock_condition_blockchains(mocker):
"""adds testerchain's chain ID to permitted conditional chains"""
mocker.patch.object(
nucypher.policy.conditions.evm, "_CONDITION_CHAINS", tuple([TESTERCHAIN_CHAIN_ID])
)
@pytest.fixture()
def compound_lingo(erc721_evm_condition_balanceof,
timelock_condition,
rpc_condition,
erc20_evm_condition_balanceof):
"""depends on contract deployments"""
lingo = ConditionLingo(
conditions=[
erc721_evm_condition_balanceof,
OR,
timelock_condition,
OR,
rpc_condition,
AND,
erc20_evm_condition_balanceof,
]
)
return lingo
@pytest.fixture()
@pytest.mark.usefixtures('agency')
def erc20_evm_condition_balanceof(test_registry):
token = ContractAgency.get_agent(NucypherTokenAgent, registry=test_registry)
condition = ContractCondition(
contract_address=token.contract.address,
method="balanceOf",
standard_contract_type="ERC20",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", 0),
parameters=[USER_ADDRESS_CONTEXT],
)
return condition
@pytest.fixture
def erc721_contract(testerchain, test_registry):
solidity_root = Path(__file__).parent / "contracts"
source_bundle = SourceBundle(base_path=solidity_root)
compiled_contracts = multiversion_compile([source_bundle], True)
testerchain._raw_contract_cache = compiled_contracts
origin, *everybody_else = testerchain.client.accounts
transacting_power = TransactingPower(
account=origin, signer=Web3Signer(testerchain.client)
)
contract, receipt = testerchain.deploy_contract(
transacting_power=transacting_power,
registry=test_registry,
contract_name="ConditionNFT",
)
# mint an NFT with tokenId = 1
tx = contract.functions.mint(origin, 1).transact({"from": origin})
testerchain.wait_for_receipt(tx)
return contract
@pytest.fixture
def erc721_evm_condition_owner(erc721_contract):
condition = ContractCondition(
contract_address=erc721_contract.address,
method="ownerOf",
standard_contract_type="ERC721",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", ":userAddress"),
parameters=[
":tokenId",
],
)
return condition
@pytest.fixture
def erc721_evm_condition_balanceof(erc721_contract):
condition = ContractCondition(
contract_address=erc721_contract.address,
method="balanceOf",
standard_contract_type="ERC721",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest(">", 0),
parameters=[
":userAddress",
],
)
return condition
@pytest.fixture
def subscription_manager_get_policy_zeroized_policy_struct_condition(
test_registry, agency
):
subscription_manager = ContractAgency.get_agent(
SubscriptionManagerAgent, registry=test_registry
)
condition = ContractCondition(
contract_address=subscription_manager.contract.address,
function_abi=subscription_manager.contract.get_function_by_name("getPolicy").abi,
method="getPolicy",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", ":expectedPolicyStruct"),
parameters=[":hrac"],
)
return condition
@pytest.fixture
def subscription_manager_is_active_policy_condition(test_registry, agency):
subscription_manager = ContractAgency.get_agent(
SubscriptionManagerAgent,
registry=test_registry
)
condition = ContractCondition(
contract_address=subscription_manager.contract.address,
function_abi=subscription_manager.contract.get_function_by_name("isPolicyActive").abi,
method="isPolicyActive",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", True),
parameters=[":hrac"],
)
return condition
@pytest.fixture
def custom_context_variable_erc20_condition(
test_registry, agency, testerchain, mock_condition_blockchains
):
token = ContractAgency.get_agent(NucypherTokenAgent, registry=test_registry)
condition = ContractCondition(
contract_address=token.contract.address,
method="balanceOf",
standard_contract_type="ERC20",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", 0),
parameters=[":addressToUse"],
)
return condition

65
tests/acceptance/blockchain/conditions/test_conditions.py
View File

@ -29,51 +29,6 @@ from tests.constants import TESTERCHAIN_CHAIN_ID
from tests.integration.characters.test_bob_handles_frags import _make_message_kits
@pytest.fixture()
def condition_providers(testerchain):
providers = {testerchain.client.chain_id: testerchain.provider}
return providers
@pytest.fixture(scope='module')
def valid_user_address_context():
return {
USER_ADDRESS_CONTEXT: {
"signature": "0x488a7acefdc6d098eedf73cdfd379777c0f4a4023a660d350d3bf309a51dd4251abaad9cdd11b71c400cfb4625c14ca142f72b39165bd980c8da1ea32892ff071c",
"address": "0x5ce9454909639D2D17A3F753ce7d93fa0b9aB12E",
"typedData": {
"primaryType": "Wallet",
"types": {
"EIP712Domain": [
{"name": "name", "type": "string"},
{"name": "version", "type": "string"},
{"name": "chainId", "type": "uint256"},
{"name": "salt", "type": "bytes32"},
],
"Wallet": [
{"name": "address", "type": "string"},
{"name": "blockNumber", "type": "uint256"},
{"name": "blockHash", "type": "bytes32"},
{"name": "signatureText", "type": "string"},
],
},
"domain": {
"name": "tDec",
"version": "1",
"chainId": 80001,
"salt": "0x3e6365d35fd4e53cbc00b080b0742b88f8b735352ea54c0534ed6a2e44a83ff0",
},
"message": {
"address": "0x5ce9454909639D2D17A3F753ce7d93fa0b9aB12E",
"blockNumber": 28117088,
"blockHash": "0x104dfae58be4a9b15d59ce447a565302d5658914f1093f10290cd846fbe258b7",
"signatureText": "I'm the owner of address 0x5ce9454909639D2D17A3F753ce7d93fa0b9aB12E as of block number 28117088",
},
},
}
}
def _dont_validate_user_address(context_variable: str, **context):
if context_variable == USER_ADDRESS_CONTEXT:
return context[USER_ADDRESS_CONTEXT]["address"]
@ -220,16 +175,16 @@ def test_rpc_condition_evaluation_with_context_var_in_return_value_test(
side_effect=_dont_validate_user_address,
)
def test_erc20_evm_condition_evaluation(
get_context_value_mock, testerchain, erc20_evm_condition, condition_providers
get_context_value_mock, testerchain, erc20_evm_condition_balanceof, condition_providers
):
context = {USER_ADDRESS_CONTEXT: {"address": testerchain.unassigned_accounts[0]}}
condition_result, call_result = erc20_evm_condition.verify(
condition_result, call_result = erc20_evm_condition_balanceof.verify(
providers=condition_providers, **context
)
assert condition_result is True
context[USER_ADDRESS_CONTEXT]["address"] = testerchain.etherbase_account
condition_result, call_result = erc20_evm_condition.verify(
condition_result, call_result = erc20_evm_condition_balanceof.verify(
providers=condition_providers, **context
)
assert condition_result is False
@ -550,16 +505,9 @@ def test_time_condition_evaluation(testerchain, timelock_condition, condition_pr
assert condition_result is True
def test_simple_compound_conditions_evaluation(testerchain):
def test_simple_compound_conditions_evaluation(testerchain, compound_timelock_lingo):
# TODO Improve internals of evaluation here (natural vs recursive approach)
conditions = [
{'returnValueTest': {'value': '0', 'comparator': '>'}, 'method': 'timelock'},
{'operator': 'and'},
{'returnValueTest': {'value': '99999999999999999', 'comparator': '<'}, 'method': 'timelock'},
{'operator': 'and'},
{'returnValueTest': {'value': '0', 'comparator': '>'}, 'method': 'timelock'}
]
conditions = json.dumps(conditions)
conditions = json.dumps(compound_timelock_lingo)
lingo = ConditionLingo.from_json(conditions)
result = lingo.eval()
assert result is True
@ -569,10 +517,11 @@ def test_simple_compound_conditions_evaluation(testerchain):
"nucypher.policy.conditions.evm.get_context_value",
side_effect=_dont_validate_user_address,
)
@pytest.mark.usefixtures("agency")
def test_onchain_conditions_lingo_evaluation(
get_context_value_mock,
testerchain,
compound_lingo,
compound_lingo,
condition_providers
):

32
tests/acceptance/network/test_middleware_retries.py Normal file
View File

@ -0,0 +1,32 @@
from unittest.mock import Mock
import maya
from web3.types import RPCEndpoint
from nucypher.blockchain.middleware.retry import RetryRequestMiddleware
from tests.constants import RPC_TOO_MANY_REQUESTS
# TODO - since this test does exponential backoff it takes >= 2^1 = 2s, should we only run on CI?
def test_request_with_retry_exponential_backoff():
retries = 1
make_request = Mock()
# Retry Case - RPCResponse fails due to limits, and retry required
make_request.return_value = RPC_TOO_MANY_REQUESTS
retry_middleware = RetryRequestMiddleware(make_request=make_request,
w3=Mock(),
retries=1,
exponential_backoff=True)
start = maya.now()
retry_response = retry_middleware(RPCEndpoint('web3_clientVersion'), None)
end = maya.now()
assert retry_response == RPC_TOO_MANY_REQUESTS
assert make_request.call_count == (retries + 1) # initial call, and then the number of retries
# check exponential backoff
delta = end - start
assert delta.total_seconds() >= 2**retries

19
tests/constants.py
View File

@ -132,3 +132,22 @@ CLI_TEST_ENV = {NUCYPHER_ENVVAR_KEYSTORE_PASSWORD: INSECURE_DEVELOPMENT_PASSWORD
CLI_ENV = {NUCYPHER_ENVVAR_KEYSTORE_PASSWORD: INSECURE_DEVELOPMENT_PASSWORD,
NUCYPHER_ENVVAR_OPERATOR_ETH_PASSWORD: INSECURE_DEVELOPMENT_PASSWORD}
TESTERCHAIN_CHAIN_ID = 131277322940537
#
# Network
#
RPC_TOO_MANY_REQUESTS = {
"jsonrpc": "2.0",
"error": {
"code": 429,
"message": "Too many concurrent requests"
}
}
RPC_SUCCESSFUL_RESPONSE = {
"jsonrpc": "2.0",
"id": 1,
"result": "Geth/v1.9.20-stable-979fc968/linux-amd64/go1.15"
}

269
tests/fixtures.py
View File

@ -24,7 +24,6 @@ from nucypher.blockchain.eth.actors import Operator
from nucypher.blockchain.eth.agents import (
ContractAgency,
NucypherTokenAgent,
SubscriptionManagerAgent,
)
from nucypher.blockchain.eth.agents import (
PREApplicationAgent,
@ -40,8 +39,6 @@ from nucypher.blockchain.eth.registry import (
LocalContractRegistry,
)
from nucypher.blockchain.eth.signers.software import Web3Signer
from nucypher.blockchain.eth.sol.compile.compile import multiversion_compile
from nucypher.blockchain.eth.sol.compile.types import SourceBundle
from nucypher.characters.lawful import Enrico
from nucypher.config.characters import (
AliceConfiguration,
@ -55,7 +52,7 @@ from nucypher.network.nodes import TEACHER_NODES
from nucypher.policy.conditions.context import USER_ADDRESS_CONTEXT
from nucypher.policy.conditions.context import USER_ADDRESS_CONTEXT
from nucypher.policy.conditions.evm import ContractCondition, RPCCondition
from nucypher.policy.conditions.lingo import AND, OR, ConditionLingo, ReturnValueTest
from nucypher.policy.conditions.lingo import ReturnValueTest
from nucypher.policy.conditions.time import TimeCondition
from nucypher.policy.payment import SubscriptionManagerPayment
from nucypher.utilities.emitters import StdoutEmitter
@ -929,32 +926,6 @@ def mock_rest_middleware():
#
@pytest.fixture(scope='module')
def random_context():
context = {
USER_ADDRESS_CONTEXT: {
"signature": "16b15f88bbd2e0a22d1d0084b8b7080f2003ea83eab1a00f80d8c18446c9c1b6224f17aa09eaf167717ca4f355bb6dc94356e037edf3adf6735a86fc3741f5231b",
"address": "0x03e75d7DD38CCE2e20FfEE35EC914C57780A8e29",
"typedMessage": {
"domain": {
"name": "tDec",
"version": "1",
"chainId": 1,
"salt": "0xf2d857f4a3edcb9b78b4d503bfe733db1e3f6cdc2b7971ee739626c97e86a558",
},
"message": {
"address": "0x03e75d7DD38CCE2e20FfEE35EC914C57780A8e29",
"blockNumber": 15440685,
"blockHash": "0x2220da8b777767df526acffd5375ebb340fc98e53c1040b25ad1a8119829e3bd",
"signatureText": "I'm the owner of address 0x03e75d7dd38cce2e20ffee35ec914c57780a8e29 as of block number 15440685",
},
},
}
}
return context
@pytest.fixture(scope='session')
def conditions_test_data():
test_conditions = Path(tests.__file__).parent / "data" / "test_conditions.json"
@ -982,40 +953,15 @@ def timelock_condition():
return condition
@pytest.fixture()
def erc1155_balance_condition_data(conditions_test_data):
data = json.dumps(conditions_test_data['ERC1155_balance'])
return data
@pytest.fixture()
def erc1155_balance_condition(erc1155_balance_condition_data):
data = erc1155_balance_condition_data
condition = ContractCondition.from_json(data)
return condition
@pytest.fixture()
def erc20_balance_condition_data(conditions_test_data):
data = json.dumps(conditions_test_data['ERC20_balance'])
return data
@pytest.fixture()
def erc20_balance_condition(erc20_balance_condition_data):
data = erc20_balance_condition_data
condition = ContractCondition.from_json(data)
return condition
@pytest.fixture()
def t_staking_data(conditions_test_data):
return json.dumps(conditions_test_data["TStaking"])
@pytest.fixture()
def custom_abi_with_multiple_parameters(conditions_test_data):
return json.dumps(conditions_test_data["customABIMultipleParameters"])
@pytest.fixture
def compound_timelock_lingo():
return [
{'returnValueTest': {'value': '0', 'comparator': '>'}, 'method': 'timelock'},
{'operator': 'and'},
{'returnValueTest': {'value': '99999999999999999', 'comparator': '<'}, 'method': 'timelock'},
{'operator': 'and'},
{'returnValueTest': {'value': '0', 'comparator': '>'}, 'method': 'timelock'}
]
@pytest.fixture
@ -1029,162 +975,41 @@ def rpc_condition():
return condition
@pytest.fixture
def erc20_evm_condition(test_registry, agency):
token = ContractAgency.get_agent(NucypherTokenAgent, registry=test_registry)
condition = ContractCondition(
contract_address=token.contract.address,
method="balanceOf",
standard_contract_type="ERC20",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", 0),
parameters=[USER_ADDRESS_CONTEXT],
)
return condition
@pytest.fixture(scope='module')
def valid_user_address_context():
return {
USER_ADDRESS_CONTEXT: {
"signature": "0x488a7acefdc6d098eedf73cdfd379777c0f4a4023a660d350d3bf309a51dd4251abaad9cdd11b71c400cfb4625c14ca142f72b39165bd980c8da1ea32892ff071c",
"address": "0x5ce9454909639D2D17A3F753ce7d93fa0b9aB12E",
"typedData": {
"primaryType": "Wallet",
"types": {
"EIP712Domain": [
{"name": "name", "type": "string"},
{"name": "version", "type": "string"},
{"name": "chainId", "type": "uint256"},
{"name": "salt", "type": "bytes32"},
],
"Wallet": [
{"name": "address", "type": "string"},
{"name": "blockNumber", "type": "uint256"},
{"name": "blockHash", "type": "bytes32"},
{"name": "signatureText", "type": "string"},
],
},
"domain": {
"name": "tDec",
"version": "1",
"chainId": 80001,
"salt": "0x3e6365d35fd4e53cbc00b080b0742b88f8b735352ea54c0534ed6a2e44a83ff0",
},
"message": {
"address": "0x5ce9454909639D2D17A3F753ce7d93fa0b9aB12E",
"blockNumber": 28117088,
"blockHash": "0x104dfae58be4a9b15d59ce447a565302d5658914f1093f10290cd846fbe258b7",
"signatureText": "I'm the owner of address 0x5ce9454909639D2D17A3F753ce7d93fa0b9aB12E as of block number 28117088",
},
},
}
}
@pytest.fixture
def custom_context_variable_erc20_condition(
test_registry, agency, testerchain, mock_condition_blockchains
):
token = ContractAgency.get_agent(NucypherTokenAgent, registry=test_registry)
condition = ContractCondition(
contract_address=token.contract.address,
method="balanceOf",
standard_contract_type="ERC20",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", 0),
parameters=[":addressToUse"],
)
return condition
@pytest.fixture
def erc721_contract(testerchain, test_registry):
solidity_root = Path(tests.__file__).parent / 'acceptance' / 'blockchain' / 'conditions' / "contracts"
source_bundle = SourceBundle(base_path=solidity_root)
compiled_contracts = multiversion_compile([source_bundle], True)
testerchain._raw_contract_cache = compiled_contracts
origin, *everybody_else = testerchain.client.accounts
transacting_power = TransactingPower(
account=origin, signer=Web3Signer(testerchain.client)
)
contract, receipt = testerchain.deploy_contract(
transacting_power=transacting_power,
registry=test_registry,
contract_name="ConditionNFT",
)
# mint an NFT with tokenId = 1
tx = contract.functions.mint(origin, 1).transact({"from": origin})
testerchain.wait_for_receipt(tx)
return contract
@pytest.fixture
def erc721_evm_condition_owner(erc721_contract):
condition = ContractCondition(
contract_address=erc721_contract.address,
method="ownerOf",
standard_contract_type="ERC721",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", ":userAddress"),
parameters=[
":tokenId",
],
)
return condition
@pytest.fixture
def erc721_evm_condition(test_registry):
condition = ContractCondition(
contract_address="0xaDD9D957170dF6F33982001E4c22eCCdd5539118",
method="ownerOf",
standard_contract_type="ERC721",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", ":userAddress"),
parameters=[
5954,
]
)
return condition
@pytest.fixture
def erc721_evm_condition_balanceof(erc721_contract):
condition = ContractCondition(
contract_address=erc721_contract.address,
method="balanceOf",
standard_contract_type="ERC721",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest(">", 0),
parameters=[
":userAddress",
],
)
return condition
@pytest.fixture
def subscription_manager_is_active_policy_condition(test_registry, agency):
subscription_manager = ContractAgency.get_agent(
SubscriptionManagerAgent,
registry=test_registry
)
condition = ContractCondition(
contract_address=subscription_manager.contract.address,
function_abi=subscription_manager.contract.get_function_by_name("isPolicyActive").abi,
method="isPolicyActive",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", True),
parameters=[":hrac"],
)
return condition
@pytest.fixture
def subscription_manager_get_policy_zeroized_policy_struct_condition(
test_registry, agency
):
subscription_manager = ContractAgency.get_agent(
SubscriptionManagerAgent, registry=test_registry
)
condition = ContractCondition(
contract_address=subscription_manager.contract.address,
function_abi=subscription_manager.contract.get_function_by_name("getPolicy").abi,
method="getPolicy",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", ":expectedPolicyStruct"),
parameters=[":hrac"],
)
return condition
@pytest.fixture
def timelock_condition():
condition = TimeCondition(
return_value_test=ReturnValueTest('>', 0)
)
return condition
@pytest.fixture()
def compound_lingo(erc721_evm_condition_balanceof,
timelock_condition,
rpc_condition,
erc20_evm_condition):
lingo = ConditionLingo(
conditions=[
erc721_evm_condition_balanceof,
OR,
timelock_condition,
OR,
rpc_condition,
AND,
erc20_evm_condition,
]
)
return lingo

94
tests/unit/conditions/conftest.py Normal file
View File

@ -0,0 +1,94 @@
import json
import pytest
from nucypher.policy.conditions.context import USER_ADDRESS_CONTEXT
from nucypher.policy.conditions.evm import ContractCondition
from nucypher.policy.conditions.lingo import ReturnValueTest, ConditionLingo, OR, AND
from tests.constants import TESTERCHAIN_CHAIN_ID
@pytest.fixture()
def compound_lingo(erc721_evm_condition,
timelock_condition,
rpc_condition,
erc20_evm_condition):
"""does not depend on contract deployments"""
lingo = ConditionLingo(
conditions=[
erc721_evm_condition,
OR,
timelock_condition,
OR,
rpc_condition,
AND,
erc20_evm_condition,
]
)
return lingo
@pytest.fixture()
def erc1155_balance_condition_data(conditions_test_data):
data = json.dumps(conditions_test_data['ERC1155_balance'])
return data
@pytest.fixture()
def erc1155_balance_condition(erc1155_balance_condition_data):
data = erc1155_balance_condition_data
condition = ContractCondition.from_json(data)
return condition
@pytest.fixture()
def erc20_balance_condition_data(conditions_test_data):
data = json.dumps(conditions_test_data['ERC20_balance'])
return data
@pytest.fixture()
def erc20_balance_condition(erc20_balance_condition_data):
data = erc20_balance_condition_data
condition = ContractCondition.from_json(data)
return condition
@pytest.fixture()
def t_staking_data(conditions_test_data):
return json.dumps(conditions_test_data["TStaking"])
@pytest.fixture()
def custom_abi_with_multiple_parameters(conditions_test_data):
return json.dumps(conditions_test_data["customABIMultipleParameters"])
@pytest.fixture
def erc20_evm_condition(test_registry):
condition = ContractCondition(
contract_address="0xaDD9D957170dF6F33982001E4c22eCCdd5539118",
method="balanceOf",
standard_contract_type="ERC20",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", 0),
parameters=[
USER_ADDRESS_CONTEXT,
],
)
return condition
@pytest.fixture
def erc721_evm_condition(test_registry):
condition = ContractCondition(
contract_address="0xaDD9D957170dF6F33982001E4c22eCCdd5539118",
method="ownerOf",
standard_contract_type="ERC721",
chain=TESTERCHAIN_CHAIN_ID,
return_value_test=ReturnValueTest("==", ":userAddress"),
parameters=[
5954,
]
)
return condition

3
tests/unit/conditions/test_evm_condition_serializers.py
View File

@ -46,8 +46,7 @@ def test_type_resolution_from_json(
conditions = (
timelock_condition,
rpc_condition,
erc20_evm_condition,
erc721_evm_condition,
erc20_evm_condition
)
for condition in conditions:
condition_json = condition.to_json()

60
tests/unit/test_web3_middleware.py
View File

@ -1,29 +1,16 @@
from typing import Any
from unittest.mock import Mock
import maya
import pytest
from requests import HTTPError
from web3.types import RPCResponse, RPCError, RPCEndpoint
from nucypher.blockchain.middleware.retry import RetryRequestMiddleware, AlchemyRetryRequestMiddleware, \
from nucypher.blockchain.middleware.retry import (
RetryRequestMiddleware,
AlchemyRetryRequestMiddleware,
InfuraRetryRequestMiddleware
TOO_MANY_REQUESTS = {
"jsonrpc": "2.0",
"error": {
"code": 429,
"message": "Too many concurrent requests"
}
}
SUCCESSFUL_RESPONSE = {
"jsonrpc": "2.0",
"id": 1,
"result": "Geth/v1.9.20-stable-979fc968/linux-amd64/go1.15"
}
)
from tests.constants import RPC_TOO_MANY_REQUESTS, RPC_SUCCESSFUL_RESPONSE
RETRY_REQUEST_CLASSES = (RetryRequestMiddleware, AlchemyRetryRequestMiddleware, InfuraRetryRequestMiddleware)
@ -33,12 +20,12 @@ def test_is_request_result_retry(retry_middleware_class):
# base checks
retry_middleware = retry_middleware_class(make_request=Mock(), w3=Mock())
assert retry_middleware.is_request_result_retry(result=TOO_MANY_REQUESTS)
assert retry_middleware.is_request_result_retry(result=RPC_TOO_MANY_REQUESTS)
http_error = HTTPError(response=Mock(status_code=429))
assert retry_middleware.is_request_result_retry(result=http_error)
assert not retry_middleware.is_request_result_retry(result=SUCCESSFUL_RESPONSE)
assert not retry_middleware.is_request_result_retry(result=RPC_SUCCESSFUL_RESPONSE)
@pytest.mark.parametrize('retry_middleware_class', RETRY_REQUEST_CLASSES)
@ -51,10 +38,10 @@ def test_request_with_retry(retry_middleware_class):
exponential_backoff=False)
# Retry Case - RPCResponse fails due to limits, and retry required
make_request.return_value = TOO_MANY_REQUESTS
make_request.return_value = RPC_TOO_MANY_REQUESTS
retry_response = retry_middleware(method=RPCEndpoint('web3_clientVersion'), params=None)
assert retry_response == TOO_MANY_REQUESTS
assert retry_response == RPC_TOO_MANY_REQUESTS
assert make_request.call_count == (retries + 1) # initial call, and then the number of retries
@ -76,42 +63,17 @@ def test_request_with_non_retry_exception(retry_middleware_class):
def test_request_success_with_no_retry(retry_middleware_class):
# Success Case - retry not needed
make_request = Mock()
make_request.return_value = SUCCESSFUL_RESPONSE
make_request.return_value = RPC_SUCCESSFUL_RESPONSE
retry_middleware = retry_middleware_class(make_request=make_request,
w3=Mock(),
retries=10,
exponential_backoff=False)
retry_response = retry_middleware(method=RPCEndpoint('web3_clientVersion'), params=None)
assert retry_response == SUCCESSFUL_RESPONSE
assert retry_response == RPC_SUCCESSFUL_RESPONSE
assert make_request.call_count == 1 # first call was successful, no need for retries
# TODO - since this test does exponential backoff it takes >= 2^1 = 2s, should we only run on circleci?
def test_request_with_retry_exponential_backoff():
retries = 1
make_request = Mock()
# Retry Case - RPCResponse fails due to limits, and retry required
make_request.return_value = TOO_MANY_REQUESTS
retry_middleware = RetryRequestMiddleware(make_request=make_request,
w3=Mock(),
retries=1,
exponential_backoff=True)
start = maya.now()
retry_response = retry_middleware(RPCEndpoint('web3_clientVersion'), None)
end = maya.now()
assert retry_response == TOO_MANY_REQUESTS
assert make_request.call_count == (retries + 1) # initial call, and then the number of retries
# check exponential backoff
delta = end - start
assert delta.total_seconds() >= 2**retries
def test_alchemy_request_with_retry():
retries = 4