test(2472): offline unittests

Author: zyzniewski-reef

tests/unit_tests/conftest.py

@@ -1,5 +1,9 @@
 import pytest
 from aioresponses import aioresponses
+from substrateinterface.base import SubstrateInterface
+from websockets.sync.client import ClientConnection
+
+import bittensor.core.subtensor
 
 
 @pytest.fixture
@@ -11,3 +15,46 @@ def force_legacy_torch_compatible_api(monkeypatch):
 def mock_aio_response():
     with aioresponses() as m:
         yield m
+
+
+@pytest.fixture
+def websockets_client_connection(mocker):
+    return mocker.Mock(
+        autospec=ClientConnection,
+        **{
+            "close_code": None,
+            "socket.getsockopt.return_value": 0,
+        },
+    )
+
+
+@pytest.fixture
+def subtensor(websockets_client_connection, mock_substrate_interface):
+    return bittensor.core.subtensor.Subtensor(
+        websocket=websockets_client_connection,
+    )
+
+
+@pytest.fixture
+def mock_substrate_interface(websockets_client_connection, mocker):
+    mocked = mocker.MagicMock(
+        autospec=SubstrateInterface,
+        **{
+            "websocket": websockets_client_connection,
+        },
+    )
+
+    mocker.patch("bittensor.core.subtensor.SubstrateInterface", return_value=mocked)
+
+    return mocked
+
+
+@pytest.fixture
+def mock_get_external_ip(mocker):
+    mocked = mocker.Mock(
+        return_value="192.168.1.1",
+    )
+
+    mocker.patch("bittensor.utils.networking.get_external_ip", mocked)
+
+    return mocked

tests/unit_tests/extrinsics/test_async_commit_reveal.py

@@ -10,7 +10,7 @@
 @pytest.fixture
 def subtensor(mocker):
     fake_substrate = mocker.AsyncMock()
-    fake_substrate.websocket.sock.getsockopt.return_value = 0
+    fake_substrate.websocket.socket.getsockopt.return_value = 0
     mocker.patch.object(
         subtensor_module, "AsyncSubstrateInterface", return_value=fake_substrate
     )

tests/unit_tests/extrinsics/test_commit_reveal.py

@@ -1,20 +1,10 @@
-from bittensor.core import subtensor as subtensor_module
-from bittensor.core.chain_data import SubnetHyperparameters
-from bittensor.core.subtensor import Subtensor
-from bittensor.core.extrinsics import commit_reveal
+import numpy as np
 import pytest
 import torch
-import numpy as np
-
 
-@pytest.fixture
-def subtensor(mocker):
-    fake_substrate = mocker.MagicMock()
-    fake_substrate.websocket.sock.getsockopt.return_value = 0
-    mocker.patch.object(
-        subtensor_module, "SubstrateInterface", return_value=fake_substrate
-    )
-    yield Subtensor()
+from bittensor.core import subtensor as subtensor_module
+from bittensor.core.chain_data import SubnetHyperparameters
+from bittensor.core.extrinsics import commit_reveal
 
 
 @pytest.fixture

tests/unit_tests/extrinsics/test_commit_weights.py

@@ -1,22 +1,9 @@
-import pytest
-
 from bittensor.core import subtensor as subtensor_module
-from bittensor.core.settings import version_as_int
-from bittensor.core.subtensor import Subtensor
 from bittensor.core.extrinsics.commit_weights import (
     do_commit_weights,
     do_reveal_weights,
 )
-
-
-@pytest.fixture
-def subtensor(mocker):
-    fake_substrate = mocker.MagicMock()
-    fake_substrate.websocket.sock.getsockopt.return_value = 0
-    mocker.patch.object(
-        subtensor_module, "SubstrateInterface", return_value=fake_substrate
-    )
-    return Subtensor()
+from bittensor.core.settings import version_as_int
 
 
 def test_do_commit_weights(subtensor, mocker):

tests/unit_tests/extrinsics/test_transfer.py

@@ -1,21 +1,8 @@
-import pytest
-
 from bittensor.core import subtensor as subtensor_module
 from bittensor.core.extrinsics.transfer import do_transfer
-from bittensor.core.subtensor import Subtensor
 from bittensor.utils.balance import Balance
 
 
-@pytest.fixture
-def subtensor(mocker):
-    fake_substrate = mocker.MagicMock()
-    fake_substrate.websocket.sock.getsockopt.return_value = 0
-    mocker.patch.object(
-        subtensor_module, "SubstrateInterface", return_value=fake_substrate
-    )
-    return Subtensor()
-
-
 def test_do_transfer_is_success_true(subtensor, mocker):
     """Successful do_transfer call."""
     # Prep

tests/unit_tests/test_axon.py

