From 848d23069e158e4568c40b50d9df122bfb34a6bc Mon Sep 17 00:00:00 2001 From: Yosuke Otosu Date: Wed, 30 Jul 2025 13:19:18 +0900 Subject: [PATCH] Add watchers for ChangeToRedeemed and ChangeToCanceled events with notification processing --- app/model/db/notification.py | 2 + batch/processor_Notifications_Token.py | 142 ++- .../notification_Notifications_GET_test.py | 4 +- .../processor_Notifications_Token_test.py | 852 ++++++++++++++++++ tests/contract_modules.py | 18 + 5 files changed, 1006 insertions(+), 12 deletions(-) diff --git a/app/model/db/notification.py b/app/model/db/notification.py index 3d19f4cec..98f0ae81f 100644 --- a/app/model/db/notification.py +++ b/app/model/db/notification.py @@ -203,6 +203,8 @@ class NotificationType(StrEnum): CANCEL_TRANSFER = "CancelTransfer" FORCE_LOCK = "ForceLock" FORCE_UNLOCK = "ForceUnlock" + CHANGE_TO_REDEEMED = "ChangeToRedeemed" + CHANGE_TO_CANCELED = "ChangeToCanceled" class NotificationBlockNumber(Base): diff --git a/batch/processor_Notifications_Token.py b/batch/processor_Notifications_Token.py index 1153b9599..1d3f74ad1 100644 --- a/batch/processor_Notifications_Token.py +++ b/batch/processor_Notifications_Token.py @@ -45,6 +45,7 @@ NotificationBlockNumber, NotificationType, ) +from app.model.schema.base import TokenType from app.utils.asyncio_utils import SemaphoreTaskGroup from app.utils.company_list import CompanyList from app.utils.web3_utils import AsyncWeb3Wrapper @@ -70,10 +71,21 @@ class Watcher: contract_cache: dict[str, Web3AsyncContract] = {} - def __init__(self, filter_name: str, filter_params: dict, notification_type: str): + def __init__( + self, + filter_name: str, + filter_params: dict, + notification_type: str, + token_type_list: list[TokenType] = None, + skip_past_data_on_initial_sync: bool = False, + ): + if token_type_list is None: + token_type_list = list([]) self.filter_name = filter_name self.filter_params = filter_params self.notification_type = notification_type + self.token_type_list = token_type_list + self.skip_past_data_on_initial_sync = skip_past_data_on_initial_sync @staticmethod def _gen_notification_id(entry, option_type=0): @@ -91,15 +103,19 @@ async def _gen_block_timestamp(entry): ).replace(tzinfo=None) @staticmethod - async def _get_token_all_list(db_session: AsyncSession): + async def _get_token_all_list( + db_session: AsyncSession, token_type_list: list[TokenType] + ): _tokens = [] + + stmt = select(IDXTokenListRegister).join( + Listing, + and_(Listing.token_address == IDXTokenListRegister.token_address), + ) + if len(token_type_list) != 0: + stmt = stmt.where(IDXTokenListRegister.token_template.in_(token_type_list)) registered_tokens: Sequence[IDXTokenListRegister] = ( - await db_session.scalars( - select(IDXTokenListRegister).join( - Listing, - and_(Listing.token_address == IDXTokenListRegister.token_address), - ) - ) + await db_session.scalars(stmt) ).all() for registered_token in registered_tokens: _tokens.append( @@ -126,7 +142,9 @@ async def loop(self): try: # Get listed tokens - _token_list = await self._get_token_all_list(db_session) + _token_list = await self._get_token_all_list( + db_session, self.token_type_list + ) latest_block_number = await async_web3.eth.block_number for _token in _token_list: @@ -136,6 +154,8 @@ async def loop(self): db_session=db_session, contract_address=_token["token"].token_address, notification_type=self.notification_type, + latest_block_number=latest_block_number, + skip_past_data_on_initial_sync=self.skip_past_data_on_initial_sync, ) + 1 ) @@ -210,7 +230,11 @@ async def loop(self): @staticmethod async def __get_synchronized_block_number( - db_session: AsyncSession, contract_address: str, notification_type: str + db_session: AsyncSession, + contract_address: str, + notification_type: str, + latest_block_number: int, + skip_past_data_on_initial_sync: bool, ): """Get latest synchronized blockNumber""" notification_block_number: NotificationBlockNumber | None = ( @@ -222,6 +246,8 @@ async def __get_synchronized_block_number( ) ).first() if notification_block_number is None: + if skip_past_data_on_initial_sync is True: + return latest_block_number - 1 return -1 else: return notification_block_number.latest_block_number @@ -515,6 +541,100 @@ async def db_merge( await db_session.merge(notification) +class WatchChangeToRedeemed(Watcher): + """Watch ChangeToRedeemed Event for Bond Token + + - Process for registering a notification when a token status is changed to redeemed. + """ + + def __init__(self): + super().__init__( + filter_name="ChangeToRedeemed", + filter_params={}, + notification_type=NotificationType.CHANGE_TO_REDEEMED, + token_type_list=[TokenType.IbetStraightBond], + skip_past_data_on_initial_sync=True, + ) + + async def db_merge( + self, + db_session: AsyncSession, + token_contract: Web3AsyncContract, + token_type: str, + log_entries: list[EventData], + token_owner_address: str, + ): + company_list = await CompanyList.get() + token_name = await AsyncContract.call_function( + contract=token_contract, function_name="name", args=(), default_returns="" + ) + for entry in log_entries: + company = company_list.find(token_owner_address) + metadata = { + "company_name": company.corporate_name, + "token_address": entry["address"], + "token_name": token_name, + "exchange_address": "", + "token_type": token_type, + } + notification = Notification() + notification.notification_id = self._gen_notification_id(entry) + notification.notification_type = self.notification_type + notification.priority = 0 + notification.address = None + notification.block_timestamp = await self._gen_block_timestamp(entry) + notification.args = dict(entry["args"]) + notification.metainfo = metadata + await db_session.merge(notification) + + +class WatchChangeToCanceled(Watcher): + """Watch ChangeToCanceled Event for Share Token + + - Process for registering a notification when a token status is changed to canceled. + """ + + def __init__(self): + super().__init__( + filter_name="ChangeToCanceled", + filter_params={}, + notification_type=NotificationType.CHANGE_TO_CANCELED, + token_type_list=[TokenType.IbetShare], + skip_past_data_on_initial_sync=True, + ) + + async def db_merge( + self, + db_session: AsyncSession, + token_contract: Web3AsyncContract, + token_type: str, + log_entries: list[EventData], + token_owner_address: str, + ): + company_list = await CompanyList.get() + token_name = await AsyncContract.call_function( + contract=token_contract, function_name="name", args=(), default_returns="" + ) + for entry in log_entries: + company = company_list.find(token_owner_address) + metadata = { + "company_name": company.corporate_name, + "token_address": entry["address"], + "token_name": token_name, + "exchange_address": "", + "token_type": token_type, + } + notification = Notification() + notification.notification_id = self._gen_notification_id(entry) + notification.notification_type = self.notification_type + notification.priority = 0 + notification.address = None + notification.block_timestamp = await self._gen_block_timestamp(entry) + notification.args = dict(entry["args"]) + notification.metainfo = metadata + await db_session.merge(notification) + + # メイン処理 async def main(): watchers = [ @@ -524,6 +644,8 @@ async def main(): WatchCancelTransfer(), WatchForceLock(), WatchForceUnlock(), + WatchChangeToRedeemed(), + WatchChangeToCanceled(), ] LOG.info("Service started successfully") diff --git a/tests/app/notification_Notifications_GET_test.py b/tests/app/notification_Notifications_GET_test.py index c3281ab6c..c27d966ba 100644 --- a/tests/app/notification_Notifications_GET_test.py +++ b/tests/app/notification_Notifications_GET_test.py @@ -544,10 +544,10 @@ def test_error_1(self, client: TestClient, session: Session): { "type": "enum", "loc": ["query", "notification_type"], - "msg": "Input should be 'NewOrder', 'NewOrderCounterpart', 'CancelOrder', 'CancelOrderCounterpart', 'ForceCancelOrder', 'BuyAgreement', 'BuySettlementOK', 'BuySettlementNG', 'SellAgreement', 'SellSettlementOK', 'SellSettlementNG', 'Transfer', 'ApplyForTransfer', 'ApproveTransfer', 'CancelTransfer', 'ForceLock' or 'ForceUnlock'", + "msg": "Input should be 'NewOrder', 'NewOrderCounterpart', 'CancelOrder', 'CancelOrderCounterpart', 'ForceCancelOrder', 'BuyAgreement', 'BuySettlementOK', 'BuySettlementNG', 'SellAgreement', 'SellSettlementOK', 'SellSettlementNG', 'Transfer', 'ApplyForTransfer', 'ApproveTransfer', 'CancelTransfer', 'ForceLock', 'ForceUnlock', 'ChangeToRedeemed' or 'ChangeToCanceled'", "input": "hoge", "ctx": { - "expected": "'NewOrder', 'NewOrderCounterpart', 'CancelOrder', 'CancelOrderCounterpart', 'ForceCancelOrder', 'BuyAgreement', 'BuySettlementOK', 'BuySettlementNG', 'SellAgreement', 'SellSettlementOK', 'SellSettlementNG', 'Transfer', 'ApplyForTransfer', 'ApproveTransfer', 'CancelTransfer', 'ForceLock' or 'ForceUnlock'" + "expected": "'NewOrder', 'NewOrderCounterpart', 'CancelOrder', 'CancelOrderCounterpart', 'ForceCancelOrder', 'BuyAgreement', 'BuySettlementOK', 'BuySettlementNG', 'SellAgreement', 'SellSettlementOK', 'SellSettlementNG', 'Transfer', 'ApplyForTransfer', 'ApproveTransfer', 'CancelTransfer', 'ForceLock', 'ForceUnlock', 'ChangeToRedeemed' or 'ChangeToCanceled'" }, }, { diff --git a/tests/batch/processor_Notifications_Token_test.py b/tests/batch/processor_Notifications_Token_test.py index daab5aab3..6293133dc 100644 --- a/tests/batch/processor_Notifications_Token_test.py +++ b/tests/batch/processor_Notifications_Token_test.py @@ -42,16 +42,20 @@ from tests.account_config import eth_account from tests.conftest import DeployedContract, SharedContract, UnitTestAccount from tests.contract_modules import ( + bond_change_to_redeemed, coupon_register_list, coupon_transfer_to_exchange, coupon_withdraw_from_exchange, + issue_bond_token, issue_coupon_token, issue_share_token, + register_bond_list, register_personalinfo, register_share_list, share_apply_for_transfer, share_approve_transfer, share_cancel_transfer, + share_change_to_canceled, share_force_lock, share_force_unlock, share_set_transfer_approval_required, @@ -121,6 +125,64 @@ async def prepare_coupon_token( return token +async def prepare_bond_token( + issuer: UnitTestAccount, + exchange: DeployedContract, + token_list: DeployedContract, + personal_info: DeployedContract, + async_session: AsyncSession, +): + # Issue token + args = { + "name": "テスト債券", + "symbol": "BOND", + "totalSupply": 1000000, + "tradableExchange": exchange["address"], + "faceValue": 10000, + "interestRate": 602, + "interestPaymentDate1": "0101", + "interestPaymentDate2": "0201", + "interestPaymentDate3": "0301", + "interestPaymentDate4": "0401", + "interestPaymentDate5": "0501", + "interestPaymentDate6": "0601", + "interestPaymentDate7": "0701", + "interestPaymentDate8": "0801", + "interestPaymentDate9": "0901", + "interestPaymentDate10": "1001", + "interestPaymentDate11": "1101", + "interestPaymentDate12": "1201", + "redemptionDate": "20191231", + "redemptionValue": 10000, + "returnDate": "20191231", + "returnAmount": "商品券をプレゼント", + "purpose": "新商品の開発資金として利用。", + "memo": "メモ", + "contactInformation": "問い合わせ先", + "privacyPolicy": "プライバシーポリシー", + "personalInfoAddress": personal_info["address"], + "transferable": True, + "isRedeemed": False, + "faceValueCurrency": "JPY", + "interestPaymentCurrency": "JPY", + "redemptionValueCurrency": "JPY", + "baseFxRate": "", + } + token = issue_bond_token(issuer, args) + register_bond_list(issuer, token, token_list) + + _listing = Listing() + _listing.token_address = token["address"] + _listing.is_public = True + _listing.max_holding_quantity = 1000000 + _listing.max_sell_amount = 1000000 + _listing.owner_address = issuer["account_address"] + async_session.add(_listing) + await async_session.commit() + + return token + + async def prepare_share_token( issuer: UnitTestAccount, exchange: DeployedContract, @@ -2241,3 +2303,793 @@ async def test_error_1( await async_session.scalars(select(NotificationBlockNumber).limit(1)) ).first() assert _notification_block_number is None + + +@pytest.mark.asyncio +class TestWatchChangeToRedeemed: + issuer = eth_account["issuer"] + lock_account = eth_account["user1"] + + ########################################################################### + # Normal Case + ########################################################################### + + # + # Initial Sync + async def test_normal_1( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToRedeemed") + exchange_contract = shared_contract["IbetStraightBondExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_bond_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetStraightBond" + async_session.add(idx_token_list_item) + await async_session.commit() + + # Run target process + await watcher.loop() + + # Assertion + block_number = web3.eth.block_number + + _notification = ( + await async_session.scalars( + select(Notification) + .where( + Notification.notification_id + == "0x{:012x}{:06x}{:06x}{:02x}".format(block_number, 0, 0, 0) + ) + .limit(1) + ) + ).first() + assert _notification is None + + _notification_block_number: NotificationBlockNumber = ( + await async_session.scalars( + select(NotificationBlockNumber) + .where( + and_( + NotificationBlockNumber.notification_type + == NotificationType.CHANGE_TO_REDEEMED, + NotificationBlockNumber.contract_address == token["address"], + ) + ) + .limit(1) + ) + ).first() + assert _notification_block_number.latest_block_number == block_number + + # + # Single event logs + async def test_normal_2( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToRedeemed") + exchange_contract = shared_contract["IbetStraightBondExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_bond_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetStraightBond" + async_session.add(idx_token_list_item) + + notification_block_number = NotificationBlockNumber() + notification_block_number.notification_type = ( + NotificationType.CHANGE_TO_REDEEMED + ) + notification_block_number.contract_address = token["address"] + notification_block_number.latest_block_number = web3.eth.block_number + async_session.add(notification_block_number) + await async_session.commit() + + # Emit ChangeToRedeemed event + bond_change_to_redeemed( + invoker=self.issuer, + token=token, + ) + + # Run target process + await watcher.loop() + + # Assertion + block_number = web3.eth.block_number + + async_session.expunge_all() + + _notification_list = (await async_session.scalars(select(Notification))).all() + assert len(_notification_list) == 1 + assert ( + _notification_list[0].notification_type + == NotificationType.CHANGE_TO_REDEEMED + ) + assert _notification_list[0].priority == 0 + assert _notification_list[0].address is None + assert _notification_list[0].args == {} + assert _notification_list[0].metainfo == { + "company_name": "株式会社DEMO", + "exchange_address": "", + "token_address": token["address"], + "token_name": "テスト債券", + "token_type": "IbetStraightBond", + } + + _notification_block_number: NotificationBlockNumber = ( + await async_session.scalars( + select(NotificationBlockNumber) + .where( + and_( + NotificationBlockNumber.notification_type + == NotificationType.CHANGE_TO_REDEEMED, + NotificationBlockNumber.contract_address == token["address"], + ) + ) + .limit(1) + ) + ).first() + assert _notification_block_number.latest_block_number == block_number + + # + # Multi event logs + async def test_normal_3( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToRedeemed") + exchange_contract = shared_contract["IbetStraightBondExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_bond_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetStraightBond" + async_session.add(idx_token_list_item) + + notification_block_number = NotificationBlockNumber() + notification_block_number.notification_type = ( + NotificationType.CHANGE_TO_REDEEMED + ) + notification_block_number.contract_address = token["address"] + notification_block_number.latest_block_number = web3.eth.block_number + async_session.add(notification_block_number) + await async_session.commit() + + # Emit ChangeToRedeemed event + bond_change_to_redeemed( + invoker=self.issuer, + token=token, + ) + bond_change_to_redeemed( + invoker=self.issuer, + token=token, + ) + + # Run target process + await watcher.loop() + + # Assertion + block_number = web3.eth.block_number + + async_session.expunge_all() + + _notification_list = (await async_session.scalars(select(Notification))).all() + assert len(_notification_list) == 2 + assert ( + _notification_list[0].notification_type + == NotificationType.CHANGE_TO_REDEEMED + ) + assert _notification_list[0].priority == 0 + assert _notification_list[0].address is None + assert _notification_list[0].args == {} + assert _notification_list[0].metainfo == { + "company_name": "株式会社DEMO", + "exchange_address": "", + "token_address": token["address"], + "token_name": "テスト債券", + "token_type": "IbetStraightBond", + } + assert ( + _notification_list[1].notification_type + == NotificationType.CHANGE_TO_REDEEMED + ) + assert _notification_list[1].priority == 0 + assert _notification_list[1].address is None + assert _notification_list[1].args == {} + assert _notification_list[1].metainfo == { + "company_name": "株式会社DEMO", + "exchange_address": "", + "token_address": token["address"], + "token_name": "テスト債券", + "token_type": "IbetStraightBond", + } + + _notification_block_number: NotificationBlockNumber = ( + await async_session.scalars( + select(NotificationBlockNumber) + .where( + and_( + NotificationBlockNumber.notification_type + == NotificationType.CHANGE_TO_REDEEMED, + NotificationBlockNumber.contract_address == token["address"], + ) + ) + .limit(1) + ) + ).first() + assert _notification_block_number.latest_block_number == block_number + + # + # No event logs + async def test_normal_4( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToRedeemed") + exchange_contract = shared_contract["IbetStraightBondExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_bond_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetStraightBond" + async_session.add(idx_token_list_item) + + notification_block_number = NotificationBlockNumber() + notification_block_number.notification_type = ( + NotificationType.CHANGE_TO_REDEEMED + ) + notification_block_number.contract_address = token["address"] + notification_block_number.latest_block_number = web3.eth.block_number + async_session.add(notification_block_number) + await async_session.commit() + + # Not emit ChangeToRedeemed event + pass + + # Run target process + await watcher.loop() + + # Assertion + block_number = web3.eth.block_number + + _notification_list = (await async_session.scalars(select(Notification))).all() + assert len(_notification_list) == 0 + + _notification_block_number = ( + await async_session.scalars(select(NotificationBlockNumber).limit(1)) + ).first() + assert _notification_block_number.latest_block_number == block_number + + # + # Skip past data on initial sync + async def test_normal_5( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToRedeemed") + exchange_contract = shared_contract["IbetStraightBondExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_bond_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetStraightBond" + async_session.add(idx_token_list_item) + await async_session.commit() + + # Emit ChangeToRedeemed event + bond_change_to_redeemed( + invoker=self.issuer, + token=token, + ) + web3.provider.make_request(RPCEndpoint("evm_mine"), []) + + # Run target process + await watcher.loop() + + # Assertion + block_number = web3.eth.block_number + + _notification_list = (await async_session.scalars(select(Notification))).all() + assert len(_notification_list) == 0 + + _notification_block_number = ( + await async_session.scalars(select(NotificationBlockNumber).limit(1)) + ).first() + assert _notification_block_number.latest_block_number == block_number + + # ########################################################################### + # # Error Case + # ########################################################################### + + # + # Error occur + @mock.patch( + "web3.eth.async_eth.AsyncEth.get_logs", + MagicMock(side_effect=Exception()), + ) + async def test_error_1( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToRedeemed") + exchange_contract = shared_contract["IbetStraightBondExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_bond_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetStraightBond" + async_session.add(idx_token_list_item) + await async_session.commit() + + # Run target process + await watcher.loop() + + # Assertion + _notification = ( + await async_session.scalars( + select(Notification).order_by(Notification.created).limit(1) + ) + ).first() + assert _notification is None + + _notification_block_number = ( + await async_session.scalars(select(NotificationBlockNumber).limit(1)) + ).first() + assert _notification_block_number is None + + +@pytest.mark.asyncio +class TestWatchChangeToCanceled: + issuer = eth_account["issuer"] + lock_account = eth_account["user1"] + + ########################################################################### + # Normal Case + ########################################################################### + + # + # Initial Sync + async def test_normal_1( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToCanceled") + exchange_contract = shared_contract["IbetShareExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_share_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetShare" + async_session.add(idx_token_list_item) + await async_session.commit() + + # Run target process + await watcher.loop() + + # Assertion + block_number = web3.eth.block_number + + _notification = ( + await async_session.scalars( + select(Notification) + .where( + Notification.notification_id + == "0x{:012x}{:06x}{:06x}{:02x}".format(block_number, 0, 0, 0) + ) + .limit(1) + ) + ).first() + assert _notification is None + + _notification_block_number: NotificationBlockNumber = ( + await async_session.scalars( + select(NotificationBlockNumber) + .where( + and_( + NotificationBlockNumber.notification_type + == NotificationType.CHANGE_TO_CANCELED, + NotificationBlockNumber.contract_address == token["address"], + ) + ) + .limit(1) + ) + ).first() + assert _notification_block_number.latest_block_number == block_number + + # + # Single event logs + async def test_normal_2( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToCanceled") + exchange_contract = shared_contract["IbetShareExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_share_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetShare" + async_session.add(idx_token_list_item) + + notification_block_number = NotificationBlockNumber() + notification_block_number.notification_type = ( + NotificationType.CHANGE_TO_CANCELED + ) + notification_block_number.contract_address = token["address"] + notification_block_number.latest_block_number = web3.eth.block_number + async_session.add(notification_block_number) + await async_session.commit() + + # Emit ChangeToCanceled event + share_change_to_canceled( + invoker=self.issuer, + token=token, + ) + + # Run target process + await watcher.loop() + + # Assertion + block_number = web3.eth.block_number + + async_session.expunge_all() + + _notification_list = (await async_session.scalars(select(Notification))).all() + assert len(_notification_list) == 1 + assert ( + _notification_list[0].notification_type + == NotificationType.CHANGE_TO_CANCELED + ) + assert _notification_list[0].priority == 0 + assert _notification_list[0].address is None + assert _notification_list[0].args == {} + assert _notification_list[0].metainfo == { + "company_name": "株式会社DEMO", + "exchange_address": "", + "token_address": token["address"], + "token_name": "テスト株式", + "token_type": "IbetShare", + } + + _notification_block_number: NotificationBlockNumber = ( + await async_session.scalars( + select(NotificationBlockNumber) + .where( + and_( + NotificationBlockNumber.notification_type + == NotificationType.CHANGE_TO_CANCELED, + NotificationBlockNumber.contract_address == token["address"], + ) + ) + .limit(1) + ) + ).first() + assert _notification_block_number.latest_block_number == block_number + + # + # Multi event logs + async def test_normal_3( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToCanceled") + exchange_contract = shared_contract["IbetShareExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_share_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetShare" + async_session.add(idx_token_list_item) + + notification_block_number = NotificationBlockNumber() + notification_block_number.notification_type = ( + NotificationType.CHANGE_TO_CANCELED + ) + notification_block_number.contract_address = token["address"] + notification_block_number.latest_block_number = web3.eth.block_number + async_session.add(notification_block_number) + await async_session.commit() + + # Emit ChangeToCanceled event + share_change_to_canceled( + invoker=self.issuer, + token=token, + ) + share_change_to_canceled( + invoker=self.issuer, + token=token, + ) + + # Run target process + await watcher.loop() + + # Assertion + block_number = web3.eth.block_number + + async_session.expunge_all() + + _notification_list = (await async_session.scalars(select(Notification))).all() + assert len(_notification_list) == 2 + assert ( + _notification_list[0].notification_type + == NotificationType.CHANGE_TO_CANCELED + ) + assert _notification_list[0].priority == 0 + assert _notification_list[0].address is None + assert _notification_list[0].args == {} + assert _notification_list[0].metainfo == { + "company_name": "株式会社DEMO", + "exchange_address": "", + "token_address": token["address"], + "token_name": "テスト株式", + "token_type": "IbetShare", + } + assert ( + _notification_list[1].notification_type + == NotificationType.CHANGE_TO_CANCELED + ) + assert _notification_list[1].priority == 0 + assert _notification_list[1].address is None + assert _notification_list[1].args == {} + assert _notification_list[1].metainfo == { + "company_name": "株式会社DEMO", + "exchange_address": "", + "token_address": token["address"], + "token_name": "テスト株式", + "token_type": "IbetShare", + } + + _notification_block_number: NotificationBlockNumber = ( + await async_session.scalars( + select(NotificationBlockNumber) + .where( + and_( + NotificationBlockNumber.notification_type + == NotificationType.CHANGE_TO_CANCELED, + NotificationBlockNumber.contract_address == token["address"], + ) + ) + .limit(1) + ) + ).first() + assert _notification_block_number.latest_block_number == block_number + + # + # No event logs + async def test_normal_4( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToCanceled") + exchange_contract = shared_contract["IbetShareExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_share_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetShare" + async_session.add(idx_token_list_item) + + notification_block_number = NotificationBlockNumber() + notification_block_number.notification_type = ( + NotificationType.CHANGE_TO_CANCELED + ) + notification_block_number.contract_address = token["address"] + notification_block_number.latest_block_number = web3.eth.block_number + async_session.add(notification_block_number) + await async_session.commit() + + # Not emit ChangeToCanceled event + pass + + # Run target process + await watcher.loop() + + # Assertion + block_number = web3.eth.block_number + + _notification_list = (await async_session.scalars(select(Notification))).all() + assert len(_notification_list) == 0 + + _notification_block_number = ( + await async_session.scalars(select(NotificationBlockNumber).limit(1)) + ).first() + assert _notification_block_number.latest_block_number == block_number + + # + # Skip past data on initial sync + async def test_normal_5( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToCanceled") + exchange_contract = shared_contract["IbetShareExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_share_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetShare" + async_session.add(idx_token_list_item) + await async_session.commit() + + # Emit ChangeToCanceled event + share_change_to_canceled( + invoker=self.issuer, + token=token, + ) + web3.provider.make_request(RPCEndpoint("evm_mine"), []) + + # Run target process + await watcher.loop() + + # Assertion + block_number = web3.eth.block_number + + _notification_list = (await async_session.scalars(select(Notification))).all() + assert len(_notification_list) == 0 + + _notification_block_number = ( + await async_session.scalars(select(NotificationBlockNumber).limit(1)) + ).first() + assert _notification_block_number.latest_block_number == block_number + + # ########################################################################### + # # Error Case + # ########################################################################### + + # + # Error occur + @mock.patch( + "web3.eth.async_eth.AsyncEth.get_logs", + MagicMock(side_effect=Exception()), + ) + async def test_error_1( + self, watcher_factory, async_session, shared_contract, mocked_company_list + ): + watcher = watcher_factory("WatchChangeToCanceled") + exchange_contract = shared_contract["IbetShareExchange"] + token_list_contract = shared_contract["TokenList"] + personal_info_contract = shared_contract["PersonalInfo"] + + # Issue token + token = await prepare_share_token( + self.issuer, + exchange_contract, + token_list_contract, + personal_info_contract, + async_session, + ) + + idx_token_list_item = IDXTokenListRegister() + idx_token_list_item.token_address = token["address"] + idx_token_list_item.owner_address = self.issuer["account_address"] + idx_token_list_item.token_template = "IbetShare" + async_session.add(idx_token_list_item) + await async_session.commit() + + # Run target process + await watcher.loop() + + # Assertion + _notification = ( + await async_session.scalars( + select(Notification).order_by(Notification.created).limit(1) + ) + ).first() + assert _notification is None + + _notification_block_number = ( + await async_session.scalars(select(NotificationBlockNumber).limit(1)) + ).first() + assert _notification_block_number is None diff --git a/tests/contract_modules.py b/tests/contract_modules.py index 309b38558..ab2899b42 100644 --- a/tests/contract_modules.py +++ b/tests/contract_modules.py @@ -381,6 +381,15 @@ def bond_set_tradable_exchange(invoker, token, exchange_address: str): ) +# BONDトークン：償還状態に変更 +def bond_change_to_redeemed(invoker, token): + web3.eth.default_account = invoker["account_address"] + TokenContract = Contract.get_contract("IbetStraightBond", token["address"]) + TokenContract.functions.changeToRedeemed().transact( + {"from": invoker["account_address"]} + ) + + ############################################################### # Share Token ############################################################### @@ -665,6 +674,15 @@ def share_set_tradable_exchange(invoker, token, exchange_address: str): ) +# SHAREトークン：消却状態に変更 +def share_change_to_canceled(invoker, token): + web3.eth.default_account = invoker["account_address"] + TokenContract = Contract.get_contract("IbetShare", token["address"]) + TokenContract.functions.changeToCanceled().transact( + {"from": invoker["account_address"]} + ) + + ############################################################### # Coupon Token ###############################################################