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