test: wallet, add coverage for addressbook migration

Github-Pull: bitcoin#28038
Rebased-From: 7ecc29a

Author: furszy

test/functional/wallet_migration.py

@@ -67,6 +67,15 @@ def assert_list_txs_equal(self, received_list_txs, expected_list_txs):
                 del d["parent_descs"]
         assert_equal(received_list_txs, expected_list_txs)
 
+    def check_address(self, wallet, addr, is_mine, is_change, label):
+        addr_info = wallet.getaddressinfo(addr)
+        assert_equal(addr_info['ismine'], is_mine)
+        assert_equal(addr_info['ischange'], is_change)
+        if label is not None:
+            assert_equal(addr_info['labels'], [label]),
+        else:
+            assert_equal(addr_info['labels'], []),
+
     def test_basic(self):
         default = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
 
@@ -470,6 +479,132 @@ def test_unloaded_by_path(self):
 
         assert_equal(bals, wallet.getbalances())
 
+    def test_addressbook(self):
+        df_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
+
+        self.log.info("Test migration of address book data")
+        wallet = self.create_legacy_wallet("legacy_addrbook")
+        df_wallet.sendtoaddress(wallet.getnewaddress(), 3)
+
+        # Import watch-only script to create a watch-only wallet after migration
+        watch_addr = df_wallet.getnewaddress()
+        wallet.importaddress(watch_addr)
+        df_wallet.sendtoaddress(watch_addr, 2)
+
+        # Import solvable script
+        multi_addr1 = wallet.getnewaddress()
+        multi_addr2 = wallet.getnewaddress()
+        multi_addr3 = df_wallet.getnewaddress()
+        wallet.importpubkey(df_wallet.getaddressinfo(multi_addr3)["pubkey"])
+        ms_addr_info = wallet.addmultisigaddress(2, [multi_addr1, multi_addr2, multi_addr3])
+
+        self.generate(self.nodes[0], 1)
+
+        # Test vectors
+        addr_external = {
+            "addr": df_wallet.getnewaddress(),
+            "is_mine": False,
+            "is_change": False,
+            "label": ""
+        }
+        addr_external_with_label = {
+            "addr": df_wallet.getnewaddress(),
+            "is_mine": False,
+            "is_change": False,
+            "label": "external"
+        }
+        addr_internal = {
+            "addr": wallet.getnewaddress(),
+            "is_mine": True,
+            "is_change": False,
+            "label": ""
+        }
+        addr_internal_with_label = {
+            "addr": wallet.getnewaddress(),
+            "is_mine": True,
+            "is_change": False,
+            "label": "internal"
+        }
+        change_address = {
+            "addr": wallet.getrawchangeaddress(),
+            "is_mine": True,
+            "is_change": True,
+            "label": None
+        }
+        watch_only_addr = {
+            "addr": watch_addr,
+            "is_mine": False,
+            "is_change": False,
+            "label": "imported"
+        }
+        ms_addr = {
+            "addr": ms_addr_info['address'],
+            "is_mine": False,
+            "is_change": False,
+            "label": "multisig"
+        }
+
+        # To store the change address in the addressbook need to send coins to it
+        wallet.send(outputs=[{wallet.getnewaddress(): 2}], options={"change_address": change_address['addr']})
+        self.generate(self.nodes[0], 1)
+
+        # Util wrapper func for 'addr_info'
+        def check(info, node):
+            self.check_address(node, info['addr'], info['is_mine'], info['is_change'], info["label"])
+
+        # Pre-migration: set label and perform initial checks
+        for addr_info in [addr_external, addr_external_with_label, addr_internal, addr_internal_with_label, change_address, watch_only_addr, ms_addr]:
+            if not addr_info['is_change']:
+                wallet.setlabel(addr_info['addr'], addr_info["label"])
+            check(addr_info, wallet)
+
+        # Migrate wallet
+        info_migration = wallet.migratewallet()
+        wallet_wo = self.nodes[0].get_wallet_rpc(info_migration["watchonly_name"])
+        wallet_solvables = self.nodes[0].get_wallet_rpc(info_migration["solvables_name"])
+
+        #########################
+        # Post migration checks #
+        #########################
+
+        # First check the main wallet
+        for addr_info in [addr_external, addr_external_with_label, addr_internal, addr_internal_with_label, change_address, ms_addr]:
+            check(addr_info, wallet)
+
+        # Watch-only wallet will contain the watch-only entry (with 'is_mine=True') and all external addresses ('send')
+        self.check_address(wallet_wo, watch_only_addr['addr'], is_mine=True, is_change=watch_only_addr['is_change'], label=watch_only_addr["label"])
+        for addr_info in [addr_external, addr_external_with_label, ms_addr]:
+            check(addr_info, wallet_wo)
+
+        # Solvables wallet will contain the multisig entry (with 'is_mine=True') and all external addresses ('send')
+        self.check_address(wallet_solvables, ms_addr['addr'], is_mine=True, is_change=ms_addr['is_change'], label=ms_addr["label"])
+        for addr_info in [addr_external, addr_external_with_label]:
+            check(addr_info, wallet_solvables)
+
+        ########################################################################################
+        # Now restart migrated wallets and verify that the addressbook entries are still there #
+        ########################################################################################
+
+        # First the main wallet
+        self.nodes[0].unloadwallet("legacy_addrbook")
+        self.nodes[0].loadwallet("legacy_addrbook")
+        for addr_info in [addr_external, addr_external_with_label, addr_internal, addr_internal_with_label, change_address, ms_addr]:
+            check(addr_info, wallet)
+
+        # Watch-only wallet
+        self.nodes[0].unloadwallet(info_migration["watchonly_name"])
+        self.nodes[0].loadwallet(info_migration["watchonly_name"])
+        self.check_address(wallet_wo, watch_only_addr['addr'], is_mine=True, is_change=watch_only_addr['is_change'], label=watch_only_addr["label"])
+        for addr_info in [addr_external, addr_external_with_label, ms_addr]:
+            check(addr_info, wallet_wo)
+
+        # Solvables wallet
+        self.nodes[0].unloadwallet(info_migration["solvables_name"])
+        self.nodes[0].loadwallet(info_migration["solvables_name"])
+        self.check_address(wallet_solvables, ms_addr['addr'], is_mine=True, is_change=ms_addr['is_change'], label=ms_addr["label"])
+        for addr_info in [addr_external, addr_external_with_label]:
+            check(addr_info, wallet_solvables)
+
     def run_test(self):
         self.generate(self.nodes[0], 101)
 
@@ -482,6 +617,7 @@ def run_test(self):
         self.test_encrypted()
         self.test_unloaded()
         self.test_unloaded_by_path()
+        self.test_addressbook()
 
 if __name__ == '__main__':
     WalletMigrationTest().main()