test: refactor: introduce replace_in_config helper

Author: theStack

test/functional/p2p_permissions.py

@@ -56,12 +56,12 @@ def run_test(self):
         # For this, we need to use whitebind instead of bind
         # by modifying the configuration file.
         ip_port = "127.0.0.1:{}".format(p2p_port(1))
-        self.replaceinconfig(1, "bind=127.0.0.1", "whitebind=bloomfilter,forcerelay@" + ip_port)
+        self.nodes[1].replace_in_config([("bind=127.0.0.1", "whitebind=bloomfilter,forcerelay@" + ip_port)])
         self.checkpermission(
             ["[email protected]"],
             # Check parameter interaction forcerelay should activate relay
             ["noban", "bloomfilter", "forcerelay", "relay", "download"])
-        self.replaceinconfig(1, "whitebind=bloomfilter,forcerelay@" + ip_port, "bind=127.0.0.1")
+        self.nodes[1].replace_in_config([("whitebind=bloomfilter,forcerelay@" + ip_port, "bind=127.0.0.1")])
 
         self.checkpermission(
             # legacy whitelistrelay should be ignored
@@ -138,12 +138,6 @@ def checkpermission(self, args, expectedPermissions):
             if p not in peerinfo['permissions']:
                 raise AssertionError("Expected permissions %r is not granted." % p)
 
-    def replaceinconfig(self, nodeid, old, new):
-        with open(self.nodes[nodeid].bitcoinconf, encoding="utf8") as f:
-            newText = f.read().replace(old, new)
-        with open(self.nodes[nodeid].bitcoinconf, 'w', encoding="utf8") as f:
-            f.write(newText)
-
 
 if __name__ == '__main__':
     P2PPermissionsTests().main()

test/functional/test_framework/test_framework.py

@@ -533,11 +533,7 @@ def get_bin_from_version(version, bin_name, bin_default):
             self.nodes.append(test_node_i)
             if not test_node_i.version_is_at_least(170000):
                 # adjust conf for pre 17
-                conf_file = test_node_i.bitcoinconf
-                with open(conf_file, 'r', encoding='utf8') as conf:
-                    conf_data = conf.read()
-                with open(conf_file, 'w', encoding='utf8') as conf:
-                    conf.write(conf_data.replace('[regtest]', ''))
+                test_node_i.replace_in_config([('[regtest]', '')])
 
     def start_node(self, i, *args, **kwargs):
         """Start a bitcoind"""

test/functional/test_framework/test_node.py

@@ -387,6 +387,21 @@ def is_node_stopped(self):
     def wait_until_stopped(self, timeout=BITCOIND_PROC_WAIT_TIMEOUT):
         wait_until_helper(self.is_node_stopped, timeout=timeout, timeout_factor=self.timeout_factor)
 
+    def replace_in_config(self, replacements):
+        """
+        Perform replacements in the configuration file.
+        The substitutions are passed as a list of search-replace-tuples, e.g.
+            [("old", "new"), ("foo", "bar"), ...]
+        """
+        with open(self.bitcoinconf, 'r', encoding='utf8') as conf:
+            conf_data = conf.read()
+        for replacement in replacements:
+            assert_equal(len(replacement), 2)
+            old, new = replacement[0], replacement[1]
+            conf_data = conf_data.replace(old, new)
+        with open(self.bitcoinconf, 'w', encoding='utf8') as conf:
+            conf.write(conf_data)
+
     @property
     def chain_path(self) -> Path:
         return Path(self.datadir) / self.chain

test/functional/wallet_crosschain.py

@@ -25,11 +25,7 @@ def setup_network(self):
         # Switch node 1 to testnet before starting it.
         self.nodes[1].chain = 'testnet3'
         self.nodes[1].extra_args = ['-maxconnections=0', '-prune=550'] # disable testnet sync
-        with open(self.nodes[1].bitcoinconf, 'r', encoding='utf8') as conf:
-            conf_data = conf.read()
-        with open (self.nodes[1].bitcoinconf, 'w', encoding='utf8') as conf:
-            conf.write(conf_data.replace('regtest=', 'testnet=').replace('[regtest]', '[test]'))
-
+        self.nodes[1].replace_in_config([('regtest=', 'testnet='), ('[regtest]', '[test]')])
         self.start_nodes()
 
     def run_test(self):