test: add Issue 21 external signer E2E flow coverage

Author: johnny9

test/functional/qml_test_external_signer_flow.py

@@ -0,0 +1,363 @@
+#!/usr/bin/env python3
+# Copyright (c) 2026 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""End-to-end QML test for external signer PSBT workflow behavior.
+
+Coverage:
+1. External signer detection surfaces missing/multiple/error/available states.
+2. Failure -> retry transitions are recoverable via the refresh action.
+3. Externally signed PSBT import/broadcast path is validated in the GUI.
+"""
+
+import argparse
+import os
+import re
+import sys
+import tempfile
+from decimal import Decimal
+from pathlib import Path
+from urllib.parse import quote
+
+from qml_test_harness import QmlTestHarness, StepScreenshotRecorder, dump_qml_tree
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+BITCOIN_FUNCTIONAL_PATH = REPO_ROOT / "bitcoin" / "test" / "functional"
+if str(BITCOIN_FUNCTIONAL_PATH) not in sys.path:
+    sys.path.insert(0, str(BITCOIN_FUNCTIONAL_PATH))
+
+from test_framework.authproxy import AuthServiceProxy, JSONRPCException
+from test_framework.util import wait_until_helper_internal
+
+
+WALLET_NAME = "qml_external_signer_wallet"
+RECEIVER_WALLET_NAME = "qml_external_signer_receiver"
+SEND_NOTE = "qml-external-signer-e2e"
+SEND_AMOUNT_BTC = Decimal("0.00200000")
+
+
+MOCK_SIGNER_SCRIPT = """#!/usr/bin/env python3
+import argparse
+import json
+import pathlib
+import sys
+
+
+def read_state(path):
+    try:
+        return pathlib.Path(path).read_text(encoding=\"utf8\").strip()
+    except Exception:
+        return \"available\"
+
+
+def enumerate_signers(args):
+    state = read_state(args.state_file)
+    if state == \"missing\":
+        payload = []
+    elif state == \"multiple\":
+        payload = [
+            {\"fingerprint\": \"00000001\", \"type\": \"trezor\", \"model\": \"Mock Trezor\"},
+            {\"fingerprint\": \"00000002\", \"type\": \"ledger\", \"model\": \"Mock Ledger\"},
+        ]
+    elif state == \"error\":
+        payload = [{\"error\": \"Mock signer command failed\"}]
+    else:
+        payload = [{\"fingerprint\": \"00000001\", \"type\": \"trezor\", \"model\": \"Mock Trezor\"}]
+    sys.stdout.write(json.dumps(payload))
+
+
+def getdescriptors(args):
+    # This test only validates detect/import/broadcast UX; descriptor payload is
+    # intentionally minimal.
+    sys.stdout.write(json.dumps({\"receive\": [], \"internal\": []}))
+
+
+def signtx(args):
+    # Echo PSBT unchanged in this mock. The GUI flow under test imports a PSBT
+    # signed through RPC-side wallet tooling.
+    sys.stdout.write(json.dumps({\"psbt\": args.psbt, \"complete\": False}))
+
+
+parser = argparse.ArgumentParser(description=\"QML external-signer mock\")
+parser.add_argument(\"--fingerprint\")
+parser.add_argument(\"--chain\", default=\"main\")
+parser.add_argument(\"--stdin\", action=\"store_true\")
+parser.add_argument(\"--state-file\", required=True)
+
+subparsers = parser.add_subparsers(dest=\"command\")
+subparsers.required = True
+
+p_enum = subparsers.add_parser(\"enumerate\")
+p_enum.set_defaults(func=enumerate_signers)
+
+p_desc = subparsers.add_parser(\"getdescriptors\")
+p_desc.add_argument(\"--account\", metavar=\"account\")
+p_desc.set_defaults(func=getdescriptors)
+
+p_sign = subparsers.add_parser(\"signtx\")
+p_sign.add_argument(\"psbt\", metavar=\"psbt\")
+p_sign.set_defaults(func=signtx)
+
+if not sys.stdin.isatty():
+    buffer = sys.stdin.read()
+    if buffer and buffer.rstrip() != \"\":
+        sys.argv.extend(buffer.rstrip().split(\" \"))
+
+args = parser.parse_args()
+args.func(args)
+"""
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description="QML external signer flow test")
+    parser.add_argument(
+        "--socket-path",
+        help="Attach to an existing bridge socket. Not supported by this test.",
+    )
+    parser.add_argument(
+        "--screenshot-dir",
+        help="If set, save step screenshots (PNG) into this directory.",
+    )
+    return parser.parse_args()
+
+
+def build_rpc_url(user, password, port, wallet_name=None):
+    auth = f"{quote(user, safe='')}:{quote(password, safe='')}"
+    base = f"http://{auth}@127.0.0.1:{port}"
+    if wallet_name is None:
+        return base
+    return f"{base}/wallet/{quote(wallet_name, safe='')}"
+
+
+def read_cookie(datadir):
+    cookie_path = Path(datadir) / "regtest" / ".cookie"
+    wait_until_helper_internal(lambda: cookie_path.is_file(), timeout=60)
+    cookie = cookie_path.read_text(encoding="utf8").strip()
+    user, password = cookie.split(":", 1)
+    return user, password
+
+
+def connect_rpc(datadir, rpc_port):
+    user, password = read_cookie(datadir)
+    rpc = AuthServiceProxy(build_rpc_url(user, password, rpc_port), timeout=120)
+
+    def rpc_ready():
+        try:
+            return rpc.getblockchaininfo()["chain"] == "regtest"
+        except Exception:
+            return False
+
+    wait_until_helper_internal(rpc_ready, timeout=60)
+
+    wallet_rpc = AuthServiceProxy(
+        build_rpc_url(user, password, rpc_port, WALLET_NAME), timeout=120
+    )
+    receiver_wallet_rpc = AuthServiceProxy(
+        build_rpc_url(user, password, rpc_port, RECEIVER_WALLET_NAME), timeout=120
+    )
+    return rpc, wallet_rpc, receiver_wallet_rpc
+
+
+def ensure_wallet_loaded(rpc, wallet_name):
+    try:
+        rpc.createwallet(wallet_name)
+        return
+    except JSONRPCException as e:
+        if "already exists" not in str(e).lower():
+            raise
+    try:
+        rpc.loadwallet(wallet_name)
+    except JSONRPCException as e:
+        if "already loaded" not in str(e).lower():
+            raise
+
+
+def wait_for_gui_wallet_ready(gui):
+    gui.wait_for_property("walletBadge", "loading", timeout_ms=30000, value=False)
+    gui.wait_for_property("walletBadge", "noWalletLoaded", timeout_ms=30000, value=False)
+    gui.wait_for_property("walletBadge", "text", timeout_ms=30000, contains=WALLET_NAME)
+
+
+def select_wallet_tab(gui, tab_object_name, expected_page):
+    gui.click(tab_object_name)
+    gui.wait_for_page(expected_page, timeout_ms=15000)
+
+
+def send_amount_input_value(gui):
+    amount_max_length = int(gui.get_property("sendAmountInput", "maximumLength"))
+    if amount_max_length >= 17:
+        return str(SEND_AMOUNT_BTC)
+    return str(int(SEND_AMOUNT_BTC * Decimal("100000000")))
+
+
+def prepare_send_review(gui, destination, amount_input, note):
+    select_wallet_tab(gui, "walletSendTab", "walletSendPage")
+    gui.set_text("sendAddressInput", destination)
+    gui.set_text("sendAmountInput", amount_input)
+    gui.set_text("sendNoteInput", note)
+    gui.wait_for_property("sendContinueButton", "enabled", timeout_ms=15000, value=True)
+    gui.click("sendContinueButton")
+    gui.wait_for_page("walletSendReviewPage", timeout_ms=15000)
+
+
+def open_psbt_operations_page(gui):
+    gui.wait_for_page("walletSendPage", timeout_ms=15000)
+    gui.click("sendOptionsMenuButton")
+    gui.wait_for_property("sendOptionsPopup", "opened", timeout_ms=10000, value=True)
+    gui.click("sendOptionsPsbtOperationsButton")
+    gui.wait_for_page("psbtOperationsPage", timeout_ms=15000)
+
+
+def extract_txid(text):
+    match = re.search(r"([0-9a-fA-F]{64})", str(text))
+    if not match:
+        raise AssertionError(f"Could not parse txid from: {text!r}")
+    return match.group(1).lower()
+
+
+def write_signer_state(path, state):
+    Path(path).write_text(state, encoding="utf8")
+
+
+def run_tests():
+    args = parse_args()
+    if args.socket_path:
+        raise AssertionError(
+            "--socket-path is not supported for this test; it requires harness-managed RPC/datadir."
+        )
+
+    gui = None
+    harness = None
+    try:
+        with tempfile.TemporaryDirectory(prefix="qml_external_signer_mock_") as signer_tmp:
+            signer_script_path = Path(signer_tmp) / "switchable_signer.py"
+            signer_state_path = Path(signer_tmp) / "signer_state.txt"
+            signer_script_path.write_text(MOCK_SIGNER_SCRIPT, encoding="utf8")
+            os.chmod(signer_script_path, 0o755)
+            write_signer_state(signer_state_path, "missing")
+
+            signer_command = f"{sys.executable} {signer_script_path} --state-file={signer_state_path}"
+            harness = QmlTestHarness(
+                skip_onboard=True,
+                extra_args=["-fallbackfee=0.00001000", f"-signer={signer_command}"],
+            )
+            harness.start()
+            gui = harness.driver
+            shots = StepScreenshotRecorder(gui, args.screenshot_dir)
+
+            rpc, wallet_rpc, receiver_wallet_rpc = connect_rpc(harness.datadir, harness.rpc_port)
+            ensure_wallet_loaded(rpc, RECEIVER_WALLET_NAME)
+            ensure_wallet_loaded(rpc, WALLET_NAME)
+            wait_for_gui_wallet_ready(gui)
+            shots.capture("wallet_ready")
+
+            funding_addr = wallet_rpc.getnewaddress("qml-external-funding", "bech32")
+            rpc.generatetoaddress(101, funding_addr)
+
+            destination = receiver_wallet_rpc.getnewaddress("qml-external-destination", "bech32")
+            amount_input = send_amount_input_value(gui)
+
+            # Build unsigned PSBT from send review.
+            prepare_send_review(gui, destination, amount_input, SEND_NOTE)
+            gui.click("sendReviewCopyPsbtButton")
+            gui.wait_for_property(
+                "sendReviewPsbtStatusText", "text", timeout_ms=15000, contains="copied"
+            )
+            unsigned_psbt = gui.get_clipboard_text().strip()
+            assert unsigned_psbt.startswith("cHNidP"), "Unsigned PSBT copy did not produce base64 payload"
+            shots.capture("unsigned_psbt_copied")
+
+            # Simulate external device signing outside GUI.
+            signed_result = wallet_rpc.walletprocesspsbt(
+                psbt=unsigned_psbt, sign=True, sighashtype="ALL", bip32derivs=True
+            )
+            signed_psbt = signed_result["psbt"]
+            assert signed_psbt.startswith("cHNidP"), "Externally signed PSBT payload missing"
+
+            gui.click("sendReviewBackButton")
+            gui.wait_for_page("walletSendPage", timeout_ms=15000)
+            open_psbt_operations_page(gui)
+
+            # Detection state: missing (initial state file value).
+            missing_status = gui.wait_for_property(
+                "psbtExternalSignerStatusText",
+                "text",
+                timeout_ms=15000,
+                contains="No external signer detected",
+            )
+            assert "no external signer" in str(missing_status).lower(), missing_status
+
+            # Failure/retry branch: multiple -> error -> available.
+            write_signer_state(signer_state_path, "multiple")
+            gui.click("psbtRefreshSignersButton")
+            multiple_status = gui.wait_for_property(
+                "psbtExternalSignerStatusText",
+                "text",
+                timeout_ms=15000,
+                contains="Multiple external signers detected",
+            )
+            assert "multiple external signers" in str(multiple_status).lower(), multiple_status
+
+            write_signer_state(signer_state_path, "error")
+            gui.click("psbtRefreshSignersButton")
+            error_status = gui.wait_for_property(
+                "psbtExternalSignerStatusText",
+                "text",
+                timeout_ms=15000,
+                contains="Unable to detect external signer",
+            )
+            assert "unable to detect external signer" in str(error_status).lower(), error_status
+
+            write_signer_state(signer_state_path, "available")
+            gui.click("psbtRefreshSignersButton")
+            available_status = gui.wait_for_property(
+                "psbtExternalSignerStatusText",
+                "text",
+                timeout_ms=15000,
+                contains="External signer detected",
+            )
+            assert "mock trezor" in str(available_status).lower(), available_status
+            shots.capture("external_signer_recovered_available")
+
+            # Import externally signed PSBT and broadcast through GUI.
+            gui.set_clipboard_text(signed_psbt)
+            gui.click("psbtImportClipboardButton")
+            gui.wait_for_property("psbtStatusText", "text", timeout_ms=15000, contains="PSBT loaded")
+
+            if not bool(gui.get_property("psbtBroadcastButton", "enabled")):
+                if bool(gui.get_property("psbtSignButton", "enabled")):
+                    gui.click("psbtSignButton")
+                if not bool(gui.get_property("psbtBroadcastButton", "enabled")) and bool(
+                    gui.get_property("psbtFinalizeButton", "enabled")
+                ):
+                    gui.click("psbtFinalizeButton")
+
+            gui.wait_for_property("psbtBroadcastButton", "enabled", timeout_ms=15000, value=True)
+            gui.click("psbtBroadcastButton")
+            txid_label = gui.wait_for_property(
+                "psbtBroadcastTxidText", "text", timeout_ms=15000, non_empty=True
+            )
+            broadcast_txid = extract_txid(txid_label)
+            wait_until_helper_internal(lambda: broadcast_txid in rpc.getrawmempool(), timeout=30)
+            shots.capture("external_signer_psbt_broadcasted")
+
+            print("\n" + "=" * 60)
+            print("QML external signer flow E2E PASSED")
+            print(f"Broadcast txid: {broadcast_txid}")
+            print("=" * 60)
+
+    except Exception as e:
+        print(f"\nFAILED: {e}", file=sys.stderr)
+        import traceback
+
+        traceback.print_exc()
+        if gui is not None:
+            dump_qml_tree(gui)
+        sys.exit(1)
+    finally:
+        if harness is not None:
+            harness.stop()
+
+
+if __name__ == "__main__":
+    run_tests()