new test.cmdtest_d.include.proxy module

test/cmdtest_d/ethdev.py

@@ -59,6 +59,7 @@ from .include.common import (
 	Ctrl_U,
 	cleanup_env,
 	thorchain_router_addr_file)
+from .include.proxy import TestProxy
 
 from .base import CmdTestBase
 from .shared import CmdTestShared
@@ -743,6 +744,8 @@ class CmdTestEthdev(CmdTestEthdevMethods, CmdTestBase, CmdTestShared):
 		self.message = 'attack at dawn'
 		self.spawn_env['MMGEN_BOGUS_SEND'] = ''
 
+		TestProxy(cfg)
+
 	@property
 	async def rpc(self):
 		from mmgen.rpc import rpc_init
@@ -1091,7 +1094,7 @@ class CmdTestEthdev(CmdTestEthdevMethods, CmdTestBase, CmdTestShared):
 	def txsend_etherscan(self):
 		if self.proto.coin == 'ETC':
 			return 'skip'
-		return self.txsend(add_args=['--tx-proxy=ethersc'])
+		return self.txsend(add_args=['--tx-proxy=ethersc', f'--proxy=localhost:{TestProxy.port}'])
 
 	def etherscan_server_stop(self):
 		if self.proto.coin == 'ETC':

test/cmdtest_d/ethswap.py

@@ -21,6 +21,7 @@ from mmgen.fileutil import get_data_from_file
 from ..include.common import imsg, chk_equal
 
 from .include.common import dfl_sid, eth_inbound_addr, thorchain_router_addr_file
+from .include.proxy import TestProxy
 from .httpd.thornode.swap import ThornodeSwapServer
 
 from .regtest import CmdTestRegtest
@@ -262,6 +263,8 @@ class CmdTestEthSwap(CmdTestSwapMethods, CmdTestRegtest):
 		self.swap_server = ThornodeSwapServer()
 		self.swap_server.start()
 
+		TestProxy(cfg)
+
 	def swaptxcreate1(self):
 		t = self._swaptxcreate(['BTC', '8.765', 'ETH'])
 		t.expect('OK? (Y/n): ', 'y')
@@ -280,8 +283,11 @@ class CmdTestEthSwap(CmdTestSwapMethods, CmdTestRegtest):
 	def swaptxsend1(self):
 		return self._swaptxsend()
 
+	def swaptxsend2(self):
+		return self._swaptxsend(add_opts=[f'--proxy=localhost:{TestProxy.port}'])
+
 	swaptxsign3 = swaptxsign2 = swaptxsign1
-	swaptxsend3 = swaptxsend2 = swaptxsend1
+	swaptxsend3 = swaptxsend1
 
 	def swaptxbump1(self): # create one-output TX back to self to rescue funds
 		return self._swaptxbump('40s', output_args=[f'{dfl_sid}:B:1'])
