diff --git a/bittensor/core/async_subtensor.py b/bittensor/core/async_subtensor.py index e2c75d108..d1f61533a 100644 --- a/bittensor/core/async_subtensor.py +++ b/bittensor/core/async_subtensor.py @@ -147,13 +147,13 @@ async def __aexit__(self, exc_type, exc_val, exc_tb): async def encode_params( self, - call_definition: list["ParamWithTypes"], + call_definition: dict[str, list["ParamWithTypes"]], params: Union[list[Any], dict[str, Any]], ) -> str: """Returns a hex encoded string of the params using their types.""" param_data = scalecodec.ScaleBytes(b"") - for i, param in enumerate(call_definition["params"]): # type: ignore + for i, param in enumerate(call_definition["params"]): scale_obj = await self.substrate.create_scale_object(param["type"]) if isinstance(params, list): param_data += scale_obj.encode(params[i]) @@ -440,7 +440,7 @@ async def query_runtime_api( return_type = call_definition["type"] - as_scale_bytes = scalecodec.ScaleBytes(json_result["result"]) # type: ignore + as_scale_bytes = scalecodec.ScaleBytes(json_result["result"]) rpc_runtime_config = RuntimeConfiguration() rpc_runtime_config.update_type_registry(load_type_registry_preset("legacy")) diff --git a/bittensor/core/subtensor.py b/bittensor/core/subtensor.py index 252414362..67888cd99 100644 --- a/bittensor/core/subtensor.py +++ b/bittensor/core/subtensor.py @@ -389,15 +389,15 @@ def add_args(cls, parser: "argparse.ArgumentParser", prefix: Optional[str] = Non @networking.ensure_connected def _encode_params( self, - call_definition: list["ParamWithTypes"], + call_definition: dict[str, list["ParamWithTypes"]], params: Union[list[Any], dict[str, Any]], ) -> str: """Returns a hex encoded string of the params using their types.""" param_data = scalecodec.ScaleBytes(b"") - for i, param in enumerate(call_definition["params"]): # type: ignore + for i, param in enumerate(call_definition["params"]): scale_obj = self.substrate.create_scale_object(param["type"]) - if type(params) is list: + if isinstance(params, list): param_data += scale_obj.encode(params[i]) else: if param["name"] not in params: @@ -1232,7 +1232,7 @@ def get_subnet_hyperparameters( else: bytes_result = bytes.fromhex(hex_bytes_result) - return SubnetHyperparameters.from_vec_u8(bytes_result) # type: ignore + return SubnetHyperparameters.from_vec_u8(bytes_result) # Community uses this method # Returns network ImmunityPeriod hyper parameter. diff --git a/tests/unit_tests/test_async_subtensor.py b/tests/unit_tests/test_async_subtensor.py new file mode 100644 index 000000000..6b58684f7 --- /dev/null +++ b/tests/unit_tests/test_async_subtensor.py @@ -0,0 +1,293 @@ +from pickle import FALSE + +import pytest + +from bittensor.core import async_subtensor + + +@pytest.fixture +def subtensor(mocker): + fake_async_substrate = mocker.AsyncMock( + autospec=async_subtensor.AsyncSubstrateInterface + ) + mocker.patch.object( + async_subtensor, "AsyncSubstrateInterface", return_value=fake_async_substrate + ) + return async_subtensor.AsyncSubtensor() + + +def test_decode_ss58_tuples_in_proposal_vote_data(mocker): + """Tests that ProposalVoteData instance instantiation works properly,""" + # Preps + mocked_decode_account_id = mocker.patch.object(async_subtensor, "decode_account_id") + fake_proposal_dict = { + "index": "0", + "threshold": 1, + "ayes": ("0 line", "1 line"), + "nays": ("2 line", "3 line"), + "end": 123, + } + + # Call + async_subtensor.ProposalVoteData(fake_proposal_dict) + + # Asserts + assert mocked_decode_account_id.call_count == len(fake_proposal_dict["ayes"]) + len( + fake_proposal_dict["nays"] + ) + assert mocked_decode_account_id.mock_calls == [ + mocker.call("0"), + mocker.call("1"), + mocker.call("2"), + mocker.call("3"), + ] + + +@pytest.mark.asyncio +async def test_encode_params(subtensor, mocker): + """Tests encode_params happy path.""" + # Preps + subtensor.substrate.create_scale_object = mocker.AsyncMock( + autospec=async_subtensor.AsyncSubstrateInterface.create_scale_object + ) + subtensor.substrate.create_scale_object.return_value.encode = mocker.Mock( + return_value=b"" + ) + + call_definition = { + "params": [ + {"name": "coldkey", "type": "Vec"}, + {"name": "uid", "type": "u16"}, + ] + } + params = ["coldkey", "uid"] + + # Call + decoded_params = await subtensor.encode_params( + call_definition=call_definition, params=params + ) + + # Asserts + subtensor.substrate.create_scale_object.call_args( + mocker.call("coldkey"), + mocker.call("Vec"), + mocker.call("uid"), + mocker.call("u16"), + ) + assert decoded_params == "0x" + + +@pytest.mark.asyncio +async def test_encode_params_raises_error(subtensor, mocker): + """Tests encode_params with raised error.""" + # Preps + subtensor.substrate.create_scale_object = mocker.AsyncMock( + autospec=async_subtensor.AsyncSubstrateInterface.create_scale_object + ) + subtensor.substrate.create_scale_object.return_value.encode = mocker.Mock( + return_value=b"" + ) + + call_definition = { + "params": [ + {"name": "coldkey", "type": "Vec"}, + ] + } + params = {"undefined param": "some value"} + + # Call and assert + with pytest.raises(ValueError): + await subtensor.encode_params(call_definition=call_definition, params=params) + + subtensor.substrate.create_scale_object.return_value.encode.assert_not_called() + + +@pytest.mark.asyncio +async def test_get_current_block(subtensor): + """Tests get_current_block method.""" + # Call + result = await subtensor.get_current_block() + + # Asserts + subtensor.substrate.get_block_number.assert_called_once() + assert result == subtensor.substrate.get_block_number.return_value + + +@pytest.mark.asyncio +async def test_get_block_hash_without_block_id_aka_none(subtensor): + """Tests get_block_hash method without passed block_id.""" + # Call + result = await subtensor.get_block_hash() + + # Asserts + assert result == subtensor.substrate.get_chain_head.return_value + + +@pytest.mark.asyncio +async def test_get_block_hash_with_block_id(subtensor): + """Tests get_block_hash method with passed block_id.""" + # Call + result = await subtensor.get_block_hash(block_id=1) + + # Asserts + assert result == subtensor.substrate.get_block_hash.return_value + + +@pytest.mark.asyncio +async def test_is_hotkey_registered_any(subtensor, mocker): + """Tests is_hotkey_registered_any method.""" + # Preps + mocked_get_netuids_for_hotkey = mocker.AsyncMock( + return_value=[1, 2], autospec=subtensor.get_netuids_for_hotkey + ) + subtensor.get_netuids_for_hotkey = mocked_get_netuids_for_hotkey + + # Call + result = await subtensor.is_hotkey_registered_any( + hotkey_ss58="hotkey", block_hash="FAKE_HASH" + ) + + # Asserts + assert result is (len(mocked_get_netuids_for_hotkey.return_value) > 0) + + +@pytest.mark.asyncio +async def test_get_subnet_burn_cost(subtensor, mocker): + """Tests get_subnet_burn_cost method.""" + # Preps + mocked_query_runtime_api = mocker.AsyncMock(autospec=subtensor.query_runtime_api) + subtensor.query_runtime_api = mocked_query_runtime_api + fake_block_hash = None + + # Call + result = await subtensor.get_subnet_burn_cost(block_hash=fake_block_hash) + + # Assert + assert result == mocked_query_runtime_api.return_value + mocked_query_runtime_api.assert_called_once_with( + runtime_api="SubnetRegistrationRuntimeApi", + method="get_network_registration_cost", + params=[], + block_hash=fake_block_hash, + ) + + +@pytest.mark.asyncio +async def test_get_total_subnets(subtensor, mocker): + """Tests get_total_subnets method.""" + # Preps + mocked_substrate_query = mocker.AsyncMock( + autospec=async_subtensor.AsyncSubstrateInterface.query + ) + subtensor.substrate.query = mocked_substrate_query + fake_block_hash = None + + # Call + result = await subtensor.get_total_subnets(block_hash=fake_block_hash) + + # Assert + assert result == mocked_substrate_query.return_value + mocked_substrate_query.assert_called_once_with( + module="SubtensorModule", + storage_function="TotalNetworks", + params=[], + block_hash=fake_block_hash, + ) + + +@pytest.mark.parametrize( + "records, response", + [([(0, True), (1, False), (3, False), (3, True)], [0, 3]), ([], [])], + ids=["with records", "empty-records"], +) +@pytest.mark.asyncio +async def test_get_subnets(subtensor, mocker, records, response): + """Tests get_subnets method with any return.""" + # Preps + fake_result = mocker.AsyncMock(autospec=list) + fake_result.records = records + fake_result.__aiter__.return_value = iter(records) + + mocked_substrate_query_map = mocker.AsyncMock( + autospec=async_subtensor.AsyncSubstrateInterface.query_map, + return_value=fake_result, + ) + + subtensor.substrate.query_map = mocked_substrate_query_map + fake_block_hash = None + + # Call + result = await subtensor.get_subnets(block_hash=fake_block_hash) + + # Asserts + mocked_substrate_query_map.assert_called_once_with( + module="SubtensorModule", + storage_function="NetworksAdded", + block_hash=fake_block_hash, + reuse_block_hash=True, + ) + assert result == response + + +@pytest.mark.parametrize( + "hotkey_ss58_in_result", + [True, False], + ids=["hotkey-exists", "hotkey-doesnt-exist"], +) +@pytest.mark.asyncio +async def test_is_hotkey_delegate(subtensor, mocker, hotkey_ss58_in_result): + """Tests is_hotkey_delegate method with any return.""" + # Preps + fake_hotkey_ss58 = "hotkey_58" + mocked_get_delegates = mocker.AsyncMock( + return_value=[ + mocker.Mock(hotkey_ss58=fake_hotkey_ss58 if hotkey_ss58_in_result else "") + ] + ) + subtensor.get_delegates = mocked_get_delegates + + # Call + result = await subtensor.is_hotkey_delegate( + hotkey_ss58=fake_hotkey_ss58, block_hash=None, reuse_block=True + ) + + # Asserts + assert result == hotkey_ss58_in_result + mocked_get_delegates.assert_called_once_with(block_hash=None, reuse_block=True) + + +@pytest.mark.parametrize( + "fake_hex_bytes_result, response", [(None, []), ("0xaabbccdd", b"\xaa\xbb\xcc\xdd")] +) +@pytest.mark.asyncio +async def test_get_delegates(subtensor, mocker, fake_hex_bytes_result, response): + """Tests get_delegates method.""" + # Preps + mocked_query_runtime_api = mocker.AsyncMock( + autospec=subtensor.query_runtime_api, return_value=fake_hex_bytes_result + ) + subtensor.query_runtime_api = mocked_query_runtime_api + mocked_delegate_info_list_from_vec_u8 = mocker.Mock() + async_subtensor.DelegateInfo.list_from_vec_u8 = ( + mocked_delegate_info_list_from_vec_u8 + ) + + # Call + result = await subtensor.get_delegates(block_hash=None, reuse_block=True) + + # Asserts + if fake_hex_bytes_result: + assert result == mocked_delegate_info_list_from_vec_u8.return_value + mocked_delegate_info_list_from_vec_u8.assert_called_once_with( + bytes.fromhex(fake_hex_bytes_result[2:]) + ) + else: + assert result == response + + mocked_query_runtime_api.assert_called_once_with( + runtime_api="DelegateInfoRuntimeApi", + method="get_delegates", + params=[], + block_hash=None, + reuse_block=True, + )