echonet: add get_timestamp_of_block_by_number func (#10263)

Author: ayeletstarkware

echonet/l1_client.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import List
+from typing import List, Optional
 
 import logging
 import requests
@@ -96,3 +96,50 @@ def get_logs(from_block: int, to_block: int, api_key: str) -> List["L1Client.Log
             )
             for result in results
         ]
+
+    @staticmethod
+    def get_timestamp_of_block(block_number: int, api_key: str) -> Optional[int]:
+        """
+        Get block timestamp by block number using eth_getBlockByNumber RPC method.
+        Tries up to RETRIES_COUNT times. On failure, logs an error and returns None.
+        """
+        rpc_url = L1Client.L1_MAINNET_URL.format(api_key=api_key)
+
+        payload = {
+            "jsonrpc": "2.0",
+            "method": "eth_getBlockByNumber",
+            "params": [hex(block_number), False],
+            "id": 1,
+        }
+
+        for attempt in range(L1Client.RETRIES_COUNT):
+            try:
+                response = requests.post(rpc_url, json=payload, timeout=10)
+                response.raise_for_status()
+                result = response.json()
+                logger.debug(
+                    f"get_timestamp_of_block succeeded on attempt {attempt + 1}",
+                    extra={"url": rpc_url, "block_number": block_number},
+                )
+                break  # success -> exit loop
+            except (requests.RequestException, ValueError) as exc:
+                logger.debug(
+                    f"get_timestamp_of_block attempt {attempt + 1}/{L1Client.RETRIES_COUNT} failed",
+                    extra={"url": rpc_url, "block_number": block_number},
+                    exc_info=True,
+                )
+
+        else:
+            logger.error(
+                f"get_timestamp_of_block failed after {L1Client.RETRIES_COUNT} attempts, returning None",
+                extra={"url": rpc_url, "block_number": block_number},
+            )
+            return None
+
+        block = result.get("result")
+        if block is None:
+            # Block not found
+            return None
+
+        # Timestamp is hex string, convert to int.
+        return int(block["timestamp"], 16)

echonet/tests/test_l1_client.py

@@ -170,6 +170,40 @@ def test_get_logs_when_rpc_result_is_empty(self, mock_post):
         self.assertEqual(mock_post.call_count, 1)
         self.assertEqual(logs, [])
 
+    @patch("l1_client.requests.post")
+    def test_get_timestamp_of_block_retries_after_failure_and_succeeds(self, mock_post):
+        request_exception = requests.RequestException("some error")
+
+        successful_response = Mock()
+        successful_response.raise_for_status.return_value = None
+        successful_response.json.return_value = {"result": {"timestamp": "0x20"}}  # 32
+
+        mock_post.side_effect = [request_exception, successful_response]
+
+        result = L1Client.get_timestamp_of_block(
+            block_number=123,
+            api_key="api_key",
+        )
+
+        self.assertEqual(mock_post.call_count, 2)
+        self.assertEqual(result, 32)
+
+    @patch("l1_client.requests.post")
+    def test_get_timestamp_of_block_returns_none_when_rpc_result_is_empty(self, mock_post):
+        response_ok = Mock()
+        response_ok.raise_for_status.return_value = None
+        response_ok.json.return_value = {"result": None}
+
+        mock_post.return_value = response_ok
+
+        result = L1Client.get_timestamp_of_block(
+            block_number=123,
+            api_key="api_key",
+        )
+
+        self.assertEqual(mock_post.call_count, 1)
+        self.assertIsNone(result)
+
 
 if __name__ == "__main__":
     unittest.main()