diff --git a/packages/komodo_defi_sdk/example/lib/screens/asset_page.dart b/packages/komodo_defi_sdk/example/lib/screens/asset_page.dart index 88de54c99..ed421c007 100644 --- a/packages/komodo_defi_sdk/example/lib/screens/asset_page.dart +++ b/packages/komodo_defi_sdk/example/lib/screens/asset_page.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:flutter/material.dart'; import 'package:flutter_bloc/flutter_bloc.dart'; import 'package:kdf_sdk_example/widgets/asset/addresses_section_widget.dart'; @@ -22,31 +24,55 @@ class _AssetPageState extends State { String? _error; late final _sdk = context.read(); + StreamSubscription? _pubkeysSubscription; @override void initState() { super.initState(); _refreshUnavailableReasons().ignore(); - _loadPubkeys(); + _startWatchingPubkeys(); + } + + void _startWatchingPubkeys() { + setState(() => _isLoading = true); + _pubkeysSubscription?.cancel(); + _pubkeysSubscription = _sdk.pubkeys + .watchPubkeys(widget.asset) + .listen( + (pubkeys) { + if (!mounted) return; + setState(() { + _pubkeys = pubkeys; + _error = null; + _isLoading = false; + }); + }, + onError: (Object e) { + if (!mounted) return; + setState(() { + _error = e.toString(); + _isLoading = false; + }); + }, + ); } - Future _loadPubkeys() async { + Future _forceRefreshPubkeys() async { setState(() => _isLoading = true); try { - final pubkeys = await _sdk.pubkeys.getPubkeys(widget.asset); - _pubkeys = pubkeys; + await _sdk.pubkeys.precachePubkeys(widget.asset); } catch (e) { - _error = e.toString(); + if (mounted) setState(() => _error = e.toString()); } finally { if (mounted) setState(() => _isLoading = false); - _refreshUnavailableReasons().ignore(); + await _refreshUnavailableReasons(); } } Future _generateNewAddress() async { setState(() => _isLoading = true); try { - final stream = _sdk.pubkeys.createNewPubkeyStream(widget.asset); + final stream = _sdk.pubkeys.watchCreateNewPubkey(widget.asset); final newPubkey = await showDialog( context: context, @@ -80,7 +106,10 @@ class _AssetPageState extends State { appBar: AppBar( title: Text(widget.asset.id.name), actions: [ - IconButton(icon: const Icon(Icons.refresh), onPressed: _loadPubkeys), + IconButton( + icon: const Icon(Icons.refresh), + onPressed: _forceRefreshPubkeys, + ), ], ), body: @@ -114,4 +143,11 @@ class _AssetPageState extends State { ), ); } + + @override + void dispose() { + _pubkeysSubscription?.cancel(); + _pubkeysSubscription = null; + super.dispose(); + } } diff --git a/packages/komodo_defi_sdk/lib/src/activation/activation_manager.dart b/packages/komodo_defi_sdk/lib/src/activation/activation_manager.dart index df4e7a478..55099e2fa 100644 --- a/packages/komodo_defi_sdk/lib/src/activation/activation_manager.dart +++ b/packages/komodo_defi_sdk/lib/src/activation/activation_manager.dart @@ -209,7 +209,7 @@ class ActivationManager { await _customTokenHistory.addAssetToWallet(user.walletId, asset); } // Pre-cache balance for the activated asset - await _balanceManager.preCacheBalance(asset); + await _balanceManager.precacheBalance(asset); } } diff --git a/packages/komodo_defi_sdk/lib/src/balances/balance_manager.dart b/packages/komodo_defi_sdk/lib/src/balances/balance_manager.dart index 4a9d46a35..d07846e2c 100644 --- a/packages/komodo_defi_sdk/lib/src/balances/balance_manager.dart +++ b/packages/komodo_defi_sdk/lib/src/balances/balance_manager.dart @@ -1,11 +1,12 @@ import 'dart:async'; -import 'package:flutter/foundation.dart'; // Add this import for debugPrint + import 'package:komodo_defi_local_auth/komodo_defi_local_auth.dart'; import 'package:komodo_defi_sdk/src/activation/activation_manager.dart'; import 'package:komodo_defi_sdk/src/activation/shared_activation_coordinator.dart'; import 'package:komodo_defi_sdk/src/assets/asset_lookup.dart'; import 'package:komodo_defi_sdk/src/pubkeys/pubkey_manager.dart'; import 'package:komodo_defi_types/komodo_defi_types.dart'; +import 'package:logging/logging.dart'; /// Interface defining the contract for balance management operations abstract class IBalanceManager { @@ -41,7 +42,7 @@ abstract class IBalanceManager { /// Pre-caches the balance for an asset. /// This is an internal method used during activation to optimize initial balance fetches. - Future preCacheBalance(Asset asset); + Future precacheBalance(Asset asset); } /// Implementation of the [IBalanceManager] interface for managing asset balances. @@ -65,7 +66,9 @@ class BalanceManager implements IBalanceManager { _auth = auth { // Listen for auth state changes _authSubscription = _auth.authStateChanges.listen(_handleAuthStateChanged); + _logger.fine('Initialized'); } + static final Logger _logger = Logger('BalanceManager'); SharedActivationCoordinator? _activationCoordinator; PubkeyManager? _pubkeyManager; @@ -111,6 +114,9 @@ class BalanceManager implements IBalanceManager { if (_isDisposed) return; final newWalletId = user?.walletId; // If the wallet ID has changed, reset all state + _logger.fine( + 'Auth state changed. wallet: $_currentWalletId -> $newWalletId', + ); if (_currentWalletId != newWalletId) { await _resetState(); _currentWalletId = newWalletId; @@ -119,6 +125,7 @@ class BalanceManager implements IBalanceManager { /// Reset all internal state when wallet changes Future _resetState() async { + _logger.fine('Resetting state'); // Cancel all active watchers for (final subscription in _activeWatchers.values) { await subscription.cancel(); @@ -200,8 +207,16 @@ class BalanceManager implements IBalanceManager { final controller = _balanceControllers.putIfAbsent( assetId, () => StreamController.broadcast( - onListen: () => _startWatchingBalance(assetId, activateIfNeeded), - onCancel: () => _stopWatchingBalance(assetId), + onListen: () { + _logger.fine( + 'onListen: ${assetId.name}, activateIfNeeded: $activateIfNeeded', + ); + _startWatchingBalance(assetId, activateIfNeeded); + }, + onCancel: () { + _logger.fine('onCancel: ${assetId.name}'); + _stopWatchingBalance(assetId); + }, ), ); @@ -212,7 +227,7 @@ class BalanceManager implements IBalanceManager { Future _ensureAssetActivated(Asset asset, bool activateIfNeeded) async { // Check if activationCoordinator is initialized if (_activationCoordinator == null) { - debugPrint( + _logger.fine( 'SharedActivationCoordinator not initialized, cannot activate asset', ); return false; @@ -232,7 +247,7 @@ class BalanceManager implements IBalanceManager { final result = await _activationCoordinator!.activateAsset(asset); return result.isSuccess; } catch (e) { - debugPrint('Failed to activate asset ${asset.id.name}: $e'); + _logger.fine('Failed to activate asset ${asset.id.name}: $e'); return false; } } @@ -271,16 +286,21 @@ class BalanceManager implements IBalanceManager { final user = await _auth.currentUser; if (user == null) { // Don't throw an error, just wait for authentication + _logger.fine( + 'Delaying balance watcher start for ${assetId.name}: unauthenticated', + ); return; } // Keep track of the wallet ID this balance is for _currentWalletId = user.walletId; + _logger.fine('Starting balance watcher for ${assetId.name}'); // Emit the last known balance immediately if available final maybeKnownBalance = lastKnown(assetId); if (maybeKnownBalance != null) { controller.add(maybeKnownBalance); + _logger.fine('Emitted initial balance for ${assetId.name}'); } try { @@ -342,6 +362,7 @@ class BalanceManager implements IBalanceManager { }, onDone: () { _stopWatchingBalance(assetId); + _logger.fine('Stopped watching ${assetId.name}'); }, cancelOnError: false, ); @@ -356,6 +377,7 @@ class BalanceManager implements IBalanceManager { if (watcher != null) { watcher.cancel(); _activeWatchers.remove(assetId); + _logger.fine('Stopped watcher for ${assetId.name}'); } // Don't close the controller here, just remove the watcher // The controller will be closed when all listeners are gone @@ -374,34 +396,66 @@ class BalanceManager implements IBalanceManager { if (_isDisposed) return; _isDisposed = true; - // Cancel auth subscription - await _authSubscription?.cancel(); + // Take snapshots to avoid concurrent modification while cancelling/closing + final StreamSubscription? authSub = _authSubscription; _authSubscription = null; - // Cancel all active watchers - for (final subscription in _activeWatchers.values) { - await subscription.cancel(); - } + final List> watcherSubs = + List>.from(_activeWatchers.values); _activeWatchers.clear(); - // Close all stream controllers - for (final controller in _balanceControllers.values) { - await controller.close(); + // Cancel auth subscription and all watchers concurrently; swallow errors + final List> cancelFutures = >[]; + if (authSub != null) { + cancelFutures.add( + authSub.cancel().catchError((Object e, StackTrace s) { + _logger.warning('Error cancelling auth subscription', e, s); + }), + ); } + for (final StreamSubscription sub in watcherSubs) { + cancelFutures.add( + sub.cancel().catchError((Object e, StackTrace s) { + _logger.warning('Error cancelling balance watcher', e, s); + }), + ); + } + if (cancelFutures.isNotEmpty) { + await Future.wait(cancelFutures); + } + + // Snapshot controllers and close all concurrently; swallow errors + final List> controllers = + List>.from(_balanceControllers.values); _balanceControllers.clear(); + final List> closeFutures = >[]; + for (final StreamController controller in controllers) { + if (!controller.isClosed) { + closeFutures.add( + controller.close().catchError((Object e, StackTrace s) { + _logger.warning('Error closing balance controller', e, s); + }), + ); + } + } + if (closeFutures.isNotEmpty) { + await Future.wait(closeFutures); + } + // Clear all other resources _balanceCache.clear(); _currentWalletId = null; + _logger.fine('Disposed'); } @override - Future preCacheBalance(Asset asset) async { + Future precacheBalance(Asset asset) async { if (_isDisposed) return; // Check if pubkeyManager is initialized if (_pubkeyManager == null) { - debugPrint('Cannot pre-cache balance: PubkeyManager not initialized'); + _logger.fine('Cannot pre-cache balance: PubkeyManager not initialized'); return; } @@ -435,15 +489,15 @@ class BalanceManager implements IBalanceManager { errorStr.contains('invalid coin'); if (isCoinNotFound && !isLastAttempt) { - debugPrint( - 'Balance pre-cache failed (attempt ${attempt + 1}): Coin ${asset.id.name} not yet available, retrying...', + _logger.fine( + 'Balance pre-cache retry ${attempt + 1}: ${asset.id.name} not yet available', ); - await Future.delayed(baseDelay * (attempt + 1)); + await Future.delayed(baseDelay * (attempt + 1)); continue; } // Either not a timing issue or final attempt - fail silently - debugPrint('Failed to pre-cache balance for ${asset.id.name}: $e'); + _logger.fine('Failed to pre-cache balance for ${asset.id.name}: $e'); return; } } diff --git a/packages/komodo_defi_sdk/lib/src/pubkeys/pubkey_manager.dart b/packages/komodo_defi_sdk/lib/src/pubkeys/pubkey_manager.dart index 9552bfe4e..75e9ad7da 100644 --- a/packages/komodo_defi_sdk/lib/src/pubkeys/pubkey_manager.dart +++ b/packages/komodo_defi_sdk/lib/src/pubkeys/pubkey_manager.dart @@ -1,17 +1,70 @@ +import 'dart:async'; + import 'package:komodo_defi_local_auth/komodo_defi_local_auth.dart'; import 'package:komodo_defi_sdk/src/_internal_exports.dart'; import 'package:komodo_defi_types/komodo_defi_type_utils.dart'; import 'package:komodo_defi_types/komodo_defi_types.dart'; +import 'package:logging/logging.dart'; + +/// Interface defining the contract for pubkey management operations +abstract class IPubkeyManager { + /// Get pubkeys for a given asset, handling HD/non-HD differences internally + Future getPubkeys(Asset asset); + + /// Watch pubkeys for a given asset, emitting the initial state if available + /// and polling for updates at a fixed interval. Optionally activates asset. + Stream watchPubkeys( + Asset asset, { + bool activateIfNeeded = true, + }); + + /// Get the last known pubkeys for an asset without triggering a refresh. + /// Returns null if no pubkeys have been fetched yet. + AssetPubkeys? lastKnown(AssetId assetId); + + /// Create a new pubkey for an asset if supported + Future createNewPubkey(Asset asset); + + /// Streamed version of [createNewPubkey] + Stream watchCreateNewPubkey(Asset asset); + + /// Unban pubkeys according to [unbanBy] criteria + Future unbanPubkeys(UnbanBy unbanBy); + + /// Pre-caches pubkeys for an asset to warm the cache and notify listeners + Future precachePubkeys(Asset asset); + + /// Dispose of any resources + Future dispose(); +} /// Manager responsible for handling pubkey operations across different assets -class PubkeyManager { - PubkeyManager(this._client, this._auth, this._activationCoordinator); +class PubkeyManager implements IPubkeyManager { + PubkeyManager(this._client, this._auth, this._activationCoordinator) { + _authSubscription = _auth.authStateChanges.listen(_handleAuthStateChanged); + _logger.fine('Initialized'); + } + static final Logger _logger = Logger('PubkeyManager'); final ApiClient _client; final KomodoDefiLocalAuth _auth; final SharedActivationCoordinator _activationCoordinator; + // Internal state for watching pubkeys per asset + final Map _pubkeysCache = {}; + final Map> _activeWatchers = {}; + final Map> _pubkeysControllers = {}; + // Track the Asset for each AssetId that has an associated controller so that + // we can restart watchers after auth changes without requiring new listeners + final Map _watchedAssets = {}; + + StreamSubscription? _authSubscription; + WalletId? _currentWalletId; + bool _isDisposed = false; + final Duration _defaultPollingInterval = const Duration(seconds: 30); + /// Get pubkeys for a given asset, handling HD/non-HD differences internally + @override Future getPubkeys(Asset asset) async { await retry(() => _activationCoordinator.activateAsset(asset)); final strategy = await _resolvePubkeyStrategy(asset); @@ -19,6 +72,7 @@ class PubkeyManager { } /// Create a new pubkey for an asset if supported + @override Future createNewPubkey(Asset asset) async { await retry(() => _activationCoordinator.activateAsset(asset)); final strategy = await _resolvePubkeyStrategy(asset); @@ -31,7 +85,8 @@ class PubkeyManager { } /// Streamed version of [createNewPubkey] - Stream createNewPubkeyStream(Asset asset) async* { + @override + Stream watchCreateNewPubkey(Asset asset) async* { await retry(() => _activationCoordinator.activateAsset(asset)); final strategy = await _resolvePubkeyStrategy(asset); if (!strategy.supportsMultipleAddresses) { @@ -44,6 +99,7 @@ class PubkeyManager { } /// Unban pubkeys according to [unbanBy] criteria + @override Future unbanPubkeys(UnbanBy unbanBy) async { final response = await _client.rpc.wallet.unbanPubkeys(unbanBy: unbanBy); return response.result; @@ -57,8 +113,265 @@ class PubkeyManager { return asset.pubkeyStrategy(kdfUser: currentUser); } + /// Stream of pubkeys per asset. Polls pubkeys (not balances) and emits updates. + /// Emits the initial known state if available. + @override + Stream watchPubkeys( + Asset asset, { + bool activateIfNeeded = true, + }) async* { + if (_isDisposed) { + throw StateError('PubkeyManager has been disposed'); + } + + // Emit last known pubkeys immediately if available + final lastKnown = _pubkeysCache[asset.id]; + if (lastKnown != null) { + yield lastKnown; + } + + final controller = _pubkeysControllers.putIfAbsent( + asset.id, + () => StreamController.broadcast( + onListen: () { + _logger.fine( + 'onListen: ${asset.id.name}, activateIfNeeded: $activateIfNeeded', + ); + _startWatchingPubkeys(asset, activateIfNeeded); + }, + onCancel: () { + _logger.fine('onCancel: ${asset.id.name}'); + _stopWatchingPubkeys(asset.id); + _watchedAssets.remove(asset.id); + }, + ), + ); + // Remember the Asset so we can restart the watcher after a reset + _watchedAssets[asset.id] = asset; + + yield* controller.stream; + } + + @override + AssetPubkeys? lastKnown(AssetId assetId) { + if (_isDisposed) { + throw StateError('PubkeyManager has been disposed'); + } + return _pubkeysCache[assetId]; + } + + Future _startWatchingPubkeys(Asset asset, bool activateIfNeeded) async { + final controller = _pubkeysControllers[asset.id]; + if (controller == null || _isDisposed) return; + + // Cancel any existing watcher for this asset + await _activeWatchers[asset.id]?.cancel(); + _activeWatchers.remove(asset.id); + + // Ensure user is authenticated + final user = await _auth.currentUser; + if (user == null) { + // Do not emit an error; wait for authentication changes + _logger.fine( + 'Delaying watcher start for ${asset.id.name}: unauthenticated', + ); + return; + } + _currentWalletId = user.walletId; + _logger.fine('Starting watcher for ${asset.id.name}'); + + // Emit last known immediately if available + final maybeKnown = _pubkeysCache[asset.id]; + if (maybeKnown != null && !controller.isClosed) { + controller.add(maybeKnown); + } + + try { + // Ensure activation if requested, otherwise only proceed if already active + bool isActive = await _activationCoordinator.isAssetActive(asset.id); + if (!isActive && activateIfNeeded) { + final activationResult = await _activationCoordinator.activateAsset( + asset, + ); + isActive = activationResult.isSuccess; + } + + if (isActive) { + final first = await getPubkeys(asset); + _pubkeysCache[asset.id] = first; + if (!controller.isClosed) controller.add(first); + _logger.fine('Emitted initial pubkeys for ${asset.id.name}'); + } + + // Periodic polling for pubkeys updates + final periodicStream = Stream.periodic(_defaultPollingInterval); + _activeWatchers[asset.id] = periodicStream + .asyncMap((_) async { + if (_isDisposed) return null; + + // Check that user is still authenticated and wallet hasn't changed + final currentUser = await _auth.currentUser; + if (currentUser == null || + currentUser.walletId != _currentWalletId) { + return null; + } + + try { + bool active = await _activationCoordinator.isAssetActive( + asset.id, + ); + if (!active && activateIfNeeded) { + final activationResult = await _activationCoordinator + .activateAsset(asset); + active = activationResult.isSuccess; + } + if (active) { + final pubkeys = await getPubkeys(asset); + _pubkeysCache[asset.id] = pubkeys; + return pubkeys; + } + } catch (_) { + // Swallow transient errors; continue with last known state + } + return _pubkeysCache[asset.id]; + }) + .listen( + (AssetPubkeys? pubkeys) { + if (pubkeys != null && !controller.isClosed) { + controller.add(pubkeys); + } + }, + onError: (Object error) { + if (!controller.isClosed) controller.addError(error); + }, + onDone: () => _stopWatchingPubkeys(asset.id), + cancelOnError: false, + ); + } catch (e) { + if (!controller.isClosed) controller.addError(e); + } + } + + void _stopWatchingPubkeys(AssetId assetId) { + final watcher = _activeWatchers[assetId]; + if (watcher != null) { + watcher.cancel(); + _activeWatchers.remove(assetId); + _logger.fine('Stopped watcher for ${assetId.name}'); + } + } + + @override + Future precachePubkeys(Asset asset) async { + if (_isDisposed) return; + + final user = await _auth.currentUser; + if (user == null) return; + + try { + final pubkeys = await getPubkeys(asset); + _pubkeysCache[asset.id] = pubkeys; + + final controller = _pubkeysControllers[asset.id]; + if (controller != null && !controller.isClosed) { + controller.add(pubkeys); + } + } catch (_) { + // Fail silently; this is a best-effort cache warm-up + } + } + + Future _handleAuthStateChanged(KdfUser? user) async { + if (_isDisposed) return; + final newWalletId = user?.walletId; + _logger.fine( + 'Auth state changed. wallet: $_currentWalletId -> $newWalletId', + ); + if (_currentWalletId != newWalletId) { + await _resetState(); + _currentWalletId = newWalletId; + } + } + + /// Called when authentication state changes to do the following: + /// - clear active watchers + /// - indicate disconnection with state error to controllers + /// - restart the pubkey watchers for the active controllers + Future _resetState() async { + _logger.fine('Resetting state'); + // Cancel all active watchers + for (final subscription in _activeWatchers.values) { + await subscription.cancel(); + } + _activeWatchers.clear(); + + // Notify existing controllers with an error to signal reconnection + for (final controller in _pubkeysControllers.values) { + if (!controller.isClosed) { + controller.addError( + StateError('Wallet changed, reconnecting pubkey watchers'), + ); + } + } + + // Clear caches + _pubkeysCache.clear(); + + // Restart pubkey watchers for controllers that remain open + final existingControllers = + Map>.from(_pubkeysControllers); + for (final entry in existingControllers.entries) { + final controller = entry.value; + if (controller.isClosed) continue; + final assetId = entry.key; + final asset = _watchedAssets[assetId]; + if (asset != null) { + await _startWatchingPubkeys(asset, true); + } + } + } + /// Dispose of any resources + @override Future dispose() async { - // No cleanup needed currently + if (_isDisposed) return; + _isDisposed = true; + + // Collect all async cleanup operations and run them concurrently. + final List> pending = >[]; + + final StreamSubscription? authSub = _authSubscription; + _authSubscription = null; + if (authSub != null) { + pending.add(authSub.cancel()); + } + + final List> watcherSubs = + _activeWatchers.values.toList(); + _activeWatchers.clear(); + for (final StreamSubscription subscription in watcherSubs) { + pending.add(subscription.cancel()); + } + + final List> controllers = + _pubkeysControllers.values.toList(); + _pubkeysControllers.clear(); + for (final StreamController controller in controllers) { + pending.add(controller.close()); + } + + try { + if (pending.isNotEmpty) { + await Future.wait(pending); + } + } catch (error, stackTrace) { + // Swallow errors during disposal to ensure best-effort cleanup + _logger.warning('Error during PubkeyManager disposal', error, stackTrace); + } + + _pubkeysCache.clear(); + _watchedAssets.clear(); + _currentWalletId = null; + _logger.fine('Disposed'); } } diff --git a/packages/komodo_defi_sdk/pubspec.yaml b/packages/komodo_defi_sdk/pubspec.yaml index 0313a03d7..4ea0c34d5 100644 --- a/packages/komodo_defi_sdk/pubspec.yaml +++ b/packages/komodo_defi_sdk/pubspec.yaml @@ -45,6 +45,7 @@ dependencies: logging: any dev_dependencies: index_generator: ^4.0.1 + fake_async: ^1.3.3 mocktail: ^1.0.4 # test: ^1.25.7 test: ^1.25.7 diff --git a/packages/komodo_defi_sdk/test/balances/balance_manager_test.dart b/packages/komodo_defi_sdk/test/balances/balance_manager_test.dart new file mode 100644 index 000000000..f6aaf8ec7 --- /dev/null +++ b/packages/komodo_defi_sdk/test/balances/balance_manager_test.dart @@ -0,0 +1,212 @@ +import 'dart:async'; + +import 'package:decimal/decimal.dart'; +import 'package:komodo_defi_local_auth/komodo_defi_local_auth.dart'; +import 'package:komodo_defi_sdk/src/activation/shared_activation_coordinator.dart'; +import 'package:komodo_defi_sdk/src/assets/asset_lookup.dart'; +import 'package:komodo_defi_sdk/src/balances/balance_manager.dart'; +import 'package:komodo_defi_sdk/src/pubkeys/pubkey_manager.dart'; +import 'package:komodo_defi_types/komodo_defi_types.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:test/test.dart'; + +class _MockAuth extends Mock implements KomodoDefiLocalAuth {} + +class _MockActivationCoordinator extends Mock + implements SharedActivationCoordinator {} + +class _MockPubkeyManager extends Mock implements PubkeyManager {} + +class _MockAssetLookup extends Mock implements IAssetLookup {} + +void main() { + group('Dispose behavior for BalanceManager', () { + late _MockAuth auth; + late _MockActivationCoordinator activation; + late _MockPubkeyManager pubkeyManager; + late _MockAssetLookup assetLookup; + + setUp(() { + registerFallbackValue( + AssetId( + id: 'ATOM', + name: 'Cosmos', + symbol: AssetSymbol(assetConfigId: 'ATOM'), + chainId: AssetChainId(chainId: 118, decimalsValue: 6), + derivationPath: null, + subClass: CoinSubClass.tendermint, + ), + ); + auth = _MockAuth(); + activation = _MockActivationCoordinator(); + pubkeyManager = _MockPubkeyManager(); + assetLookup = _MockAssetLookup(); + }); + + test('dispose swallows cancel/close errors and is idempotent', () async { + // Arrange auth stream with throwing-cancel subscription + when( + () => auth.authStateChanges, + ).thenAnswer((_) => _StreamWithThrowingCancel()); + + final manager = BalanceManager( + assetLookup: assetLookup, + auth: auth, + pubkeyManager: pubkeyManager, + activationCoordinator: activation, + ); + + await manager.dispose(); + await manager.dispose(); + }); + + test('dispose during active watch stops further emissions', () async { + // Normal auth stream + final authChanges = StreamController.broadcast(); + when(() => auth.authStateChanges).thenAnswer((_) => authChanges.stream); + when(() => auth.currentUser).thenAnswer( + (_) async => const KdfUser( + walletId: WalletId( + name: 'w', + authOptions: AuthOptions(derivationMethod: DerivationMethod.iguana), + ), + isBip39Seed: false, + ), + ); + + // Asset and lookup + final assetId = AssetId( + id: 'ATOM', + name: 'Cosmos', + symbol: AssetSymbol(assetConfigId: 'ATOM'), + chainId: AssetChainId(chainId: 118, decimalsValue: 6), + derivationPath: null, + subClass: CoinSubClass.tendermint, + ); + final asset = Asset( + id: assetId, + protocol: TendermintProtocol.fromJson({ + 'type': 'Tendermint', + 'rpc_urls': [ + {'url': 'http://localhost:26657'}, + ], + }), + isWalletOnly: false, + signMessagePrefix: null, + ); + when(() => assetLookup.fromId(assetId)).thenReturn(asset); + + // Activation + when( + () => activation.isAssetActive(assetId), + ).thenAnswer((_) async => true); + + // Pubkey manager returns balance + when(() => pubkeyManager.getPubkeys(asset)).thenAnswer( + (_) async => AssetPubkeys( + assetId: assetId, + keys: [ + PubkeyInfo( + address: 'cosmos1pre', + derivationPath: null, + chain: null, + balance: BalanceInfo( + total: Decimal.zero, + spendable: Decimal.zero, + unspendable: Decimal.zero, + ), + coinTicker: assetId.id, + ), + ], + availableAddressesCount: 1, + syncStatus: SyncStatusEnum.success, + ), + ); + + final manager = BalanceManager( + assetLookup: assetLookup, + auth: auth, + pubkeyManager: pubkeyManager, + activationCoordinator: activation, + ); + + addTearDown(() async { + await manager.dispose(); + await authChanges.close(); + }); + + final events = []; + final sub = manager.watchBalance(assetId).listen(events.add); + + // Let initial microtasks run + await Future.delayed(const Duration(milliseconds: 10)); + + await manager.dispose(); + + // Change underlying return; should not emit anymore + when(() => pubkeyManager.getPubkeys(asset)).thenAnswer( + (_) async => AssetPubkeys( + assetId: assetId, + keys: [ + PubkeyInfo( + address: 'cosmos1new', + derivationPath: null, + chain: null, + balance: BalanceInfo( + total: Decimal.zero, + spendable: Decimal.zero, + unspendable: Decimal.zero, + ), + coinTicker: assetId.id, + ), + ], + availableAddressesCount: 1, + syncStatus: SyncStatusEnum.success, + ), + ); + + await Future.delayed(const Duration(seconds: 1)); + + expect(events, isNotEmpty); + await sub.cancel(); + }); + }); +} + +class _ThrowingCancelSubscription implements StreamSubscription { + @override + Future asFuture([E? futureValue]) => Completer().future; + + @override + Future cancel() => Future.error(Exception('cancel failed')); + + @override + bool get isPaused => false; + + @override + void onData(void Function(T data)? handleData) {} + + @override + void onDone(void Function()? handleDone) {} + + @override + void onError(Function? handleError) {} + + @override + void pause([Future? resumeSignal]) {} + + @override + void resume() {} +} + +class _StreamWithThrowingCancel extends Stream { + @override + StreamSubscription listen( + void Function(T event)? onData, { + Function? onError, + void Function()? onDone, + bool? cancelOnError, + }) { + return _ThrowingCancelSubscription(); + } +} diff --git a/packages/komodo_defi_sdk/test/pubkeys/pubkey_manager_test.dart b/packages/komodo_defi_sdk/test/pubkeys/pubkey_manager_test.dart new file mode 100644 index 000000000..f5e5c3c03 --- /dev/null +++ b/packages/komodo_defi_sdk/test/pubkeys/pubkey_manager_test.dart @@ -0,0 +1,744 @@ +// ignore_for_file: prefer_const_constructors + +import 'dart:async'; + +import 'package:decimal/decimal.dart'; +import 'package:fake_async/fake_async.dart'; +import 'package:komodo_defi_local_auth/komodo_defi_local_auth.dart'; +import 'package:komodo_defi_sdk/src/activation/shared_activation_coordinator.dart'; +import 'package:komodo_defi_sdk/src/pubkeys/pubkey_manager.dart'; +import 'package:komodo_defi_types/komodo_defi_types.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:test/test.dart'; + +class _MockApiClient extends Mock implements ApiClient {} + +class _MockAuth extends Mock implements KomodoDefiLocalAuth {} + +class _MockActivationCoordinator extends Mock + implements SharedActivationCoordinator {} + +void main() { + group('User stories and edge cases for PubkeyManager', () { + late _MockApiClient client; + late _MockAuth auth; + late _MockActivationCoordinator activation; + late StreamController authChanges; + late PubkeyManager manager; + + // Common test asset: single-address protocol (Tendermint) + late Asset tendermintAsset; + + setUpAll(() { + registerFallbackValue({}); + registerFallbackValue( + AssetId( + id: 'DUMMY', + name: 'Dummy', + symbol: AssetSymbol(assetConfigId: 'DUMMY'), + chainId: AssetChainId(chainId: 0, decimalsValue: 0), + derivationPath: null, + subClass: CoinSubClass.tendermint, + ), + ); + registerFallbackValue( + Asset( + id: AssetId( + id: 'DUMMY', + name: 'Dummy', + symbol: AssetSymbol(assetConfigId: 'DUMMY'), + chainId: AssetChainId(chainId: 0, decimalsValue: 0), + derivationPath: null, + subClass: CoinSubClass.tendermint, + ), + protocol: TendermintProtocol.fromJson({ + 'type': 'Tendermint', + 'rpc_urls': [ + {'url': 'http://localhost:26657'}, + ], + }), + isWalletOnly: false, + signMessagePrefix: null, + ), + ); + }); + + setUp(() { + client = _MockApiClient(); + auth = _MockAuth(); + activation = _MockActivationCoordinator(); + authChanges = StreamController.broadcast(); + + when(() => auth.authStateChanges).thenAnswer((_) => authChanges.stream); + + manager = PubkeyManager(client, auth, activation); + + // Minimal Tendermint asset (single-address) + final assetId = AssetId( + id: 'ATOM', + name: 'Cosmos', + symbol: AssetSymbol(assetConfigId: 'ATOM'), + chainId: AssetChainId(chainId: 118, decimalsValue: 6), + derivationPath: null, + subClass: CoinSubClass.tendermint, + ); + final protocol = TendermintProtocol.fromJson({ + 'type': 'Tendermint', + 'rpc_urls': [ + {'url': 'http://localhost:26657'}, + ], + }); + tendermintAsset = Asset( + id: assetId, + protocol: protocol, + isWalletOnly: false, + signMessagePrefix: null, + ); + }); + + tearDown(() async { + await manager.dispose(); + await authChanges.close(); + }); + + KdfUser nonHdUser() => KdfUser( + walletId: WalletId( + name: 'test-wallet', + authOptions: AuthOptions(derivationMethod: DerivationMethod.iguana), + ), + isBip39Seed: false, + ); + + Future stubActivationAlwaysActive(Asset asset) async { + when( + () => activation.isAssetActive(asset.id), + ).thenAnswer((_) async => true); + when( + () => activation.activateAsset(asset), + ).thenAnswer((_) async => ActivationResult.success(asset.id)); + } + + void stubWalletMyBalance({ + required String address, + required String coin, + Decimal? total, + Decimal? unspendable, + }) { + when(() => client.executeRpc(any())).thenAnswer((invocation) async { + final req = + invocation.positionalArguments.first as Map; + final method = req['method'] as String?; + if (method == 'my_balance') { + return { + 'address': address, + 'balance': (total ?? Decimal.zero).toString(), + 'unspendable_balance': (unspendable ?? Decimal.zero).toString(), + 'coin': coin, + }; + } + if (method == 'unban_pubkeys') { + return { + 'result': { + 'still_banned': {}, + 'unbanned': {}, + 'were_not_banned': [], + }, + }; + } + // Default minimal success for other RPCs that might appear + return {'result': {}}; + }); + } + + test( + 'getPubkeys returns single address for single-address protocol', + () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + await stubActivationAlwaysActive(tendermintAsset); + + stubWalletMyBalance(address: 'cosmos1abc', coin: tendermintAsset.id.id); + + final result = await manager.getPubkeys(tendermintAsset); + expect(result.assetId, tendermintAsset.id); + expect(result.keys, hasLength(1)); + expect(result.keys.first.address, 'cosmos1abc'); + }, + ); + + test( + 'createNewPubkey throws UnsupportedError for single-address assets', + () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + await stubActivationAlwaysActive(tendermintAsset); + + stubWalletMyBalance(address: 'cosmos1abc', coin: tendermintAsset.id.id); + + expect( + () => manager.createNewPubkey(tendermintAsset), + throwsA(isA()), + ); + }, + ); + + test( + 'createNewPubkeyStream yields error for single-address assets', + () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + await stubActivationAlwaysActive(tendermintAsset); + + stubWalletMyBalance(address: 'cosmos1abc', coin: tendermintAsset.id.id); + + final states = + await manager + .watchCreateNewPubkey(tendermintAsset) + .take(1) + .toList(); + expect(states.single.status, NewAddressStatus.error); + }, + ); + + test('unbanPubkeys delegates to RPC and returns result', () async { + // auth not required here + stubWalletMyBalance(address: 'cosmos1abc', coin: tendermintAsset.id.id); + + final res = await manager.unbanPubkeys(const UnbanBy.all()); + expect(res.isEmpty, isTrue); + }); + + test( + 'watchPubkeys emits last known immediately, then same via controller, then refreshed value', + () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + await stubActivationAlwaysActive(tendermintAsset); + + // First response used for preCache + stubWalletMyBalance(address: 'cosmos1pre', coin: tendermintAsset.id.id); + await manager.precachePubkeys(tendermintAsset); + + // Update the stub to simulate a new address on refresh + stubWalletMyBalance(address: 'cosmos1new', coin: tendermintAsset.id.id); + + final stream = manager.watchPubkeys(tendermintAsset); + + // First emit is immediate lastKnown, second is same from controller, third is refreshed value + final firstThree = await stream.take(3).toList(); + expect(firstThree[0].keys.first.address, 'cosmos1pre'); + expect(firstThree[2].keys.first.address, 'cosmos1new'); + }, + ); + + test('watchPubkeys respects polling interval (~30s)', () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + await stubActivationAlwaysActive(tendermintAsset); + + // Initial cache + stubWalletMyBalance(address: 'cosmos1pre', coin: tendermintAsset.id.id); + await manager.precachePubkeys(tendermintAsset); + + fakeAsync((FakeAsync async) { + // After start, we set a different address for the immediate refresh + stubWalletMyBalance( + address: 'cosmos1poll1', + coin: tendermintAsset.id.id, + ); + + final emitted = []; + final sub = manager.watchPubkeys(tendermintAsset).listen((e) { + emitted.add(e.keys.first.address); + }); + + // Allow the immediate refresh to occur + async.flushMicrotasks(); + expect(emitted.contains('cosmos1poll1'), isTrue); + + // Prepare next poll result and ensure it's not emitted before 30s + stubWalletMyBalance( + address: 'cosmos1poll2', + coin: tendermintAsset.id.id, + ); + async + ..elapse(Duration(seconds: 29)) + ..flushMicrotasks(); + expect(emitted.contains('cosmos1poll2'), isFalse); + + // Hitting 30s should emit the next poll + async + ..elapse(Duration(seconds: 1)) + ..flushMicrotasks(); + expect(emitted.contains('cosmos1poll2'), isTrue); + + unawaited(sub.cancel()); + }); + }); + + test('watchPubkeys stops and new watches throw after dispose', () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + await stubActivationAlwaysActive(tendermintAsset); + + stubWalletMyBalance(address: 'cosmos1pre', coin: tendermintAsset.id.id); + await manager.precachePubkeys(tendermintAsset); + + final received = []; + final sub = manager.watchPubkeys(tendermintAsset).listen((e) { + received.add(e.keys.first.address); + }); + + // Cancel current subscription before disposing + await sub.cancel(); + await manager.dispose(); + + // After dispose, starting a new watch should throw on listen + expect( + () => manager.watchPubkeys(tendermintAsset).first, + throwsA(isA()), + ); + }); + + test('watchPubkeys updates lastKnown after emission', () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + await stubActivationAlwaysActive(tendermintAsset); + + stubWalletMyBalance(address: 'cosmos1pre', coin: tendermintAsset.id.id); + await manager.precachePubkeys(tendermintAsset); + + // Change to a new address which should update via immediate get in start + stubWalletMyBalance(address: 'cosmos1start', coin: tendermintAsset.id.id); + + final first = await manager.watchPubkeys(tendermintAsset).first; + expect(first.keys.first.address, isNotEmpty); + + // lastKnown should be updated to latest emitted value + final cached = manager.lastKnown(tendermintAsset.id); + expect(cached, isNotNull); + expect(cached!.keys.first.address, first.keys.first.address); + }); + + test( + 'watchPubkeys with activateIfNeeded=false only emits last known if inactive', + () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + + // Pre-cache with initial address + when( + () => activation.activateAsset(tendermintAsset), + ).thenAnswer((_) async => ActivationResult.success(tendermintAsset.id)); + when( + () => activation.isAssetActive(tendermintAsset.id), + ).thenAnswer((_) async => true); + stubWalletMyBalance(address: 'cosmos1pre', coin: tendermintAsset.id.id); + await manager.precachePubkeys(tendermintAsset); + + // Now simulate inactive asset and disable activation on watch + when( + () => activation.isAssetActive(tendermintAsset.id), + ).thenAnswer((_) async => false); + // If it were to fetch, it would get this new address, but it should not + stubWalletMyBalance(address: 'cosmos1new', coin: tendermintAsset.id.id); + + final stream = manager.watchPubkeys( + tendermintAsset, + activateIfNeeded: false, + ); + + // Give stream a brief moment to potentially emit more; should only emit one + final received = + await stream.timeout(Duration(milliseconds: 200)).take(1).toList(); + expect(received.single.keys.first.address, 'cosmos1pre'); + }, + ); + + test( + 'lastKnown returns null when no cache; updates after preCache', + () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + when( + () => activation.isAssetActive(tendermintAsset.id), + ).thenAnswer((_) async => true); + when( + () => activation.activateAsset(tendermintAsset), + ).thenAnswer((_) async => ActivationResult.success(tendermintAsset.id)); + + expect(manager.lastKnown(tendermintAsset.id), isNull); + + stubWalletMyBalance(address: 'cosmos1pre', coin: tendermintAsset.id.id); + await manager.precachePubkeys(tendermintAsset); + + final cached = manager.lastKnown(tendermintAsset.id); + expect(cached, isNotNull); + expect(cached!.keys.first.address, 'cosmos1pre'); + }, + ); + + test('auth wallet change resets state and clears cache', () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + when( + () => activation.isAssetActive(tendermintAsset.id), + ).thenAnswer((_) async => true); + when( + () => activation.activateAsset(tendermintAsset), + ).thenAnswer((_) async => ActivationResult.success(tendermintAsset.id)); + + stubWalletMyBalance(address: 'cosmos1pre', coin: tendermintAsset.id.id); + await manager.precachePubkeys(tendermintAsset); + expect(manager.lastKnown(tendermintAsset.id), isNotNull); + + // Emit new user with different wallet ID + final newUser = KdfUser( + walletId: WalletId( + name: 'other', + authOptions: AuthOptions(derivationMethod: DerivationMethod.iguana), + ), + isBip39Seed: false, + ); + authChanges.add(newUser); + await Future.delayed(Duration(milliseconds: 50)); + + expect(manager.lastKnown(tendermintAsset.id), isNull); + }); + + test( + 'auth wallet change emits error and restarts watching on same subscription', + () async { + // Arrange: setup auth to return a mutable current user + final user1 = nonHdUser(); + KdfUser current = user1; + when(() => auth.currentUser).thenAnswer((_) async => current); + await stubActivationAlwaysActive(tendermintAsset); + + // Prime cache and first fetches + stubWalletMyBalance(address: 'cosmos1pre', coin: tendermintAsset.id.id); + await manager.precachePubkeys(tendermintAsset); + stubWalletMyBalance( + address: 'cosmos1first', + coin: tendermintAsset.id.id, + ); + + final emitted = []; + final errors = []; + final sub = manager + .watchPubkeys(tendermintAsset) + .listen( + (pubkeys) => emitted.add(pubkeys.keys.first.address), + onError: errors.add, + ); + + // Allow immediate refresh + await Future.delayed(Duration(milliseconds: 10)); + + // Act: change wallet and ensure new value is fetched on the same subscription + stubWalletMyBalance( + address: 'cosmos1afterChange', + coin: tendermintAsset.id.id, + ); + final user2 = KdfUser( + walletId: WalletId( + name: 'other', + authOptions: AuthOptions(derivationMethod: DerivationMethod.iguana), + ), + isBip39Seed: false, + ); + current = user2; // update what auth.currentUser returns + authChanges.add(user2); + + // Assert: receive an error and then a new emission without re-subscribing + await Future.delayed(Duration(milliseconds: 80)); + expect(errors.whereType(), isNotEmpty); + // The controller remains open and should emit after restart + expect(emitted.contains('cosmos1afterChange'), isTrue); + + await sub.cancel(); + }, + ); + + test( + 'watchPubkeys second subscriber receives immediate lastKnown when controller exists (due to immediate yield)', + () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + await stubActivationAlwaysActive(tendermintAsset); + + fakeAsync((async) { + // Seed cache + stubWalletMyBalance( + address: 'cosmos1pre', + coin: tendermintAsset.id.id, + ); + // First fetch result for immediate refresh + stubWalletMyBalance( + address: 'cosmos1first', + coin: tendermintAsset.id.id, + ); + + final s1Events = []; + final sub1 = manager.watchPubkeys(tendermintAsset).listen((e) { + s1Events.add(e.keys.first.address); + }); + + // Let initial get happen + async.flushMicrotasks(); + + // Prepare next poll result + stubWalletMyBalance( + address: 'cosmos1poll', + coin: tendermintAsset.id.id, + ); + + // Second subscriber joins AFTER controller already active + final s2Events = []; + final sub2 = manager.watchPubkeys(tendermintAsset).listen((e) { + s2Events.add(e.keys.first.address); + }); + + // With immediate yield reintroduced, second subscriber sees immediate lastKnown + async.flushMicrotasks(); + expect(s2Events, isNotEmpty); + + // Only after the next polling tick (~30s) should second subscriber receive a new value + async + ..elapse(const Duration(seconds: 30)) + ..flushMicrotasks(); + + expect(s2Events, contains('cosmos1poll')); + + unawaited(sub1.cancel()); + unawaited(sub2.cancel()); + }); + }, + ); + + test( + 'watchPubkeys activateIfNeeded is sticky per controller (first subscriber decides)', + () async { + final user = nonHdUser(); + when(() => auth.currentUser).thenAnswer((_) async => user); + + // Start as inactive; do NOT allow activation on first subscriber + when( + () => activation.isAssetActive(tendermintAsset.id), + ).thenAnswer((_) async => false); + when( + () => activation.activateAsset(tendermintAsset), + ).thenAnswer((_) async => ActivationResult.success(tendermintAsset.id)); + + // Do NOT pre-cache; we want to ensure no activation occurs and no emissions happen + + // First subscriber: activateIfNeeded=false (controller is created here) + final s1Events = []; + final s1 = manager + .watchPubkeys(tendermintAsset, activateIfNeeded: false) + .listen((e) { + s1Events.add(e.keys.first.address); + }); + + // Allow initial onListen to run + await Future.delayed(const Duration(milliseconds: 10)); + + // Second subscriber: activateIfNeeded=true but controller already exists + final s2Events = []; + final s2 = manager.watchPubkeys(tendermintAsset).listen((e) { + s2Events.add(e.keys.first.address); + }); + + // Give listeners a brief moment + await Future.delayed(const Duration(milliseconds: 10)); + + // Because the controller was created with activateIfNeeded=false, no activation should occur + verifyNever(() => activation.activateAsset(tendermintAsset)); + + // No emissions should occur since activation is disabled and asset inactive + expect(s1Events, isEmpty); + expect(s2Events, isEmpty); + + // Clean up + await s1.cancel(); + await s2.cancel(); + }, + ); + + test('dispose prevents further access', () async { + await manager.dispose(); + expect( + () => manager.lastKnown(tendermintAsset.id), + throwsA(isA()), + ); + }); + }); + + group('Dispose behavior for PubkeyManager', () { + late _MockApiClient client; + late _MockAuth auth; + late _MockActivationCoordinator activation; + + setUp(() { + client = _MockApiClient(); + auth = _MockAuth(); + activation = _MockActivationCoordinator(); + }); + + test( + 'dispose swallows auth subscription cancel errors and is idempotent', + () async { + // Arrange auth stream that returns a subscription whose cancel throws + when( + () => auth.authStateChanges, + ).thenAnswer((_) => _StreamWithThrowingCancel()); + + final manager = PubkeyManager(client, auth, activation); + + // Act + Assert: dispose does not throw even if cancel throws + await manager.dispose(); + // Idempotent + await manager.dispose(); + }, + ); + + test( + 'dispose during active watch stops further emissions (no race with timers)', + () async { + // Normal auth stream + final authChanges = StreamController.broadcast(); + when(() => auth.authStateChanges).thenAnswer((_) => authChanges.stream); + when(() => auth.currentUser).thenAnswer( + (_) async => KdfUser( + walletId: WalletId( + name: 'w', + authOptions: AuthOptions( + derivationMethod: DerivationMethod.iguana, + ), + ), + isBip39Seed: false, + ), + ); + + final manager = PubkeyManager(client, auth, activation); + addTearDown(() async { + await manager.dispose(); + await authChanges.close(); + }); + + // Active asset + when( + () => activation.isAssetActive(any()), + ).thenAnswer((_) async => true); + when(() => activation.activateAsset(any())).thenAnswer(( + invocation, + ) async { + final assetArg = invocation.positionalArguments.first as Asset; + return ActivationResult.success(assetArg.id); + }); + + // Provide a minimal single-address asset and RPC stub + final assetId = AssetId( + id: 'ATOM', + name: 'Cosmos', + symbol: AssetSymbol(assetConfigId: 'ATOM'), + chainId: AssetChainId(chainId: 118, decimalsValue: 6), + derivationPath: null, + subClass: CoinSubClass.tendermint, + ); + final asset = Asset( + id: assetId, + protocol: TendermintProtocol.fromJson({ + 'type': 'Tendermint', + 'rpc_urls': [ + {'url': 'http://localhost:26657'}, + ], + }), + isWalletOnly: false, + signMessagePrefix: null, + ); + + when(() => client.executeRpc(any())).thenAnswer((invocation) async { + final req = + invocation.positionalArguments.first as Map; + final method = req['method'] as String?; + if (method == 'my_balance') { + return { + 'address': 'cosmos1pre', + 'balance': '0', + 'unspendable_balance': '0', + 'coin': assetId.id, + }; + } + return {'result': {}}; + }); + + // Start watch + final events = []; + final sub = manager.watchPubkeys(asset).listen(events.add); + + // Allow initial microtasks + await Future.delayed(const Duration(milliseconds: 10)); + final initial = events.length; + + // Now dispose while timer could schedule next polls + await manager.dispose(); + + // Change RPC response that would be observed if polling still alive + when(() => client.executeRpc(any())).thenAnswer((invocation) async { + return { + 'address': 'cosmos1new', + 'balance': '0', + 'unspendable_balance': '0', + 'coin': assetId.id, + }; + }); + + // Wait longer than polling interval to ensure nothing else emitted + await Future.delayed(const Duration(seconds: 1)); + + // Assert: stream should not emit after dispose + expect(events.length, initial); + await sub.cancel(); + }, + ); + }); +} + +class _ThrowingCancelSubscription implements StreamSubscription { + @override + Future asFuture([E? futureValue]) => Completer().future; + + @override + Future cancel() => Future.error(Exception('cancel failed')); + + @override + bool get isPaused => false; + + @override + void onData(void Function(T data)? handleData) {} + + @override + void onDone(void Function()? handleDone) {} + + @override + void onError(Function? handleError) {} + + @override + void pause([Future? resumeSignal]) {} + + @override + void resume() {} +} + +class _StreamWithThrowingCancel extends Stream { + @override + StreamSubscription listen( + void Function(T event)? onData, { + Function? onError, + void Function()? onDone, + bool? cancelOnError, + }) { + return _ThrowingCancelSubscription(); + } +} diff --git a/packages/komodo_defi_sdk/test/src/komodo_defi_sdk_test.dart b/packages/komodo_defi_sdk/test/src/komodo_defi_sdk_test.dart deleted file mode 100644 index c09a30759..000000000 --- a/packages/komodo_defi_sdk/test/src/komodo_defi_sdk_test.dart +++ /dev/null @@ -1,11 +0,0 @@ -// ignore_for_file: prefer_const_constructors -import 'package:komodo_defi_sdk/komodo_defi_sdk.dart'; -import 'package:test/test.dart'; - -void main() { - group('KomodoDefiSdk', () { - test('can be instantiated', () { - expect(KomodoDefiSdk(), isNotNull); - }); - }); -}