Merge branch 'staging' into tests/zyzniewski/proper_mock

Author: thewhaleking

bittensor/core/async_subtensor.py

@@ -1,5 +1,6 @@
 import asyncio
 import copy
+from datetime import datetime, timezone
 import ssl
 from functools import partial
 from typing import Optional, Any, Union, Iterable, TYPE_CHECKING
@@ -27,6 +28,7 @@
     decode_account_id,
     DynamicInfo,
 )
+from bittensor.core.chain_data.delegate_info import DelegatedInfo
 from bittensor.core.chain_data.utils import decode_metadata
 from bittensor.core.config import Config
 from bittensor.core.errors import SubstrateRequestException
@@ -1219,7 +1221,7 @@ async def get_delegated(
         if not result:
             return []
 
-        return DelegateInfo.delegated_list_from_dicts(result)
+        return DelegatedInfo.list_from_dicts(result)
 
     async def get_delegates(
         self,
@@ -2744,6 +2746,36 @@ async def weights_rate_limit(
         )
         return None if call is None else int(call)
 
+    async def get_timestamp(
+        self,
+        block: Optional[int] = None,
+        block_hash: Optional[str] = None,
+        reuse_block: bool = False,
+    ) -> datetime:
+        """
+        Retrieves the datetime timestamp for a given block
+
+        Arguments:
+            block: The blockchain block number for the query. Do not specify if specifying block_hash or reuse_block.
+            block_hash: The blockchain block_hash representation of the block id. Do not specify if specifying block
+                or reuse_block.
+            reuse_block: Whether to reuse the last-used blockchain block hash. Do not specify if specifying block or
+                block_hash.
+
+        Returns:
+            datetime object for the timestamp of the block
+        """
+        unix = (
+            await self.query_module(
+                "Timestamp",
+                "Now",
+                block=block,
+                block_hash=block_hash,
+                reuse_block=reuse_block,
+            )
+        ).value
+        return datetime.fromtimestamp(unix / 1000, tz=timezone.utc)
+
     # Extrinsics helper ================================================================================================
 
     async def sign_and_send_extrinsic(

bittensor/core/subtensor.py

@@ -1,4 +1,6 @@
 import copy
+from datetime import datetime, timezone
+
 from functools import lru_cache
 from typing import TYPE_CHECKING, Any, Iterable, Optional, Union, cast
 
@@ -2070,6 +2072,19 @@ def weights_rate_limit(
         )
         return None if call is None else int(call)
 
+    def get_timestamp(self, block: Optional[int] = None) -> datetime:
+        """
+        Retrieves the datetime timestamp for a given block
+
+        Arguments:
+            block: The blockchain block number for the query.
+
+        Returns:
+            datetime object for the timestamp of the block
+        """
+        unix = cast(ScaleObj, self.query_module("Timestamp", "Now", block=block)).value
+        return datetime.fromtimestamp(unix / 1000, tz=timezone.utc)
+
     # Extrinsics helper ================================================================================================
 
     def sign_and_send_extrinsic(

tests/e2e_tests/test_cross_subtensor_compatibility.py

@@ -0,0 +1,18 @@
+from datetime import datetime
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_get_timestamp(subtensor, async_subtensor, local_chain):
+    with subtensor:
+        block_number = subtensor.get_current_block()
+        assert isinstance(
+            subtensor.get_timestamp(), datetime
+        )  # verify it works with no block number specified
+        sync_result = subtensor.get_timestamp(
+            block=block_number
+        )  # verify it works with block number specified
+    async with async_subtensor:
+        assert isinstance(await async_subtensor.get_timestamp(), datetime)
+        async_result = await async_subtensor.get_timestamp(block=block_number)
+    assert sync_result == async_result

tests/unit_tests/test_async_subtensor.py

@@ -1,6 +1,8 @@
+import datetime
 import unittest.mock as mock
 
 import pytest
+from async_substrate_interface.types import ScaleObj
 from bittensor_wallet import Wallet
 
 from bittensor import u64_normalized_float
@@ -1231,9 +1233,9 @@ async def test_get_delegated_no_block_hash_no_reuse(subtensor, mocker):
     # Preps
     fake_coldkey_ss58 = "fake_ss58_address"
 
-    mocked_delegated_list_from_dicts = mocker.Mock()
-    async_subtensor.DelegateInfo.delegated_list_from_dicts = (
-        mocked_delegated_list_from_dicts
+    mocked_delegated_list_from_dicts = mocker.patch.object(
+        async_subtensor.DelegatedInfo,
+        "list_from_dicts",
     )
 
     # Call
@@ -1259,9 +1261,9 @@ async def test_get_delegated_with_block_hash(subtensor, mocker):
     fake_coldkey_ss58 = "fake_ss58_address"
     fake_block_hash = "fake_block_hash"
 
-    mocked_delegated_list_from_dicts = mocker.Mock()
-    async_subtensor.DelegateInfo.delegated_list_from_dicts = (
-        mocked_delegated_list_from_dicts
+    mocked_delegated_list_from_dicts = mocker.patch.object(
+        async_subtensor.DelegatedInfo,
+        "list_from_dicts",
     )
 
     # Call
@@ -1289,9 +1291,9 @@ async def test_get_delegated_with_reuse_block(subtensor, mocker):
     fake_coldkey_ss58 = "fake_ss58_address"
     reuse_block = True
 
-    mocked_delegated_list_from_dicts = mocker.Mock()
-    async_subtensor.DelegateInfo.delegated_list_from_dicts = (
-        mocked_delegated_list_from_dicts
+    mocked_delegated_list_from_dicts = mocker.patch.object(
+        async_subtensor.DelegatedInfo,
+        "list_from_dicts",
     )
 
     # Call
@@ -2714,3 +2716,15 @@ async def test_get_all_neuron_certificates(mocker, subtensor):
         block_hash=None,
         reuse_block_hash=False,
     )
+
+
+@pytest.mark.asyncio
+async def test_get_timestamp(mocker, subtensor):
+    fake_block = 1000
+    mocked_query = mocker.AsyncMock(return_value=ScaleObj(1740586018 * 1000))
+    mocker.patch.object(subtensor.substrate, "query", mocked_query)
+    expected_result = datetime.datetime(
+        2025, 2, 26, 16, 6, 58, tzinfo=datetime.timezone.utc
+    )
+    actual_result = await subtensor.get_timestamp(block=fake_block)
+    assert expected_result == actual_result

tests/unit_tests/test_subtensor.py

@@ -17,11 +17,13 @@
 
 import argparse
 import unittest.mock as mock
+import datetime
 from unittest.mock import MagicMock
 
 import pytest
 from bittensor_wallet import Wallet
 from async_substrate_interface import sync_substrate
+from async_substrate_interface.types import ScaleObj
 import websockets
 
 from bittensor import StakeInfo
@@ -3095,3 +3097,14 @@ def test_get_all_neuron_certificates(mocker, subtensor):
         params=[fake_netuid],
         block_hash=None,
     )
+
+
+def test_get_timestamp(mocker, subtensor):
+    fake_block = 1000
+    mocked_query = mocker.MagicMock(return_value=ScaleObj(1740586018 * 1000))
+    mocker.patch.object(subtensor.substrate, "query", mocked_query)
+    expected_result = datetime.datetime(
+        2025, 2, 26, 16, 6, 58, tzinfo=datetime.timezone.utc
+    )
+    actual_result = subtensor.get_timestamp(block=fake_block)
+    assert expected_result == actual_result