Best Python code snippet using playwright-python
test_stellar_wallet.py
Source:test_stellar_wallet.py
1import pytest2from aiohttp import web3from aiohttp.test_utils import unittest_run_loop4from asynctest import patch5from tests.test_utils import BaseTestClass6from conf import settings7from stellar.wallet import (8 get_stellar_wallet,9 get_transaction,10 get_wallet_effect,11 get_operations_of_transaction,12 get_transaction_by_wallet,13 submit_transaction,14)15HORIZON_URL = settings['HORIZON_URL']16class TestHorizonApi(BaseTestClass):17 class SuccessResponse:18 def __init__(self, content_type):19 self.status = 20020 self.content_type = content_type21 async def json(self):22 return {23 'account_id': 'test',24 'balances': 'test',25 'sequence': 'test',26 'data': 'test',27 'signers': 'test',28 'thresholds': 'test',29 'flags': 'test',30 }31 class GetOperationsOfTransactionSuccess:32 def __init__(self, content_type):33 self.status = 20034 self.operation_records = [35 {36 "_links": {37 "self": {"href": "test-self-link"},38 "transaction": {"href": "test-link"},39 "effects": {"href": "test-self-link/effects"},40 "succeeds": {"href": "success-link"},41 "precedes": {"href": "precedes-link"},42 },43 "id": "test-id",44 "paging_token": "test-paging-token",45 "source_account": "test-source-account-address",46 "type": "create_account",47 "type_i": 0,48 "created_at": "2018-10-30T06:18:30Z",49 "transaction_hash": "test-hash",50 "starting_balance": "10000.0000000",51 "funder": "test-funder-address",52 "account": "test-account-address",53 }54 ]55 self.content_type = content_type56 async def json(self):57 return {58 "_links": {"self": {"href": "self-link"}, "next": {"href": "next-link"}, "prev": {"href": "prev-link"}},59 "_embedded": {"records": self.operation_records},60 }61 class GetTransactionBywalletSuccess:62 def __init__(self, content_type):63 self.status = 20064 self.transaction_records = [65 {66 '_links': {67 'self': {'href': 'test-self-link'},68 'account': {'href': 'test-account-link'},69 'ledger': {'href': 'test-ledger_link'},70 'operations': {'href': 'test-operation-link}', 'templated': True},71 'effects': {'href': 'test-effect-link', 'templated': True},72 'precedes': {'href': 'test-precedes-link'},73 'succeeds': {'href': 'test-succeed'},74 },75 'id': 'test-id',76 'paging_token': 'test-paging_token',77 'hash': 'test-hash',78 'ledger': 434_802,79 'created_at': '2018-10-30T06:18:30Z',80 'source_account': 'test-source-account-address',81 'source_account_sequence': '803158929876',82 'fee_paid': 100,83 'operation_count': 1,84 'envelope_xdr': 'test-envelop-xdr',85 'result_xdr': 'test-result-xdr',86 'result_meta_xdr': 'test-result-meta-xdr',87 'fee_meta_xdr': 'test-fee-meta-xdr',88 'memo_type': 'none',89 'signatures': ['test-signature'],90 }91 ]92 self.content_type = content_type93 async def json(self):94 return {95 '_links': {'self': {'href': 'self-link'}, 'next': {'href': 'next-link'}, 'prev': {'href': 'prev-link'}},96 '_embedded': {'records': self.transaction_records},97 }98 class NotFoundResponse:99 def __init__(self, content_type):100 self.status = 404101 self.content_type = content_type102 async def json(self):103 return {104 'type': 'https://stellar.org/horizon-errors/not_found',105 'title': 'Resource Missing',106 'status': 404,107 'detail': 'test-detail',108 }109 class BadRequestResponse:110 def __init__(self, content_type):111 self.status = 400112 self.content_type = content_type113 async def json(self):114 return {'detail': 'test-detail'}115 async def setUpAsync(self):116 self.wallet_address = 'test-address'117 self.transaction_hash = 'test-hash'118 @unittest_run_loop119 @patch('stellar.wallet.ClientSession.get')120 async def test_get_stellar_wallet_success(self, mock_get):121 session = mock_get.return_value122 session.__aenter__.return_value = self.SuccessResponse('application/hal+json')123 wallet = await get_stellar_wallet(self.wallet_address)124 url = f'{HORIZON_URL}/accounts/{self.wallet_address}'125 mock_get.assert_called_once_with(url)126 @unittest_run_loop127 @patch('stellar.wallet.ClientSession.get')128 async def test_get_stellar_wallet_fail(self, mock_get):129 session = mock_get.return_value130 session.__aenter__.return_value = self.NotFoundResponse('application/problem+json')131 with pytest.raises(web.HTTPNotFound):132 wallet = await get_stellar_wallet(self.wallet_address)133 url = f'{HORIZON_URL}/accounts/{self.wallet_address}'134 mock_get.assert_called_once_with(url)135 @unittest_run_loop136 @patch('stellar.wallet.ClientSession.get')137 async def test_get_stellar_wallet_upstream_fail(self, mock_get):138 session = mock_get.return_value139 session.__aenter__.return_value = self.NotFoundResponse('document/html')140 with pytest.raises(web.HTTPInternalServerError):141 wallet = await get_stellar_wallet(self.wallet_address)142 url = f'{HORIZON_URL}/accounts/{self.wallet_address}'143 mock_get.assert_called_once_with(url)144 @unittest_run_loop145 @patch('stellar.wallet.ClientSession.get')146 async def test_get_transaction_success(self, mock_get):147 session = mock_get.return_value148 session.__aenter__.return_value = self.SuccessResponse('application/hal+json')149 transaction = await get_transaction(self.transaction_hash)150 url = f'{HORIZON_URL}/transactions/{self.transaction_hash}'151 mock_get.assert_called_once_with(url)152 @unittest_run_loop153 @patch('stellar.wallet.ClientSession.get')154 async def test_get_transaction_fail(self, mock_get):155 session = mock_get.return_value156 session.__aenter__.return_value = self.NotFoundResponse('application/problem+json')157 with pytest.raises(web.HTTPNotFound):158 transaction = await get_transaction(self.transaction_hash)159 url = f'{HORIZON_URL}/transactions/{self.transaction_hash}'160 mock_get.assert_called_once_with(url)161 @unittest_run_loop162 @patch('stellar.wallet.ClientSession.get')163 async def test_get_transaction_upstream_fail(self, mock_get):164 session = mock_get.return_value165 session.__aenter__.return_value = self.NotFoundResponse('document/html')166 with pytest.raises(web.HTTPInternalServerError):167 transaction = await get_transaction(self.transaction_hash)168 url = f'{HORIZON_URL}/transactions/{self.transaction_hash}'169 mock_get.assert_called_once_with(url)170 @unittest_run_loop171 @patch('stellar.wallet.ClientSession.get')172 async def test_get_wallet_effect_success(self, mock_get):173 session = mock_get.return_value174 session.__aenter__.return_value = self.SuccessResponse('application/hal+json')175 effect = await get_wallet_effect(self.wallet_address, limit=2, offset='test-cursor')176 url = f'{HORIZON_URL}/accounts/{self.wallet_address}/effects?order=asc&limit=2&cursor=test-cursor'177 mock_get.assert_called_once_with(url)178 @unittest_run_loop179 @patch('stellar.wallet.ClientSession.get')180 async def test_get_wallet_effect_not_found(self, mock_get):181 session = mock_get.return_value182 session.__aenter__.return_value = self.NotFoundResponse('application/problem+json')183 with pytest.raises(web.HTTPNotFound):184 effect = await get_wallet_effect(self.wallet_address)185 url = f'{HORIZON_URL}/accounts/{self.wallet_address}/effects?order=asc'186 mock_get.assert_called_once_with(url)187 @unittest_run_loop188 @patch('stellar.wallet.ClientSession.get')189 async def test_get_wallet_effect_bad_request(self, mock_get):190 session = mock_get.return_value191 session.__aenter__.return_value = self.BadRequestResponse('application/problem+json')192 with pytest.raises(web.HTTPBadRequest):193 effect = await get_wallet_effect(self.wallet_address)194 url = f'{HORIZON_URL}/accounts/{self.wallet_address}/effects?order=asc'195 mock_get.assert_called_once_with(url)196 @unittest_run_loop197 @patch('stellar.wallet.ClientSession.get')198 async def test_get_wallet_effect_wrong_parameter(self, mock_get):199 session = mock_get.return_value200 session.__aenter__.return_value = self.BadRequestResponse('application/problem+json')201 with pytest.raises(ValueError):202 effect = await get_wallet_effect(self.wallet_address, sort='test-sort')203 mock_get.assert_not_called()204 @unittest_run_loop205 @patch('stellar.wallet.ClientSession.get')206 async def test_get_wallet_effect_upstream_fail(self, mock_get):207 session = mock_get.return_value208 session.__aenter__.return_value = self.BadRequestResponse('document/html')209 with pytest.raises(ValueError):210 effect = await get_wallet_effect(self.wallet_address, sort='test-sort')211 mock_get.assert_not_called()212 @unittest_run_loop213 @patch('stellar.wallet.ClientSession.get')214 async def test_get_get_transaction_by_wallet_success(self, mock_get):215 session = mock_get.return_value216 session.__aenter__.return_value = self.GetTransactionBywalletSuccess('application/hal+json')217 transactions = await get_transaction_by_wallet(self.wallet_address, limit=2, offset='test-cursor')218 url = f'{HORIZON_URL}/accounts/{self.wallet_address}/transactions?order=asc&limit=2&cursor=test-cursor'219 mock_get.assert_called_once_with(url)220 assert transactions == session.__aenter__.return_value.transaction_records221 @unittest_run_loop222 @patch('stellar.wallet.ClientSession.get')223 async def test_get_get_transaction_by_wallet_not_found(self, mock_get):224 session = mock_get.return_value225 session.__aenter__.return_value = self.NotFoundResponse('application/problem+json')226 with pytest.raises(web.HTTPNotFound):227 transactions = await get_transaction_by_wallet(self.wallet_address)228 @unittest_run_loop229 @patch('stellar.wallet.ClientSession.get')230 async def test_get_get_transaction_by_wallet_upstream_fail(self, mock_get):231 session = mock_get.return_value232 session.__aenter__.return_value = self.NotFoundResponse('document/html')233 with pytest.raises(web.HTTPInternalServerError):234 transactions = await get_transaction_by_wallet(self.wallet_address)235 @unittest_run_loop236 async def test_get_get_transaction_by_wallet_wrong_parameter(self):237 with pytest.raises(ValueError):238 transactions = await get_transaction_by_wallet(self.wallet_address, sort='test-sort')239 @unittest_run_loop240 @patch('stellar.wallet.ClientSession.get')241 async def test_get_operations_of_transaction_success(self, mock_get):242 session = mock_get.return_value243 session.__aenter__.return_value = self.GetOperationsOfTransactionSuccess('application/hal+json')244 transactions = await get_operations_of_transaction(self.transaction_hash, limit=2, offset='test-cursor')245 url = f'{HORIZON_URL}/transactions/{self.transaction_hash}/operations?order=asc&limit=2&cursor=test-cursor'246 mock_get.assert_called_once_with(url)247 assert transactions == session.__aenter__.return_value.operation_records248 @unittest_run_loop249 @patch('stellar.wallet.ClientSession.get')250 async def test_get_operations_of_transaction_not_found(self, mock_get):251 session = mock_get.return_value252 session.__aenter__.return_value = self.NotFoundResponse('application/problem+json')253 with pytest.raises(web.HTTPNotFound):254 transactions = await get_operations_of_transaction(self.wallet_address)255 @unittest_run_loop256 @patch('stellar.wallet.ClientSession.get')257 async def test_get_operations_of_transaction_upstream_fail(self, mock_get):258 session = mock_get.return_value259 session.__aenter__.return_value = self.NotFoundResponse('document/html')260 with pytest.raises(web.HTTPInternalServerError):261 transactions = await get_operations_of_transaction(self.wallet_address)262 @unittest_run_loop263 async def test_get_operations_of_transaction_wrong_parameter(self):264 with pytest.raises(ValueError):265 transactions = await get_operations_of_transaction(self.wallet_address, sort='test-sort')266 @unittest_run_loop267 @patch('transaction.transaction.aiohttp.ClientSession.post')268 async def test_submit_transaction_success(self, mock_post) -> None:269 session = mock_post.return_value270 session.__aenter__.return_value = self.SuccessResponse('application/hal+json')271 signed_xdr = 'Testtest'272 result = await submit_transaction(signed_xdr)273 expect = self.SuccessResponse('application/hal+json')274 assert result == await expect.json()275 @unittest_run_loop276 @patch('transaction.transaction.aiohttp.ClientSession.post')277 async def test_submit_transaction_fail_not_found(self, mock_post) -> None:278 session = mock_post.return_value279 session.__aenter__.return_value = self.BadRequestResponse('application/hal+json')280 with pytest.raises(web.HTTPBadRequest):281 signed_xdr = 'Testtest'282 result = await submit_transaction(signed_xdr)283 @unittest_run_loop284 @patch('transaction.transaction.aiohttp.ClientSession.post')285 async def test_submit_transaction_fail_bad_request(self, mock_post) -> None:286 session = mock_post.return_value287 session.__aenter__.return_value = self.NotFoundResponse('application/problem+json')288 with pytest.raises(web.HTTPNotFound):289 signed_xdr = 'Testtest'290 result = await submit_transaction(signed_xdr)291 @unittest_run_loop292 @patch('transaction.transaction.aiohttp.ClientSession.post')293 async def test_submit_transaction_upstream_fail(self, mock_post) -> None:294 session = mock_post.return_value295 session.__aenter__.return_value = self.NotFoundResponse('document/html')296 with pytest.raises(web.HTTPInternalServerError):297 signed_xdr = 'Testtest'...
context.py
Source:context.py
...11 return inner12# class AioExecutionContext(_aio_callable_context_manager, ExecutionContext):13# def __enter__(self):14# raise NotImplementedError()15# async def __aenter__(self):16# async with self.database._conn_lock:17# self.database.push_execution_context(self)18# self.connection = await self.database._connect(19# self.database.database,20# **self.database.connect_kwargs)21# if self.with_transaction:22# self.txn = self.database.transaction()23# await self.txn.__aenter__()24# return self25# def __exit__(self, exc_type, exc_val, exc_tb):26# raise NotImplementedError()27# async def __aexit__(self, exc_type, exc_val, exc_tb):28# async with self.database._conn_lock:29# if self.connection is None:30# self.database.pop_execution_context()31# else:32# try:33# if self.with_transaction:34# if not exc_type:35# self.txn.commit(False)36# await self.txn.__aexit__(exc_type, exc_val, exc_tb)37# finally:38# self.database.pop_execution_context()39# await self.database._close(self.connection)40# class AioUsing(AioExecutionContext, Using):41# def __enter__(self):42# raise NotImplementedError()43# async def __aenter__(self):44# self._orig = []45# for model in self.models:46# self._orig.append(model._meta.database)47# model._meta.database = self.database48# return super(Using, self).__aenter__()49# def __exit__(self, exc_type, exc_val, exc_tb):50# raise NotImplementedError()51# async def __aexit__(self, exc_type, exc_val, exc_tb):52# await super(Using, self).__aexit__(exc_type, exc_val, exc_tb)53# for i, model in enumerate(self.models):54# model._meta.database = self._orig[i]55class _aio_atomic(_aio_callable_context_manager):56 __slots__ = ('conn', 'transaction_type', 'context_manager')57 def __init__(self, conn, transaction_type=None):58 self.conn = conn59 self.transaction_type = transaction_type60 async def __aenter__(self):61 await self.conn.__aenter__()62 if self.conn.transaction_depth() == 0:63 self.context_manager = self.conn.transaction(self.transaction_type)64 else:65 self.context_manager = self.conn.savepoint()66 return await self.context_manager.__aenter__()67 async def __aexit__(self, exc_type, exc_val, exc_tb):68 await self.context_manager.__aexit__(exc_type, exc_val, exc_tb)69 await self.conn.__aexit__(exc_type, exc_val, exc_tb)70class aio_transaction(_aio_callable_context_manager):71 __slots__ = ('conn', 'autocommit', 'transaction_type')72 def __init__(self, conn, transaction_type=None):73 self.conn = conn74 self.transaction_type = transaction_type75 async def _begin(self):76 if self.transaction_type:77 await self.conn.begin(self.transaction_type)78 else:79 await self.conn.begin()80 async def commit(self, begin=True):81 await self.conn.commit()82 if begin:83 await self._begin()84 async def rollback(self, begin=True):85 await self.conn.rollback()86 if begin:87 await self._begin()88 async def __aenter__(self):89 self.autocommit = self.conn.autocommit90 self.conn.autocommit = False91 if self.conn.transaction_depth() == 0:92 await self._begin()93 self.conn.push_transaction(self)94 return self95 async def __aexit__(self, exc_type, exc_val, exc_tb):96 try:97 if exc_type:98 await self.rollback(False)99 elif self.conn.transaction_depth() == 1:100 try:101 await self.commit(False)102 except:103 await self.rollback(False)104 raise105 finally:106 self.conn.autocommit = self.autocommit107 self.conn.pop_transaction()108class aio_savepoint(_aio_callable_context_manager):109 __slots__ = ('conn', 'sid', 'quoted_sid', 'autocommit')110 def __init__(self, conn, sid=None):111 self.conn = conn112 self.sid = sid or uuid.uuid4().hex113 _compiler = conn.compiler() # TODO: breing the compiler here somehow114 self.quoted_sid = _compiler.quote(self.sid)115 async def _execute(self, query):116 await self.conn.execute_sql(query, require_commit=False)117 async def _begin(self):118 await self._execute('SAVEPOINT %s;' % self.quoted_sid)119 async def commit(self, begin=True):120 await self._execute('RELEASE SAVEPOINT %s;' % self.quoted_sid)121 if begin:122 await self._begin()123 async def rollback(self):124 await self._execute('ROLLBACK TO SAVEPOINT %s;' % self.quoted_sid)125 def __enter__(self):126 raise NotImplementedError()127 async def __aenter__(self):128 self.autocommit = self.conn.get_autocommit()129 self.conn.set_autocommit(False)130 await self._begin()131 return self132 def __exit__(self, exc_type, exc_val, exc_tb):133 raise NotImplementedError()134 async def __aexit__(self, exc_type, exc_val, exc_tb):135 try:136 if exc_type:137 await self.rollback()138 else:139 try:140 await self.commit(begin=False)141 except:...
test_redis.py
Source:test_redis.py
...59 res = await get_redis_key(redis=redis, key='test key')60 redis.client.return_value.__aenter__.return_value.get.assert_called_once_with('test key')61 assert res62class AMagicMock(MagicMock):63 async def __aenter__(self):64 val = mock.MagicMock()65 f = Future()66 f.set_result(True)67 val.set = MagicMock(return_value=f)68 val.get = MagicMock(return_value=f)69 return val70 async def __aexit__(self, exc_type, exc_val, exc_tb):71 pass72@pytest.mark.skipif(sys.version_info > (3, 8) or sys.version_info <= (3, 7), reason="requires python3.7")73async def test_set_redis_key_py_3_7():74 redis = mock.MagicMock()75 redis.client.return_value = AMagicMock()76 res = await set_redis_key(redis=redis, key='test key', value='test value')77 assert res...
test_github_releases_data.py
Source:test_github_releases_data.py
1from unittest.mock import AsyncMock, MagicMock, call2import aiohttp3from randovania.interface_common import github_releases_data4async def test_download_from_github_success(mocker):5 # Setup6 page_size = 57 response_a = AsyncMock()8 response_b = AsyncMock()9 responses = [response_a, response_b]10 mock_get_response: MagicMock = mocker.patch.object(aiohttp.ClientSession, 'get')11 mock_get_response.side_effect = responses12 for r in responses:13 r.__aenter__.return_value.raise_for_status = MagicMock()14 response_a.__aenter__.return_value.json = AsyncMock(return_value=[1, 2, 3, 4, 5])15 response_b.__aenter__.return_value.json = AsyncMock(return_value=[6, 7])16 # Run17 returned_value = await github_releases_data._download_from_github(page_size=page_size)18 # Assert19 mock_get_response.assert_has_calls([20 call(21 github_releases_data._RELEASES_URL,22 params={"page": 1, "per_page": page_size},23 ),24 call(25 github_releases_data._RELEASES_URL,26 params={"page": 2, "per_page": page_size},27 ),28 ])29 for r in responses:30 r.__aenter__.return_value.raise_for_status.assert_called_once_with()31 r.__aenter__.return_value.json.assert_awaited()32 assert returned_value == [1, 2, 3, 4, 5, 6, 7]33async def test_download_from_github_bad_response_is_caught(mocker):34 # Setup35 mock_get_response: MagicMock = mocker.patch.object(aiohttp.ClientSession, 'get')36 mock_get_response.return_value.__aenter__.return_value.raise_for_status = MagicMock(37 side_effect=aiohttp.ClientResponseError(None, None))38 # Run39 returned_value = await github_releases_data._download_from_github()40 # Assert41 mock_get_response.assert_called()42 mock_get_response.return_value.__aenter__.return_value.raise_for_status.assert_called()43 assert returned_value is None44async def test_download_from_github_connection_failure_is_caught(mocker):45 # Setup46 mock_get_response: MagicMock = mocker.patch.object(aiohttp.ClientSession, 'get',47 side_effect=aiohttp.ClientConnectionError())48 # Run49 returned_value = await github_releases_data._download_from_github()50 # Assert51 mock_get_response.assert_called()...
LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Get 100 minutes of automation test minutes FREE!!