@@ -450,7 +456,8 @@ class CmdTestEthSwapEth(CmdTestEthSwapMethods, CmdTestSwapMethods, CmdTestEthdev
 		return self._swaptxsend_eth_proxy(test=True)
 
 	def swaptxsend5a(self):
-		return self._swaptxsend_eth_proxy(add_opts=['--txhex-idx=1'])
+		return self._swaptxsend_eth_proxy(
+			add_opts = ['--txhex-idx=1', f'--proxy=localhost:{TestProxy.port}'])
 
 	def swaptxsend5b(self):
 		return self._swaptxsend_eth_proxy(add_opts=['--txhex-idx=2'])

test/cmdtest_d/include/proxy.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+#
+# MMGen Wallet, a terminal-based cryptocurrency wallet
+# Copyright (C)2013-2025 The MMGen Project <mmgen@tuta.io>
+# Licensed under the GNU General Public License, Version 3:
+#   https://www.gnu.org/licenses
+# Public project repositories:
+#   https://github.com/mmgen/mmgen-wallet
+#   https://gitlab.com/mmgen/mmgen-wallet
+
+"""
+test.cmdtest_d.include.proxy: SSH SOCKS proxy runner for the cmdtest.py test suite
+"""
+
+import sys, atexit
+from subprocess import run, PIPE
+
+from mmgen.util import msg, die, fmt
+from mmgen.util2 import port_in_use
+
+from ...include.common import omsg
+
+class TestProxy:
+
+	port = 49237
+	no_ssh_errmsg = """
+		The SSH daemon must be running and listening on localhost in order to test
+		XMR TX relaying via SOCKS proxy.  If sshd is not running, please start it.
+		Otherwise, add the line 'ListenAddress 127.0.0.1' to your sshd_config, and
+		then restart the daemon.
+	"""
+	bad_perm_errmsg = """
+		In order to test XMR TX relaying via SOCKS proxy, it’s desirable to enable
+		SSH to localhost without a password, which is not currently supported by
+		your configuration.  Your possible courses of action:
+
+		1. Continue by answering 'y' at this prompt, and enter your system password
+		   at the following prompt;
+
+		2. Exit the test here, add your user SSH public key to your user
+		   'authorized_keys' file, and restart the test; or
+
+		3. Exit the test here, start the SSH SOCKS proxy manually by entering the
+		   following command, and restart the test:
+
+			  {}
+	"""
+	need_start_errmsg = """
+		Please start the SSH SOCKS proxy by entering the following command:
+
+			{}
+
+		Then restart the test.
+	"""
+
+	def kill_proxy(self, args):
+		if sys.platform in ('linux', 'darwin'):
+			omsg(f'Killing SSH SOCKS server at localhost:{self.port}')
+			cmd = ['pkill', '-f', ' '.join(args)]
+			run(cmd)
+
+	def __init__(self, cfg, external_call=False):
+
+		def start_proxy():
+			if external_call or not cfg.no_daemon_autostart:
+				run(a + b2)
+				omsg(f'SSH SOCKS server started, listening at localhost:{self.port}')
+
+		a = ['ssh', '-x', '-o', 'ExitOnForwardFailure=True', '-D', f'localhost:{self.port}']
+		b0 = ['-o', 'PasswordAuthentication=False']
+		b1 = ['localhost', 'true']
+		b2 = ['-fN', '-E', 'txrelay-proxy.debug', 'localhost']
+
+		if port_in_use(self.port):
+			omsg(f'Port {self.port} already in use. Assuming SSH SOCKS server is running')
+		else:
+			cp = run(a + b0 + b1, stdout=PIPE, stderr=PIPE)
+			if err := cp.stderr.decode():
+				omsg(err)
+			if cp.returncode == 0:
+				start_proxy()
+			elif 'onnection refused' in err:
+				die(2, fmt(self.no_ssh_errmsg, indent='    '))
+			elif 'ermission denied' in err:
+				msg(fmt(self.bad_perm_errmsg.format(' '.join(a + b2)), indent='    ', strip_char='\t'))
+				from mmgen.ui import keypress_confirm
+				keypress_confirm(cfg, 'Continue?', do_exit=True)
+				start_proxy()
+			else:
+				die(2, fmt(self.need_start_errmsg.format(' '.join(a + b2)), indent='    '))
+
+		if not (external_call or cfg.no_daemon_stop):
+			atexit.unregister(self.kill_proxy)
+			atexit.register(self.kill_proxy, a + b2)

test/cmdtest_d/rune.py

@@ -13,6 +13,7 @@ test.cmdtest_d.rune: THORChain RUNE tests for the cmdtest.py test suite
 """
 
 from .include.common import dfl_sid, dfl_words_file
+from .include.proxy import TestProxy
 from .httpd.thornode.rpc import ThornodeRPCServer
 from .ethdev import CmdTestEthdevMethods
 from .base import CmdTestBase
@@ -66,6 +67,8 @@ class CmdTestRune(CmdTestEthdevMethods, CmdTestBase, CmdTestShared):
 		self.rpc_server = ThornodeRPCServer()
 		self.rpc_server.start()
 
+		TestProxy(cfg)
+
 	def addrgen(self):
 		return self._addrgen()
 
@@ -106,7 +109,7 @@ class CmdTestRune(CmdTestEthdevMethods, CmdTestBase, CmdTestShared):
 			has_label = True)
 
 	def txsend1_test(self):
-		return self._txsend(add_args=['--test'])
+		return self._txsend(add_opts=['--test', f'--proxy=localhost:{TestProxy.port}'], test=True)
 
 	def txsend1(self):
 		return self._txsend()

test/cmdtest_d/runeswap.py

@@ -13,6 +13,7 @@ test.cmdtest_d.runeswap: THORChain swap tests for the cmdtest.py test suite
 """
 
 from .httpd.thornode.swap import ThornodeSwapServer
+from .include.proxy import TestProxy
 
 from .regtest import CmdTestRegtest
 from .swap import CmdTestSwapMethods, create_cross_methods
@@ -75,6 +76,8 @@ class CmdTestRuneSwap(CmdTestSwapMethods, CmdTestRegtest):
 		self.swap_server = ThornodeSwapServer()
 		self.swap_server.start()
 
+		TestProxy(cfg)
+
 	def swap_server_stop(self):
 		return self._thornode_server_stop()
 
@@ -104,7 +107,7 @@ class CmdTestRuneSwapRune(CmdTestSwapMethods, CmdTestRune):
 		return self._swaptxsign()
 
 	def swaptxsend1(self):
-		return self._swaptxsend()
+		return self._swaptxsend(add_opts=[f'--proxy=localhost:{TestProxy.port}'])
 
 	def swaptxstatus1(self):
 		return self._swaptxsend(add_opts=['--verbose', '--status'], status=True)

test/cmdtest_d/xmrwallet.py

@@ -24,8 +24,7 @@ import sys, os, time, re, atexit, asyncio, shutil
 from subprocess import run, PIPE
 from collections import namedtuple
 
-from mmgen.util import msg, fmt, async_run, capfirst, is_int, die, list_gen
-from mmgen.util2 import port_in_use
+from mmgen.util import async_run, capfirst, is_int, die, list_gen
 from mmgen.obj import MMGenRange
 from mmgen.amt import XMRAmt
 from mmgen.addrlist import ViewKeyAddrList, KeyAddrList, AddrIdxList
@@ -43,6 +42,7 @@ from ..include.common import (
 	strip_ansi_escapes
 )
 from .include.common import get_file_with_ext
+from .include.proxy import TestProxy
 from .base import CmdTestBase
 
 # atexit functions:
@@ -53,12 +53,6 @@ def stop_daemons(self):
 def stop_miner_wallet_daemon(self):
 	async_run(self.users['miner'].wd_rpc.stop_daemon())
 
-def kill_proxy(cls, args):
-	if sys.platform in ('linux', 'darwin'):
-		omsg(f'Killing SSH SOCKS server at localhost:{cls.socks_port}')
-		cmd = ['pkill', '-f', ' '.join(args)]
-		run(cmd)
-
 class CmdTestXMRWallet(CmdTestBase):
 	"""
 	Monero wallet operations
@@ -69,7 +63,6 @@ class CmdTestXMRWallet(CmdTestBase):
 	tmpdir_nums = [29]
 	dfl_random_txs = 3
 	color = True
-	socks_port = 49237
 	# Bob’s daemon is stopped via process kill, not RPC, so put Bob last in list:
 	#    user     sid      autosign  shift kal_range add_coind_args
 	user_data = (
@@ -134,7 +127,8 @@ class CmdTestXMRWallet(CmdTestBase):
 
 		self.tx_relay_daemon_parm = 'localhost:{}'.format(self.users[self.tx_relay_user].md.rpc_port)
 		self.tx_relay_daemon_proxy_parm = (
-			self.tx_relay_daemon_parm + f':127.0.0.1:{self.socks_port}') # must be IP, not 'localhost'
+			# must be IP, not 'localhost':
+			self.tx_relay_daemon_parm + f':127.0.0.1:{TestProxy.port}')
 
 		if not cfg.no_daemon_stop:
 			atexit.register(stop_daemons, self)
@@ -146,80 +140,13 @@ class CmdTestXMRWallet(CmdTestBase):
 			if os.path.exists(self.datadir_base):
 				shutil.rmtree(self.datadir_base)
 			os.makedirs(self.datadir_base)
+			TestProxy(cfg)
 			self.start_daemons()
-			self.init_proxy(cfg)
 
 		self.balance = None
 
 	# init methods
 
-	@classmethod
-	def init_proxy(cls, cfg, external_call=False):
-
-		def start_proxy():
-			if external_call or not cfg.no_daemon_autostart:
-				run(a+b2)
-				omsg(f'SSH SOCKS server started, listening at localhost:{cls.socks_port}')
-
-		debug_file = os.path.join('' if external_call else cls.datadir_base, 'txrelay-proxy.debug')
-		a = ['ssh', '-x', '-o', 'ExitOnForwardFailure=True', '-D', f'localhost:{cls.socks_port}']
-		b0 = ['-o', 'PasswordAuthentication=False']
-		b1 = ['localhost', 'true']
-		b2 = ['-fN', '-E', debug_file, 'localhost']
-
-		if port_in_use(cls.socks_port):
-			omsg(f'Port {cls.socks_port} already in use.  Assuming SSH SOCKS server is running')
-		else:
-			cp = run(a+b0+b1, stdout=PIPE, stderr=PIPE)
-			err = cp.stderr.decode()
-			if err:
-				omsg(err)
-
-			if cp.returncode == 0:
-				start_proxy()
-			elif 'onnection refused' in err:
-				die(2, fmt("""
-					The SSH daemon must be running and listening on localhost in order to test
-					XMR TX relaying via SOCKS proxy.  If sshd is not running, please start it.
-					Otherwise, add the line 'ListenAddress 127.0.0.1' to your sshd_config, and
-					then restart the daemon.
-				""", indent='    '))
-			elif 'ermission denied' in err:
-				msg(fmt(f"""
-					In order to test XMR TX relaying via SOCKS proxy, it’s desirable to enable
-					SSH to localhost without a password, which is not currently supported by
-					your configuration.  Your possible courses of action:
-
-					1. Continue by answering 'y' at this prompt, and enter your system password
-					   at the following prompt;
-
-					2. Exit the test here, add your user SSH public key to your user
-					   'authorized_keys' file, and restart the test; or
-
-					3. Exit the test here, start the SSH SOCKS proxy manually by entering the
-					   following command, and restart the test:
-
-					      {' '.join(a+b2)}
-				""", indent='    ', strip_char='\t'))
-
-				from mmgen.ui import keypress_confirm
-				keypress_confirm(cfg, 'Continue?', do_exit=True)
-				start_proxy()
-			else:
-				die(2, fmt(f"""
-					Please start the SSH SOCKS proxy by entering the following command:
-
-						{' '.join(a+b2)}
-
-					Then restart the test.
-				""", indent='    '))
-
-		if not (external_call or cfg.no_daemon_stop):
-			atexit.unregister(kill_proxy)
-			atexit.register(kill_proxy, cls, a + b2)
-
-		return True
-
 	def init_users(self):
 		from mmgen.daemon import CoinDaemon
 		from mmgen.proto.xmr.daemon import MoneroWalletDaemon

test/modtest_d/testdep.py

@@ -65,5 +65,5 @@ class unit_tests:
 		return True
 
 	def ssh_socks_proxy(self, name, ut):
-		from test.cmdtest_d.xmrwallet import CmdTestXMRWallet
-		return CmdTestXMRWallet.init_proxy(cfg, external_call=True)
+		from test.cmdtest_d.include.proxy import TestProxy
+		return TestProxy(cfg, external_call=True)