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/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/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
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/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/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'))
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)
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()