10793540350

Committed 10 Sep 2024 01:31PM UTC coverage: 93.551% (-0.3%) from 93.892%

Build # 10793540350

Build Type

push

github

Committed by

falvaradorodriguez

Commit Message

Fix cowswap test

Run Details

8777 of 9382 relevant lines covered (93.55%)

3.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.06

/safe_eth/eth/clients/sourcify_client.py

import os
from functools import cache
from typing import Any, Dict, List, Optional
from urllib.parse import urljoin

from ...util.http import prepare_http_session
from .. import EthereumNetwork
from ..utils import fast_is_checksum_address
from .contract_metadata import ContractMetadata


class SourcifyClientException(Exception):
    pass


class SourcifyClientConfigurationProblem(Exception):
    pass


class SourcifyClient:
    """
    Get contract metadata from Sourcify. Matches can be full or partial:

      - Full: Both the source files as well as the meta data files were an exact match between the deployed bytecode
        and the published files.
      - Partial: Source code compiles to the same bytecode and thus the contract behaves in the same way,
        but the source code can be different: Variables can have misleading names,
        comments can be different and especially the NatSpec comments could have been modified.

    """

    def __init__(
        self,
        network: EthereumNetwork = EthereumNetwork.MAINNET,
        base_url_api: str = "https://sourcify.dev",
        base_url_repo: str = "https://repo.sourcify.dev/",
        request_timeout: int = int(
            os.environ.get("SOURCIFY_CLIENT_REQUEST_TIMEOUT", 10)
        ),
    ):
        self.network = network
        self.base_url_api = base_url_api
        self.base_url_repo = base_url_repo
        self.http_session = prepare_http_session(10, 100)
        self.request_timeout = request_timeout

        if not self.is_chain_supported(network.value):
            raise SourcifyClientConfigurationProblem(
                f"Network {network.name} - {network.value} not supported"
            )

    def _get_abi_from_metadata(self, metadata: Dict[str, Any]) -> List[Dict[str, Any]]:
        return metadata["output"]["abi"]

    def _get_name_from_metadata(self, metadata: Dict[str, Any]) -> Optional[str]:
        values = list(metadata["settings"].get("compilationTarget", {}).values())
        if values:
            return values[0]
        return None

    def _do_request(self, url: str) -> Optional[Dict[str, Any]]:
        response = self.http_session.get(url, timeout=self.request_timeout)
        if not response.ok:
            return None

        return response.json()

    def is_chain_supported(self, chain_id: int) -> bool:
        chains = self.get_chains()
        if not chains:
            raise IOError("Cannot get chains for SourcifyClient")
        for chain in chains:
            if not isinstance(chain, dict):
                continue
            chain_id_str = chain.get("chainId")
            if chain_id_str is None:
                continue
            try:
                if chain_id == int(chain_id_str):
                    return True
            except ValueError:
                continue
        return False

    @cache
    def get_chains(self) -> Dict[str, Any]:
        url = urljoin(self.base_url_api, "/server/chains")
        result = self._do_request(url)
        return result or {}

    def get_contract_metadata(
        self, contract_address: str
    ) -> Optional[ContractMetadata]:
        assert fast_is_checksum_address(
            contract_address
        ), "Expecting a checksummed address"

        for match_type in ("full_match", "partial_match"):
            url = urljoin(
                self.base_url_repo,
                f"/contracts/{match_type}/{self.network.value}/{contract_address}/metadata.json",
            )
            metadata = self._do_request(url)
            if metadata:
                abi = self._get_abi_from_metadata(metadata)
                name = self._get_name_from_metadata(metadata)
                return ContractMetadata(name, abi, match_type == "partial_match")
        return None

