Add tests

Author: thewhaleking

tests/helpers/settings.py

@@ -34,3 +34,5 @@
 )
 
 ARCHIVE_ENTRYPOINT = "wss://archive.chain.opentensor.ai:443"
+
+LATENT_LITE_ENTRYPOINT = "wss://lite.sub.latent.to:443"

tests/unit_tests/asyncio_/test_substrate_interface.py

@@ -2,11 +2,12 @@
 from unittest.mock import AsyncMock, MagicMock, ANY
 
 import pytest
+from scalecodec import ss58_encode
 from websockets.exceptions import InvalidURI
 
 from async_substrate_interface.async_substrate import AsyncSubstrateInterface
 from async_substrate_interface.types import ScaleObj
-from tests.helpers.settings import ARCHIVE_ENTRYPOINT
+from tests.helpers.settings import ARCHIVE_ENTRYPOINT, LATENT_LITE_ENTRYPOINT
 
 
 @pytest.mark.asyncio
@@ -100,7 +101,7 @@ async def test_runtime_call(monkeypatch):
 @pytest.mark.asyncio
 async def test_websocket_shutdown_timer():
     # using default ws shutdown timer of 5.0 seconds
-    async with AsyncSubstrateInterface("wss://lite.sub.latent.to:443") as substrate:
+    async with AsyncSubstrateInterface(LATENT_LITE_ENTRYPOINT) as substrate:
         await substrate.get_chain_head()
         await asyncio.sleep(6)
         assert (
@@ -141,3 +142,43 @@ async def test_legacy_decoding():
             block_hash=block_hash,
         )
         assert timestamp.value == 1716358476004
+
+
+@pytest.mark.asyncio
+async def test_ss58_conversion():
+    async with AsyncSubstrateInterface(
+        LATENT_LITE_ENTRYPOINT, ss58_format=42, decode_ss58=False
+    ) as substrate:
+        block_hash = await substrate.get_chain_finalised_head()
+        qm = await substrate.query_map(
+            "SubtensorModule",
+            "OwnedHotkeys",
+            block_hash=block_hash,
+        )
+        # only do the first page, bc otherwise this will be massive
+        for key, value in qm.records:
+            assert isinstance(key, tuple)
+            assert isinstance(value, ScaleObj)
+            assert isinstance(value.value, list)
+            assert len(key) == 1
+            for key_tuple in value.value:
+                assert len(key_tuple[0]) == 32
+                random_key = key_tuple[0]
+
+        ss58_of_key = ss58_encode(bytes(random_key), substrate.ss58_format)
+        assert isinstance(ss58_of_key, str)
+
+        substrate.decode_ss58 = True  # change to decoding True
+
+        qm = await substrate.query_map(
+            "SubtensorModule",
+            "OwnedHotkeys",
+            block_hash=block_hash,
+        )
+        for key, value in qm.records:
+            assert isinstance(key, str)
+            assert isinstance(value, ScaleObj)
+            assert isinstance(value.value, list)
+            if len(value.value) > 0:
+                for decoded_key in value.value:
+                    assert isinstance(decoded_key, str)

tests/unit_tests/sync/test_substrate_interface.py

@@ -1,9 +1,11 @@
 from unittest.mock import MagicMock
 
+from scalecodec import ss58_encode
+
 from async_substrate_interface.sync_substrate import SubstrateInterface
 from async_substrate_interface.types import ScaleObj
 
-from tests.helpers.settings import ARCHIVE_ENTRYPOINT
+from tests.helpers.settings import ARCHIVE_ENTRYPOINT, LATENT_LITE_ENTRYPOINT
 
 
 def test_runtime_call(monkeypatch):
@@ -101,3 +103,42 @@ def test_legacy_decoding():
             block_hash=block_hash,
         )
         assert timestamp.value == 1716358476004
+
+
+def test_ss58_conversion():
+    with SubstrateInterface(
+        LATENT_LITE_ENTRYPOINT, ss58_format=42, decode_ss58=False
+    ) as substrate:
+        block_hash = substrate.get_chain_finalised_head()
+        qm = substrate.query_map(
+            "SubtensorModule",
+            "OwnedHotkeys",
+            block_hash=block_hash,
+        )
+        # only do the first page, bc otherwise this will be massive
+        for key, value in qm.records:
+            assert isinstance(key, tuple)
+            assert isinstance(value, ScaleObj)
+            assert isinstance(value.value, list)
+            assert len(key) == 1
+            for key_tuple in value.value:
+                assert len(key_tuple[0]) == 32
+                random_key = key_tuple[0]
+
+        ss58_of_key = ss58_encode(bytes(random_key), substrate.ss58_format)
+        assert isinstance(ss58_of_key, str)
+
+        substrate.decode_ss58 = True  # change to decoding True
+
+        qm = substrate.query_map(
+            "SubtensorModule",
+            "OwnedHotkeys",
+            block_hash=block_hash,
+        )
+        for key, value in qm.records:
+            assert isinstance(key, str)
+            assert isinstance(value, ScaleObj)
+            assert isinstance(value.value, list)
+            if len(value.value) > 0:
+                for decoded_key in value.value:
+                    assert isinstance(decoded_key, str)