@@ -47,7 +47,7 @@
 )
 
 
-def test_attach_initial():
+def test_attach_initial(mock_get_external_ip):
     # Create a mock AxonServer instance
     server = Axon()
 
@@ -92,7 +92,7 @@ def wrong_verify_fn(synapse: TestSynapse) -> bool:
         server.attach(forward_fn, blacklist_fn, priority_fn, wrong_verify_fn)
 
 
-def test_attach():
+def test_attach(mock_get_external_ip):
     # Create a mock AxonServer instance
     server = Axon()
 
@@ -165,7 +165,7 @@ def mock_request():
 
 
 @pytest.fixture
-def axon_instance():
+def axon_instance(mock_get_external_ip):
     axon = Axon()
     axon.required_hash_fields = {"test_endpoint": ["field1", "field2"]}
     axon.forward_class_types = {
@@ -352,7 +352,7 @@ async def test_verify_body_integrity_error_cases(
         (MockInfo(), "MockInfoString", "edge_case_empty_string"),
     ],
 )
-def test_to_string(info_return, expected_output, test_id):
+def test_to_string(info_return, expected_output, test_id, mock_get_external_ip):
     # Arrange
     axon = Axon()
     with patch.object(axon, "info", return_value=info_return):
@@ -381,7 +381,9 @@ def test_to_string(info_return, expected_output, test_id):
         ),
     ],
 )
-def test_valid_ipv4_and_ipv6_address(ip, port, expected_ip_type, test_id):
+def test_valid_ipv4_and_ipv6_address(
+    ip, port, expected_ip_type, test_id, mock_get_external_ip
+):
     # Arrange
     hotkey = MockHotkey("5EemgxS7cmYbD34esCFoBgUZZC8JdnGtQvV5Qw3QFUCRRtGP")
     coldkey = MockHotkey("5EemgxS7cmYbD34esCFoBgUZZC8JdnGtQvV5Qw3QFUCRRtGP")
@@ -454,7 +456,14 @@ def test_invalid_ip_address(ip, port, expected_exception):
     ],
 )
 def test_axon_str_representation(
-    ip, port, ss58_address, started, forward_fns, expected_str, test_id
+    ip,
+    port,
+    ss58_address,
+    started,
+    forward_fns,
+    expected_str,
+    test_id,
+    mock_get_external_ip,
 ):
     # Arrange
     hotkey = MockHotkey(ss58_address)

tests/unit_tests/test_dendrite.py

@@ -48,7 +48,7 @@ def dummy(synapse: SynapseDummy) -> SynapseDummy:
 
 
 @pytest.fixture
-def setup_dendrite():
+def setup_dendrite(mock_get_external_ip):
     # Assuming bittensor.Wallet() returns a wallet object
     user_wallet = get_mock_wallet()
     dendrite_obj = Dendrite(user_wallet)
@@ -70,7 +70,10 @@ def axon_info():
 @pytest.fixture(scope="session")
 def setup_axon():
     wallet = get_mock_wallet()
-    axon = Axon(wallet)
+    axon = Axon(
+        wallet,
+        external_ip="192.168.1.1",
+    )
     axon.attach(forward_fn=dummy)
     axon.start()
     yield axon
@@ -122,15 +125,15 @@ def __await__(self):
         return self().__await__()
 
 
-def test_dendrite_create_wallet():
+def test_dendrite_create_wallet(mock_get_external_ip):
     d = Dendrite(get_mock_wallet())
     d = Dendrite(get_mock_wallet().hotkey)
     d = Dendrite(get_mock_wallet().coldkeypub)
     assert d.__str__() == d.__repr__()
 
 
 @pytest.mark.asyncio
-async def test_forward_many():
+async def test_forward_many(mock_get_external_ip):
     n = 10
     d = Dendrite(wallet=get_mock_wallet())
     d.call = AsyncMock()
@@ -147,7 +150,7 @@ async def test_forward_many():
     assert len([resp]) == 1
 
 
-def test_pre_process_synapse():
+def test_pre_process_synapse(mock_get_external_ip):
     d = Dendrite(wallet=get_mock_wallet())
     s = Synapse()
     synapse = d.preprocess_synapse_for_request(

tests/unit_tests/test_subtensor.py

@@ -97,9 +97,7 @@ def test_serve_axon_with_external_ip_set():
     assert axon_info.ip == external_ip
 
 
-def test_serve_axon_with_external_port_set():
-    external_ip: str = "2001:0db8:85a3:0000:0000:8a2e:0370:7334"
-
+def test_serve_axon_with_external_port_set(mock_get_external_ip):
     internal_port: int = 1234
     external_port: int = 5678
 
@@ -134,14 +132,10 @@ def test_serve_axon_with_external_port_set():
         config=mock_config,
     )
 
