Merge branch 'master' into fix/run-tests-script

gnosis/eth/ethereum_client.py

@@ -1193,6 +1193,8 @@ def __init__(
         self.ethereum_node_url: str = ethereum_node_url
         self.timeout = provider_timeout
         self.slow_timeout = slow_provider_timeout
+        self.use_caching_middleware = use_caching_middleware
+
         self.w3_provider = HTTPProvider(
             self.ethereum_node_url,
             request_kwargs={"timeout": provider_timeout},
@@ -1205,26 +1207,31 @@ def __init__(
         )
         self.w3: Web3 = Web3(self.w3_provider)
         self.slow_w3: Web3 = Web3(self.w3_slow_provider)
-        # Remove not needed middlewares
+
+        # Adjust Web3.py middleware
         for w3 in self.w3, self.slow_w3:
+            # Don't spend resources con converting dictionaries to attribute dictionaries
             w3.middleware_onion.remove("attrdict")
+            # Disable web3 automatic retry, it's handled on our HTTP Session
+            w3.provider.middlewares = []
+            if self.use_caching_middleware:
+                w3.middleware_onion.add(simple_cache_middleware)
 
-        self.erc20: Erc20Manager = Erc20Manager(self)
-        self.erc721: Erc721Manager = Erc721Manager(self)
-        self.tracing: TracingManager = TracingManager(self)
-        self.batch_call_manager: BatchCallManager = BatchCallManager(self)
+        # The geth_poa_middleware is required to connect to geth --dev or the Goerli public network.
+        # It may also be needed for other EVM compatible blockchains like Polygon or BNB Chain (Binance Smart Chain).
         try:
             if self.get_network() != EthereumNetwork.MAINNET:
-                self.w3.middleware_onion.inject(geth_poa_middleware, layer=0)
+                for w3 in self.w3, self.slow_w3:
+                    w3.middleware_onion.inject(geth_poa_middleware, layer=0)
+        except (IOError, OSError):
             # For tests using dummy connections (like IPC)
-        except (IOError, FileNotFoundError):
-            self.w3.middleware_onion.inject(geth_poa_middleware, layer=0)
-
-        self.use_caching_middleware = use_caching_middleware
-        if self.use_caching_middleware:
-            self.w3.middleware_onion.add(simple_cache_middleware)
-            self.slow_w3.middleware_onion.add(simple_cache_middleware)
+            for w3 in self.w3, self.slow_w3:
+                w3.middleware_onion.inject(geth_poa_middleware, layer=0)
 
+        self.erc20: Erc20Manager = Erc20Manager(self)
+        self.erc721: Erc721Manager = Erc721Manager(self)
+        self.tracing: TracingManager = TracingManager(self)
+        self.batch_call_manager: BatchCallManager = BatchCallManager(self)
         self.batch_request_max_size = batch_request_max_size
 
     def __str__(self):
@@ -1238,10 +1245,19 @@ def _prepare_http_session(self, retry_count: int) -> requests.Session:
         https://web3py.readthedocs.io/en/stable/providers.html#httpprovider
         """
         session = requests.Session()
+        retry_conf = (
+            requests.adapters.Retry(
+                total=retry_count,
+                backoff_factor=0.3,
+            )
+            if retry_count
+            else 0
+        )
+
         adapter = requests.adapters.HTTPAdapter(
             pool_connections=1,  # Doing all the connections to the same url
             pool_maxsize=100,  # Number of concurrent connections
-            max_retries=retry_count,  # Nodes are not very responsive some times
+            max_retries=retry_conf,  # Nodes are not very responsive some times
             pool_block=False,
         )
         session.mount("http://", adapter)

gnosis/eth/tests/utils.py

@@ -1,3 +1,4 @@
+import functools
 import json
 import os
 from copy import deepcopy
@@ -74,6 +75,29 @@ def just_test_if_polygon_node() -> str:
     return polygon_node_url
 
 
+def skip_on(exception, reason="Test skipped due to a controlled exception"):
+    """
+    Decorator to skip a test if an exception is raised instead of failing it
+
+    :param exception:
+    :param reason:
+    :return:
+    """
+
+    def decorator_func(f):
+        @functools.wraps(f)
+        def wrapper(*args, **kwargs):
+            try:
+                # Run the test
+                return f(*args, **kwargs)
+            except exception:
+                pytest.skip(reason)
+
+        return wrapper
+
+    return decorator_func
+
+
 # TODO Move this to EthereumClient
 def send_tx(w3: Web3, tx: TxParams, account: LocalAccount) -> bytes:
     tx["from"] = account.address

gnosis/safe/api/transaction_service_api.py

@@ -103,6 +103,11 @@ def parse_signatures(cls, raw_tx: Dict[str, Any]) -> Optional[bytes]:
             )
 
     def get_balances(self, safe_address: str) -> List[Dict[str, Any]]:
+        """
+
+        :param safe_address:
+        :return: a list of balances for provided Safe
+        """
         response = self._get_request(f"/api/v1/safes/{safe_address}/balances/")
         if not response.ok:
             raise SafeAPIException(f"Cannot get balances: {response.content}")
@@ -152,6 +157,11 @@ def get_safe_transaction(
         return safe_tx, tx_hash
 
     def get_transactions(self, safe_address: ChecksumAddress) -> List[Dict[str, Any]]:
+        """
+
+        :param safe_address:
+        :return: a list of transactions for provided Safe
+        """
         response = self._get_request(
             f"/api/v1/safes/{safe_address}/multisig-transactions/"
         )
@@ -160,12 +170,36 @@ def get_transactions(self, safe_address: ChecksumAddress) -> List[Dict[str, Any]
         return response.json().get("results", [])
 
     def get_delegates(self, safe_address: ChecksumAddress) -> List[Dict[str, Any]]:
+        """
+
+        :param safe_address:
+        :return: a list of delegates for provided Safe
+        """
         response = self._get_request(f"/api/v1/delegates/?safe={safe_address}")
         if not response.ok:
             raise SafeAPIException(f"Cannot get delegates: {response.content}")
         return response.json().get("results", [])
 
+    def get_safes_for_owner(
+        self, owner_address: ChecksumAddress
+    ) -> List[ChecksumAddress]:
+        """
+
+        :param owner_address:
+        :return: a List of Safe addresses which the owner_address is an owner
+        """
+        response = self._get_request(f"/api/v1/owners/{owner_address}/safes/")
+        if not response.ok:
+            raise SafeAPIException(f"Cannot get delegates: {response.content}")
+        return response.json().get("safes", [])
+
     def post_signatures(self, safe_tx_hash: bytes, signatures: bytes) -> bool:
+        """
+        Create a new confirmation with provided signature for the given safe_tx_hash
+        :param safe_tx_hash:
+        :param signatures:
+        :return: True if new confirmation was created
+        """
         safe_tx_hash = HexBytes(safe_tx_hash).hex()
         response = self._post_request(
             f"/api/v1/multisig-transactions/{safe_tx_hash}/confirmations/",

gnosis/safe/tests/api/test_transaction_service_api.py

@@ -102,3 +102,8 @@ def test_get_balances(self):
     def test_get_transactions(self):
         transactions = self.transaction_service_api.get_transactions(self.safe_address)
         self.assertIsInstance(transactions, list)
+
+    def test_get_safes_for_owner(self):
+        owner_address = "0x5aC255889882aCd3da2aA939679E3f3d4cea221e"
+        safes = self.transaction_service_api.get_safes_for_owner(owner_address)
+        self.assertIn(self.safe_address, safes)

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = safe-eth-py
-version = 6.0.0b7
+version = 6.0.0b8
 description = Safe Ecosystem Foundation utilities for Ethereum projects
 long_description = file: README.rst
 long_description_content_type = text/x-rst; charset=UTF-8