1	import os	4✔
2	from functools import cache	4✔
3	from typing import Any, Dict, List, Optional	4✔
4	from urllib.parse import urljoin	4✔
5
6	from ...util.http import prepare_http_session	4✔
7	from .. import EthereumNetwork	4✔
8	from ..utils import fast_is_checksum_address	4✔
9	from .contract_metadata import ContractMetadata	4✔
10
11
12	class SourcifyClientException(Exception):	4✔
13	pass	4✔
14
15
16	class SourcifyClientConfigurationProblem(Exception):	4✔
17	pass	4✔
18
19
20	class SourcifyClient:	4✔
21	"""
22	Get contract metadata from Sourcify. Matches can be full or partial:
23
24	- Full: Both the source files as well as the meta data files were an exact match between the deployed bytecode
25	and the published files.
26	- Partial: Source code compiles to the same bytecode and thus the contract behaves in the same way,
27	but the source code can be different: Variables can have misleading names,
28	comments can be different and especially the NatSpec comments could have been modified.
29
30	"""
31
32	def __init__(	4✔
33	self,
34	network: EthereumNetwork = EthereumNetwork.MAINNET,
35	base_url_api: str = "https://sourcify.dev",
36	base_url_repo: str = "https://repo.sourcify.dev/",
37	request_timeout: int = int(
38	os.environ.get("SOURCIFY_CLIENT_REQUEST_TIMEOUT", 10)
39	),
40	):
41	self.network = network	4✔
42	self.base_url_api = base_url_api	4✔
43	self.base_url_repo = base_url_repo	4✔
44	self.http_session = prepare_http_session(10, 100)	4✔
45	self.request_timeout = request_timeout	4✔
46
47	if not self.is_chain_supported(network.value):	4✔
48	raise SourcifyClientConfigurationProblem(	4✔
49	f"Network {network.name} - {network.value} not supported"
50	)
51
52	def _get_abi_from_metadata(self, metadata: Dict[str, Any]) -> List[Dict[str, Any]]:	4✔
53	return metadata["output"]["abi"]	4✔
54
55	def _get_name_from_metadata(self, metadata: Dict[str, Any]) -> Optional[str]:	4✔
56	values = list(metadata["settings"].get("compilationTarget", {}).values())	4✔
57	if values:	4✔
58	return values[0]	4✔
59	return None	×
60
61	def _do_request(self, url: str) -> Optional[Dict[str, Any]]:	4✔
62	response = self.http_session.get(url, timeout=self.request_timeout)	4✔
63	if not response.ok:	4✔
64	return None	4✔
65
66	return response.json()	4✔
67
68	def is_chain_supported(self, chain_id: int) -> bool:	4✔
69	chains = self.get_chains()	4✔
70	if not chains:	4✔
71	raise IOError("Cannot get chains for SourcifyClient")	×
72	for chain in chains:	4✔
73	if not isinstance(chain, dict):	4✔
74	continue	×
75	chain_id_str = chain.get("chainId")	4✔
76	if chain_id_str is None:	4✔
77	continue	×
78	try:	4✔
79	if chain_id == int(chain_id_str):	4✔
80	return True	4✔
81	except ValueError:	×
82	continue	×
83	return False	4✔
84
85	@cache	4✔
86	def get_chains(self) -> Dict[str, Any]:	4✔
87	url = urljoin(self.base_url_api, "/server/chains")	4✔
88	result = self._do_request(url)	4✔
89	return result or {}	4✔
90
91	def get_contract_metadata(	4✔
92	self, contract_address: str
93	) -> Optional[ContractMetadata]:
94	assert fast_is_checksum_address(	4✔
95	contract_address
96	), "Expecting a checksummed address"
97
98	for match_type in ("full_match", "partial_match"):	4✔
99	url = urljoin(	4✔
100	self.base_url_repo,
101	f"/contracts/{match_type}/{self.network.value}/{contract_address}/metadata.json",
102	)
103	metadata = self._do_request(url)	4✔
104	if metadata:	4✔
105	abi = self._get_abi_from_metadata(metadata)	4✔
106	name = self._get_name_from_metadata(metadata)	4✔
107	return ContractMetadata(name, abi, match_type == "partial_match")	4✔
108	return None	×

safe-global / safe-eth-py / 10793540350

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous