How to use test_max_retries method in responses

Best Python code snippet using responses

rpc_test.py

Source:rpc_test.py

1""" SKALE RPCWallet test """2from http import HTTPStatus3import pytest4import mock5from hexbytes import HexBytes6from skale.wallets import RPCWallet7from skale.utils.exceptions import RPCWalletError8from tests.constants import (9    ENDPOINT, EMPTY_HEX_STR, ETH_PRIVATE_KEY,10    NOT_EXISTING_RPC_WALLET_URL, TEST_SGX_ENDPOINT, TEST_RPC_WALLET_URL11)12from tests.helper import response_mock, request_mock13from tests.wallets.utils import SgxClient14from skale.utils.web3_utils import (15    init_web3,16    private_key_to_address, private_key_to_public,17    to_checksum_address18)19TX_DICT = {20    'to': '0x1057dc7277a319927D3eB43e05680B75a00eb5f4',21    'value': 9,22    'gas': 200000,23    'gasPrice': 1,24    'nonce': 7,25    'chainId': None,26    'data': '0x0'27}28EMPTY_ETH_ACCOUNT = '0x0000000000000000000000000000000000000000'29TEST_MAX_RETRIES = 330@pytest.fixture31def web3():32    return init_web3(ENDPOINT)33@pytest.fixture34@mock.patch('skale.wallets.sgx_wallet.SgxClient', new=SgxClient)35def wallet(web3):36    return RPCWallet(TEST_RPC_WALLET_URL, TEST_SGX_ENDPOINT, web3,37                     retry_if_failed=True)38@mock.patch('skale.wallets.sgx_wallet.SgxClient', new=SgxClient)39def test_rpc_not_available(web3):40    wallet = RPCWallet(NOT_EXISTING_RPC_WALLET_URL, TEST_SGX_ENDPOINT, web3)41    with pytest.raises(RPCWalletError):42        wallet.sign_and_send({})43def test_rpc_sign_and_send(wallet):44    res_mock = response_mock(HTTPStatus.OK,45                             {'data': {'transaction_hash': EMPTY_HEX_STR},46                              'error': None})47    with mock.patch('requests.post', new=request_mock(res_mock)):48        tx_hash = wallet.sign_and_send(TX_DICT)49        assert tx_hash == EMPTY_HEX_STR50def test_rpc_sign_and_send_fails(wallet):51    cnt = 052    def post_mock(*args, **kwargs):53        nonlocal cnt54        response_mock = mock.Mock()55        if cnt < TEST_MAX_RETRIES:56            rv = {'data': None, 'error': object()}57            cnt += 158        else:59            rv = {'data': 'test', 'error': ''}60        response_mock.json = mock.Mock(return_value=rv)61        return response_mock62    with mock.patch('requests.post', post_mock):63        with pytest.raises(RPCWalletError):64            wallet.sign_and_send(TX_DICT)65        assert cnt == TEST_MAX_RETRIES66@mock.patch('skale.wallets.sgx_wallet.SgxClient', new=SgxClient)67def test_rpc_sign_and_send_sgx_unreachable_no_retries(web3):68    res_mock = response_mock(HTTPStatus.BAD_REQUEST,69                             {'data': None,70                              'error': 'Sgx server is unreachable'})71    wallet = RPCWallet(NOT_EXISTING_RPC_WALLET_URL, TEST_SGX_ENDPOINT, web3)72    with mock.patch('requests.post', new=request_mock(res_mock)):73        with pytest.raises(RPCWalletError):74            wallet.sign_and_send(TX_DICT)75            assert res_mock.call_count == 176def test_rpc_sign_and_send_sgx_unreachable(wallet):77    cnt = 078    def post_mock(*args, **kwargs):79        nonlocal cnt80        response_mock = mock.Mock()81        if cnt < TEST_MAX_RETRIES:82            rv = {'data': None, 'error': 'Sgx server is unreachable'}83            cnt += 184        else:85            rv = {'data': 'test', 'error': ''}86        response_mock.json = mock.Mock(return_value=rv)87        return response_mock88    with mock.patch('requests.post', post_mock):89        with pytest.raises(RPCWalletError):90            wallet.sign_and_send(TX_DICT)91        assert cnt == TEST_MAX_RETRIES92def test_rpc_sign_hash(wallet):93    unsigned_hash = '0x31323331'94    signed_message = wallet.sign_hash(unsigned_hash)95    assert signed_message.signature == HexBytes('0x6161616161613131313131')96def test_sign_and_send_dry_run(wallet):97    cnt = 098    def post_mock(*args, **kwargs):99        nonlocal cnt100        response_mock = mock.Mock()101        if cnt < TEST_MAX_RETRIES:102            rv = {'data': None, 'error': 'Dry run failed: {"Test"}'}103            cnt += 1104        else:105            rv = {'data': 'test', 'error': ''}106        response_mock.json = mock.Mock(return_value=rv)107        return response_mock108    with mock.patch('requests.post', post_mock):109        with pytest.raises(RPCWalletError):110            wallet.sign_and_send(TX_DICT)111        assert cnt == 1112def test_rpc_address(wallet):113    address = to_checksum_address(114        private_key_to_address(ETH_PRIVATE_KEY)115    )116    assert wallet.address == address117def test_rpc_public_key(wallet):118    pk = private_key_to_public(ETH_PRIVATE_KEY)...

test_api.py

Source:test_api.py

...3import requests4import time5from api import get6class TestGetRequest(unittest.TestCase):7    def test_max_retries(self):8        with patch('requests.get', side_effect=requests.exceptions.RequestException) as requests_get:9            with self.assertRaises(requests.exceptions.RequestException):10                get('http://example.com', max_retries=3, backoff_factor=0)11        self.assertEqual(requests_get.call_count, 3)12    13    def test_backoff_factor(self):14        backoff_factor = 0.515        max_retries = 416        total_delay = sum([backoff_factor * (2 ** n) for n in range(max_retries-1)])17        start_time = time.time()18        with patch('requests.get', side_effect=requests.exceptions.RequestException) as requests_get:19            with self.assertRaises(requests.exceptions.RequestException):20                get('http://example.com', max_retries=max_retries, backoff_factor=backoff_factor)21        end_time = time.time()22        time_diff = end_time - start_time23        24        self.assertAlmostEqual(time_diff, total_delay, places=0)25        self.assertEqual(requests_get.call_count, max_retries)26    def test_raises_on_timeout_error(self):27        with patch('requests.get', side_effect=requests.exceptions.ReadTimeout):28            with self.assertRaises(requests.exceptions.ReadTimeout):29                get('http://example.com', max_retries=1)30    def test_raises_on_http_error(self):31        with patch('requests.get', side_effect=requests.exceptions.HTTPError):32            with self.assertRaises(requests.exceptions.HTTPError):33                get('http://example.com', max_retries=1)34    def test_raises_on_server_internal_error(self):35        with patch('requests.get', side_effect=requests.exceptions.ConnectionError):36            with self.assertRaises(requests.exceptions.ConnectionError):37                get('http://example.com', max_retries=1)38m = TestGetRequest()39#m.test_backoff_factor()40m.test_max_retries()41m.test_raises_on_http_error()42m.test_raises_on_server_internal_error()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.