-    with mock.patch(
-        "bittensor.utils.networking.get_external_ip", return_value=external_ip
-    ):
-        # mock the get_external_ip function to return the external ip
-        mock_subtensor.serve_axon(
-            netuid=-1,
-            axon=mock_axon_with_external_port_set,
-        )
+    mock_subtensor.serve_axon(
+        netuid=-1,
+        axon=mock_axon_with_external_port_set,
+    )
 
     mock_serve_axon.assert_called_once()
     # verify that the axon is served to the network with the external port
@@ -249,19 +243,6 @@ def test_determine_chain_endpoint_and_network(
     assert result_endpoint == expected_endpoint
 
 
-@pytest.fixture
-def subtensor(mocker):
-    fake_substrate = mocker.MagicMock()
-    fake_substrate.websocket.sock.getsockopt.return_value = 0
-    mocker.patch.object(
-        subtensor_module, "SubstrateInterface", return_value=fake_substrate
-    )
-    fake_websocket = mocker.MagicMock()
-    fake_websocket.client.connect.return_value = 0
-    mocker.patch.object(subtensor_module, "ws_client", return_value=fake_websocket)
-    return Subtensor()
-
-
 @pytest.fixture
 def mock_logger():
     with mock.patch.object(logging, "warning") as mock_warning:
@@ -1925,37 +1906,26 @@ def test_reveal_weights_false(subtensor, mocker):
     assert mocked_extrinsic.call_count == 5
 
 
-def test_connect_without_substrate(mocker):
-    """Ensure re-connection is called when using an alive substrate."""
+def test_connect_without_substrate(subtensor, websockets_client_connection, mocker):
+    """Ensure re-connection is called when using a disconnected substrate."""
     # Prep
-    fake_substrate = mocker.MagicMock()
-    fake_substrate.websocket.sock.getsockopt.return_value = 1
-    mocker.patch.object(
-        subtensor_module, "SubstrateInterface", return_value=fake_substrate
-    )
-    fake_subtensor = Subtensor()
+    websockets_client_connection.socket.getsockopt.return_value = 1
     spy_get_substrate = mocker.spy(Subtensor, "_get_substrate")
 
     # Call
-    _ = fake_subtensor.block
+    _ = subtensor.block
 
     # Assertions
     assert spy_get_substrate.call_count == 1
 
 
-def test_connect_with_substrate(mocker):
+def test_connect_with_substrate(subtensor, mocker):
     """Ensure re-connection is non called when using an alive substrate."""
     # Prep
-    fake_substrate = mocker.MagicMock()
-    fake_substrate.websocket.socket.getsockopt.return_value = 0
-    mocker.patch.object(
-        subtensor_module, "SubstrateInterface", return_value=fake_substrate
-    )
-    fake_subtensor = Subtensor()
     spy_get_substrate = mocker.spy(Subtensor, "_get_substrate")
 
     # Call
-    _ = fake_subtensor.block
+    _ = subtensor.block
 
     # Assertions
     assert spy_get_substrate.call_count == 0

tests/unit_tests/utils/test_networking.py

@@ -83,13 +83,42 @@ def test_int_to_ip6_underflow():
 
 
 # Test getting external IP address
-def test_get_external_ip():
+def test_get_external_ip(mocker):
     """Test getting the external IP address."""
-    assert utils.networking.get_external_ip()
+    mocked_requests_get = mock.Mock(
+        return_value=mock.Mock(
+            **{
+                "text": "192.168.1.1",
+            },
+        ),
+    )
+
+    mocker.patch.object(
+        requests,
+        "get",
+        mocked_requests_get,
+    )
+
+    assert utils.networking.get_external_ip() == "192.168.1.1"
+
+    mocked_requests_get.assert_called_once_with("https://checkip.amazonaws.com")
 
 
-def test_get_external_ip_os_broken():
+def test_get_external_ip_os_broken(mocker):
     """Test getting the external IP address when os.popen is broken."""
+    mocked_requests_get = mock.Mock(
+        return_value=mock.Mock(
+            **{
+                "text": "192.168.1.1",
+            },
+        ),
+    )
+
+    mocker.patch.object(
+        requests,
+        "get",
+        mocked_requests_get,
+    )
 
     class FakeReadline:
         def readline(self):
@@ -99,7 +128,9 @@ def mock_call():
         return FakeReadline()
 
     with mock.patch.object(os, "popen", new=mock_call):
-        assert utils.networking.get_external_ip()
+        assert utils.networking.get_external_ip() == "192.168.1.1"
+
+    mocked_requests_get.assert_called_once_with("https://checkip.amazonaws.com")
 
 
 def test_get_external_ip_os_request_urllib_broken():