PK!U8''aioetherscan/__init__.pyfrom aioetherscan.client import Client PK! [0aioetherscan/client.pyimport asyncio from aioetherscan.modules.account import Account from aioetherscan.modules.block import Block from aioetherscan.modules.contract import Contract from aioetherscan.modules.logs import Logs from aioetherscan.modules.proxy import Proxy from aioetherscan.modules.stats import Stats from aioetherscan.modules.transaction import Transaction from aioetherscan.modules.utils import Utils from aioetherscan.network import Network class Client: def __init__(self, api_key: str, network: str = 'main', loop: asyncio.AbstractEventLoop = None) -> None: self._http = Network(api_key, network, loop) self.account = Account(self) self.block = Block(self) self.contract = Contract(self) self.transaction = Transaction(self) self.stats = Stats(self) self.logs = Logs(self) self.proxy = Proxy(self) self.utils = Utils(self) async def close(self): await self._http.close() PK!@@aioetherscan/common.pyfrom typing import Union def check_hex(number: Union[str, int]) -> str: if isinstance(number, int): return hex(number) try: int(number, 16) except ValueError as e: raise ValueError(f'Invalid hex parameter {number!r}: {e}') else: return number def check_tag(tag: Union[str, int]) -> str: _TAGS = ( 'earliest', # the earliest/genesis block 'latest', # the latest mined block 'pending' # for the pending state/transactions ) if tag in _TAGS: return tag return check_hex(tag) PK!YkFFaioetherscan/exceptions.pyclass EtherscanClientError(Exception): pass class EtherscanClientContentTypeError(EtherscanClientError): def __init__(self, status, content): self.status = status self.content = content def __str__(self): return f'[{self.status}] {self.content!r}' class EtherscanClientApiError(EtherscanClientError): def __init__(self, message, result): self.message = message self.result = result def __str__(self): return f'[{self.message}] {self.result}' class EtherscanClientProxyError(EtherscanClientError): """JSON-RPC 2.0 Specification https://www.jsonrpc.org/specification#error_object """ def __init__(self, code, message): self.code = code self.message = message def __str__(self): return f'[{self.code}] {self.message}' PK!4lLLaioetherscan/modules/account.pyfrom typing import Iterable, Optional, List, Dict, Tuple from aioetherscan.common import check_tag from aioetherscan.modules.base import BaseModule class Account(BaseModule): """Account & token APIs https://etherscan.io/apis#accounts https://etherscan.io/apis#tokens """ _SORT_ORDERS = ( 'asc', # ascending order 'desc' # descending order ) _BLOCK_TYPES = ( 'blocks', # full blocks only 'uncles' # uncle blocks only ) @property def _module(self) -> str: return 'account' async def balance(self, address: str, tag: str = 'latest') -> str: """Get Ether Balance for a single Address.""" return await self._get( action='balance', address=address, tag=check_tag(tag) ) async def balances(self, addresses: Iterable[str], tag: str = 'latest') -> List[Dict]: """Get Ether Balance for multiple Addresses in a single call.""" return await self._get( action='balancemulti', address=','.join(addresses), tag=check_tag(tag) ) async def normal_txs( self, address: str, start_block: Optional[int] = None, end_block: Optional[int] = None, sort: Optional[str] = None, page: Optional[int] = None, offset: Optional[int] = None ) -> List[Dict]: """Get a list of 'Normal' Transactions By Address.""" return await self._get( action='txlist', address=address, startblock=start_block, endblock=end_block, sort=self._check_sort_direction(sort), page=page, offset=offset ) async def internal_txs( self, address: str, start_block: Optional[int] = None, end_block: Optional[int] = None, sort: Optional[str] = None, page: Optional[int] = None, offset: Optional[int] = None, txhash: Optional[str] = None ) -> List[Dict]: """Get a list of 'Internal' Transactions by Address or Transaction Hash.""" return await self._get( action='txlistinternal', address=address, startblock=start_block, endblock=end_block, sort=self._check_sort_direction(sort), page=page, offset=offset, txhash=txhash ) async def token_transfers( self, address: Optional[str] = None, contract_address: Optional[str] = None, start_block: Optional[int] = None, end_block: Optional[int] = None, sort: Optional[str] = None, page: Optional[int] = None, offset: Optional[int] = None, ) -> List[Dict]: """Get a list of "ERC20 - Token Transfer Events" by Address.""" if not address and not contract_address: raise ValueError('At least one of address or contract_address must be specified.') return await self._get( action='tokentx', address=address, startblock=start_block, endblock=end_block, sort=self._check_sort_direction(sort), page=page, offset=offset, contractaddress=contract_address ) async def mined_blocks( self, address: str, blocktype: str = 'blocks', page: Optional[int] = None, offset: Optional[int] = None ) -> List: """Get list of Blocks Mined by Address.""" return await self._get( action='getminedblocks', address=address, blocktype=self._check_blocktype(blocktype), page=page, offset=offset ) async def token_balance(self, address: str, contract_address: str, tag: str = 'latest') -> str: """Get ERC20-Token Account Balance for TokenContractAddress.""" return await self._get( action='tokenbalance', address=address, contractaddress=contract_address, tag=check_tag(tag) ) def _check(self, value: str, values: Tuple[str, ...]): if value and value.lower() not in values: raise ValueError(f'Invalid value {value!r}, only {values} are supported.') return value def _check_sort_direction(self, sort: str) -> str: return self._check(sort, self._SORT_ORDERS) def _check_blocktype(self, blocktype: str) -> str: return self._check(blocktype, self._BLOCK_TYPES) PK!)aioetherscan/modules/base.pyfrom abc import ABC, abstractmethod class BaseModule(ABC): def __init__(self, client): self._client = client @property @abstractmethod def _module(self) -> str: '''Returns API module name.''' async def _get(self, **params): return await self._client._http.get(params={**dict(module=self._module), **params}) async def _post(self, **params): return await self._client._http.post(data={**dict(module=self._module), **params}) PK!Daioetherscan/modules/block.pyfrom typing import Dict from aioetherscan.modules.base import BaseModule class Block(BaseModule): """Block APIs https://etherscan.io/apis#blocks """ @property def _module(self) -> str: return 'block' async def block_reward(self, blockno: int) -> Dict: """[BETA] Get Block And Uncle Rewards by BlockNo.""" return await self._get( action='getblockreward', blockno=blockno ) PK!L  aioetherscan/modules/contract.pyfrom typing import Dict, List from aioetherscan.modules.base import BaseModule class Contract(BaseModule): """Contract APIs https://etherscan.io/apis#contracts """ @property def _module(self) -> str: return 'contract' async def contract_abi(self, address: str) -> str: """Get Contract ABI for Verified Contract Source Codes https://etherscan.io/contractsVerified. """ return await self._get( action='getabi', address=address ) async def contract_source_code(self, address: str) -> List[Dict]: """Get Contract Source Code for Verified Contract Source Codes https://etherscan.io/contractsVerified. """ return await self._get( action='getsourcecode', address=address ) async def verify_contract_source_code( self, contract_address: str, source_code: str, contract_name: str, compiler_version: str, optimization_used: bool = False, runs: int = 200, constructor_arguements: str = None, libraries: Dict[str, str] = None ) -> str: """[BETA] Verify Source Code""" return await self._post( module='contract', action='verifysourcecode', contractaddress=contract_address, sourceCode=source_code, contractname=contract_name, compilerversion=compiler_version, optimizationUsed=1 if optimization_used else 0, runs=runs, constructorArguements=constructor_arguements, **self._parse_libraries(libraries or {}) ) async def check_verification_status(self, guid: str) -> str: """Check Source code verification submission status""" return await self._get( action='checkverifystatus', guid=guid ) @staticmethod def _parse_libraries(libraries: Dict[str, str]) -> Dict[str, str]: return dict( part for i, (name, address) in enumerate(libraries.items(), start=1) for part in ( (f'libraryname{i}', name), (f'libraryaddress{i}', address) ) ) PK!%h aioetherscan/modules/logs.pyfrom typing import Union, Optional, List, Dict from aioetherscan.modules.base import BaseModule class Logs(BaseModule): """Event logs https://etherscan.io/apis#logs """ _TOPIC_OPERATORS = ('and', 'or') _BLOCKS = ('latest',) @property def _module(self) -> str: return 'logs' async def get_logs( self, from_block: Union[int, str], to_block: Union[int, str], address: str, topics: List[str], topic_operators: Optional[List[str]] = None ) -> List[Dict]: """[Beta] The Event Log API was designed to provide an alternative to the native eth_getLogs https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getlogs. """ return await self._get( action='getLogs', fromBlock=self._check_block(from_block), toBlock=self._check_block(to_block), address=address, **self._fill_topics(topics, topic_operators) ) def _check_block(self, block: Union[str, int]) -> Union[str, int]: if isinstance(block, int): return block if block in self._BLOCKS: return block raise ValueError(f'Invalid value {block!r}, only integers or {self._BLOCKS} are supported.') def _fill_topics(self, topics: List[str], topic_operators: List[str]): if len(topics) > 1: self._check_topics(topics, topic_operators) topic_params = { f'topic{idx}': value for idx, value in enumerate(topics) } topic_operator_params = { f'topic{idx}_{idx + 1}_opr': value for idx, value in enumerate(topic_operators) } return {**topic_params, **topic_operator_params} else: return {'topic0': topics[0]} def _check_topics(self, topics: List[str], topic_operators: List[str]) -> None: if not topic_operators: raise ValueError('Topic operators are required when more than 1 topic passed.') for op in topic_operators: if op not in self._TOPIC_OPERATORS: raise ValueError(f'Invalid topic operator {op!r}, must be one of: {self._TOPIC_OPERATORS}') if len(topics) - len(topic_operators) != 1: raise ValueError(f'Invalid length of topic_operators list, must be len(topics) - 1.') PK!{w1aioetherscan/modules/proxy.pyfrom typing import Union, Dict from aioetherscan.common import check_hex, check_tag from aioetherscan.modules.base import BaseModule class Proxy(BaseModule): """Geth/Parity Proxy APIs https://etherscan.io/apis#proxy https://github.com/ethereum/wiki/wiki/JSON-RPC """ @property def _module(self) -> str: return 'proxy' async def block_number(self) -> str: """Returns the number of most recent block.""" return await self._get(action='eth_blockNumber') async def block_by_number(self, full: bool, tag: Union[int, str] = 'latest') -> Dict: """Returns information about a block by block number.""" return await self._get( action='eth_getBlockByNumber', boolean=full, tag=check_tag(tag), ) async def uncle_block_by_number_and_index(self, index: Union[int, str], tag: Union[int, str] = 'latest') -> Dict: """Returns information about a uncle by block number.""" return await self._get( action='eth_getUncleByBlockNumberAndIndex', index=check_hex(index), tag=check_tag(tag), ) async def block_tx_count_by_number(self, tag: Union[int, str] = 'latest') -> str: """Returns the number of transactions in a block from a block matching the given block number.""" return await self._get( action='eth_getBlockTransactionCountByNumber', tag=check_tag(tag), ) async def tx_by_hash(self, txhash: Union[int, str]) -> Dict: """Returns the information about a transaction requested by transaction hash.""" return await self._get( action='eth_getTransactionByHash', txhash=check_hex(txhash), ) async def tx_by_number_and_index(self, index: Union[int, str], tag: Union[int, str] = 'latest') -> Dict: """Returns information about a transaction by block number and transaction index position.""" return await self._get( action='eth_getTransactionByBlockNumberAndIndex', index=check_hex(index), tag=check_tag(tag), ) async def tx_count(self, address: str, tag: Union[int, str] = 'latest') -> str: """Returns the number of transactions sent from an address.""" return await self._get( action='eth_getTransactionCount', address=address, tag=check_tag(tag), ) async def send_raw_tx(self, raw_hex: str) -> Dict: """Creates new message call transaction or a contract creation for signed transactions.""" return await self._post( module='proxy', action='eth_sendRawTransaction', hex=raw_hex ) async def tx_receipt(self, txhash: str) -> Dict: """Returns the receipt of a transaction by transaction hash.""" return await self._get( action='eth_getTransactionReceipt', txhash=check_hex(txhash), ) async def call(self, to: str, data: str, tag: Union[int, str] = 'latest') -> str: """Executes a new message call immediately without creating a transaction on the block chain.""" return await self._get( action='eth_call', to=check_hex(to), data=check_hex(data), tag=check_tag(tag), ) async def code(self, address: str, tag: Union[int, str] = 'latest') -> str: """Returns code at a given address.""" return await self._get( action='eth_getCode', address=address, tag=check_tag(tag), ) async def storage_at(self, address: str, position: str, tag: Union[int, str] = 'latest') -> str: """Returns the value from a storage position at a given address.""" return await self._get( action='eth_getStorageAt', address=address, position=position, tag=check_tag(tag), ) async def gas_price(self) -> str: """Returns the current price per gas in wei.""" return await self._get(action='eth_gasPrice', ) async def estimate_gas(self, to: str, value: str, gas_price: str, gas: str) -> str: """Makes a call or transaction, which won't be added to the blockchain and returns the used gas. Can be used for estimating the used gas. """ return await self._get( action='eth_estimateGas', to=check_hex(to), value=value, gasPrice=gas_price, gas=gas, ) PK!խaioetherscan/modules/stats.pyfrom typing import Dict from aioetherscan.modules.base import BaseModule class Stats(BaseModule): """General stats https://etherscan.io/apis#stats """ @property def _module(self) -> str: return 'stats' async def eth_supply(self) -> str: """Get Total Supply of Ether""" return await self._get(action='ethsupply') async def eth_price(self) -> Dict: """Get ETHER LastPrice Price""" return await self._get(action='ethprice') PK!8**#aioetherscan/modules/transaction.pyfrom typing import Dict from aioetherscan.modules.base import BaseModule class Transaction(BaseModule): """Transaction APIs https://etherscan.io/apis#transactions """ @property def _module(self) -> str: return 'transaction' async def contract_execution_status(self, txhash: str) -> Dict: """[BETA] Check Contract Execution Status (if there was an error during contract execution) """ return await self._get( action='getstatus', txhash=txhash ) async def tx_receipt_status(self, txhash: str) -> Dict: """[BETA] Check Transaction Receipt Status (Only applicable for Post Byzantium fork transactions) """ return await self._get( action='gettxreceiptstatus', txhash=txhash ) PK!. aioetherscan/modules/utils.pyimport asyncio import itertools from typing import Tuple, Dict, List, Iterator, AsyncIterator from aioetherscan.exceptions import EtherscanClientApiError class Utils: """Helper methods which use the combination of documented APIs.""" def __init__(self, client): self._client = client async def token_transfers_generator( self, contract_address: str, be_polite: bool = True, block_limit: int = 50, offset: int = 3, start_block: int = 0, end_block: int = None, ) -> AsyncIterator[Dict]: if not end_block: end_block = int(await self._client.proxy.block_number(), 16) if be_polite: for sblock, eblock in self._generate_intervals(start_block, end_block, block_limit): result = await self._parse_by_pages(contract_address, sblock, eblock, offset) for t in result: yield t else: tasks = [ self._parse_by_pages(contract_address, sblock, eblock, offset) for sblock, eblock in self._generate_intervals(start_block, end_block, block_limit) ] result = await asyncio.gather(*tasks) for t in itertools.chain.from_iterable(result): yield t async def token_transfers( self, contract_address: str, be_polite: bool = True, block_limit: int = 50, offset: int = 3, start_block: int = 0, end_block: int = None, ) -> List[Dict]: kwargs = {k: v for k, v in locals().items() if k != 'self' and not k.startswith('_')} return [t async for t in self.token_transfers_generator(**kwargs)] async def _parse_by_pages(self, contract_address: str, start_block: int, end_block: int, offset: int) -> List[Dict]: page, result = 1, [] while True: try: transfers = await self._client.account.token_transfers( contract_address=contract_address, start_block=start_block, end_block=end_block, page=page, offset=offset ) except EtherscanClientApiError as e: if e.message == 'No transactions found': break raise else: result.extend(transfers) page += 1 return result @staticmethod def _generate_intervals(from_number: int, to_number: int, count: int) -> Iterator[Tuple[int, int]]: for i in range(from_number, to_number + 1, count): yield (i, min(i + count - 1, to_number)) PK!oxl{{aioetherscan/network.pyimport asyncio import logging from enum import Enum from typing import Union, Dict, List import aiohttp from asyncio_throttle import Throttler from aioetherscan.exceptions import EtherscanClientContentTypeError, EtherscanClientError, EtherscanClientApiError, \ EtherscanClientProxyError class HttpMethod(Enum): GET = 'get' POST = 'post' class Network: _NETWORKS = { 'main': 'https://api.etherscan.io/api', 'ropsten': 'https://api-ropsten.etherscan.io/api', # ROPSTEN (Revival) TESTNET 'kovan': 'https://api-kovan.etherscan.io/api', # KOVAN (POA) TESTNET 'rinkeby': 'https://api-rinkeby.etherscan.io/api', # RINKEBY (CLIQUE) TESTNET 'tobalaba': 'https://api-tobalaba.etherscan.com/api' # TOBALABA NETWORK } def __init__(self, api_key: str, network: str, loop: asyncio.AbstractEventLoop) -> None: self._API_KEY = api_key self._set_network(network) self._loop = loop or asyncio.get_event_loop() self._session = aiohttp.ClientSession(loop=self._loop) self._throttler = Throttler(rate_limit=5) self._logger = logging.getLogger(__name__) async def close(self): await self._session.close() async def get(self, params: Dict = None) -> Union[Dict, List, str]: return await self._request(HttpMethod.GET, params=self._filter_and_sign(params)) async def post(self, data: Dict = None) -> Union[Dict, List, str]: return await self._request(HttpMethod.POST, data=self._filter_and_sign(data)) async def _request(self, method: HttpMethod, data: Dict = None, params: Dict = None) -> Union[Dict, List, str]: session_method = getattr(self._session, method.value) async with self._throttler: async with session_method(self._API_URL, params=params, data=data) as response: self._logger.debug('[%s] %r %r %s', method.name, str(response.url), data, response.status) return await self._handle_response(response) async def _handle_response(self, response: aiohttp.ClientResponse) -> Union[Dict, list, str]: try: response_json = await response.json() except aiohttp.ContentTypeError: raise EtherscanClientContentTypeError(response.status, await response.text()) except Exception as e: raise EtherscanClientError(e) else: self._logger.debug('Response: %r', response_json) self._raise_if_error(response_json) return response_json['result'] @staticmethod def _raise_if_error(response_json: Dict): if 'status' in response_json and response_json['status'] != '1': message, result = response_json.get('message'), response_json.get('result') raise EtherscanClientApiError(message, result) if 'error' in response_json: err = response_json['error'] code, message = err.get('code'), err.get('message') raise EtherscanClientProxyError(code, message) def _filter_and_sign(self, params: Dict): return self._sign( self._filter_params(params or {}) ) def _sign(self, params: Dict) -> Dict: if not params: params = {} params['apikey'] = self._API_KEY return params @staticmethod def _filter_params(params: Dict) -> Dict: return {k: v for k, v in params.items() if v is not None} def _set_network(self, network: str) -> None: if network not in self._NETWORKS: raise ValueError(f'Incorrect network {network!r}, supported only: {", ".join(self._NETWORKS.keys())}') self._API_URL = self._NETWORKS[network] PK!aioetherscan/tests/__init__.pyPK!·!!"aioetherscan/tests/test_account.pyfrom unittest.mock import patch import pytest from asynctest import CoroutineMock from aioetherscan import Client @pytest.fixture() async def account(): c = Client('TestApiKey') yield c.account await c.close() @pytest.mark.asyncio async def test_balance(account): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.balance('addr') mock.assert_called_once_with(params=dict(module='account', action='balance', address='addr', tag='latest')) with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.balance('addr', 123) mock.assert_called_once_with(params=dict(module='account', action='balance', address='addr', tag='0x7b')) @pytest.mark.asyncio async def test_balances(account): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.balances(['a1', 'a2']) mock.assert_called_once_with( params=dict(module='account', action='balancemulti', address='a1,a2', tag='latest')) with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.balances(['a1', 'a2'], 123) mock.assert_called_once_with(params=dict(module='account', action='balancemulti', address='a1,a2', tag='0x7b')) @pytest.mark.asyncio async def test_normal_txs(account): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.normal_txs('addr') mock.assert_called_once_with( params=dict( module='account', action='txlist', address='addr', startblock=None, endblock=None, sort=None, page=None, offset=None ) ) with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.normal_txs( address='addr', start_block=1, end_block=2, sort='asc', page=3, offset=4 ) mock.assert_called_once_with( params=dict( module='account', action='txlist', address='addr', startblock=1, endblock=2, sort='asc', page=3, offset=4 ) ) with pytest.raises(ValueError): await account.normal_txs( address='addr', sort='wrong', ) @pytest.mark.asyncio async def test_internal_txs(account): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.internal_txs('addr') mock.assert_called_once_with( params=dict( module='account', action='txlistinternal', address='addr', startblock=None, endblock=None, sort=None, page=None, offset=None, txhash=None ) ) with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.internal_txs( address='addr', start_block=1, end_block=2, sort='asc', page=3, offset=4, txhash='0x123' ) mock.assert_called_once_with( params=dict( module='account', action='txlistinternal', address='addr', startblock=1, endblock=2, sort='asc', page=3, offset=4, txhash='0x123' ) ) with pytest.raises(ValueError): await account.internal_txs( address='addr', sort='wrong', ) @pytest.mark.asyncio async def test_token_transfers(account): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.token_transfers('addr') mock.assert_called_once_with( params=dict( module='account', action='tokentx', address='addr', startblock=None, endblock=None, sort=None, page=None, offset=None, contractaddress=None ) ) with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.token_transfers(contract_address='addr') mock.assert_called_once_with( params=dict( module='account', action='tokentx', address=None, startblock=None, endblock=None, sort=None, page=None, offset=None, contractaddress='addr' ) ) with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.token_transfers( address='addr', start_block=1, end_block=2, sort='asc', page=3, offset=4, contract_address='0x123' ) mock.assert_called_once_with( params=dict( module='account', action='tokentx', address='addr', startblock=1, endblock=2, sort='asc', page=3, offset=4, contractaddress='0x123' ) ) with pytest.raises(ValueError): await account.token_transfers( address='addr', sort='wrong', ) with pytest.raises(ValueError): await account.token_transfers() @pytest.mark.asyncio async def test_mined_blocks(account): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.mined_blocks('addr') mock.assert_called_once_with( params=dict( module='account', action='getminedblocks', address='addr', blocktype='blocks', page=None, offset=None, ) ) with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.mined_blocks( address='addr', blocktype='uncles', page=1, offset=2 ) mock.assert_called_once_with( params=dict( module='account', action='getminedblocks', address='addr', blocktype='uncles', page=1, offset=2 ) ) with pytest.raises(ValueError): await account.mined_blocks( address='addr', blocktype='wrong', ) @pytest.mark.asyncio async def test_token_balance(account): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.token_balance('addr', 'contractaddr') mock.assert_called_once_with( params=dict( module='account', action='tokenbalance', address='addr', contractaddress='contractaddr', tag='latest', ) ) with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await account.token_balance('addr', 'contractaddr', 123) mock.assert_called_once_with( params=dict( module='account', action='tokenbalance', address='addr', contractaddress='contractaddr', tag='0x7b', ) ) def test_check(account): assert account._check('a', ('a', 'b')) == 'a' assert account._check('A', ('a', 'b')) == 'A' with pytest.raises(ValueError): account._check('c', ('a', 'b')) with pytest.raises(ValueError): account._check('C', ('a', 'b')) def test_check_sort_direction(account): assert account._check_sort_direction('asc') == 'asc' assert account._check_sort_direction('desc') == 'desc' with pytest.raises(ValueError): account._check_sort_direction('wrong') def test_check_blocktype(account): assert account._check_blocktype('blocks') == 'blocks' assert account._check_blocktype('uncles') == 'uncles' with pytest.raises(ValueError): account._check_blocktype('wrong') PK!oyl aioetherscan/tests/test_block.pyfrom unittest.mock import patch import pytest from asynctest import CoroutineMock from aioetherscan import Client @pytest.fixture() async def block(): c = Client('TestApiKey') yield c.block await c.close() @pytest.mark.asyncio async def test_block_reward(block): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await block.block_reward(123) mock.assert_called_once_with(params=dict(module='block', action='getblockreward', blockno=123)) PK!£22!aioetherscan/tests/test_client.pyfrom unittest.mock import patch import pytest from asynctest import CoroutineMock from aioetherscan import Client from aioetherscan.modules.account import Account from aioetherscan.modules.block import Block from aioetherscan.modules.contract import Contract from aioetherscan.modules.logs import Logs from aioetherscan.modules.proxy import Proxy from aioetherscan.modules.stats import Stats from aioetherscan.modules.transaction import Transaction from aioetherscan.modules.utils import Utils @pytest.fixture() async def client(): c = Client('TestApiKey') yield c await c.close() def test_api_key(): with pytest.raises(TypeError): c = Client() def test_init(client): assert isinstance(client.account, Account) assert isinstance(client.block, Block) assert isinstance(client.contract, Contract) assert isinstance(client.transaction, Transaction) assert isinstance(client.stats, Stats) assert isinstance(client.logs, Logs) assert isinstance(client.proxy, Proxy) assert isinstance(client.utils, Utils) assert isinstance(client.account._client, Client) assert isinstance(client.block._client, Client) assert isinstance(client.contract._client, Client) assert isinstance(client.transaction._client, Client) assert isinstance(client.stats._client, Client) assert isinstance(client.logs._client, Client) assert isinstance(client.proxy._client, Client) assert isinstance(client.utils._client, Client) @pytest.mark.asyncio async def test_close_session(client): with patch('aioetherscan.network.Network.close', new_callable=CoroutineMock) as m: await client.close() m.assert_called_once_with() @pytest.mark.asyncio async def test_networks(): with patch('aioetherscan.network.Network._set_network') as m: c = Client('TestApiKey') m.assert_called_once_with('main') await c.close() with patch('aioetherscan.network.Network._set_network') as m: c = Client('TestApiKey', 'kovan') m.assert_called_once_with('kovan') await c.close() #PK!Fv2XX!aioetherscan/tests/test_common.pyfrom unittest.mock import Mock, patch import pytest from aioetherscan.common import check_hex, check_tag def test_check_hex(): assert check_hex(123) == '0x7b' assert check_hex(0x7b) == '0x7b' assert check_hex('0x7b') == '0x7b' with pytest.raises(ValueError): check_hex('wrong') def test_check_tag(): assert check_tag('latest') == 'latest' assert check_tag('earliest') == 'earliest' assert check_tag('pending') == 'pending' with patch('aioetherscan.common.check_hex', new=Mock()) as mock: check_tag(123) mock.assert_called_once_with(123) PK! =U44#aioetherscan/tests/test_contract.pyfrom unittest.mock import patch import pytest from asynctest import CoroutineMock from aioetherscan import Client @pytest.fixture() async def contract(): c = Client('TestApiKey') yield c.contract await c.close() @pytest.mark.asyncio async def test_contract_abi(contract): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await contract.contract_abi('0x012345') mock.assert_called_once_with(params=dict(module='contract', action='getabi', address='0x012345')) @pytest.mark.asyncio async def test_contract_source_code(contract): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await contract.contract_source_code('0x012345') mock.assert_called_once_with(params=dict(module='contract', action='getsourcecode', address='0x012345')) @pytest.mark.asyncio async def test_verify_contract_source_code(contract): with patch('aioetherscan.network.Network.post', new=CoroutineMock()) as mock: await contract.verify_contract_source_code( contract_address='0x012345', source_code='some source code\ntest', contract_name='some contract name', compiler_version='1.0.0', optimization_used=False, runs=123, constructor_arguements='some args' ) mock.assert_called_once_with( data=dict( module='contract', action='verifysourcecode', contractaddress='0x012345', sourceCode='some source code\ntest', contractname='some contract name', compilerversion='1.0.0', optimizationUsed=0, runs=123, constructorArguements='some args' ) ) with patch('aioetherscan.network.Network.post', new=CoroutineMock()) as mock: await contract.verify_contract_source_code( contract_address='0x012345', source_code='some source code\ntest', contract_name='some contract name', compiler_version='1.0.0', optimization_used=False, runs=123, constructor_arguements='some args', libraries={'one_name': 'one_addr', 'two_name': 'two_addr'} ) mock.assert_called_once_with( data=dict( module='contract', action='verifysourcecode', contractaddress='0x012345', sourceCode='some source code\ntest', contractname='some contract name', compilerversion='1.0.0', optimizationUsed=0, runs=123, constructorArguements='some args', libraryname1='one_name', libraryaddress1='one_addr', libraryname2='two_name', libraryaddress2='two_addr', ) ) @pytest.mark.asyncio async def test_check_verification_status(contract): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await contract.check_verification_status('some_guid') mock.assert_called_once_with(params=dict(module='contract', action='checkverifystatus', guid='some_guid')) def test_parse_libraries(contract): mydict = { 'lib1': 'addr1', 'lib2': 'addr2', } expected = { 'libraryname1': 'lib1', 'libraryaddress1': 'addr1', 'libraryname2': 'lib2', 'libraryaddress2': 'addr2' } assert contract._parse_libraries(mydict) == expected mydict = { 'lib1': 'addr1', } expected = { 'libraryname1': 'lib1', 'libraryaddress1': 'addr1', } assert contract._parse_libraries(mydict) == expected mydict = {} expected = {} assert contract._parse_libraries(mydict) == expected PK!i55%aioetherscan/tests/test_exceptions.pyfrom aioetherscan.exceptions import EtherscanClientContentTypeError, EtherscanClientApiError, EtherscanClientProxyError def test_content_type_error(): e = EtherscanClientContentTypeError(1, 2) assert e.status == 1 assert e.content == 2 assert str(e) == '[1] 2' def test_api_error(): e = EtherscanClientApiError(1, 2) assert e.message == 1 assert e.result == 2 assert str(e) == '[1] 2' def test_proxy_error(): e = EtherscanClientProxyError(1, 2) assert e.code == 1 assert e.message == 2 assert str(e) == '[1] 2' PK! xO22aioetherscan/tests/test_logs.pyfrom unittest.mock import patch, Mock, call import pytest from asynctest import CoroutineMock from aioetherscan import Client @pytest.fixture() async def logs(): c = Client('TestApiKey') yield c.logs await c.close() @pytest.mark.asyncio async def test_balance(logs): with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await logs.get_logs( from_block=1, to_block=2, address='addr', topics=['topic', ] ) mock.assert_called_once_with( params=dict( module='logs', action='getLogs', fromBlock=1, toBlock=2, address='addr', topic0='topic' ) ) with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: await logs.get_logs( from_block='latest', to_block='latest', address='addr', topics=['topic', ] ) mock.assert_called_once_with( params=dict( module='logs', action='getLogs', fromBlock='latest', toBlock='latest', address='addr', topic0='topic' ) ) with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: with patch('aioetherscan.modules.logs.Logs._check_block', new=Mock()) as block_mock: await logs.get_logs( from_block=1, to_block=2, address='addr', topics=['top1', 'top2'], topic_operators=['and'] ) block_mock.assert_has_calls([call(1), call(2)]) mock.assert_called_once() with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: with patch('aioetherscan.modules.logs.Logs._fill_topics', new=Mock()) as topic_mock: topic_mock.return_value = {} await logs.get_logs( from_block=1, to_block=2, address='addr', topics=['topic', ] ) topic_mock.assert_called_once_with(['topic', ], None) mock.assert_called_once() with patch('aioetherscan.network.Network.get', new=CoroutineMock()) as mock: with patch('aioetherscan.modules.logs.Logs._fill_topics', new=Mock()) as topic_mock: topic_mock.return_value = {} await logs.get_logs( from_block=1, to_block=2, address='addr', topics=['top1', 'top2'], topic_operators=['and'] ) topic_mock.assert_called_once_with(['top1', 'top2'], ['and']) mock.assert_called_once() def test_check_block(logs): assert logs._check_block(1) == 1 assert logs._check_block(0x1) == 1 assert logs._check_block('latest') == 'latest' with pytest.raises(ValueError): logs._check_block('123') def test_fill_topics(logs): assert logs._fill_topics(['top1'], None) == {'topic0': 'top1'} topics = ['top1', 'top2'] topic_operators = ['or'] assert logs._fill_topics(topics, topic_operators) == {'topic0': 'top1', 'topic1': 'top2', 'topic0_1_opr': 'or'} topics = ['top1', 'top2', 'top3'] topic_operators = ['or', 'and'] assert logs._fill_topics(topics, topic_operators) == { 'topic0': 'top1', 'topic1': 'top2', 'topic2': 'top3', 'topic0_1_opr': 'or', 'topic1_2_opr': 'and' } with patch('aioetherscan.modules.logs.Logs._check_topics', new=Mock()) as check_topics_mock: logs._fill_topics(topics, topic_operators) check_topics_mock.assert_called_once_with(topics, topic_operators) def test_check_topics(logs): with pytest.raises(ValueError): logs._check_topics([], []) with pytest.raises(ValueError): logs._check_topics([], ['xor']) with pytest.raises(ValueError): logs._check_topics(['top1'], ['or']) assert logs._check_topics(['top1', 'top2'], ['or']) is None PK!VK"aioetherscan/tests/test_network.pyimport asyncio import json import logging from unittest.mock import patch import aiohttp import asynctest import pytest from asyncio_throttle import Throttler from asynctest import CoroutineMock, MagicMock from asynctest import patch from aioetherscan.exceptions import EtherscanClientContentTypeError, EtherscanClientError, EtherscanClientApiError, \ EtherscanClientProxyError from aioetherscan.network import Network, HttpMethod class SessionMock(CoroutineMock): @pytest.mark.asyncio async def get(self, url, params, data): return AsyncCtxMgrMock() class AsyncCtxMgrMock(MagicMock): @pytest.mark.asyncio async def __aenter__(self): return self.aenter @pytest.mark.asyncio async def __aexit__(self, *args): pass def get_loop(): return asyncio.get_event_loop() def apikey(): return 'testapikey' @pytest.fixture() async def nw(): nw = Network(apikey(), 'main', get_loop()) yield nw await nw.close() def test_init(): myloop = get_loop() n = Network(apikey(), 'main', myloop) assert n._API_KEY == apikey() assert n._loop == myloop assert isinstance(n._session, aiohttp.ClientSession) assert n._session.loop == myloop assert isinstance(n._throttler, Throttler) assert n._throttler.rate_limit == 5 assert isinstance(n._logger, logging.Logger) def test_sign(nw): assert nw._sign({}) == {'apikey': nw._API_KEY} assert nw._sign({'smth': 'smth'}) == {'smth': 'smth', 'apikey': nw._API_KEY} def test_filter_params(nw): assert nw._filter_params({}) == {} assert nw._filter_params({1: 2, 3: None}) == {1: 2} assert nw._filter_params({1: 2, 3: 0}) == {1: 2, 3: 0} assert nw._filter_params({1: 2, 3: False}) == {1: 2, 3: False} @pytest.mark.asyncio async def test_get(nw): with patch('aioetherscan.network.Network._request', new=CoroutineMock()) as mock: await nw.get() mock.assert_called_once_with(HttpMethod.GET, params={'apikey': nw._API_KEY}) @pytest.mark.asyncio async def test_post(nw): with patch('aioetherscan.network.Network._request', new=CoroutineMock()) as mock: await nw.post() mock.assert_called_once_with(HttpMethod.POST, data={'apikey': nw._API_KEY}) with patch('aioetherscan.network.Network._request', new=CoroutineMock()) as mock: await nw.post({'some': 'data'}) mock.assert_called_once_with(HttpMethod.POST, data={'apikey': nw._API_KEY, 'some': 'data'}) with patch('aioetherscan.network.Network._request', new=CoroutineMock()) as mock: await nw.post({'some': 'data', 'null': None}) mock.assert_called_once_with(HttpMethod.POST, data={'apikey': nw._API_KEY, 'some': 'data'}) @pytest.mark.asyncio async def test_request(nw): class MagicMockContext(MagicMock): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) type(self).__aenter__ = CoroutineMock(return_value=MagicMock()) type(self).__aexit__ = CoroutineMock(return_value=MagicMock()) with asynctest.mock.patch('aiohttp.ClientSession.get', new_callable=MagicMockContext) as m: with patch('aioetherscan.network.Network._handle_response', new=CoroutineMock()) as h: await nw._request(HttpMethod.GET) m.assert_called_once_with('https://api.etherscan.io/api', data=None, params=None) h.assert_called_once() with asynctest.mock.patch('aiohttp.ClientSession.post', new_callable=MagicMockContext) as m: with patch('aioetherscan.network.Network._handle_response', new=CoroutineMock()) as h: await nw._request(HttpMethod.POST) m.assert_called_once_with('https://api.etherscan.io/api', data=None, params=None) h.assert_called_once() with asynctest.mock.patch('aiohttp.ClientSession.post', new_callable=MagicMockContext) as m: with patch('aioetherscan.network.Network._handle_response', new=CoroutineMock()) as h: with patch('asyncio_throttle.Throttler.acquire', new=CoroutineMock()) as t: await nw._request(HttpMethod.POST) t.assert_called_once() # noinspection PyTypeChecker @pytest.mark.asyncio async def test_handle_response(nw): class MockResponse: def __init__(self, data, raise_exc=None): self.data = data self.raise_exc = raise_exc @property def status(self): return 200 async def text(self): return 'some text' async def json(self): if self.raise_exc: raise self.raise_exc return json.loads(self.data) with pytest.raises(EtherscanClientContentTypeError) as e: await nw._handle_response(MockResponse('qweasd', aiohttp.ContentTypeError('info', 'hist'))) assert e.value.status == 200 assert e.value.content == 'some text' with pytest.raises(EtherscanClientError, match='some exc'): await nw._handle_response(MockResponse('qweasd', Exception('some exc'))) with pytest.raises(EtherscanClientApiError) as e: await nw._handle_response(MockResponse('{"status": "0", "message": "NOTOK", "result": "res"}')) assert e.value.message == 'NOTOK' assert e.value.result == 'res' with pytest.raises(EtherscanClientProxyError) as e: await nw._handle_response(MockResponse('{"error": {"code": "100", "message": "msg"}}')) assert e.value.code == '100' assert e.value.message == 'msg' assert await nw._handle_response(MockResponse('{"result": "someresult"}')) == 'someresult' @pytest.mark.asyncio async def test_close_session(nw): with patch('aiohttp.ClientSession.close', new_callable=CoroutineMock) as m: await nw.close() m.assert_called_once_with() def test_test_network(): nw = Network(apikey(), 'tobalaba', get_loop()) assert nw._API_URL == 'https://api-tobalaba.etherscan.com/api' def test_invalid_network(): with pytest.raises(ValueError): Network(apikey(), 'wrong', get_loop()) PK!4ɛpK>X;baP>PK!HAkd%aioetherscan-0.1.2.dist-info/METADATAMO0 >݇ DHL`Wd:FZ>HRP=I@'nkgM0{&&|PBjS/OخU ]'H@YNWBc4|:72ulB^y-CӾʨ-M/fɰHȯOlvzp3R(u$ǶJG>6!rŌ7IQmBn |>q|Ҳqv5Pm\)ú5/翆oӑ,0Ҽƙ-ѢPK!H@;b7 #aioetherscan-0.1.2.dist-info/RECORD}ɶHy~ d04 >iV/k|^{DL<8j~aclWd#0[)t #hOoz9RS㊐܏虋"i'l2GI93ܺki;΢8T<&r9/kK䥵h8eZ4Uyudm0,=GCvyę+} &BʢgvR c(P,`|}BUb2v8AINjE:ڹv7P@  Z zcE cE|V+@ Ab{mI j:4:/q{]x"8jl@"EѶI _,(ZT4ܤmV7OXV1ϪeslĂ~}= _.ՎPj3<-A~hA..۫T`T> u.Zaq==E|^]Q0-; 9 ew:qTK2/3`$ *w{y|( !IUTRԪd"$:c2ûvIےU7BÃVQZNkMFO Cwύm7+͒yҗ0NTj=:qR0PGu W3✅sRk:n>ӻ'qrֲdaHBcGj50re+MƘt]Oi.f58Zw{r1y>*RȬ !֗~82)n{ET>)4toi?yWmlL h,G|߶jRp{v=zEffYڴVT Kh$xBlxu,4;πx9N[]O_caN~M)82v0jHֈ#n;$eР *f /G|U]d-4uR*jefEi ZOj۶9Rp -|;llZ\XqN¯)u1P wܑEˑǿPK!U8''aioetherscan/__init__.pyPK! [0]aioetherscan/client.pyPK!@@Saioetherscan/common.pyPK!YkFFaioetherscan/exceptions.pyPK!4lLLE aioetherscan/modules/account.pyPK!)aioetherscan/modules/base.pyPK!Daioetherscan/modules/block.pyPK!L  aioetherscan/modules/contract.pyPK!%h 3*aioetherscan/modules/logs.pyPK!{w13aioetherscan/modules/proxy.pyPK!խ Faioetherscan/modules/stats.pyPK!8**#4Haioetherscan/modules/transaction.pyPK!. Kaioetherscan/modules/utils.pyPK!oxl{{Vaioetherscan/network.pyPK!Reaioetherscan/tests/__init__.pyPK!·!!"eaioetherscan/tests/test_account.pyPK!oyl aioetherscan/tests/test_block.pyPK!£22!͉aioetherscan/tests/test_client.pyPK!Fv2XX!>aioetherscan/tests/test_common.pyPK! =U44#Քaioetherscan/tests/test_contract.pyPK!i55%Jaioetherscan/tests/test_exceptions.pyPK! xO22¦aioetherscan/tests/test_logs.pyPK!VK"1aioetherscan/tests/test_network.pyPK!