From bf7ac79ae16ed9b88cdceaa37156d032f1daf548 Mon Sep 17 00:00:00 2001 From: Liquid369 Date: Tue, 16 Jun 2026 07:39:42 +0700 Subject: [PATCH 1/4] Add explorer server storage, defaults and selection persistence Introduce an EXPLORER_SERVERS table with add/edit/remove/get helpers and seed the default explorers: explorer.duddino.com, testnet.duddino.com and zkbitcoin.com as a mainnet fallback. Defaults are seeded by URL so an upgrade never collides with the ids of explorers a user added previously, and NULL network metadata left behind by an older schema is backfilled. Remember the selected explorer per network (keyed by URL) in the settings cache so switching networks never remaps to an unrelated server. --- src/constants.py | 15 ++++ src/database.py | 203 ++++++++++++++++++++++++++++++++--------------- src/misc.py | 4 + 3 files changed, 157 insertions(+), 65 deletions(-) diff --git a/src/constants.py b/src/constants.py index 222b648..9752291 100644 --- a/src/constants.py +++ b/src/constants.py @@ -31,6 +31,11 @@ log_File = os.path.join(user_dir, 'debug.log') database_File = os.path.join(user_dir, 'application.db') +# Default explorers (url). zkbitcoin acts as a mainnet fallback if the primary +# explorer fails. Defined here so DefaultCache can reference them. +DEFAULT_MAINNET_EXPLORER = "https://explorer.duddino.com/" +DEFAULT_TESTNET_EXPLORER = "https://testnet.duddino.com/" + DefaultCache = { "lastAddress": "", "window_width": starting_width, @@ -38,6 +43,10 @@ "splitter_x": 342, "splitter_y": 133, "console_hidden": False, + # Selected explorer is stored per network, keyed by URL (stable across + # reordering), so switching networks never remaps to a different server. + "selectedExplorer_mainnet": DEFAULT_MAINNET_EXPLORER, + "selectedExplorer_testnet": DEFAULT_TESTNET_EXPLORER, "selectedHW_index": 0, "selectedRPC_index": 0, "isTestnetRPC": False, @@ -52,6 +61,12 @@ ["https", "latvia.fuzzbawls.pw:8080", "spmtUser", "8X88u7TuefPm7mQaJY52"], ["https", "charlotte.fuzzbawls.pw:8080", "spmtUser", "ZyD936tm9dvqmMP8A777"]] +# Default explorer rows (url, isTestnet, isCustom). +trusted_explorers = [ + [DEFAULT_MAINNET_EXPLORER, False, False], + [DEFAULT_TESTNET_EXPLORER, True, False], + ["https://zkbitcoin.com/", False, False] +] HW_devices = [ # (model name, api index) diff --git a/src/database.py b/src/database.py index 098ddd0..a99f0a9 100644 --- a/src/database.py +++ b/src/database.py @@ -1,22 +1,12 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -# Copyright (c) 2017-2019 Random.Zebra (https://github.com/random-zebra/) -# Distributed under the MIT software license, see the accompanying -# file LICENSE.txt or http://www.opensource.org/licenses/mit-license.php. - import logging import sqlite3 import threading -from constants import database_File, trusted_RPC_Servers +from constants import database_File, trusted_RPC_Servers, trusted_explorers from misc import printDbg, getCallerName, getFunctionName, printException class Database: - - ''' - class methods - ''' def __init__(self, app): printDbg("DB: Initializing...") self.app = app @@ -118,23 +108,45 @@ def initTables(self): # Tables for RPC Servers cursor.execute("CREATE TABLE IF NOT EXISTS PUBLIC_RPC_SERVERS(" - " id INTEGER PRIMARY KEY, protocol TEXT, host TEXT," - " user TEXT, pass TEXT)") + " id INTEGER PRIMARY KEY, protocol TEXT, host TEXT," + " user TEXT, pass TEXT)") cursor.execute("CREATE TABLE IF NOT EXISTS CUSTOM_RPC_SERVERS(" - " id INTEGER PRIMARY KEY, protocol TEXT, host TEXT," - " user TEXT, pass TEXT)") + " id INTEGER PRIMARY KEY, protocol TEXT, host TEXT," + " user TEXT, pass TEXT)") + + # Table for Explorers + cursor.execute("CREATE TABLE IF NOT EXISTS EXPLORER_SERVERS(" + " id INTEGER PRIMARY KEY, url TEXT, isTestnet BOOLEAN, isCustom BOOLEAN)") + + # Add the isTestnet and isCustom columns if they don't exist + try: + cursor.execute("ALTER TABLE EXPLORER_SERVERS ADD COLUMN isTestnet BOOLEAN") + except sqlite3.OperationalError as e: + if 'duplicate column name: isTestnet' in str(e): + pass # The column already exists + else: + raise + + try: + cursor.execute("ALTER TABLE EXPLORER_SERVERS ADD COLUMN isCustom BOOLEAN") + except sqlite3.OperationalError as e: + if 'duplicate column name: isCustom' in str(e): + pass # The column already exists + else: + raise self.initTable_RPC(cursor) + self.initTable_Explorer(cursor) # Tables for Utxos cursor.execute("CREATE TABLE IF NOT EXISTS UTXOS(" - " tx_hash TEXT, tx_ouput_n INTEGER, satoshis INTEGER, confirmations INTEGER," - " script TEXT, receiver TEXT, staker TEXT, coinstake BOOLEAN," - " PRIMARY KEY (tx_hash, tx_ouput_n))") + " tx_hash TEXT, tx_ouput_n INTEGER, satoshis INTEGER, confirmations INTEGER," + " script TEXT, receiver TEXT, staker TEXT, coinstake BOOLEAN," + " PRIMARY KEY (tx_hash, tx_ouput_n))") cursor.execute("CREATE TABLE IF NOT EXISTS RAWTXES(" - " tx_hash TEXT PRIMARY KEY, rawtx TEXT, lastfetch INTEGER)") + " tx_hash TEXT PRIMARY KEY, rawtx TEXT, lastfetch INTEGER)") printDbg("DB: Tables initialized") @@ -142,6 +154,27 @@ def initTables(self): err_msg = 'error initializing tables' printException(getCallerName(), getFunctionName(), err_msg, e.args) + def initTable_Explorer(self, cursor): + # Ensure each default explorer exists with correct network metadata. + # Keying on URL (rather than a hardcoded id) avoids colliding with the + # auto-assigned ids of explorers a user may have added previously, which + # would otherwise silently skip a new default such as the zkbitcoin + # mainnet fallback. + for url, isTestnet, isCustom in trusted_explorers: + # On databases upgraded from an older schema the ALTERs above may + # have added isTestnet/isCustom as NULL on a pre-existing default + # row. Backfill those (NULL only, so we never clobber a user's + # custom entry) so the network-filtered dropdown/fallback don't + # treat a NULL testnet default as mainnet (bool(None) == False). + cursor.execute("UPDATE EXPLORER_SERVERS SET isTestnet = ?, isCustom = ?" + " WHERE url = ? AND (isTestnet IS NULL OR isCustom IS NULL)", + (isTestnet, isCustom, url)) + # Insert the default if it isn't present yet. + cursor.execute("INSERT INTO EXPLORER_SERVERS (url, isTestnet, isCustom)" + " SELECT ?, ?, ?" + " WHERE NOT EXISTS (SELECT 1 FROM EXPLORER_SERVERS WHERE url = ?)", + (url, isTestnet, isCustom, url)) + def initTable_RPC(self, cursor): s = trusted_RPC_Servers # Insert Default public trusted servers @@ -285,12 +318,11 @@ def removeRPCServer(self, id): removed_RPC = False try: cursor = self.getCursor() - cursor.execute("DELETE FROM CUSTOM_RPC_SERVERS" - " WHERE id=?", (id,)) + cursor.execute("DELETE FROM CUSTOM_RPC_SERVERS WHERE id=?", (id,)) removed_RPC = True except Exception as e: - err_msg = 'error removing RPC servers from database' + err_msg = 'error removing RPC server from database' printException(getCallerName(), getFunctionName(), err_msg, e.args) finally: @@ -298,6 +330,82 @@ def removeRPCServer(self, id): if removed_RPC: self.app.sig_changed_rpcServers.emit() + ''' + Explorer servers methods + ''' + + def addExplorerServer(self, url, isTestnet): + printDbg("DB: Adding new Explorer server...") + try: + cursor = self.getCursor() + cursor.execute("INSERT OR IGNORE INTO EXPLORER_SERVERS (url, isTestnet, isCustom) VALUES (?, ?, ?)", + (url, isTestnet, True)) + printDbg("DB: Explorer server added or already exists") + + except Exception as e: + err_msg = 'error adding Explorer server entry to DB' + printException(getCallerName(), getFunctionName(), err_msg, e.args) + finally: + self.releaseCursor() + self.app.sig_ExplorerListReloaded.emit() + + def editExplorerServer(self, url, isTestnet, id): + printDbg("DB: Editing Explorer server with id %d" % id) + try: + cursor = self.getCursor() + cursor.execute("UPDATE EXPLORER_SERVERS SET url = ?, isTestnet = ? WHERE id = ?", + (url, isTestnet, id)) + + except Exception as e: + err_msg = 'error editing Explorer server entry to DB' + printException(getCallerName(), getFunctionName(), err_msg, e.args) + finally: + self.releaseCursor() + self.app.sig_ExplorerListReloaded.emit() + + def getExplorerServers(self, isTestnet=None): + tableName = "EXPLORER_SERVERS" + printDbg("DB: Getting Explorer servers from table %s" % tableName) + try: + cursor = self.getCursor() + if isTestnet is None: + cursor.execute("SELECT * FROM %s" % tableName) + else: + cursor.execute("SELECT * FROM %s WHERE isTestnet = ?" % tableName, (isTestnet,)) + rows = cursor.fetchall() + + except Exception as e: + err_msg = 'error getting Explorer servers from database' + printException(getCallerName(), getFunctionName(), err_msg, e.args) + rows = [] + finally: + self.releaseCursor() + + server_list = [] + for row in rows: + server = {} + server["id"] = row[0] + server["url"] = row[1] + server["isTestnet"] = row[2] + server["isCustom"] = row[3] + server_list.append(server) + + return server_list + + def removeExplorerServer(self, id): + printDbg("DB: Remove Explorer server with id %d" % id) + try: + cursor = self.getCursor() + cursor.execute("DELETE FROM EXPLORER_SERVERS WHERE id = ?", (id,)) + printDbg("DB: Explorer server removed") + + except Exception as e: + err_msg = 'error removing Explorer server from database' + printException(getCallerName(), getFunctionName(), err_msg, e.args) + finally: + self.releaseCursor(vacuum=True) + self.app.sig_ExplorerListReloaded.emit() + ''' UTXOS methods ''' @@ -306,7 +414,6 @@ def rewards_from_rows(self, rows): rewards = [] for row in rows: - # fetch masternode item utxo = {} utxo['txid'] = row[0] utxo['vout'] = row[1] @@ -316,7 +423,6 @@ def rewards_from_rows(self, rows): utxo['receiver'] = row[5] utxo['coinstake'] = row[6] utxo['staker'] = row[7] - # add to list rewards.append(utxo) return rewards @@ -325,17 +431,13 @@ def addReward(self, utxo): logging.debug("DB: Adding reward") try: cursor = self.getCursor() - cursor.execute("INSERT OR REPLACE INTO UTXOS " "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (utxo['txid'], utxo['vout'], utxo['satoshis'], utxo['confirmations'], - utxo['script'], utxo['receiver'], utxo['coinstake'], utxo['staker']) - ) - + utxo['script'], utxo['receiver'], utxo['coinstake'], utxo['staker'])) except Exception as e: err_msg = 'error adding reward UTXO to DB' printException(getCallerName(), getFunctionName(), err_msg, e) - finally: self.releaseCursor() @@ -344,7 +446,6 @@ def deleteReward(self, tx_hash, tx_ouput_n): try: cursor = self.getCursor() cursor.execute("DELETE FROM UTXOS WHERE tx_hash = ? AND tx_ouput_n = ?", (tx_hash, tx_ouput_n)) - except Exception as e: err_msg = 'error deleting UTXO from DB' printException(getCallerName(), getFunctionName(), err_msg, e.args) @@ -355,11 +456,8 @@ def getReward(self, tx_hash, tx_ouput_n): logging.debug("DB: Getting reward") try: cursor = self.getCursor() - - cursor.execute("SELECT * FROM UTXOS" - " WHERE tx_hash = ? AND tx_ouput_n = ?", (tx_hash, tx_ouput_n)) + cursor.execute("SELECT * FROM UTXOS WHERE tx_hash = ? AND tx_ouput_n = ?", (tx_hash, tx_ouput_n)) rows = cursor.fetchall() - except Exception as e: err_msg = 'error getting reward %s-%d' % (tx_hash, tx_ouput_n) printException(getCallerName(), getFunctionName(), err_msg, e) @@ -367,14 +465,13 @@ def getReward(self, tx_hash, tx_ouput_n): finally: self.releaseCursor() - if len(rows) > 0: + if rows: return self.rewards_from_rows(rows)[0] return None def getRewardsList(self, receiver=None): try: cursor = self.getCursor() - if receiver is None: printDbg("DB: Getting rewards of all masternodes") cursor.execute("SELECT * FROM UTXOS") @@ -382,14 +479,12 @@ def getRewardsList(self, receiver=None): printDbg("DB: Getting rewards of %s" % receiver) cursor.execute("SELECT * FROM UTXOS WHERE receiver = ?", (receiver,)) rows = cursor.fetchall() - except Exception as e: err_msg = 'error getting rewards list for %s' % receiver printException(getCallerName(), getFunctionName(), err_msg, e) rows = [] finally: self.releaseCursor() - return self.rewards_from_rows(rows) """ @@ -398,78 +493,56 @@ def getRewardsList(self, receiver=None): def txes_from_rows(self, rows): txes = [] - for row in rows: - # fetch tx item tx = {} tx['txid'] = row[0] tx['rawtx'] = row[1] - # add to list txes.append(tx) - return txes - def addRawTx(self, tx_hash, rawtx, lastfetch=0): logging.debug("DB: Adding rawtx for %s" % tx_hash) try: cursor = self.getCursor() - - cursor.execute("INSERT OR REPLACE INTO RAWTXES " - "VALUES (?, ?, ?)", - (tx_hash, rawtx, lastfetch) - ) - + cursor.execute("INSERT OR REPLACE INTO RAWTXES VALUES (?, ?, ?)", (tx_hash, rawtx, lastfetch)) except Exception as e: err_msg = 'error adding rawtx to DB' printException(getCallerName(), getFunctionName(), err_msg, e) - finally: self.releaseCursor() - def deleteRawTx(self, tx_hash): logging.debug("DB: Deleting rawtx for %s" % tx_hash) try: cursor = self.getCursor() - cursor.execute("DELETE FROM RAWTXES WHERE tx_hash = ?", (tx_hash, )) - + cursor.execute("DELETE FROM RAWTXES WHERE tx_hash = ?", (tx_hash,)) except Exception as e: err_msg = 'error deleting rawtx from DB' printException(getCallerName(), getFunctionName(), err_msg, e.args) finally: self.releaseCursor(vacuum=True) - def getRawTx(self, tx_hash): logging.debug("DB: Getting rawtx for %s" % tx_hash) try: cursor = self.getCursor() - - cursor.execute("SELECT * FROM RAWTXES" - " WHERE tx_hash = ?", (tx_hash, )) + cursor.execute("SELECT * FROM RAWTXES WHERE tx_hash = ?", (tx_hash,)) rows = cursor.fetchall() - except Exception as e: err_msg = 'error getting raw tx for %s' % tx_hash printException(getCallerName(), getFunctionName(), err_msg, e) rows = [] finally: self.releaseCursor() - - if len(rows) > 0: + if rows: return self.txes_from_rows(rows)[0] return None def clearRawTxes(self, minTime): - ''' - removes txes with lastfetch older than mintime - ''' printDbg("Pruning table RAWTXES") try: cursor = self.getCursor() - cursor.execute("DELETE FROM RAWTXES WHERE lastfetch < ?", (minTime, )) - + cursor.execute("DELETE FROM RAWTXES WHERE lastfetch < ?", (minTime,)) except Exception as e: err_msg = 'error deleting rawtx from DB' printException(getCallerName(), getFunctionName(), err_msg, e.args) diff --git a/src/misc.py b/src/misc.py index 6e377ec..437dee6 100644 --- a/src/misc.py +++ b/src/misc.py @@ -283,6 +283,8 @@ def readCacheSettings(): cache["console_hidden"] = settings.value('cache_consoleHidden', DefaultCache["console_hidden"], type=bool) cache["selectedHW_index"] = settings.value('cache_HWindex', DefaultCache["selectedHW_index"], type=int) cache["selectedRPC_index"] = settings.value('cache_RPCindex', DefaultCache["selectedRPC_index"], type=int) + cache["selectedExplorer_mainnet"] = settings.value('cache_ExplorerMainnet', DefaultCache["selectedExplorer_mainnet"], type=str) + cache["selectedExplorer_testnet"] = settings.value('cache_ExplorerTestnet', DefaultCache["selectedExplorer_testnet"], type=str) cache["isTestnetRPC"] = settings.value('cache_isTestnetRPC', DefaultCache["isTestnetRPC"], type=bool) cache["hwAcc"] = settings.value('cache_hwAcc', DefaultCache["hwAcc"], type=int) cache["spathFrom"] = settings.value('cache_spathFrom', DefaultCache["spathFrom"], type=int) @@ -309,6 +311,8 @@ def saveCacheSettings(cache): settings.setValue('cache_consoleHidden', cache.get('console_hidden')) settings.setValue('cache_HWindex', cache.get('selectedHW_index')) settings.setValue('cache_RPCindex', cache.get('selectedRPC_index')) + settings.setValue('cache_ExplorerMainnet', cache.get('selectedExplorer_mainnet')) + settings.setValue('cache_ExplorerTestnet', cache.get('selectedExplorer_testnet')) settings.setValue('cache_isTestnetRPC', cache.get('isTestnetRPC')) settings.setValue('cache_hwAcc', cache.get('hwAcc')) settings.setValue('cache_spathFrom', cache.get('spathFrom')) From c3d1e59416211acbc71e751e01f8db0cf96d03de Mon Sep 17 00:00:00 2001 From: Liquid369 Date: Tue, 16 Jun 2026 07:39:50 +0700 Subject: [PATCH 2/4] Drive block explorer requests from the selected explorer with fallback BlockBookClient now resolves its URL from the active explorer selection. When the primary explorer fails, the request is retried against the other explorers configured for the active network (e.g. zkbitcoin) before the error propagates. ApiClient exposes updateExplorerUrl() to switch the active explorer at runtime. --- src/apiClient.py | 14 +++++++--- src/blockbookClient.py | 60 +++++++++++++++++++++++++++--------------- 2 files changed, 49 insertions(+), 25 deletions(-) diff --git a/src/apiClient.py b/src/apiClient.py index 2205db8..b304b08 100644 --- a/src/apiClient.py +++ b/src/apiClient.py @@ -7,7 +7,7 @@ from blockbookClient import BlockBookClient from cryptoIDClient import CryptoIDClient -from misc import getCallerName, getFunctionName, printException, printError +from misc import getCallerName, getFunctionName, printException, printError, printDbg def process_api_exceptions(func): @@ -31,9 +31,15 @@ def process_api_exceptions_int(*args, **kwargs): class ApiClient: - def __init__(self, isTestnet=False): - self.isTestnet = isTestnet - self.api = BlockBookClient(isTestnet) + def __init__(self, main_wnd): + self.main_wnd = main_wnd + self.isTestnet = main_wnd.isTestnetRPC + self.api = BlockBookClient(main_wnd, self.isTestnet) + + def updateExplorerUrl(self, new_url): + # Update the explorer URL in the BlockBookClient instance + printDbg(f"Updating explorer URL to: {new_url}") + self.api.updateBaseUrl(new_url) @process_api_exceptions def getAddressUtxos(self, address): diff --git a/src/blockbookClient.py b/src/blockbookClient.py index 4348542..6eea998 100644 --- a/src/blockbookClient.py +++ b/src/blockbookClient.py @@ -5,8 +5,7 @@ # file LICENSE.txt or http://www.opensource.org/licenses/mit-license.php. import requests - -from misc import getCallerName, getFunctionName, printException +from misc import getCallerName, getFunctionName, printException, printDbg def process_blockbook_exceptions(func): @@ -15,33 +14,53 @@ def process_blockbook_exceptions_int(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: - if client.isTestnet: - new_url = "https://testnet.fuzzbawls.pw" - else: - new_url = "https://zkbitcoin.com/" - message = "BlockBook Client exception on %s\nTrying backup server %s" % (client.url, new_url) + message = "BlockBook Client exception on %s" % client.url printException(getCallerName(True), getFunctionName(True), message, str(e)) - try: - client.url = new_url - return func(*args, **kwargs) - - except Exception: - raise + # Primary explorer failed: retry against the other explorers + # configured for this network (e.g. zkbitcoin) before giving up. + for new_url in client.getFallbackUrls(): + printDbg("Trying backup explorer %s" % new_url) + try: + client.url = new_url + return func(*args, **kwargs) + except Exception: + continue + # All explorers failed: re-raise so ApiClient can fall back further. + raise return process_blockbook_exceptions_int class BlockBookClient: - - def __init__(self, isTestnet=False): + def __init__(self, main_wnd, isTestnet=False): + self.main_wnd = main_wnd self.isTestnet = isTestnet - if isTestnet: - self.url = "https://testnet.rockdev.org/" - else: - self.url = "https://explorer.rockdev.org/" + self.url = "" + self.loadURL() + + def network(self): + return 'testnet' if self.isTestnet else 'mainnet' + + def loadURL(self): + self.url = self.main_wnd.getExplorerURL(self.network()) + printDbg(f"Using Explorer URL: {self.url}") + + def getFallbackUrls(self): + # Return the other explorer URLs for this network (current one excluded). + try: + urls = self.main_wnd.getExplorerURLList(self.network()) + except Exception: + return [] + return [u for u in urls if u != self.url] + + def updateBaseUrl(self, new_url): + # Update the explorer URL + self.url = new_url + printDbg(f"Explorer URL updated to: {self.url}") def checkResponse(self, method, param=""): - url = self.url + "/api/%s" % method + # rstrip avoids a double slash when the URL already ends with '/' + url = self.url.rstrip('/') + "/api/%s" % method if param != "": url += "/%s" % param resp = requests.get(url, data={}, verify=True) @@ -53,7 +72,6 @@ def checkResponse(self, method, param=""): @process_blockbook_exceptions def getAddressUtxos(self, address): utxos = self.checkResponse("utxo", address) - # Add script for cryptoID legacy for u in utxos: u["script"] = "" return utxos From 84a8a59bc142ea83830410afb6054ea37cfae72f Mon Sep 17 00:00:00 2001 From: Liquid369 Date: Tue, 16 Jun 2026 07:39:59 +0700 Subject: [PATCH 3/4] Add explorer selection dropdown and configuration dialog Add an Explorer dropdown to the header, filtered to the active network, and an 'Explorer Servers config...' dialog to add, edit and remove custom explorers, mirroring the RPC servers UI. The selection is remembered per network and applied to the API client, and the dropdown refreshes when the RPC network changes. --- pet4l.py | 10 +- src/mainApp.py | 13 ++- src/mainWindow.py | 119 +++++++++++++++++++- src/qt/dlg_configureExplorer.py | 186 ++++++++++++++++++++++++++++++++ src/qt/guiHeader.py | 12 +++ 5 files changed, 328 insertions(+), 12 deletions(-) create mode 100644 src/qt/dlg_configureExplorer.py diff --git a/pet4l.py b/pet4l.py index 0c27a7e..8830426 100644 --- a/pet4l.py +++ b/pet4l.py @@ -33,18 +33,16 @@ from PyQt5.QtWidgets import QApplication from mainApp import App - # Create App + # Create QApplication app = QApplication(sys.argv) - # -------------- - # Create QMainWindow Widget ex = App(imgDir, app, args) - # -- Launch RPC watchdog + # Launch RPC watchdog ex.mainWindow.rpc_watchdogThread.start() - # Execute App + # Execute the application app.exec_() try: app.deleteLater() @@ -52,5 +50,3 @@ print(e) sys.exit() - - diff --git a/src/mainApp.py b/src/mainApp.py index e0b9419..7b1d8e0 100644 --- a/src/mainApp.py +++ b/src/mainApp.py @@ -19,7 +19,7 @@ from constants import user_dir, SECONDS_IN_2_MONTHS from qt.dlg_configureRPCservers import ConfigureRPCservers_dlg from qt.dlg_signmessage import SignMessage_dlg - +from qt.dlg_configureExplorer import ConfigureExplorerServers_dlg class ServiceExit(Exception): """ @@ -37,6 +37,8 @@ def service_shutdown(signum, frame): class App(QMainWindow): # Signal emitted from database sig_changed_rpcServers = pyqtSignal() + # Signal: Explorer list has been reloaded (emitted by DB) + sig_ExplorerListReloaded = pyqtSignal() def __init__(self, imgDir, app, start_args): # Create the userdir if it doesn't exist @@ -99,6 +101,9 @@ def initUI(self, imgDir): self.rpcConfMenu = QAction(self.pivx_icon, 'RPC Servers config...', self) self.rpcConfMenu.triggered.connect(self.onEditRPCServer) confMenu.addAction(self.rpcConfMenu) + self.explorerConfMenu = QAction(self.pivx_icon, 'Explorer Servers config...', self) + self.explorerConfMenu.triggered.connect(self.onEditExplorerServer) + confMenu.addAction(self.explorerConfMenu) toolsMenu = mainMenu.addMenu('Tools') self.signVerifyAction = QAction('Sign/Verify message', self) self.signVerifyAction.triggered.connect(self.onSignVerifyMessage) @@ -141,6 +146,12 @@ def onEditRPCServer(self): if ui.exec(): printDbg("Configuring RPC Servers...") + def onEditExplorerServer(self): + # Create Dialog + ui = ConfigureExplorerServers_dlg(self) + if ui.exec(): + printDbg("Configuring Explorer Servers...") + def onSignVerifyMessage(self): # Create Dialog ui = SignMessage_dlg(self.mainWindow) diff --git a/src/mainWindow.py b/src/mainWindow.py index 727b793..1b96a77 100644 --- a/src/mainWindow.py +++ b/src/mainWindow.py @@ -16,7 +16,8 @@ QFileDialog, QTextEdit, QTabWidget, QLabel, QSplitter from apiClient import ApiClient -from constants import starting_height, DefaultCache, wqueue +from constants import starting_height, DefaultCache, wqueue, \ + DEFAULT_MAINNET_EXPLORER, DEFAULT_TESTNET_EXPLORER from hwdevice import HWdevice from misc import printDbg, printException, printOK, getCallerName, getFunctionName, \ WriteStreamReceiver, now, persistCacheSetting, myPopUp_sb, getRemotePET4Lversion @@ -41,7 +42,6 @@ class MainWindow(QWidget): # signal: UTXO list loading percent (emitted by load_utxos_thread in tabRewards) sig_UTXOsLoading = pyqtSignal(int) - def __init__(self, parent, imgDir): super(QWidget, self).__init__(parent) self.parent = parent @@ -56,6 +56,7 @@ def __init__(self, parent, imgDir): self.rpcClient = None self.rpcConnected = False self.updatingRPCbox = False + self.updatingExplorerbox = False self.rpcStatusMess = "Not Connected" self.isBlockchainSynced = False # Changes when an RPC client is connected (affecting API client) @@ -71,6 +72,9 @@ def __init__(self, parent, imgDir): # -- Load RPC Servers list (init selection and self.isTestnet) self.updateRPClist() + # -- Load Explorer Servers list + self.explorerServersList = [] + self.updateExplorerList() # -- Init HW selection self.header.hwDevices.setCurrentIndex(self.parent.cache['selectedHW_index']) @@ -79,7 +83,7 @@ def __init__(self, parent, imgDir): self.hwdevice = HWdevice(self) # -- init Api Client - self.apiClient = ApiClient(self.isTestnetRPC) + self.apiClient = ApiClient(self) # -- Create Queue to redirect stdout self.queue = wqueue @@ -158,10 +162,12 @@ def connButtons(self): self.header.button_checkHw.clicked.connect(lambda: self.onCheckHw()) self.header.rpcClientsBox.currentIndexChanged.connect(self.onChangeSelectedRPC) self.header.hwDevices.currentIndexChanged.connect(self.onChangeSelectedHW) + self.header.explorerClientsBox.currentIndexChanged.connect(self.onChangeSelectedExplorer) # -- Connect signals self.sig_clearRPCstatus.connect(self.clearRPCstatus) self.sig_RPCstatusUpdated.connect(self.showRPCstatus) self.parent.sig_changed_rpcServers.connect(self.updateRPClist) + self.parent.sig_ExplorerListReloaded.connect(self.updateExplorerList) def getRPCserver(self): itemData = self.header.rpcClientsBox.itemData(self.header.rpcClientsBox.currentIndex()) @@ -425,6 +431,105 @@ def updateRPClist(self): # reload servers in configure dialog self.sig_RPClistReloaded.emit() + def explorerCacheKey(self): + # Selection is remembered per network so switching networks never + # remaps to an unrelated explorer. + return 'selectedExplorer_testnet' if self.isTestnetRPC else 'selectedExplorer_mainnet' + + def explorerSettingsKey(self): + return 'cache_ExplorerTestnet' if self.isTestnetRPC else 'cache_ExplorerMainnet' + + def setSelectedExplorer(self, url): + # Persist the selected explorer URL for the active network. + self.parent.cache[self.explorerCacheKey()] = persistCacheSetting(self.explorerSettingsKey(), url) + + def updateExplorerList(self): + # Full list (both networks) backs the configuration dialog... + self.explorerServersList = self.parent.db.getExplorerServers() + # ...while the header dropdown only offers explorers for the active + # network, so a testnet explorer isn't a selectable no-op on mainnet. + network_explorers = [e for e in self.explorerServersList + if bool(e['isTestnet']) == self.isTestnetRPC] + + # Repopulate the explorer box. Guard so that the programmatic + # clear()/addItem() calls don't fire onChangeSelectedExplorer. + self.updatingExplorerbox = True + self.header.explorerClientsBox.clear() + for explorer in network_explorers: + self.header.explorerClientsBox.addItem(explorer["url"], explorer) + + # Restore the selection saved for THIS network, matched by URL (the + # combo's item text). Indices are per-network here, so a saved URL is + # the only stable handle across networks and reordering. + saved_url = self.parent.cache.get(self.explorerCacheKey()) + index = self.header.explorerClientsBox.findText(saved_url) if saved_url else -1 + if index < 0: + index = 0 # saved explorer no longer available -> first for network + self.header.explorerClientsBox.setCurrentIndex(index) + self.updatingExplorerbox = False + + # Persist whatever ended up selected so the cache reflects reality, then + # sync the api client. We do this explicitly because the guard above + # swallowed the currentIndexChanged signal. + selected_explorer = self.header.explorerClientsBox.currentData() + if selected_explorer: + self.setSelectedExplorer(selected_explorer['url']) + self.applySelectedExplorer() + + def applySelectedExplorer(self): + selected_explorer = self.header.explorerClientsBox.currentData() + if selected_explorer: + url = selected_explorer['url'] + self.header.activeExplorerLabel.setText("Active Explorer: %s" % url) + else: + # No explorer configured for this network: fall back to the default. + network = 'testnet' if self.isTestnetRPC else 'mainnet' + url = self.getExplorerURL(network) + self.header.activeExplorerLabel.setText("Active Explorer: None") + printDbg("Active Explorer URL: %s" % url) + if getattr(self, 'apiClient', None) is not None: + self.apiClient.updateExplorerUrl(url) + + def onChangeSelectedExplorer(self, i): + # Don't react while we are programmatically repopulating the box + if self.updatingExplorerbox: + return + + selected_explorer = self.header.explorerClientsBox.itemData(i) + if selected_explorer: + explorer_url = selected_explorer.get('url', '') + # Persist the new selection for the active network + self.setSelectedExplorer(explorer_url) + # Point the api client at the newly selected explorer + if getattr(self, 'apiClient', None) is not None: + self.apiClient.updateExplorerUrl(explorer_url) + printDbg("Explorer changed to: %s" % explorer_url) + self.header.activeExplorerLabel.setText("Active Explorer: %s" % explorer_url) + else: + printDbg("No explorer selected") + self.header.activeExplorerLabel.setText("Active Explorer: None") + + def getExplorerURLList(self, network): + # All configured explorer URLs for the given network, defaults if none. + isTestnet = (network == 'testnet') + urls = [e['url'] for e in self.explorerServersList if bool(e['isTestnet']) == isTestnet] + if not urls: + printDbg("No explorers configured for %s, using default." % network) + urls = [DEFAULT_TESTNET_EXPLORER if isTestnet else DEFAULT_MAINNET_EXPLORER] + return urls + + def getExplorerURL(self, network): + # Honour the persisted per-network selection, falling back to the first + # explorer for that network. Reads only cache/list state (no Qt widget), + # so it is safe to call from the RPC worker thread when ApiClient is + # rebuilt on a network switch. + cache_key = 'selectedExplorer_testnet' if network == 'testnet' else 'selectedExplorer_mainnet' + saved_url = self.parent.cache.get(cache_key) + urls = self.getExplorerURLList(network) + if saved_url and saved_url in urls: + return saved_url + return urls[0] + def updateRPCstatus(self, ctrl, fDebug=False): rpc_index, rpc_protocol, rpc_host, rpc_user, rpc_password = self.getRPCserver() if fDebug: @@ -449,6 +554,7 @@ def updateRPCstatus(self, ctrl, fDebug=False): if rpc_index != self.header.rpcClientsBox.currentIndex(): return + networkChanged = False with self.lock: self.rpcClient = rpcClient self.rpcConnected = status @@ -460,5 +566,10 @@ def updateRPCstatus(self, ctrl, fDebug=False): if isTestnet != self.isTestnetRPC: self.isTestnetRPC = isTestnet self.parent.cache['isTestnetRPC'] = persistCacheSetting('isTestnetRPC', isTestnet) - self.apiClient = ApiClient(isTestnet) + self.apiClient = ApiClient(self) + networkChanged = True self.sig_RPCstatusUpdated.emit(rpc_index, fDebug) + # We are on a worker thread here: refresh the explorer dropdown for the + # new network via the (queued) signal so it runs on the GUI thread. + if networkChanged: + self.parent.sig_ExplorerListReloaded.emit() diff --git a/src/qt/dlg_configureExplorer.py b/src/qt/dlg_configureExplorer.py new file mode 100644 index 0000000..f8513c4 --- /dev/null +++ b/src/qt/dlg_configureExplorer.py @@ -0,0 +1,186 @@ +from PyQt5.QtWidgets import QDialog, QHBoxLayout, QVBoxLayout, QLabel, \ + QListWidget, QFrame, QFormLayout, QComboBox, QLineEdit, QListWidgetItem, \ + QWidget, QPushButton, QMessageBox + +from misc import myPopUp + + +class ConfigureExplorerServers_dlg(QDialog): + def __init__(self, main_wnd): + super().__init__(parent=main_wnd) + self.main_wnd = main_wnd.mainWindow + self.setWindowTitle('Explorer Servers Configuration') + self.changing_index = None + self.initUI() + self.loadServers() + self.main_wnd.parent.sig_ExplorerListReloaded.connect(self.loadServers) + + def clearEditFrame(self): + self.ui.url_edt.clear() + self.ui.network_select.setCurrentIndex(0) + + def initUI(self): + self.ui = Ui_ConfigureExplorerServersDlg() + self.ui.setupUi(self) + + def insert_server_list(self, server, index): + id = server['id'] + server_line = QWidget() + server_row = QHBoxLayout() + server_text = "%s" % server['url'] + if server['isCustom'] is None: + server['isCustom'] = False + if not server['isCustom']: + server_text = "%s" % server_text + server_row.addWidget(QLabel(server_text)) + server_row.addStretch(1) + # -- Edit button + editBtn = QPushButton() + editBtn.setIcon(self.main_wnd.editMN_icon) + editBtn.setToolTip("Edit server configuration") + editBtn.setEnabled(server['isCustom']) + if not server['isCustom']: + editBtn.setToolTip('Default servers are not editable') + editBtn.clicked.connect(lambda: self.onAddServer(index)) + server_row.addWidget(editBtn) + # -- Remove button + removeBtn = QPushButton() + removeBtn.setIcon(self.main_wnd.removeMN_icon) + removeBtn.setToolTip("Remove server configuration") + removeBtn.setEnabled(server['isCustom']) + if not server['isCustom']: + removeBtn.setToolTip('Cannot remove default servers') + removeBtn.clicked.connect(lambda: self.onRemoveServer(index)) + server_row.addWidget(removeBtn) + # -- + server_line.setLayout(server_row) + self.serverItems[id] = QListWidgetItem() + self.serverItems[id].setSizeHint(server_line.sizeHint()) + self.ui.serversBox.addItem(self.serverItems[id]) + self.ui.serversBox.setItemWidget(self.serverItems[id], server_line) + + def loadServers(self): + # Clear serversBox + self.ui.serversBox.clear() + # Fill serversBox. Index by list position (explorerServersList holds + # both networks; the header dropdown is filtered, so its indices differ). + self.serverItems = {} + for index, server in enumerate(self.main_wnd.explorerServersList): + self.insert_server_list(server, index) + + def loadEditFrame(self, index): + server = self.main_wnd.explorerServersList[index] + self.ui.url_edt.setText(server['url']) + if server['isTestnet']: + self.ui.network_select.setCurrentIndex(1) + else: + self.ui.network_select.setCurrentIndex(0) + + def onAddServer(self, index=None): + # Save current index (None for new entry) + self.changing_index = index + # Hide 'Add' and 'Close' buttons and disable serversBox + self.ui.addServer_btn.hide() + self.ui.close_btn.hide() + self.ui.serversBox.setEnabled(False) + # Show edit-frame + self.ui.editFrame.setHidden(False) + # If we are adding a new server, clear edit-frame + if index is None: + self.clearEditFrame() + # else pre-load data + else: + self.loadEditFrame(index) + + def onCancel(self): + # Show 'Add' and 'Close' buttons and enable serversBox + self.ui.addServer_btn.show() + self.ui.close_btn.show() + self.ui.serversBox.setEnabled(True) + # Hide edit-frame + self.ui.editFrame.setHidden(True) + # Clear edit-frame + self.clearEditFrame() + + def onClose(self): + # close dialog + self.close() + + def onRemoveServer(self, index): + mess = "Are you sure you want to remove server with index %d (%s) from list?" % ( + index, self.main_wnd.explorerServersList[index].get('url')) + ans = myPopUp(self, QMessageBox.Question, 'PET4L - remove server', mess) + if ans == QMessageBox.Yes: + # Remove entry from database + id = self.main_wnd.explorerServersList[index].get('id') + self.main_wnd.db.removeExplorerServer(id) + # Reload explorer list in main window + self.main_wnd.updateExplorerList() + self.loadServers() + + def onSave(self): + # Get new config data + url = self.ui.url_edt.text() + isTestnet = self.ui.network_select.currentIndex() == 1 + if self.changing_index is None: + self.main_wnd.db.addExplorerServer(url, isTestnet) + else: + # Edit existing entry in DB. + id = self.main_wnd.explorerServersList[self.changing_index].get('id') + self.main_wnd.db.editExplorerServer(url, isTestnet, id) + + # Reload explorer list in main window + self.main_wnd.updateExplorerList() + self.loadServers() + + # call onCancel + self.onCancel() + + +class Ui_ConfigureExplorerServersDlg(object): + def setupUi(self, ConfigureExplorerServersDlg): + ConfigureExplorerServersDlg.setModal(True) + # -- Layout + self.layout = QVBoxLayout(ConfigureExplorerServersDlg) + self.layout.setSpacing(10) + # -- Servers List + self.serversBox = QListWidget() + self.layout.addWidget(self.serversBox) + # -- 'Add Server' button + self.addServer_btn = QPushButton("Add Explorer Server") + self.layout.addWidget(self.addServer_btn) + # -- 'Close' button + hBox = QHBoxLayout() + hBox.addStretch(1) + self.close_btn = QPushButton("Close") + hBox.addWidget(self.close_btn) + self.layout.addLayout(hBox) + # -- Edit section + self.editFrame = QFrame() + frameLayout = QFormLayout() + frameLayout.setFieldGrowthPolicy(QFormLayout.AllNonFixedFieldsGrow) + frameLayout.setContentsMargins(5, 10, 5, 5) + frameLayout.setSpacing(7) + self.url_edt = QLineEdit() + frameLayout.addRow(QLabel("URL"), self.url_edt) + hBox = QHBoxLayout() + self.network_select = QComboBox() + self.network_select.addItems(['Mainnet', 'Testnet']) + hBox.addWidget(self.network_select) + frameLayout.addRow(QLabel("Network"), hBox) + hBox2 = QHBoxLayout() + self.cancel_btn = QPushButton("Cancel") + self.save_btn = QPushButton("Save") + hBox2.addWidget(self.cancel_btn) + hBox2.addWidget(self.save_btn) + frameLayout.addRow(hBox2) + self.editFrame.setLayout(frameLayout) + self.layout.addWidget(self.editFrame) + self.editFrame.setHidden(True) + ConfigureExplorerServersDlg.setMinimumWidth(500) + ConfigureExplorerServersDlg.setMinimumHeight(500) + # Connect main buttons + self.addServer_btn.clicked.connect(lambda: ConfigureExplorerServersDlg.onAddServer()) + self.close_btn.clicked.connect(lambda: ConfigureExplorerServersDlg.onClose()) + self.cancel_btn.clicked.connect(lambda: ConfigureExplorerServersDlg.onCancel()) + self.save_btn.clicked.connect(lambda: ConfigureExplorerServersDlg.onSave()) diff --git a/src/qt/guiHeader.py b/src/qt/guiHeader.py index fdaf92e..d000b9d 100644 --- a/src/qt/guiHeader.py +++ b/src/qt/guiHeader.py @@ -5,6 +5,7 @@ # file LICENSE.txt or http://www.opensource.org/licenses/mit-license.php. from PyQt5.QtWidgets import QPushButton, QLabel, QGridLayout, QHBoxLayout, QComboBox, QWidget +from PyQt5.QtCore import Qt from constants import HW_devices from PyQt5.Qt import QSizePolicy @@ -70,4 +71,15 @@ def __init__(self, caller, *args, **kwargs): self.centralBox.addWidget(self.hwLed, 1, 3) layout.addLayout(self.centralBox) layout.addStretch(1) + # Explorer Clients Box + label4 = QLabel("Explorer") + self.centralBox.addWidget(label4, 2, 0) + self.explorerClientsBox = QComboBox() + self.explorerClientsBox.setToolTip("Select Explorer Server") + self.centralBox.addWidget(self.explorerClientsBox, 2, 1) + # (currentIndexChanged is connected in MainWindow.connButtons) + # Active Explorer Label + self.activeExplorerLabel = QLabel("Active Explorer: None") + self.activeExplorerLabel.setTextFormat(Qt.RichText) + layout.addWidget(self.activeExplorerLabel) self.setLayout(layout) From b8a7b54c36c0bb2969f74a64c9bad07b664e61af Mon Sep 17 00:00:00 2001 From: Liquid369 Date: Tue, 16 Jun 2026 07:40:08 +0700 Subject: [PATCH 4/4] Add explorer customization tests Cover the explorer fallback retry, per-network selection persistence, network filtering of the dropdown, and the database seeding and upgrade migration. Ignore __pycache__ directories anywhere in the tree. --- .gitignore | 2 +- tests/test_explorer_change.py | 291 ++++++++++++++++++++++++++++++++++ 2 files changed, 292 insertions(+), 1 deletion(-) create mode 100644 tests/test_explorer_change.py diff --git a/.gitignore b/.gitignore index ec6007b..13cb233 100644 --- a/.gitignore +++ b/.gitignore @@ -4,7 +4,7 @@ /.pytest_cache/ /.idea/ /.settings/ -/__pycache__/ +__pycache__/ # Virtual environments /venv*/ diff --git a/tests/test_explorer_change.py b/tests/test_explorer_change.py new file mode 100644 index 0000000..3ef7a37 --- /dev/null +++ b/tests/test_explorer_change.py @@ -0,0 +1,291 @@ +import os +import sys +import unittest +from unittest.mock import MagicMock + +# The app runs with `src/` on sys.path (see pet4l.py), so its modules import +# each other flat (e.g. `from apiClient import ApiClient`). Mirror that here. +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) + +from blockbookClient import BlockBookClient # noqa: E402 + +# MainWindow pulls in the hardware-wallet stack (btchip/trezor), which may not +# be installed in a CI/headless environment. Import it lazily so the explorer +# fallback tests can still run on their own. +try: + import mainWindow as mainWindow_mod # noqa: E402 + from mainWindow import MainWindow # noqa: E402 + MAINWINDOW_AVAILABLE = True +except Exception: + MAINWINDOW_AVAILABLE = False + + +class FakeMainWnd: + """Minimal stand-in for MainWindow as seen by BlockBookClient.""" + def __init__(self, urls): + self.urls = urls + + def getExplorerURL(self, network): + return self.urls[0] + + def getExplorerURLList(self, network): + return list(self.urls) + + +class BlockBookFallbackTest(unittest.TestCase): + def test_loads_primary_url(self): + client = BlockBookClient(FakeMainWnd(['https://primary.example', 'https://zkbitcoin.com/'])) + self.assertEqual(client.url, 'https://primary.example') + + def test_fallback_excludes_current_url(self): + client = BlockBookClient(FakeMainWnd(['https://primary.example', 'https://zkbitcoin.com/'])) + self.assertEqual(client.getFallbackUrls(), ['https://zkbitcoin.com/']) + + def test_retries_next_explorer_on_failure(self): + # Primary fails -> the call must transparently retry on the next + # configured explorer (e.g. zkbitcoin) and succeed. + client = BlockBookClient(FakeMainWnd(['https://primary.example', 'https://zkbitcoin.com/'])) + attempted = [] + + def fake_checkResponse(method, param=""): + attempted.append(client.url) + if client.url == 'https://primary.example': + raise Exception("primary down") + return [{'txid': 'abc'}] + + client.checkResponse = fake_checkResponse + utxos = client.getAddressUtxos('myaddress') + + self.assertEqual(attempted, ['https://primary.example', 'https://zkbitcoin.com/']) + self.assertEqual(client.url, 'https://zkbitcoin.com/') + self.assertEqual(utxos[0]['script'], '') + + def test_raises_when_all_explorers_fail(self): + client = BlockBookClient(FakeMainWnd(['https://primary.example', 'https://zkbitcoin.com/'])) + client.checkResponse = MagicMock(side_effect=Exception("everything down")) + with self.assertRaises(Exception): + client.getBalance('myaddress') + # primary + every fallback was tried + self.assertEqual(client.checkResponse.call_count, 2) + + +class FakeComboBox: + """Just enough of QComboBox for the explorer methods under test.""" + def __init__(self, items=None): + self._items = list(items or []) # list of (text, data) + self._index = 0 if self._items else -1 + + def clear(self): + self._items = [] + self._index = -1 + + def addItem(self, text, data=None): + self._items.append((text, data)) + if self._index < 0: + self._index = 0 + + def count(self): + return len(self._items) + + def setCurrentIndex(self, i): + self._index = i + + def currentIndex(self): + return self._index + + def findText(self, text): + for idx, (t, _) in enumerate(self._items): + if t == text: + return idx + return -1 + + def itemData(self, i): + if 0 <= i < len(self._items): + return self._items[i][1] + return None + + def currentData(self): + return self.itemData(self._index) + + +@unittest.skipUnless(MAINWINDOW_AVAILABLE, "MainWindow deps (btchip/trezor) unavailable") +class OnChangeSelectedExplorerTest(unittest.TestCase): + def _make_window(self, items): + # Bypass the heavy __init__; we only exercise onChangeSelectedExplorer. + win = MainWindow.__new__(MainWindow) + win.updatingExplorerbox = False + win.isTestnetRPC = False + win.parent = MagicMock() + win.parent.cache = {'selectedExplorer_mainnet': '', 'selectedExplorer_testnet': ''} + win.apiClient = MagicMock() + win.header = MagicMock() + win.header.explorerClientsBox = FakeComboBox(items) + win.explorerServersList = [data for _, data in items] + return win + + def setUp(self): + self.items = [ + ("https://explorer1.com", {'id': 1, 'url': 'https://explorer1.com', 'isTestnet': False, 'isCustom': True}), + ("https://explorer2.com", {'id': 2, 'url': 'https://explorer2.com', 'isTestnet': False, 'isCustom': True}), + ] + # Avoid touching real QSettings; just echo the value back. + self._orig_persist = mainWindow_mod.persistCacheSetting + mainWindow_mod.persistCacheSetting = lambda key, value: value + + def tearDown(self): + mainWindow_mod.persistCacheSetting = self._orig_persist + + def test_change_updates_api_client(self): + win = self._make_window(self.items) + win.onChangeSelectedExplorer(1) + win.apiClient.updateExplorerUrl.assert_called_with('https://explorer2.com') + + def test_change_persists_url_for_network(self): + win = self._make_window(self.items) + win.onChangeSelectedExplorer(1) + self.assertEqual(win.parent.cache['selectedExplorer_mainnet'], 'https://explorer2.com') + + def test_guard_blocks_programmatic_change(self): + win = self._make_window(self.items) + win.updatingExplorerbox = True + win.onChangeSelectedExplorer(1) + win.apiClient.updateExplorerUrl.assert_not_called() + + +@unittest.skipUnless(MAINWINDOW_AVAILABLE, "MainWindow deps (btchip/trezor) unavailable") +class ExplorerPerNetworkSelectionTest(unittest.TestCase): + """Selection must be remembered per network, not by a shared index.""" + ALL = [ + {'id': 1, 'url': 'https://m1', 'isTestnet': False, 'isCustom': False}, + {'id': 2, 'url': 'https://m2', 'isTestnet': False, 'isCustom': False}, + {'id': 3, 'url': 'https://t1', 'isTestnet': True, 'isCustom': False}, + ] + + def setUp(self): + self._orig_persist = mainWindow_mod.persistCacheSetting + mainWindow_mod.persistCacheSetting = lambda key, value: value + self.win = MainWindow.__new__(MainWindow) + self.win.updatingExplorerbox = False + self.win.apiClient = MagicMock() + self.win.header = MagicMock() + self.win.header.explorerClientsBox = FakeComboBox() + self.win.parent = MagicMock() + self.win.parent.cache = {'selectedExplorer_mainnet': 'https://m1', + 'selectedExplorer_testnet': 'https://t1'} + self.win.parent.db.getExplorerServers.return_value = self.ALL + + def tearDown(self): + mainWindow_mod.persistCacheSetting = self._orig_persist + + def test_network_switch_does_not_inherit_index(self): + # On mainnet, user picks the 2nd mainnet explorer (m2). + self.win.isTestnetRPC = False + self.win.updateExplorerList() + self.win.onChangeSelectedExplorer(1) + self.assertEqual(self.win.parent.cache['selectedExplorer_mainnet'], 'https://m2') + + # Switch to testnet: index 1 must NOT carry over; t1 is restored. + self.win.isTestnetRPC = True + self.win.updateExplorerList() + self.assertEqual(self.win.header.explorerClientsBox.currentData()['url'], 'https://t1') + self.win.apiClient.updateExplorerUrl.assert_called_with('https://t1') + + # Switch back to mainnet: the saved m2 choice is restored, not reset. + self.win.isTestnetRPC = False + self.win.updateExplorerList() + self.assertEqual(self.win.header.explorerClientsBox.currentData()['url'], 'https://m2') + + +class InitTableExplorerTest(unittest.TestCase): + """initTable_Explorer must seed defaults and repair upgraded databases.""" + def _run(self, setup_rows): + import sqlite3 + from database import Database + from constants import trusted_explorers + conn = sqlite3.connect(':memory:') + cur = conn.cursor() + cur.execute("CREATE TABLE EXPLORER_SERVERS" + " (id INTEGER PRIMARY KEY, url TEXT, isTestnet BOOLEAN, isCustom BOOLEAN)") + for row in setup_rows: + cur.execute("INSERT INTO EXPLORER_SERVERS (id,url,isTestnet,isCustom) VALUES (?,?,?,?)", row) + Database(MagicMock()).initTable_Explorer(cur) + cur.execute("SELECT url,isTestnet,isCustom FROM EXPLORER_SERVERS ORDER BY id") + rows = cur.fetchall() + conn.close() + self.trusted_urls = [u for u, _, _ in trusted_explorers] + return {url: (isT, isC) for url, isT, isC in rows} + + def test_fresh_db_seeds_all_defaults(self): + result = self._run([]) + for url in self.trusted_urls: + self.assertIn(url, result) + # zkbitcoin is the mainnet fallback + self.assertEqual(result['https://zkbitcoin.com/'], (0, 0)) + + def test_backfills_null_metadata_on_upgrade(self): + result = self._run([ + (0, 'https://explorer.duddino.com/', None, None), + (1, 'https://testnet.duddino.com/', None, None), + ]) + # NULL testnet flag must be repaired, not left to read as mainnet + self.assertEqual(result['https://testnet.duddino.com/'], (1, 0)) + self.assertIn('https://zkbitcoin.com/', result) + + def test_no_id_collision_with_existing_custom(self): + result = self._run([ + (0, 'https://explorer.duddino.com/', 0, 0), + (1, 'https://testnet.duddino.com/', 1, 0), + (2, 'https://my.custom.explorer/', 0, 1), + ]) + # zkbitcoin still inserted despite a custom row already at id 2... + self.assertIn('https://zkbitcoin.com/', result) + # ...and the user's custom row is left untouched + self.assertEqual(result['https://my.custom.explorer/'], (0, 1)) + + +@unittest.skipUnless(MAINWINDOW_AVAILABLE, "MainWindow deps (btchip/trezor) unavailable") +class ExplorerUrlListTest(unittest.TestCase): + def _window(self, explorers): + win = MainWindow.__new__(MainWindow) + win.explorerServersList = explorers + win.header = MagicMock() + win.header.explorerClientsBox.currentData.return_value = None + return win + + def test_filters_by_network(self): + win = self._window([ + {'url': 'https://m1', 'isTestnet': False, 'isCustom': False}, + {'url': 'https://t1', 'isTestnet': True, 'isCustom': False}, + {'url': 'https://zk', 'isTestnet': False, 'isCustom': False}, + ]) + self.assertEqual(win.getExplorerURLList('mainnet'), ['https://m1', 'https://zk']) + self.assertEqual(win.getExplorerURLList('testnet'), ['https://t1']) + + def test_defaults_when_network_empty(self): + from constants import DEFAULT_MAINNET_EXPLORER, DEFAULT_TESTNET_EXPLORER + win = self._window([]) + self.assertEqual(win.getExplorerURLList('mainnet'), [DEFAULT_MAINNET_EXPLORER]) + self.assertEqual(win.getExplorerURLList('testnet'), [DEFAULT_TESTNET_EXPLORER]) + + def test_getExplorerURL_reads_cache_not_widget(self): + win = MainWindow.__new__(MainWindow) + win.explorerServersList = [ + {'url': 'https://m1', 'isTestnet': False, 'isCustom': False}, + {'url': 'https://m2', 'isTestnet': False, 'isCustom': False}, + {'url': 'https://t1', 'isTestnet': True, 'isCustom': False}, + ] + win.parent = MagicMock() + win.parent.cache = {'selectedExplorer_mainnet': 'https://m2', + 'selectedExplorer_testnet': 'https://t1'} + # No header: any Qt-widget access from this worker-thread-safe method + # would raise AttributeError and fail the test. + win.header = None + self.assertEqual(win.getExplorerURL('mainnet'), 'https://m2') + self.assertEqual(win.getExplorerURL('testnet'), 'https://t1') + # A stale/removed saved URL falls back to the first for that network. + win.parent.cache['selectedExplorer_mainnet'] = 'https://gone' + self.assertEqual(win.getExplorerURL('mainnet'), 'https://m1') + + +if __name__ == '__main__': + unittest.main()