diff --git a/app/model/schema/__init__.py b/app/model/schema/__init__.py index f428d57a..c1ccbd5e 100644 --- a/app/model/schema/__init__.py +++ b/app/model/schema/__init__.py @@ -138,10 +138,13 @@ IbetShareTransfer, IbetShareAdditionalIssue, IbetShareRedeem, + ListAllTokenLockEventsQuery, + ListAllTokenLockEventsSortItem, # Response TokenAddressResponse, IbetStraightBondResponse, - IbetShareResponse + IbetShareResponse, + ListAllTokenLockEventsResponse ) from .token_holders import ( # Request diff --git a/app/model/schema/token.py b/app/model/schema/token.py index 17b3d8ed..71e672b9 100644 --- a/app/model/schema/token.py +++ b/app/model/schema/token.py @@ -16,23 +16,32 @@ SPDX-License-Identifier: Apache-2.0 """ +from enum import Enum from typing import ( List, Optional ) import math +from fastapi import Query from pydantic import ( BaseModel, Field, validator ) +from pydantic.dataclasses import dataclass from web3 import Web3 +from . import ( + LockEventCategory +) +from .position import LockEvent from .types import ( MMDD_constr, YYYYMMDD_constr, - EMPTY_str + EMPTY_str, + SortOrder, + ResultSet ) @@ -327,6 +336,28 @@ def account_address_is_valid_address(cls, v): return v +class ListAllTokenLockEventsSortItem(str, Enum): + account_address = "account_address" + lock_address = "lock_address" + recipient_address = "recipient_address" + value = "value" + block_timestamp = "block_timestamp" + + +@dataclass +class ListAllTokenLockEventsQuery: + offset: Optional[int] = Query(default=None, description="Start position", ge=0) + limit: Optional[int] = Query(default=None, description="Number of set", ge=0) + + account_address: Optional[str] = Query(default=None, description="Account address") + lock_address: Optional[str] = Query(default=None, description="Lock address") + recipient_address: Optional[str] = Query(default=None, description="Recipient address") + category: Optional[LockEventCategory] = Query(default=None, description="Event category") + + sort_item: ListAllTokenLockEventsSortItem = Query(default=ListAllTokenLockEventsSortItem.block_timestamp, description="Sort item") + sort_order: SortOrder = Query(default=SortOrder.DESC, description="Sort order(0: ASC, 1: DESC)") + + ############################ # RESPONSE ############################ @@ -391,3 +422,9 @@ class IbetShareResponse(BaseModel): token_status: int is_canceled: bool memo: str + + +class ListAllTokenLockEventsResponse(BaseModel): + """List All Lock/Unlock events (Response)""" + result_set: ResultSet + events: List[LockEvent] = Field(description="Lock/Unlock event list") diff --git a/app/routers/bond.py b/app/routers/bond.py index 9bf79229..a9bb6d88 100644 --- a/app/routers/bond.py +++ b/app/routers/bond.py @@ -39,7 +39,10 @@ func, literal_column, cast, - String + String, + literal, + null, + column ) from sqlalchemy.orm import ( Session, @@ -64,6 +67,9 @@ UpdateTransferApprovalRequest, ListTransferHistorySortItem, ListTransferHistoryQuery, + ListAllTokenLockEventsQuery, + ListAllTokenLockEventsSortItem, + LockEventCategory, # Response IbetStraightBondResponse, TokenAddressResponse, @@ -85,6 +91,7 @@ BatchRegisterPersonalInfoUploadResponse, ListBatchRegisterPersonalInfoUploadResponse, GetBatchRegisterPersonalInfoResponse, + ListAllTokenLockEventsResponse ) from app.model.db import ( Account, @@ -111,7 +118,9 @@ BatchRegisterPersonalInfoUploadStatus, BatchRegisterPersonalInfo, TransferApprovalHistory, - TransferApprovalOperationType + TransferApprovalOperationType, + IDXLock, + IDXUnlock ) from app.model.blockchain import ( IbetStraightBondContract, @@ -1801,6 +1810,144 @@ def retrieve_batch_register_personal_info( }) +# GET: /bond/tokens/{token_address}/lock_events +@router.get( + "/tokens/{token_address}/lock_events", + summary="List all lock/unlock events related to given bond token", + response_model=ListAllTokenLockEventsResponse, + responses=get_routers_responses(422) +) +def list_all_lock_events_by_bond( + token_address: str, + issuer_address: Optional[str] = Header(None), + request_query: ListAllTokenLockEventsQuery = Depends(), + db: Session = Depends(db_session) +): + # Validate Headers + validate_headers(issuer_address=(issuer_address, address_is_valid_address)) + + # Request parameters + offset = request_query.offset + limit = request_query.limit + sort_item = request_query.sort_item + sort_order = request_query.sort_order + + # Base query + query_lock = ( + db.query( + literal(value=LockEventCategory.Lock.value, type_=String).label("category"), + IDXLock.transaction_hash.label("transaction_hash"), + IDXLock.token_address.label("token_address"), + IDXLock.lock_address.label("lock_address"), + IDXLock.account_address.label("account_address"), + null().label("recipient_address"), + IDXLock.value.label("value"), + IDXLock.data.label("data"), + IDXLock.block_timestamp.label("block_timestamp"), + Token + ). + join(Token, IDXLock.token_address == Token.token_address). + filter(Token.type == TokenType.IBET_STRAIGHT_BOND.value). + filter(Token.token_address == token_address). + filter(Token.token_status != 2) + ) + if issuer_address is not None: + query_lock = query_lock.filter(Token.issuer_address == issuer_address) + + query_unlock = ( + db.query( + literal(value=LockEventCategory.Unlock.value, type_=String).label("category"), + IDXUnlock.transaction_hash.label("transaction_hash"), + IDXUnlock.token_address.label("token_address"), + IDXUnlock.lock_address.label("lock_address"), + IDXUnlock.account_address.label("account_address"), + IDXUnlock.recipient_address.label("recipient_address"), + IDXUnlock.value.label("value"), + IDXUnlock.data.label("data"), + IDXUnlock.block_timestamp.label("block_timestamp"), + Token + ). + join(Token, IDXUnlock.token_address == Token.token_address). + filter(Token.type == TokenType.IBET_STRAIGHT_BOND.value). + filter(Token.token_address == token_address). + filter(Token.token_status != 2) + ) + if issuer_address is not None: + query_unlock = query_unlock.filter(Token.issuer_address == issuer_address) + + total = query_lock.count() + query_unlock.count() + + # Filter + match request_query.category: + case LockEventCategory.Lock.value: + query = query_lock + case LockEventCategory.Unlock.value: + query = query_unlock + case _: + query = query_lock.union_all(query_unlock) + + if request_query.account_address is not None: + query = query.filter(column("account_address") == request_query.account_address) + if request_query.lock_address is not None: + query = query.filter(column("lock_address") == request_query.lock_address) + if request_query.recipient_address is not None: + query = query.filter(column("recipient_address") == request_query.recipient_address) + + count = query.count() + + # Sort + sort_attr = column(sort_item) + if sort_order == 0: # ASC + query = query.order_by(sort_attr) + else: # DESC + query = query.order_by(desc(sort_attr)) + + if sort_item != ListAllTokenLockEventsSortItem.block_timestamp.value: + # NOTE: Set secondary sort for consistent results + query = query.order_by(desc(column(ListAllTokenLockEventsSortItem.block_timestamp.value))) + + # Pagination + if offset is not None: + query = query.offset(offset) + if limit is not None: + query = query.limit(limit) + + lock_events = query.all() + + resp_data = [] + for lock_event in lock_events: + _token = lock_event[9] + _bond = IbetStraightBondContract(_token.token_address).get() + token_name = _bond.name + + block_timestamp_utc = timezone("UTC").localize(lock_event[8]) + resp_data.append({ + "category": lock_event[0], + "transaction_hash": lock_event[1], + "issuer_address": _token.issuer_address, + "token_address": lock_event[2], + "token_type": _token.type, + "token_name": token_name, + "lock_address": lock_event[3], + "account_address": lock_event[4], + "recipient_address": lock_event[5], + "value": lock_event[6], + "data": lock_event[7], + "block_timestamp": block_timestamp_utc.astimezone(local_tz).isoformat() + }) + + data = { + "result_set": { + "count": count, + "offset": offset, + "limit": limit, + "total": total + }, + "events": resp_data + } + return json_response(data) + + # POST: /bond/transfers @router.post( "/transfers", diff --git a/app/routers/position.py b/app/routers/position.py index 4bdfcba7..b26b3605 100644 --- a/app/routers/position.py +++ b/app/routers/position.py @@ -262,6 +262,7 @@ def list_all_locked_position( } return json_response(resp) + # GET: /positions/{account_address}/lock/events @router.get( "/{account_address}/lock/events", diff --git a/app/routers/share.py b/app/routers/share.py index 6ffead6a..5f0be56e 100644 --- a/app/routers/share.py +++ b/app/routers/share.py @@ -40,7 +40,10 @@ func, literal_column, cast, - String + String, + literal, + null, + column ) from sqlalchemy.orm import ( Session, @@ -65,6 +68,9 @@ UpdateTransferApprovalRequest, ListTransferHistorySortItem, ListTransferHistoryQuery, + ListAllTokenLockEventsQuery, + LockEventCategory, + ListAllTokenLockEventsSortItem, # Response IbetShareResponse, TokenAddressResponse, @@ -86,6 +92,7 @@ BatchRegisterPersonalInfoUploadResponse, ListBatchRegisterPersonalInfoUploadResponse, GetBatchRegisterPersonalInfoResponse, + ListAllTokenLockEventsResponse ) from app.model.db import ( Account, @@ -112,7 +119,9 @@ BatchRegisterPersonalInfoUploadStatus, BatchRegisterPersonalInfo, TransferApprovalHistory, - TransferApprovalOperationType + TransferApprovalOperationType, + IDXLock, + IDXUnlock ) from app.model.blockchain import ( IbetShareContract, @@ -1790,6 +1799,144 @@ def retrieve_batch_register_personal_info( }) +# GET: /share/tokens/{token_address}/lock_events +@router.get( + "/tokens/{token_address}/lock_events", + summary="List all lock/unlock events related to given share token", + response_model=ListAllTokenLockEventsResponse, + responses=get_routers_responses(422) +) +def list_all_lock_events_by_share( + token_address: str, + issuer_address: Optional[str] = Header(None), + request_query: ListAllTokenLockEventsQuery = Depends(), + db: Session = Depends(db_session) +): + # Validate Headers + validate_headers(issuer_address=(issuer_address, address_is_valid_address)) + + # Request parameters + offset = request_query.offset + limit = request_query.limit + sort_item = request_query.sort_item + sort_order = request_query.sort_order + + # Base query + query_lock = ( + db.query( + literal(value=LockEventCategory.Lock.value, type_=String).label("category"), + IDXLock.transaction_hash.label("transaction_hash"), + IDXLock.token_address.label("token_address"), + IDXLock.lock_address.label("lock_address"), + IDXLock.account_address.label("account_address"), + null().label("recipient_address"), + IDXLock.value.label("value"), + IDXLock.data.label("data"), + IDXLock.block_timestamp.label("block_timestamp"), + Token + ). + join(Token, IDXLock.token_address == Token.token_address). + filter(Token.type == TokenType.IBET_SHARE.value). + filter(Token.token_address == token_address). + filter(Token.token_status != 2) + ) + if issuer_address is not None: + query_lock = query_lock.filter(Token.issuer_address == issuer_address) + + query_unlock = ( + db.query( + literal(value=LockEventCategory.Unlock.value, type_=String).label("category"), + IDXUnlock.transaction_hash.label("transaction_hash"), + IDXUnlock.token_address.label("token_address"), + IDXUnlock.lock_address.label("lock_address"), + IDXUnlock.account_address.label("account_address"), + IDXUnlock.recipient_address.label("recipient_address"), + IDXUnlock.value.label("value"), + IDXUnlock.data.label("data"), + IDXUnlock.block_timestamp.label("block_timestamp"), + Token + ). + join(Token, IDXUnlock.token_address == Token.token_address). + filter(Token.type == TokenType.IBET_SHARE.value). + filter(Token.token_address == token_address). + filter(Token.token_status != 2) + ) + if issuer_address is not None: + query_unlock = query_unlock.filter(Token.issuer_address == issuer_address) + + total = query_lock.count() + query_unlock.count() + + # Filter + match request_query.category: + case LockEventCategory.Lock.value: + query = query_lock + case LockEventCategory.Unlock.value: + query = query_unlock + case _: + query = query_lock.union_all(query_unlock) + + if request_query.account_address is not None: + query = query.filter(column("account_address") == request_query.account_address) + if request_query.lock_address is not None: + query = query.filter(column("lock_address") == request_query.lock_address) + if request_query.recipient_address is not None: + query = query.filter(column("recipient_address") == request_query.recipient_address) + + count = query.count() + + # Sort + sort_attr = column(sort_item) + if sort_order == 0: # ASC + query = query.order_by(sort_attr) + else: # DESC + query = query.order_by(desc(sort_attr)) + + if sort_item != ListAllTokenLockEventsSortItem.block_timestamp.value: + # NOTE: Set secondary sort for consistent results + query = query.order_by(desc(column(ListAllTokenLockEventsSortItem.block_timestamp.value))) + + # Pagination + if offset is not None: + query = query.offset(offset) + if limit is not None: + query = query.limit(limit) + + lock_events = query.all() + + resp_data = [] + for lock_event in lock_events: + _token = lock_event[9] + _share = IbetShareContract(_token.token_address).get() + token_name = _share.name + + block_timestamp_utc = timezone("UTC").localize(lock_event[8]) + resp_data.append({ + "category": lock_event[0], + "transaction_hash": lock_event[1], + "issuer_address": _token.issuer_address, + "token_address": lock_event[2], + "token_type": _token.type, + "token_name": token_name, + "lock_address": lock_event[3], + "account_address": lock_event[4], + "recipient_address": lock_event[5], + "value": lock_event[6], + "data": lock_event[7], + "block_timestamp": block_timestamp_utc.astimezone(local_tz).isoformat() + }) + + data = { + "result_set": { + "count": count, + "offset": offset, + "limit": limit, + "total": total + }, + "events": resp_data + } + return json_response(data) + + # POST: /share/transfers @router.post( "/transfers", diff --git a/tests/test_app_routers_bond_lock_events_GET.py b/tests/test_app_routers_bond_lock_events_GET.py new file mode 100644 index 00000000..092c3090 --- /dev/null +++ b/tests/test_app_routers_bond_lock_events_GET.py @@ -0,0 +1,623 @@ +""" +Copyright BOOSTRY Co., Ltd. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. + +You may obtain a copy of the License at +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +""" +from datetime import datetime +from unittest import mock +from unittest.mock import ANY, MagicMock +import pytest +from sqlalchemy.orm import Session + +from app.model.db import ( + IDXLock, + IDXUnlock, + Token, + TokenType +) +from app.model.blockchain import ( + IbetStraightBondContract +) + + +class TestAppRoutersBondLockEvents: + + # target API endpoint + base_url = "/bond/tokens/{token_address}/lock_events" + + issuer_address = "0x1234567890123456789012345678900000000100" + + account_address_1 = "0x1234567890123456789012345678900000000000" + account_address_2 = "0x1234567890123456789012345678900000000001" + + other_account_address_1 = "0x1234567890123456789012345678911111111111" + other_account_address_2 = "0x1234567890123456789012345678922222222222" + + lock_address_1 = "0x1234567890123456789012345678900000000100" + lock_address_2 = "0x1234567890123456789012345678900000000200" + + token_address_1 = "0x1234567890123456789012345678900000000010" + token_name_1 = "test_bond_1" + token_address_2 = "0x1234567890123456789012345678900000000020" + + def setup_data(self, db: Session, token_status: int = 1): + # prepare data: Token + _token = Token() + _token.token_address = self.token_address_1 + _token.issuer_address = self.issuer_address + _token.type = TokenType.IBET_STRAIGHT_BOND.value # bond + _token.tx_hash = "" + _token.abi = "" + _token.token_status = token_status + db.add(_token) + + # prepare data: Token + _token = Token() + _token.token_address = self.token_address_2 + _token.issuer_address = self.issuer_address + _token.type = TokenType.IBET_STRAIGHT_BOND.value # bond + _token.tx_hash = "" + _token.abi = "" + _token.token_status = token_status + db.add(_token) + + # prepare data: Lock events + _lock = IDXLock() + _lock.transaction_hash = "tx_hash_1" + _lock.block_number = 1 + _lock.token_address = self.token_address_1 + _lock.lock_address = self.lock_address_1 + _lock.account_address = self.account_address_1 + _lock.value = 1 + _lock.data = {"message": "locked_1"} + _lock.block_timestamp = datetime.utcnow() + db.add(_lock) + + _lock = IDXLock() + _lock.transaction_hash = "tx_hash_2" + _lock.block_number = 2 + _lock.token_address = self.token_address_1 + _lock.lock_address = self.lock_address_2 + _lock.account_address = self.account_address_2 + _lock.value = 1 + _lock.data = {"message": "locked_2"} + _lock.block_timestamp = datetime.utcnow() + db.add(_lock) + + _unlock = IDXUnlock() + _unlock.transaction_hash = "tx_hash_3" + _unlock.block_number = 3 + _unlock.token_address = self.token_address_1 + _unlock.lock_address = self.lock_address_1 + _unlock.account_address = self.account_address_1 + _unlock.recipient_address = self.other_account_address_1 + _unlock.value = 1 + _unlock.data = {"message": "unlocked_1"} + _unlock.block_timestamp = datetime.utcnow() + db.add(_unlock) + + _unlock = IDXUnlock() + _unlock.transaction_hash = "tx_hash_4" + _unlock.block_number = 4 + _unlock.token_address = self.token_address_1 + _unlock.lock_address = self.lock_address_2 + _unlock.account_address = self.account_address_2 + _unlock.recipient_address = self.other_account_address_2 + _unlock.value = 1 + _unlock.data = {"message": "unlocked_2"} + _unlock.block_timestamp = datetime.utcnow() + db.add(_unlock) + + db.commit() + + @staticmethod + def get_contract_mock_data(token_name_list: list[str]): + token_contract_list = [] + for toke_name in token_name_list: + token = IbetStraightBondContract() + token.name = toke_name + token_contract_list.append(token) + return token_contract_list + + expected_lock_1 = { + "category": "Lock", + "transaction_hash": "tx_hash_1", + "issuer_address": issuer_address, + "token_address": token_address_1, + "token_type": TokenType.IBET_STRAIGHT_BOND.value, + "token_name": token_name_1, + "lock_address": lock_address_1, + "account_address": account_address_1, + "recipient_address": None, + "value": 1, + "data": {"message": "locked_1"}, + "block_timestamp": ANY + } + expected_lock_2 = { + "category": "Lock", + "transaction_hash": "tx_hash_2", + "issuer_address": issuer_address, + "token_address": token_address_1, + "token_type": TokenType.IBET_STRAIGHT_BOND.value, + "token_name": token_name_1, + "lock_address": lock_address_2, + "account_address": account_address_2, + "recipient_address": None, + "value": 1, + "data": {"message": "locked_2"}, + "block_timestamp": ANY + } + expected_unlock_1 = { + "category": "Unlock", + "transaction_hash": "tx_hash_3", + "issuer_address": issuer_address, + "token_address": token_address_1, + "token_type": TokenType.IBET_STRAIGHT_BOND.value, + "token_name": token_name_1, + "lock_address": lock_address_1, + "account_address": account_address_1, + "recipient_address": other_account_address_1, + "value": 1, + "data": {"message": "unlocked_1"}, + "block_timestamp": ANY + } + expected_unlock_2 = { + "category": "Unlock", + "transaction_hash": "tx_hash_4", + "issuer_address": issuer_address, + "token_address": token_address_1, + "token_type": TokenType.IBET_STRAIGHT_BOND.value, + "token_name": token_name_1, + "lock_address": lock_address_2, + "account_address": account_address_2, + "recipient_address": other_account_address_2, + "value": 1, + "data": {"message": "unlocked_2"}, + "block_timestamp": ANY + } + + ########################################################################### + # Normal Case + ########################################################################### + + # Normal_1 + # 0 record + def test_normal_1(self, client, db): + # prepare data: Token + _token = Token() + _token.token_address = self.token_address_1 + _token.issuer_address = self.issuer_address + _token.type = TokenType.IBET_STRAIGHT_BOND.value + _token.tx_hash = "" + _token.abi = "" + db.add(_token) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "result_set": { + "count": 0, + "offset": None, + "limit": None, + "total": 0, + }, + "events": [] + } + + # Normal_2 + # Multiple record + @mock.patch("app.model.blockchain.token.IbetStraightBondContract.get") + def test_normal_2(self, mock_IbetStraightBondContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetStraightBondContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1, self.token_name_1, self.token_name_1, self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "result_set": { + "count": 4, + "offset": None, + "limit": None, + "total": 4 + }, + "events": [ + self.expected_unlock_2, + self.expected_unlock_1, + self.expected_lock_2, + self.expected_lock_1 + ] + } + + # Normal_3 + # Records not subject to extraction + # token_status + @mock.patch("app.model.blockchain.token.IbetStraightBondContract.get") + def test_normal_3(self, mock_IbetStraightBondContract_get, client, db): + # prepare data + self.setup_data(db=db, token_status=2) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + 'result_set': { + 'count': 0, + 'offset': None, + 'limit': None, + 'total': 0 + }, + 'events': [] + } + + # Normal_4 + # issuer_address is not None + @mock.patch("app.model.blockchain.token.IbetStraightBondContract.get") + def test_normal_4(self, mock_IbetStraightBondContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetStraightBondContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1, self.token_name_1, self.token_name_1, self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + headers={"issuer-address": self.issuer_address} + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "result_set": { + "count": 4, + "offset": None, + "limit": None, + "total": 4 + }, + "events": [ + self.expected_unlock_2, + self.expected_unlock_1, + self.expected_lock_2, + self.expected_lock_1 + ] + } + + # Normal_5_1 + # Search filter: category + @mock.patch("app.model.blockchain.token.IbetStraightBondContract.get") + def test_normal_5_1(self, mock_IbetStraightBondContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetStraightBondContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1, self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={"category": "Lock"} + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "result_set": { + "count": 2, + "offset": None, + "limit": None, + "total": 4 + }, + "events": [ + self.expected_lock_2, + self.expected_lock_1 + ] + } + + # Normal_5_2 + # Search filter: account_address + @mock.patch("app.model.blockchain.token.IbetStraightBondContract.get") + def test_normal_5_2(self, mock_IbetStraightBondContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetStraightBondContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1, self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={"account_address": self.account_address_1} + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "result_set": { + "count": 2, + "offset": None, + "limit": None, + "total": 4 + }, + "events": [ + self.expected_unlock_1, + self.expected_lock_1 + ] + } + + # Normal_5_3 + # Search filter: lock_address + @mock.patch("app.model.blockchain.token.IbetStraightBondContract.get") + def test_normal_5_3(self, mock_IbetStraightBondContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetStraightBondContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1, self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={"lock_address": self.lock_address_1} + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + 'result_set': { + 'count': 2, + 'offset': None, + 'limit': None, + 'total': 4 + }, + 'events': [ + self.expected_unlock_1, + self.expected_lock_1 + ] + } + + # Normal_5_4 + # Search filter: recipient_address + @mock.patch("app.model.blockchain.token.IbetStraightBondContract.get") + def test_normal_5_4(self, mock_IbetStraightBondContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetStraightBondContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={"recipient_address": self.other_account_address_2} + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + 'result_set': { + 'count': 1, + 'offset': None, + 'limit': None, + 'total': 4 + }, + 'events': [ + self.expected_unlock_2 + ] + } + + # Normal_6 + # Sort + @pytest.mark.parametrize("sort_item, sort_order, data, expect", [ + # ( + # "{sort_item}", {sort_order}, + # {data used to contract mock}, + # {expected result} + # ), + ( + "account_address", 0, + get_contract_mock_data([token_name_1, token_name_1, token_name_1, token_name_1]), + [expected_unlock_1, expected_lock_1, expected_unlock_2, expected_lock_2] + ), + ( + "lock_address", 0, + get_contract_mock_data([token_name_1, token_name_1, token_name_1, token_name_1]), + [expected_unlock_1, expected_lock_1, expected_unlock_2, expected_lock_2] + ), + ( + "recipient_address", 0, + get_contract_mock_data([token_name_1, token_name_1, token_name_1, token_name_1]), + [expected_unlock_1, expected_unlock_2, expected_lock_2, expected_lock_1] + ), + ( + "recipient_address", 1, + get_contract_mock_data([token_name_1, token_name_1, token_name_1, token_name_1]), + [expected_lock_2, expected_lock_1, expected_unlock_2, expected_unlock_1] + ), + ( + "value", 0, + get_contract_mock_data([token_name_1, token_name_1, token_name_1, token_name_1]), + [expected_unlock_2, expected_unlock_1, expected_lock_2, expected_lock_1] + ), + ]) + def test_normal_6(self, sort_item, sort_order, data, expect, client, db): + # prepare data + self.setup_data(db=db) + + # request target api + with mock.patch("app.model.blockchain.token.IbetStraightBondContract.get", MagicMock(side_effect=data)): + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={ + "sort_item": sort_item, + "sort_order": sort_order + } + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + 'result_set': { + 'count': 4, + 'offset': None, + 'limit': None, + 'total': 4 + }, + 'events': expect + } + + # Normal_7 + # Pagination + @mock.patch("app.model.blockchain.token.IbetStraightBondContract.get") + def test_normal_7(self, mock_IbetStraightBondContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetStraightBondContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={ + "offset": 1, + "limit": 1 + } + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + 'result_set': { + 'count': 4, + 'offset': 1, + 'limit': 1, + 'total': 4 + }, + 'events': [ + self.expected_unlock_1 + ] + } + + # ########################################################################### + # # Error Case + # ########################################################################### + + # Error_1_1 + # RequestValidationError + # header + def test_error_1_1(self, client, db): + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + headers={ + "issuer-address": "test", + } + ) + + # assertion + assert resp.status_code == 422 + assert resp.json() == { + "meta": { + "code": 1, + "title": "RequestValidationError" + }, + "detail": [ + { + "loc": ["header", "issuer-address"], + "msg": "issuer-address is not a valid address", + "type": "value_error" + } + ] + } + + # Error_1_2 + # RequestValidationError + # query(invalid value) + def test_error_1_2(self, client, db): + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={ + "category": "test", + "sort_item": "test", + "offset": "test", + "limit": "test", + } + ) + + # assertion + assert resp.status_code == 422 + assert resp.json() == { + 'meta': { + 'code': 1, + 'title': 'RequestValidationError' + }, + 'detail': [ + { + 'loc': ['query', 'offset'], + 'msg': 'value is not a valid integer', + 'type': 'type_error.integer' + }, + { + 'loc': ['query', 'limit'], + 'msg': 'value is not a valid integer', + 'type': 'type_error.integer' + }, + { + 'loc': ['query', 'category'], + 'msg': "value is not a valid enumeration member; permitted: 'Lock', 'Unlock'", + 'type': 'type_error.enum', + 'ctx': {'enum_values': ['Lock', 'Unlock']} + }, + { + 'loc': ['query', 'sort_item'], + 'msg': "value is not a valid enumeration member; permitted: 'account_address', 'lock_address', 'recipient_address', 'value', 'block_timestamp'", + 'type': 'type_error.enum', + 'ctx': {'enum_values': ['account_address', 'lock_address', 'recipient_address', 'value', 'block_timestamp']} + } + ] + } diff --git a/tests/test_app_routers_share_lock_events_GET.py b/tests/test_app_routers_share_lock_events_GET.py new file mode 100644 index 00000000..4c107755 --- /dev/null +++ b/tests/test_app_routers_share_lock_events_GET.py @@ -0,0 +1,623 @@ +""" +Copyright BOOSTRY Co., Ltd. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. + +You may obtain a copy of the License at +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +See the License for the specific language governing permissions and +limitations under the License. + +SPDX-License-Identifier: Apache-2.0 +""" +from datetime import datetime +from unittest import mock +from unittest.mock import ANY, MagicMock +import pytest +from sqlalchemy.orm import Session + +from app.model.db import ( + IDXLock, + IDXUnlock, + Token, + TokenType +) +from app.model.blockchain import ( + IbetShareContract +) + + +class TestAppRoutersShareLockEvents: + + # target API endpoint + base_url = "/share/tokens/{token_address}/lock_events" + + issuer_address = "0x1234567890123456789012345678900000000100" + + account_address_1 = "0x1234567890123456789012345678900000000000" + account_address_2 = "0x1234567890123456789012345678900000000001" + + other_account_address_1 = "0x1234567890123456789012345678911111111111" + other_account_address_2 = "0x1234567890123456789012345678922222222222" + + lock_address_1 = "0x1234567890123456789012345678900000000100" + lock_address_2 = "0x1234567890123456789012345678900000000200" + + token_address_1 = "0x1234567890123456789012345678900000000010" + token_name_1 = "test_share_1" + token_address_2 = "0x1234567890123456789012345678900000000020" + + def setup_data(self, db: Session, token_status: int = 1): + # prepare data: Token + _token = Token() + _token.token_address = self.token_address_1 + _token.issuer_address = self.issuer_address + _token.type = TokenType.IBET_SHARE.value # bond + _token.tx_hash = "" + _token.abi = "" + _token.token_status = token_status + db.add(_token) + + # prepare data: Token + _token = Token() + _token.token_address = self.token_address_2 + _token.issuer_address = self.issuer_address + _token.type = TokenType.IBET_SHARE.value # bond + _token.tx_hash = "" + _token.abi = "" + _token.token_status = token_status + db.add(_token) + + # prepare data: Lock events + _lock = IDXLock() + _lock.transaction_hash = "tx_hash_1" + _lock.block_number = 1 + _lock.token_address = self.token_address_1 + _lock.lock_address = self.lock_address_1 + _lock.account_address = self.account_address_1 + _lock.value = 1 + _lock.data = {"message": "locked_1"} + _lock.block_timestamp = datetime.utcnow() + db.add(_lock) + + _lock = IDXLock() + _lock.transaction_hash = "tx_hash_2" + _lock.block_number = 2 + _lock.token_address = self.token_address_1 + _lock.lock_address = self.lock_address_2 + _lock.account_address = self.account_address_2 + _lock.value = 1 + _lock.data = {"message": "locked_2"} + _lock.block_timestamp = datetime.utcnow() + db.add(_lock) + + _unlock = IDXUnlock() + _unlock.transaction_hash = "tx_hash_3" + _unlock.block_number = 3 + _unlock.token_address = self.token_address_1 + _unlock.lock_address = self.lock_address_1 + _unlock.account_address = self.account_address_1 + _unlock.recipient_address = self.other_account_address_1 + _unlock.value = 1 + _unlock.data = {"message": "unlocked_1"} + _unlock.block_timestamp = datetime.utcnow() + db.add(_unlock) + + _unlock = IDXUnlock() + _unlock.transaction_hash = "tx_hash_4" + _unlock.block_number = 4 + _unlock.token_address = self.token_address_1 + _unlock.lock_address = self.lock_address_2 + _unlock.account_address = self.account_address_2 + _unlock.recipient_address = self.other_account_address_2 + _unlock.value = 1 + _unlock.data = {"message": "unlocked_2"} + _unlock.block_timestamp = datetime.utcnow() + db.add(_unlock) + + db.commit() + + @staticmethod + def get_contract_mock_data(token_name_list: list[str]): + token_contract_list = [] + for toke_name in token_name_list: + token = IbetShareContract() + token.name = toke_name + token_contract_list.append(token) + return token_contract_list + + expected_lock_1 = { + "category": "Lock", + "transaction_hash": "tx_hash_1", + "issuer_address": issuer_address, + "token_address": token_address_1, + "token_type": TokenType.IBET_SHARE.value, + "token_name": token_name_1, + "lock_address": lock_address_1, + "account_address": account_address_1, + "recipient_address": None, + "value": 1, + "data": {"message": "locked_1"}, + "block_timestamp": ANY + } + expected_lock_2 = { + "category": "Lock", + "transaction_hash": "tx_hash_2", + "issuer_address": issuer_address, + "token_address": token_address_1, + "token_type": TokenType.IBET_SHARE.value, + "token_name": token_name_1, + "lock_address": lock_address_2, + "account_address": account_address_2, + "recipient_address": None, + "value": 1, + "data": {"message": "locked_2"}, + "block_timestamp": ANY + } + expected_unlock_1 = { + "category": "Unlock", + "transaction_hash": "tx_hash_3", + "issuer_address": issuer_address, + "token_address": token_address_1, + "token_type": TokenType.IBET_SHARE.value, + "token_name": token_name_1, + "lock_address": lock_address_1, + "account_address": account_address_1, + "recipient_address": other_account_address_1, + "value": 1, + "data": {"message": "unlocked_1"}, + "block_timestamp": ANY + } + expected_unlock_2 = { + "category": "Unlock", + "transaction_hash": "tx_hash_4", + "issuer_address": issuer_address, + "token_address": token_address_1, + "token_type": TokenType.IBET_SHARE.value, + "token_name": token_name_1, + "lock_address": lock_address_2, + "account_address": account_address_2, + "recipient_address": other_account_address_2, + "value": 1, + "data": {"message": "unlocked_2"}, + "block_timestamp": ANY + } + + ########################################################################### + # Normal Case + ########################################################################### + + # Normal_1 + # 0 record + def test_normal_1(self, client, db): + # prepare data: Token + _token = Token() + _token.token_address = self.token_address_1 + _token.issuer_address = self.issuer_address + _token.type = TokenType.IBET_SHARE.value + _token.tx_hash = "" + _token.abi = "" + db.add(_token) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "result_set": { + "count": 0, + "offset": None, + "limit": None, + "total": 0, + }, + "events": [] + } + + # Normal_2 + # Multiple record + @mock.patch("app.model.blockchain.token.IbetShareContract.get") + def test_normal_2(self, mock_IbetShareContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetShareContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1, self.token_name_1, self.token_name_1, self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "result_set": { + "count": 4, + "offset": None, + "limit": None, + "total": 4 + }, + "events": [ + self.expected_unlock_2, + self.expected_unlock_1, + self.expected_lock_2, + self.expected_lock_1 + ] + } + + # Normal_3 + # Records not subject to extraction + # token_status + @mock.patch("app.model.blockchain.token.IbetShareContract.get") + def test_normal_3(self, mock_IbetShareContract_get, client, db): + # prepare data + self.setup_data(db=db, token_status=2) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + 'result_set': { + 'count': 0, + 'offset': None, + 'limit': None, + 'total': 0 + }, + 'events': [] + } + + # Normal_4 + # issuer_address is not None + @mock.patch("app.model.blockchain.token.IbetShareContract.get") + def test_normal_4(self, mock_IbetShareContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetShareContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1, self.token_name_1, self.token_name_1, self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + headers={"issuer-address": self.issuer_address} + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "result_set": { + "count": 4, + "offset": None, + "limit": None, + "total": 4 + }, + "events": [ + self.expected_unlock_2, + self.expected_unlock_1, + self.expected_lock_2, + self.expected_lock_1 + ] + } + + # Normal_5_1 + # Search filter: category + @mock.patch("app.model.blockchain.token.IbetShareContract.get") + def test_normal_5_1(self, mock_IbetShareContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetShareContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1, self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={"category": "Lock"} + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "result_set": { + "count": 2, + "offset": None, + "limit": None, + "total": 4 + }, + "events": [ + self.expected_lock_2, + self.expected_lock_1 + ] + } + + # Normal_5_2 + # Search filter: account_address + @mock.patch("app.model.blockchain.token.IbetShareContract.get") + def test_normal_5_2(self, mock_IbetShareContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetShareContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1, self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={"account_address": self.account_address_1} + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "result_set": { + "count": 2, + "offset": None, + "limit": None, + "total": 4 + }, + "events": [ + self.expected_unlock_1, + self.expected_lock_1 + ] + } + + # Normal_5_3 + # Search filter: lock_address + @mock.patch("app.model.blockchain.token.IbetShareContract.get") + def test_normal_5_3(self, mock_IbetShareContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetShareContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1, self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={"lock_address": self.lock_address_1} + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + 'result_set': { + 'count': 2, + 'offset': None, + 'limit': None, + 'total': 4 + }, + 'events': [ + self.expected_unlock_1, + self.expected_lock_1 + ] + } + + # Normal_5_4 + # Search filter: recipient_address + @mock.patch("app.model.blockchain.token.IbetShareContract.get") + def test_normal_5_4(self, mock_IbetShareContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetShareContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={"recipient_address": self.other_account_address_2} + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + 'result_set': { + 'count': 1, + 'offset': None, + 'limit': None, + 'total': 4 + }, + 'events': [ + self.expected_unlock_2 + ] + } + + # Normal_6 + # Sort + @pytest.mark.parametrize("sort_item, sort_order, data, expect", [ + # ( + # "{sort_item}", {sort_order}, + # {data used to contract mock}, + # {expected result} + # ), + ( + "account_address", 0, + get_contract_mock_data([token_name_1, token_name_1, token_name_1, token_name_1]), + [expected_unlock_1, expected_lock_1, expected_unlock_2, expected_lock_2] + ), + ( + "lock_address", 0, + get_contract_mock_data([token_name_1, token_name_1, token_name_1, token_name_1]), + [expected_unlock_1, expected_lock_1, expected_unlock_2, expected_lock_2] + ), + ( + "recipient_address", 0, + get_contract_mock_data([token_name_1, token_name_1, token_name_1, token_name_1]), + [expected_unlock_1, expected_unlock_2, expected_lock_2, expected_lock_1] + ), + ( + "recipient_address", 1, + get_contract_mock_data([token_name_1, token_name_1, token_name_1, token_name_1]), + [expected_lock_2, expected_lock_1, expected_unlock_2, expected_unlock_1] + ), + ( + "value", 0, + get_contract_mock_data([token_name_1, token_name_1, token_name_1, token_name_1]), + [expected_unlock_2, expected_unlock_1, expected_lock_2, expected_lock_1] + ), + ]) + def test_normal_6(self, sort_item, sort_order, data, expect, client, db): + # prepare data + self.setup_data(db=db) + + # request target api + with mock.patch("app.model.blockchain.token.IbetShareContract.get", MagicMock(side_effect=data)): + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={ + "sort_item": sort_item, + "sort_order": sort_order + } + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + 'result_set': { + 'count': 4, + 'offset': None, + 'limit': None, + 'total': 4 + }, + 'events': expect + } + + # Normal_7 + # Pagination + @mock.patch("app.model.blockchain.token.IbetShareContract.get") + def test_normal_7(self, mock_IbetShareContract_get, client, db): + # prepare data + self.setup_data(db=db) + + # mock + mock_IbetShareContract_get.side_effect = self.get_contract_mock_data( + [self.token_name_1] + ) + + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={ + "offset": 1, + "limit": 1 + } + ) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + 'result_set': { + 'count': 4, + 'offset': 1, + 'limit': 1, + 'total': 4 + }, + 'events': [ + self.expected_unlock_1 + ] + } + + # ########################################################################### + # # Error Case + # ########################################################################### + + # Error_1_1 + # RequestValidationError + # header + def test_error_1_1(self, client, db): + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + headers={ + "issuer-address": "test", + } + ) + + # assertion + assert resp.status_code == 422 + assert resp.json() == { + "meta": { + "code": 1, + "title": "RequestValidationError" + }, + "detail": [ + { + "loc": ["header", "issuer-address"], + "msg": "issuer-address is not a valid address", + "type": "value_error" + } + ] + } + + # Error_1_2 + # RequestValidationError + # query(invalid value) + def test_error_1_2(self, client, db): + # request target api + resp = client.get( + self.base_url.format(token_address=self.token_address_1), + params={ + "category": "test", + "sort_item": "test", + "offset": "test", + "limit": "test", + } + ) + + # assertion + assert resp.status_code == 422 + assert resp.json() == { + 'meta': { + 'code': 1, + 'title': 'RequestValidationError' + }, + 'detail': [ + { + 'loc': ['query', 'offset'], + 'msg': 'value is not a valid integer', + 'type': 'type_error.integer' + }, + { + 'loc': ['query', 'limit'], + 'msg': 'value is not a valid integer', + 'type': 'type_error.integer' + }, + { + 'loc': ['query', 'category'], + 'msg': "value is not a valid enumeration member; permitted: 'Lock', 'Unlock'", + 'type': 'type_error.enum', + 'ctx': {'enum_values': ['Lock', 'Unlock']} + }, + { + 'loc': ['query', 'sort_item'], + 'msg': "value is not a valid enumeration member; permitted: 'account_address', 'lock_address', 'recipient_address', 'value', 'block_timestamp'", + 'type': 'type_error.enum', + 'ctx': {'enum_values': ['account_address', 'lock_address', 'recipient_address', 'value', 'block_timestamp']} + } + ] + }