added wallet management tests and address unsubscribe feature on wallet server

lbry/lbry/extras/daemon/Daemon.py

@@ -1134,7 +1134,7 @@ class Daemon(metaclass=JSONRPCServerType):
         return wallet
 
     @requires("wallet")
-    def jsonrpc_wallet_remove(self, wallet_id):
+    async def jsonrpc_wallet_remove(self, wallet_id):
         """
         Remove an existing wallet.
 
@@ -1146,8 +1146,10 @@ class Daemon(metaclass=JSONRPCServerType):
 
         Returns: {Wallet}
         """
-        wallet = self.wallet_manager.get_wallet_or_default(wallet_id)
+        wallet = self.wallet_manager.get_wallet_or_error(wallet_id)
         self.wallet_manager.wallets.remove(wallet)
+        for account in wallet.accounts:
+            await self.ledger.unsubscribe_account(account)
         return wallet
 
     @requires("wallet")

lbry/tests/integration/test_wallet_commands.py

@@ -4,6 +4,19 @@ from lbry.testcase import CommandTestCase
 from torba.client.wallet import ENCRYPT_ON_DISK
 
 
+class WalletCommands(CommandTestCase):
+
+    async def test_wallet_create_and_add_subscribe(self):
+        session = next(iter(self.conductor.spv_node.server.session_mgr.sessions))
+        self.assertEqual(len(session.hashX_subs), 27)
+        wallet = await self.daemon.jsonrpc_wallet_create('foo', create_account=True, single_key=True)
+        self.assertEqual(len(session.hashX_subs), 28)
+        await self.daemon.jsonrpc_wallet_remove(wallet.id)
+        self.assertEqual(len(session.hashX_subs), 27)
+        await self.daemon.jsonrpc_wallet_add(wallet.id)
+        self.assertEqual(len(session.hashX_subs), 28)
+
+
 class WalletEncryptionAndSynchronization(CommandTestCase):
 
     SEED = (

torba/torba/client/baseledger.py

@@ -392,6 +392,10 @@ class BaseLedger(metaclass=LedgerRegistry):
             await self.subscribe_addresses(address_manager, await address_manager.get_addresses())
         await account.ensure_address_gap()
 
+    async def unsubscribe_account(self, account: baseaccount.BaseAccount):
+        for address in await account.get_addresses():
+            await self.network.unsubscribe_address(address)
+
     async def announce_addresses(self, address_manager: baseaccount.AddressManager, addresses: List[str]):
         await self.subscribe_addresses(address_manager, addresses)
         await self._on_address_controller.add(

torba/torba/client/basenetwork.py

@@ -261,6 +261,9 @@ class BaseNetwork:
                 self.client.abort()
             raise asyncio.CancelledError()
 
+    def unsubscribe_address(self, address):
+        return self.rpc('blockchain.address.unsubscribe', [address], True)
+
     def get_server_features(self):
         return self.rpc('server.features', (), restricted=True)
 

torba/torba/server/session.py

@@ -882,6 +882,9 @@ class ElectrumX(SessionBase):
         self.hashX_subs[hashX] = alias
         return await self.address_status(hashX)
 
+    async def hashX_unsubscribe(self, hashX, alias):
+        del self.hashX_subs[hashX]
+
     def address_to_hashX(self, address):
         try:
             return self.coin.address_to_hashX(address)
@@ -916,6 +919,13 @@ class ElectrumX(SessionBase):
         hashX = self.address_to_hashX(address)
         return await self.hashX_subscribe(hashX, address)
 
+    async def address_unsubscribe(self, address):
+        """Unsubscribe an address.
+
+        address: the address to unsubscribe"""
+        hashX = self.address_to_hashX(address)
+        return await self.hashX_unsubscribe(hashX, address)
+
     async def get_balance(self, hashX):
         utxos = await self.db.all_utxos(hashX)
         confirmed = sum(utxo.value for utxo in utxos)
@@ -1284,6 +1294,7 @@ class ElectrumX(SessionBase):
                 'blockchain.address.get_mempool': self.address_get_mempool,
                 'blockchain.address.listunspent': self.address_listunspent,
                 'blockchain.address.subscribe': self.address_subscribe,
+                'blockchain.address.unsubscribe': self.address_unsubscribe,
             })
 
         self.request_handlers = handlers