E2E Test Patterns

tests/e2e_tests/conftest.py

@@ -0,0 +1,57 @@
+import os
+import signal
+from substrateinterface import SubstrateInterface
+import pytest
+import subprocess
+import logging
+import shlex
+import re
+import time
+
+logging.basicConfig(level=logging.INFO)
+
+
+# Fixture for setting up and tearing down a localnet.sh chain between tests
+@pytest.fixture(scope="function")
+def local_chain():
+    # Get the environment variable for the script path
+    script_path = os.getenv("LOCALNET_SH_PATH")
+
+    if not script_path:
+        # Skip the test if the localhost.sh path is not set
+        logging.warning("LOCALNET_SH_PATH env variable is not set, e2e test skipped.")
+        pytest.skip("LOCALNET_SH_PATH environment variable is not set.")
+
+    # Start new node process
+    cmds = shlex.split(script_path)
+    process = subprocess.Popen(
+        cmds, stdout=subprocess.PIPE, text=True, preexec_fn=os.setsid
+    )
+
+    # Pattern match indicates node is compiled and ready
+    pattern = re.compile(r"Successfully ran block step\.")
+
+    def wait_for_node_start(process, pattern):
+        for line in process.stdout:
+            print(line.strip())
+            if pattern.search(line):
+                print("Node started!")
+                break
+
+    wait_for_node_start(process, pattern)
+
+    # Run the test, passing in substrate interface
+    yield SubstrateInterface(url="ws://127.0.0.1:9945")
+
+    # Terminate the process group (includes all child processes)
+    os.killpg(os.getpgid(process.pid), signal.SIGTERM)
+
+    # Give some time for the process to terminate
+    time.sleep(1)
+
+    # If the process is not terminated, send SIGKILL
+    if process.poll() is None:
+        os.killpg(os.getpgid(process.pid), signal.SIGKILL)
+
+    # Ensure the process has terminated
+    process.wait()

tests/e2e_tests/multistep/test_last_tx_block.py

@@ -0,0 +1,51 @@
+from bittensor.commands.root import RootRegisterCommand
+from bittensor.commands.delegates import NominateCommand
+from bittensor.commands.network import RegisterSubnetworkCommand
+from bittensor.commands.register import RegisterCommand
+from ..utils import setup_wallet
+
+
+# Automated testing for take related tests described in
+# https://discord.com/channels/799672011265015819/1176889736636407808/1236057424134144152
+def test_takes(local_chain):
+    # Register root as Alice
+    (keypair, exec_command) = setup_wallet("//Alice")
+    exec_command(RootRegisterCommand, ["root", "register"])
+
+    # Create subnet 1 and verify created successfully
+    assert not (local_chain.query("SubtensorModule", "NetworksAdded", [1]).serialize())
+
+    exec_command(RegisterSubnetworkCommand, ["s", "create"])
+    assert local_chain.query("SubtensorModule", "NetworksAdded", [1])
+
+    assert local_chain.query("SubtensorModule", "NetworksAdded", [1]).serialize()
+
+    # Register and nominate Bob
+    (keypair, exec_command) = setup_wallet("//Bob")
+    assert (
+        local_chain.query(
+            "SubtensorModule", "LastTxBlock", [keypair.ss58_address]
+        ).serialize()
+        == 0
+    )
+
+    assert (
+        local_chain.query(
+            "SubtensorModule", "LastTxBlockDelegateTake", [keypair.ss58_address]
+        ).serialize()
+        == 0
+    )
+    exec_command(RegisterCommand, ["s", "register", "--neduid", "1"])
+    exec_command(NominateCommand, ["root", "nominate"])
+    assert (
+        local_chain.query(
+            "SubtensorModule", "LastTxBlock", [keypair.ss58_address]
+        ).serialize()
+        > 0
+    )
+    assert (
+        local_chain.query(
+            "SubtensorModule", "LastTxBlockDelegateTake", [keypair.ss58_address]
+        ).serialize()
+        > 0
+    )

tests/e2e_tests/subcommands/wallet/test_transfer.py

@@ -0,0 +1,32 @@
+from bittensor.commands.transfer import TransferCommand
+from ...utils import setup_wallet
+import bittensor
+
+
+# Example test using the local_chain fixture
+def test_transfer(local_chain):
+    (keypair, exec_command) = setup_wallet("//Alice")
+
+    acc_before = local_chain.query("System", "Account", [keypair.ss58_address])
+    exec_command(
+        TransferCommand,
+        [
+            "wallet",
+            "transfer",
+            "--amount",
+            "2",
+            "--dest",
+            "5GpzQgpiAKHMWNSH3RN4GLf96GVTDct9QxYEFAY7LWcVzTbx",
+        ],
+    )
+    acc_after = local_chain.query("System", "Account", [keypair.ss58_address])
+
+    expected_transfer = 2_000_000_000
+    tolerance = 200_000  # Tx fee tolerance
+
+    actual_difference = (
+        acc_before.value["data"]["free"] - acc_after.value["data"]["free"]
+    )
+    assert (
+        expected_transfer <= actual_difference <= expected_transfer + tolerance
+    ), f"Expected transfer with tolerance: {expected_transfer} <= {actual_difference} <= {expected_transfer + tolerance}"

tests/e2e_tests/utils.py

@@ -0,0 +1,32 @@
+from substrateinterface import Keypair
+from typing import List
+import bittensor
+
+
+def setup_wallet(uri: str):
+    keypair = Keypair.create_from_uri(uri)
+    wallet_path = "/tmp/btcli-e2e-wallet-{}".format(uri.strip("/"))
+    wallet = bittensor.wallet(path=wallet_path)
+    wallet.set_coldkey(keypair=keypair, encrypt=False, overwrite=True)
+    wallet.set_coldkeypub(keypair=keypair, encrypt=False, overwrite=True)
+    wallet.set_hotkey(keypair=keypair, encrypt=False, overwrite=True)
+
+    def exec_command(command, extra_args: List[str]):
+        parser = bittensor.cli.__create_parser__()
+        args = extra_args + [
+            "--no_prompt",
+            "--subtensor.network",
+            "local",
+            "--subtensor.chain_endpoint",
+            "ws://localhost:9945",
+            "--wallet.path",
+            wallet_path,
+        ]
+        config = bittensor.config(
+            parser=parser,
+            args=args,
+        )
+        cli_instance = bittensor.cli(config)
+        command.run(cli_instance)
+
+    return (keypair, exec_command)