diff --git a/app/exceptions.py b/app/exceptions.py index a4e83988..de4eb153 100644 --- a/app/exceptions.py +++ b/app/exceptions.py @@ -27,6 +27,9 @@ class AppError(Exception): code_list: list[int] | None = None +################################################ +# 400_BAD_REQUEST +################################################ class InvalidParameterError(AppError): status_code = status.HTTP_400_BAD_REQUEST code = 1 @@ -37,11 +40,42 @@ class SendTransactionError(AppError): code = 2 +class AuthTokenAlreadyExistsError(AppError): + status_code = status.HTTP_400_BAD_REQUEST + code = 3 + + +class ResponseLimitExceededError(AppError): + status_code = status.HTTP_400_BAD_REQUEST + code = 4 + + +class Integer64bitLimitExceededError(AppError): + status_code = status.HTTP_400_BAD_REQUEST + code = 5 + + +class OperationNotAllowedStateError(AppError): + """ + Error returned when server-side data is not ready to process the request + """ + + status_code = status.HTTP_400_BAD_REQUEST + code_list = [ + 101, # Transfer approval operations cannot be performed for accounts that do not have personal information registered. + ] + + def __init__(self, code: int, message: str = None): + self.code = code + super().__init__(message) + + class ContractRevertError(AppError): """ - * Error code is defined here - * https://github.com/BoostryJP/ibet-SmartContract/blob/dev-23.3/docs/Errors.md - * If contract doesn't throw error code, 0 is returned. + Revert error occurs from smart-contract + + - Error code: https://github.com/BoostryJP/ibet-SmartContract/blob/master/docs/Errors.md + - If contract doesn't throw error code, 0 is returned. """ status_code = status.HTTP_400_BAD_REQUEST @@ -57,26 +91,17 @@ def __repr__(self): return f"" +################################################ +# 401_UNAUTHORIZED +################################################ class AuthorizationError(AppError): status_code = status.HTTP_401_UNAUTHORIZED code = 1 +################################################ +# 503_SERVICE_UNAVAILABLE +################################################ class ServiceUnavailableError(AppError): status_code = status.HTTP_503_SERVICE_UNAVAILABLE code = 1 - - -class AuthTokenAlreadyExistsError(AppError): - status_code = status.HTTP_400_BAD_REQUEST - code = 3 - - -class ResponseLimitExceededError(AppError): - status_code = status.HTTP_400_BAD_REQUEST - code = 4 - - -class Integer64bitLimitExceededError(AppError): - status_code = status.HTTP_400_BAD_REQUEST - code = 5 diff --git a/app/main.py b/app/main.py index 2324fab5..334b0f61 100644 --- a/app/main.py +++ b/app/main.py @@ -215,6 +215,18 @@ async def response_limit_exceeded_error_handler( ) +# 400:OperationNotAllowedStateError +@app.exception_handler(OperationNotAllowedStateError) +async def operation_not_permitted_error_handler( + request: Request, exc: OperationNotAllowedStateError +): + meta = {"code": exc.code, "title": "OperationNotAllowedStateError"} + return JSONResponse( + status_code=exc.status_code, + content=jsonable_encoder({"meta": meta, "detail": exc.args[0]}), + ) + + # 400:ContractRevertError @app.exception_handler(ContractRevertError) async def contract_revert_error_handler(request: Request, exc: ContractRevertError): diff --git a/app/model/db/transfer_appoval_history.py b/app/model/db/transfer_appoval_history.py index 30894cfa..0a5ea90d 100644 --- a/app/model/db/transfer_appoval_history.py +++ b/app/model/db/transfer_appoval_history.py @@ -17,8 +17,9 @@ SPDX-License-Identifier: Apache-2.0 """ from enum import Enum +from typing import Optional -from sqlalchemy import BigInteger, String +from sqlalchemy import JSON, BigInteger, String from sqlalchemy.orm import Mapped, mapped_column from .base import Base @@ -41,6 +42,10 @@ class TransferApprovalHistory(Base): application_id: Mapped[int] = mapped_column(BigInteger, index=True, nullable=False) # Operation Type: TransferApprovalOperationType operation_type: Mapped[str] = mapped_column(String(20), index=True, nullable=False) + # From Address Personal Information (snapshot) + from_address_personal_info: Mapped[Optional[dict | list]] = mapped_column(JSON) + # To Address Personal Information (snapshot) + to_address_personal_info: Mapped[Optional[dict | list]] = mapped_column(JSON) def json(self): return { @@ -48,6 +53,8 @@ def json(self): "exchange_address": self.exchange_address, "application_id": self.application_id, "operation_type": self.operation_type, + "from_address_personal_info": self.from_address_personal_info, + "to_address_personal_info": self.to_address_personal_info, } diff --git a/app/model/schema/__init__.py b/app/model/schema/__init__.py index d7d905a4..c80fbf9a 100644 --- a/app/model/schema/__init__.py +++ b/app/model/schema/__init__.py @@ -133,6 +133,7 @@ ListTransferHistorySortItem, TransferApprovalHistoryResponse, TransferApprovalsResponse, + TransferApprovalTokenDetailResponse, TransferApprovalTokenResponse, TransferHistoryResponse, TransferResponse, diff --git a/app/model/schema/transfer.py b/app/model/schema/transfer.py index dc72cc74..0961ac53 100644 --- a/app/model/schema/transfer.py +++ b/app/model/schema/transfer.py @@ -143,6 +143,30 @@ class TransferApprovalTokenResponse(BaseModel): issuer_cancelable: bool +class TransferApprovalTokenDetailResponse(BaseModel): + """transfer approval token data""" + + id: int + token_address: str + exchange_address: str + application_id: int + from_address: str + from_address_personal_information: Optional[dict] = Field(...) + to_address: str + to_address_personal_information: Optional[dict] = Field(...) + amount: int + application_datetime: str + application_blocktimestamp: str + approval_datetime: Optional[str] = Field(...) + approval_blocktimestamp: Optional[str] = Field(...) + cancellation_blocktimestamp: Optional[str] = Field(...) + cancelled: bool + escrow_finished: bool + transfer_approved: bool + status: int + issuer_cancelable: bool + + class TransferApprovalHistoryResponse(BaseModel): """transfer approval token history""" diff --git a/app/routers/bond.py b/app/routers/bond.py index fc7aad3e..9d3ff4b0 100644 --- a/app/routers/bond.py +++ b/app/routers/bond.py @@ -48,6 +48,7 @@ AuthorizationError, ContractRevertError, InvalidParameterError, + OperationNotAllowedStateError, SendTransactionError, ) from app.model.blockchain import ( @@ -133,6 +134,7 @@ TokenUpdateOperationCategory, TransferApprovalHistoryResponse, TransferApprovalsResponse, + TransferApprovalTokenDetailResponse, TransferApprovalTokenResponse, TransferHistoryResponse, UpdateTransferApprovalOperationType, @@ -3016,6 +3018,7 @@ def list_token_transfer_approval_history( InvalidParameterError, SendTransactionError, ContractRevertError, + OperationNotAllowedStateError, ), ) def update_transfer_approval( @@ -3116,6 +3119,37 @@ def update_transfer_approval( if transfer_approval_op is not None: raise InvalidParameterError("duplicate operation") + # Check the existence of personal information data for from_address and to_address + _from_address_personal_info: IDXPersonalInfo | None = db.scalars( + select(IDXPersonalInfo) + .where( + and_( + IDXPersonalInfo.account_address == _transfer_approval.from_address, + IDXPersonalInfo.issuer_address == issuer_address, + ) + ) + .limit(1) + ).first() + if _from_address_personal_info is None: + raise OperationNotAllowedStateError( + 101, "personal information for from_address is not registered" + ) + + _to_address_personal_info: IDXPersonalInfo | None = db.scalars( + select(IDXPersonalInfo) + .where( + and_( + IDXPersonalInfo.account_address == _transfer_approval.to_address, + IDXPersonalInfo.issuer_address == issuer_address, + ) + ) + .limit(1) + ).first() + if _to_address_personal_info is None: + raise OperationNotAllowedStateError( + 101, "personal information for to_address is not registered" + ) + # Send transaction # - APPROVE -> approveTransfer # In the case of a transfer approval for a token, if the transaction is reverted, @@ -3189,6 +3223,12 @@ def update_transfer_approval( transfer_approval_op.exchange_address = _transfer_approval.exchange_address transfer_approval_op.application_id = _transfer_approval.application_id transfer_approval_op.operation_type = data.operation_type + transfer_approval_op.from_address_personal_info = ( + _from_address_personal_info.personal_info + ) + transfer_approval_op.to_address_personal_info = ( + _to_address_personal_info.personal_info + ) db.add(transfer_approval_op) db.commit() @@ -3196,7 +3236,7 @@ def update_transfer_approval( # GET: /bond/transfer_approvals/{token_address}/{id} @router.get( "/transfer_approvals/{token_address}/{id}", - response_model=TransferApprovalTokenResponse, + response_model=TransferApprovalTokenDetailResponse, responses=get_routers_responses(422, 404, InvalidParameterError), ) def retrieve_transfer_approval_history(db: DBSession, token_address: str, id: int): @@ -3333,13 +3373,51 @@ def retrieve_transfer_approval_history(db: DBSession, token_address: str, id: in else: cancellation_blocktimestamp = None + # Get personal information of account address + # NOTE: + # If the transfer approval operation has already been performed, get the data at that time. + # Otherwise, get the latest data. + if _transfer_approval_op is not None: + _from_address_personal_info = _transfer_approval_op.from_address_personal_info + _to_address_personal_info = _transfer_approval_op.to_address_personal_info + else: + _from_account: IDXPersonalInfo | None = db.scalars( + select(IDXPersonalInfo) + .where( + and_( + IDXPersonalInfo.account_address == _transfer_approval.from_address, + IDXPersonalInfo.issuer_address == _token.issuer_address, + ) + ) + .limit(1) + ).first() + _from_address_personal_info = ( + _from_account.personal_info if _from_account is not None else None + ) + + _to_account: IDXPersonalInfo | None = db.scalars( + select(IDXPersonalInfo) + .where( + and_( + IDXPersonalInfo.account_address == _transfer_approval.to_address, + IDXPersonalInfo.issuer_address == _token.issuer_address, + ) + ) + .limit(1) + ).first() + _to_address_personal_info = ( + _to_account.personal_info if _to_account is not None else None + ) + history = { "id": _transfer_approval.id, "token_address": token_address, "exchange_address": _transfer_approval.exchange_address, "application_id": _transfer_approval.application_id, "from_address": _transfer_approval.from_address, + "from_address_personal_information": _from_address_personal_info, "to_address": _transfer_approval.to_address, + "to_address_personal_information": _to_address_personal_info, "amount": _transfer_approval.amount, "application_datetime": application_datetime, "application_blocktimestamp": application_blocktimestamp, @@ -3352,7 +3430,6 @@ def retrieve_transfer_approval_history(db: DBSession, token_address: str, id: in "status": status, "issuer_cancelable": issuer_cancelable, } - return json_response(history) diff --git a/app/routers/share.py b/app/routers/share.py index 8cb5de13..9a93f41b 100644 --- a/app/routers/share.py +++ b/app/routers/share.py @@ -49,6 +49,7 @@ AuthorizationError, ContractRevertError, InvalidParameterError, + OperationNotAllowedStateError, SendTransactionError, ) from app.model.blockchain import ( @@ -133,6 +134,7 @@ TokenUpdateOperationCategory, TransferApprovalHistoryResponse, TransferApprovalsResponse, + TransferApprovalTokenDetailResponse, TransferApprovalTokenResponse, TransferHistoryResponse, UpdateTransferApprovalOperationType, @@ -3008,6 +3010,7 @@ def list_token_transfer_approval_history( InvalidParameterError, SendTransactionError, ContractRevertError, + OperationNotAllowedStateError, ), ) def update_transfer_approval( @@ -3108,6 +3111,37 @@ def update_transfer_approval( if transfer_approval_op is not None: raise InvalidParameterError("duplicate operation") + # Check the existence of personal information data for from_address and to_address + _from_address_personal_info: IDXPersonalInfo | None = db.scalars( + select(IDXPersonalInfo) + .where( + and_( + IDXPersonalInfo.account_address == _transfer_approval.from_address, + IDXPersonalInfo.issuer_address == issuer_address, + ) + ) + .limit(1) + ).first() + if _from_address_personal_info is None: + raise OperationNotAllowedStateError( + 101, "personal information for from_address is not registered" + ) + + _to_address_personal_info: IDXPersonalInfo | None = db.scalars( + select(IDXPersonalInfo) + .where( + and_( + IDXPersonalInfo.account_address == _transfer_approval.to_address, + IDXPersonalInfo.issuer_address == issuer_address, + ) + ) + .limit(1) + ).first() + if _to_address_personal_info is None: + raise OperationNotAllowedStateError( + 101, "personal information for to_address is not registered" + ) + # Send transaction # - APPROVE -> approveTransfer # In the case of a transfer approval for a token, if the transaction is reverted, @@ -3179,6 +3213,12 @@ def update_transfer_approval( transfer_approval_op.exchange_address = _transfer_approval.exchange_address transfer_approval_op.application_id = _transfer_approval.application_id transfer_approval_op.operation_type = data.operation_type + transfer_approval_op.from_address_personal_info = ( + _from_address_personal_info.personal_info + ) + transfer_approval_op.to_address_personal_info = ( + _to_address_personal_info.personal_info + ) db.add(transfer_approval_op) db.commit() @@ -3186,7 +3226,7 @@ def update_transfer_approval( # GET: /share/transfer_approvals/{token_address}/{id} @router.get( "/transfer_approvals/{token_address}/{id}", - response_model=TransferApprovalTokenResponse, + response_model=TransferApprovalTokenDetailResponse, responses=get_routers_responses(422, 404, InvalidParameterError), ) def retrieve_transfer_approval_history( @@ -3327,13 +3367,51 @@ def retrieve_transfer_approval_history( else: cancellation_blocktimestamp = None + # Get personal information of account address + # NOTE: + # If the transfer approval operation has already been performed, get the data at that time. + # Otherwise, get the latest data. + if _transfer_approval_op is not None: + _from_address_personal_info = _transfer_approval_op.from_address_personal_info + _to_address_personal_info = _transfer_approval_op.to_address_personal_info + else: + _from_account: IDXPersonalInfo | None = db.scalars( + select(IDXPersonalInfo) + .where( + and_( + IDXPersonalInfo.account_address == _transfer_approval.from_address, + IDXPersonalInfo.issuer_address == _token.issuer_address, + ) + ) + .limit(1) + ).first() + _from_address_personal_info = ( + _from_account.personal_info if _from_account is not None else None + ) + + _to_account: IDXPersonalInfo | None = db.scalars( + select(IDXPersonalInfo) + .where( + and_( + IDXPersonalInfo.account_address == _transfer_approval.to_address, + IDXPersonalInfo.issuer_address == _token.issuer_address, + ) + ) + .limit(1) + ).first() + _to_address_personal_info = ( + _to_account.personal_info if _to_account is not None else None + ) + history = { "id": _transfer_approval.id, "token_address": token_address, "exchange_address": _transfer_approval.exchange_address, "application_id": _transfer_approval.application_id, "from_address": _transfer_approval.from_address, + "from_address_personal_information": _from_address_personal_info, "to_address": _transfer_approval.to_address, + "to_address_personal_information": _to_address_personal_info, "amount": _transfer_approval.amount, "application_datetime": application_datetime, "application_blocktimestamp": application_blocktimestamp, diff --git a/migrations/versions/380a311952f4_v23_12_0_feature_555.py b/migrations/versions/380a311952f4_v23_12_0_feature_555.py new file mode 100644 index 00000000..09da4d49 --- /dev/null +++ b/migrations/versions/380a311952f4_v23_12_0_feature_555.py @@ -0,0 +1,42 @@ +"""v23_12_0_feature_555 + +Revision ID: 380a311952f4 +Revises: 53caec5f4dce +Create Date: 2023-11-08 11:20:47.891601 + +""" +from alembic import op +import sqlalchemy as sa + + +from app.database import get_db_schema + +# revision identifiers, used by Alembic. +revision = "380a311952f4" +down_revision = "53caec5f4dce" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "transfer_approval_history", + sa.Column("from_address_personal_info", sa.JSON(), nullable=True), + schema=get_db_schema(), + ) + op.add_column( + "transfer_approval_history", + sa.Column("to_address_personal_info", sa.JSON(), nullable=True), + schema=get_db_schema(), + ) + + +def downgrade(): + op.drop_column( + "transfer_approval_history", "to_address_personal_info", schema=get_db_schema() + ) + op.drop_column( + "transfer_approval_history", + "from_address_personal_info", + schema=get_db_schema(), + ) diff --git a/tests/test_app_routers_bond_transfer_approvals_{token_Address}_GET.py b/tests/test_app_routers_bond_transfer_approvals_{token_address}_GET.py similarity index 100% rename from tests/test_app_routers_bond_transfer_approvals_{token_Address}_GET.py rename to tests/test_app_routers_bond_transfer_approvals_{token_address}_GET.py diff --git a/tests/test_app_routers_bond_transfer_approvals_{token_Address}_{id}_GET.py b/tests/test_app_routers_bond_transfer_approvals_{token_address}_{id}_GET.py similarity index 58% rename from tests/test_app_routers_bond_transfer_approvals_{token_Address}_{id}_GET.py rename to tests/test_app_routers_bond_transfer_approvals_{token_address}_{id}_GET.py index 9c588952..dadc1f39 100644 --- a/tests/test_app_routers_bond_transfer_approvals_{token_Address}_{id}_GET.py +++ b/tests/test_app_routers_bond_transfer_approvals_{token_address}_{id}_GET.py @@ -22,6 +22,7 @@ import config from app.model.db import ( + IDXPersonalInfo, IDXTransferApproval, Token, TokenType, @@ -83,9 +84,9 @@ class TestAppRoutersBondTransferApprovalsTokenAddressIdGET: # Normal Case ########################################################################### - # + # # unapproved data - def test_normal_1(self, client, db): + def test_normal_1_1(self, client, db): # prepare data: Token _token = Token() _token.type = TokenType.IBET_STRAIGHT_BOND.value @@ -95,6 +96,38 @@ def test_normal_1(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXPersonalInfo + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = self.test_issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # latest data + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = self.test_issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # latest data + db.add(_personal_info_to) + + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -127,7 +160,88 @@ def test_normal_1(self, client, db): "exchange_address": config.ZERO_ADDRESS, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, + "amount": 200, + "application_datetime": self.test_application_datetime_str, + "application_blocktimestamp": self.test_application_blocktimestamp_str, + "approval_datetime": None, + "approval_blocktimestamp": None, + "cancellation_blocktimestamp": None, + "cancelled": False, + "escrow_finished": False, + "transfer_approved": False, + "status": 0, + "issuer_cancelable": True, + } + + # + # unapproved data (Personal information is not synchronized) + def test_normal_1_2(self, client, db): + # prepare data: Token + _token = Token() + _token.type = TokenType.IBET_STRAIGHT_BOND.value + _token.tx_hash = self.test_transaction_hash + _token.issuer_address = self.test_issuer_address + _token.token_address = self.test_token_address + _token.abi = {} + db.add(_token) + + # prepare data: IDXTransferApproval + id = 10 + _idx_transfer_approval = IDXTransferApproval() + _idx_transfer_approval.id = id + _idx_transfer_approval.token_address = self.test_token_address + _idx_transfer_approval.exchange_address = config.ZERO_ADDRESS + _idx_transfer_approval.application_id = 100 + _idx_transfer_approval.from_address = self.test_from_address + _idx_transfer_approval.to_address = self.test_to_address + _idx_transfer_approval.amount = 200 + _idx_transfer_approval.application_datetime = self.test_application_datetime + _idx_transfer_approval.application_blocktimestamp = ( + self.test_application_blocktimestamp + ) + _idx_transfer_approval.approval_datetime = None + _idx_transfer_approval.approval_blocktimestamp = None + _idx_transfer_approval.cancellation_blocktimestamp = None + _idx_transfer_approval.cancelled = None + _idx_transfer_approval.escrow_finished = None + _idx_transfer_approval.transfer_approved = None + db.add(_idx_transfer_approval) + + # request target API + resp = client.get(self.base_url.format(self.test_token_address, id)) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "id": 10, + "token_address": self.test_token_address, + "exchange_address": config.ZERO_ADDRESS, + "application_id": 100, + "from_address": self.test_from_address, + "from_address_personal_information": None, + "to_address": self.test_to_address, + "to_address_personal_information": None, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, @@ -154,6 +268,7 @@ def test_normal_2_1(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -169,17 +284,38 @@ def test_normal_2_1(self, client, db): ) _idx_transfer_approval.approval_datetime = None _idx_transfer_approval.approval_blocktimestamp = None - _idx_transfer_approval.cancellation_blocktimestamp = None - _idx_transfer_approval.cancelled = None + _idx_transfer_approval.cancellation_blocktimestamp = None # event synchronizing + _idx_transfer_approval.cancelled = None # event synchronizing _idx_transfer_approval.escrow_finished = None _idx_transfer_approval.transfer_approved = None db.add(_idx_transfer_approval) + # prepare data: TransferApprovalHistory _cancel_op = TransferApprovalHistory() _cancel_op.token_address = self.test_token_address _cancel_op.exchange_address = self.test_exchange_address _cancel_op.application_id = 100 _cancel_op.operation_type = TransferApprovalOperationType.CANCEL.value + _cancel_op.from_address_personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # snapshot + _cancel_op.to_address_personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # snapshot db.add(_cancel_op) # request target API @@ -193,7 +329,27 @@ def test_normal_2_1(self, client, db): "exchange_address": self.test_exchange_address, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, @@ -219,6 +375,7 @@ def test_normal_2_2(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -236,12 +393,40 @@ def test_normal_2_2(self, client, db): _idx_transfer_approval.approval_blocktimestamp = None _idx_transfer_approval.cancellation_blocktimestamp = ( self.test_cancellation_blocktimestamp - ) - _idx_transfer_approval.cancelled = True + ) # event synced + _idx_transfer_approval.cancelled = True # event synced _idx_transfer_approval.escrow_finished = None _idx_transfer_approval.transfer_approved = None db.add(_idx_transfer_approval) + # prepare data: TransferApprovalHistory + _cancel_op = TransferApprovalHistory() + _cancel_op.token_address = self.test_token_address + _cancel_op.exchange_address = self.test_exchange_address + _cancel_op.application_id = 100 + _cancel_op.operation_type = TransferApprovalOperationType.CANCEL.value + _cancel_op.from_address_personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # snapshot + _cancel_op.to_address_personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # snapshot + db.add(_cancel_op) + # request target API resp = client.get(self.base_url.format(self.test_token_address, id)) @@ -253,7 +438,27 @@ def test_normal_2_2(self, client, db): "exchange_address": self.test_exchange_address, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, @@ -279,6 +484,38 @@ def test_normal_3(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXPersonalInfo + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = self.test_issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # latest data + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = self.test_issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # latest data + db.add(_personal_info_to) + + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -296,7 +533,7 @@ def test_normal_3(self, client, db): _idx_transfer_approval.approval_blocktimestamp = None _idx_transfer_approval.cancellation_blocktimestamp = None _idx_transfer_approval.cancelled = None - _idx_transfer_approval.escrow_finished = True + _idx_transfer_approval.escrow_finished = True # escrow finished _idx_transfer_approval.transfer_approved = False db.add(_idx_transfer_approval) @@ -311,7 +548,27 @@ def test_normal_3(self, client, db): "exchange_address": self.test_exchange_address, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, @@ -338,6 +595,7 @@ def test_normal_4_1(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -358,14 +616,35 @@ def test_normal_4_1(self, client, db): _idx_transfer_approval.cancellation_blocktimestamp = None _idx_transfer_approval.cancelled = None _idx_transfer_approval.escrow_finished = True - _idx_transfer_approval.transfer_approved = None + _idx_transfer_approval.transfer_approved = None # event synchronizing db.add(_idx_transfer_approval) + # prepare data: TransferApprovalHistory _approval_op = TransferApprovalHistory() _approval_op.token_address = self.test_token_address _approval_op.exchange_address = self.test_exchange_address _approval_op.application_id = 100 _approval_op.operation_type = TransferApprovalOperationType.APPROVE.value + _approval_op.from_address_personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # snapshot + _approval_op.to_address_personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # snapshot db.add(_approval_op) # request target API @@ -379,7 +658,27 @@ def test_normal_4_1(self, client, db): "exchange_address": self.test_exchange_address, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, @@ -405,6 +704,7 @@ def test_normal_4_2(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -425,9 +725,37 @@ def test_normal_4_2(self, client, db): _idx_transfer_approval.cancellation_blocktimestamp = None _idx_transfer_approval.cancelled = None _idx_transfer_approval.escrow_finished = True - _idx_transfer_approval.transfer_approved = True + _idx_transfer_approval.transfer_approved = True # synced db.add(_idx_transfer_approval) + # prepare data: TransferApprovalHistory + _approval_op = TransferApprovalHistory() + _approval_op.token_address = self.test_token_address + _approval_op.exchange_address = self.test_exchange_address + _approval_op.application_id = 100 + _approval_op.operation_type = TransferApprovalOperationType.APPROVE.value + _approval_op.from_address_personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # snapshot + _approval_op.to_address_personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # snapshot + db.add(_approval_op) + # request target API resp = client.get(self.base_url.format(self.test_token_address, id)) @@ -439,7 +767,27 @@ def test_normal_4_2(self, client, db): "exchange_address": self.test_exchange_address, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, diff --git a/tests/test_app_routers_bond_transfer_approvals_{token_Address}_{id}_POST.py b/tests/test_app_routers_bond_transfer_approvals_{token_address}_{id}_POST.py similarity index 72% rename from tests/test_app_routers_bond_transfer_approvals_{token_Address}_{id}_POST.py rename to tests/test_app_routers_bond_transfer_approvals_{token_address}_{id}_POST.py index 53022aed..f1c688bd 100644 --- a/tests/test_app_routers_bond_transfer_approvals_{token_Address}_{id}_POST.py +++ b/tests/test_app_routers_bond_transfer_approvals_{token_address}_{id}_POST.py @@ -37,6 +37,7 @@ from app.model.db import ( Account, AuthToken, + IDXPersonalInfo, IDXTransferApproval, Token, TokenType, @@ -111,6 +112,36 @@ def test_normal_1_1(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenContract_approve_transfer = mock.patch( target="app.model.blockchain.token.IbetSecurityTokenInterface.approve_transfer", @@ -149,6 +180,26 @@ def test_normal_1_1(self, client, db): assert approval_op.exchange_address == config.ZERO_ADDRESS assert approval_op.application_id == 100 assert approval_op.operation_type == TransferApprovalOperationType.APPROVE.value + assert approval_op.from_address_personal_info == { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + assert approval_op.to_address_personal_info == { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # # APPROVE @@ -193,6 +244,36 @@ def test_normal_1_2(self, client, db): _idx_transfer_approval.escrow_finished = True db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenEscrow_approve_transfer = mock.patch( target="app.model.blockchain.exchange.IbetSecurityTokenEscrow.approve_transfer", @@ -231,6 +312,26 @@ def test_normal_1_2(self, client, db): assert approval_op.exchange_address == self.test_exchange_address assert approval_op.application_id == 100 assert approval_op.operation_type == TransferApprovalOperationType.APPROVE.value + assert approval_op.from_address_personal_info == { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + assert approval_op.to_address_personal_info == { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # # CANCEL @@ -274,6 +375,36 @@ def test_normal_2_1(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenContract_cancel_transfer = mock.patch( target="app.model.blockchain.token.IbetSecurityTokenInterface.cancel_transfer", @@ -312,6 +443,26 @@ def test_normal_2_1(self, client, db): assert cancel_op.exchange_address == config.ZERO_ADDRESS assert cancel_op.application_id == 100 assert cancel_op.operation_type == TransferApprovalOperationType.CANCEL.value + assert cancel_op.from_address_personal_info == { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + assert cancel_op.to_address_personal_info == { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # # Authorization by auth-token @@ -360,6 +511,36 @@ def test_normal_3(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenContract_approve_transfer = mock.patch( target="app.model.blockchain.token.IbetSecurityTokenInterface.approve_transfer", @@ -1054,6 +1235,36 @@ def test_error_5_1(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # request target API resp = client.post( self.base_url.format(self.test_token_address, id), @@ -1114,6 +1325,36 @@ def test_error_5_2(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenContract_approve_transfer = mock.patch( target="app.model.blockchain.token.IbetSecurityTokenInterface.approve_transfer", @@ -1192,6 +1433,36 @@ def test_error_5_3(self, client, db): _idx_transfer_approval.escrow_finished = True db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # request target API resp = client.post( self.base_url.format(self.test_token_address, id), @@ -1253,6 +1524,36 @@ def test_error_5_4(self, client, db): _idx_transfer_approval.escrow_finished = True db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenEscrow_approve_transfer = mock.patch( target="app.model.blockchain.exchange.IbetSecurityTokenEscrow.approve_transfer", @@ -1324,6 +1625,36 @@ def test_error_6_1(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # request target API resp = client.post( self.base_url.format(self.test_token_address, id), @@ -1384,6 +1715,36 @@ def test_error_6_2(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenContract_cancel_transfer = mock.patch( target="app.model.blockchain.token.IbetSecurityTokenInterface.cancel_transfer", @@ -1407,3 +1768,148 @@ def test_error_6_2(self, client, db): "meta": {"code": 120802, "title": "ContractRevertError"}, "detail": "Application is invalid.", } + + # + # InvalidParameterError + # personal information for from_address is not registered + def test_error_7_1(self, client, db): + issuer = config_eth_account("user1") + issuer_address = issuer["address"] + + # prepare data + account = Account() + account.issuer_address = issuer_address + account.keyfile = issuer["keyfile_json"] + account.eoa_password = E2EEUtils.encrypt("password") + db.add(account) + + _token = Token() + _token.type = TokenType.IBET_STRAIGHT_BOND.value + _token.tx_hash = self.test_transaction_hash + _token.issuer_address = issuer_address + _token.token_address = self.test_token_address + _token.abi = {} + db.add(_token) + + id = 10 + _idx_transfer_approval = IDXTransferApproval() + _idx_transfer_approval.id = id + _idx_transfer_approval.token_address = self.test_token_address + _idx_transfer_approval.exchange_address = config.ZERO_ADDRESS + _idx_transfer_approval.application_id = 100 + _idx_transfer_approval.from_address = self.test_from_address + _idx_transfer_approval.to_address = self.test_to_address + _idx_transfer_approval.amount = 200 + _idx_transfer_approval.application_datetime = self.test_application_datetime + _idx_transfer_approval.application_blocktimestamp = ( + self.test_application_blocktimestamp + ) + _idx_transfer_approval.approval_datetime = None + _idx_transfer_approval.approval_blocktimestamp = None + _idx_transfer_approval.cancellation_blocktimestamp = None + _idx_transfer_approval.cancelled = None + db.add(_idx_transfer_approval) + + # mock + IbetSecurityTokenContract_approve_transfer = mock.patch( + target="app.model.blockchain.token.IbetSecurityTokenInterface.approve_transfer", + return_value=("test_tx_hash", {"status": 1}), + ) + + # request target API + with IbetSecurityTokenContract_approve_transfer as mock_transfer: + resp = client.post( + self.base_url.format(self.test_token_address, id), + json={"operation_type": "approve"}, + headers={ + "issuer-address": issuer_address, + "eoa-password": E2EEUtils.encrypt("password"), + }, + ) + + # Assertion + assert resp.status_code == 400 + assert resp.json() == { + "meta": {"code": 101, "title": "OperationNotAllowedStateError"}, + "detail": "personal information for from_address is not registered", + } + + # + # InvalidParameterError + # personal information for to_address is not registered + def test_error_7_2(self, client, db): + issuer = config_eth_account("user1") + issuer_address = issuer["address"] + + # prepare data + account = Account() + account.issuer_address = issuer_address + account.keyfile = issuer["keyfile_json"] + account.eoa_password = E2EEUtils.encrypt("password") + db.add(account) + + _token = Token() + _token.type = TokenType.IBET_STRAIGHT_BOND.value + _token.tx_hash = self.test_transaction_hash + _token.issuer_address = issuer_address + _token.token_address = self.test_token_address + _token.abi = {} + db.add(_token) + + id = 10 + _idx_transfer_approval = IDXTransferApproval() + _idx_transfer_approval.id = id + _idx_transfer_approval.token_address = self.test_token_address + _idx_transfer_approval.exchange_address = config.ZERO_ADDRESS + _idx_transfer_approval.application_id = 100 + _idx_transfer_approval.from_address = self.test_from_address + _idx_transfer_approval.to_address = self.test_to_address + _idx_transfer_approval.amount = 200 + _idx_transfer_approval.application_datetime = self.test_application_datetime + _idx_transfer_approval.application_blocktimestamp = ( + self.test_application_blocktimestamp + ) + _idx_transfer_approval.approval_datetime = None + _idx_transfer_approval.approval_blocktimestamp = None + _idx_transfer_approval.cancellation_blocktimestamp = None + _idx_transfer_approval.cancelled = None + db.add(_idx_transfer_approval) + + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + # mock + IbetSecurityTokenContract_approve_transfer = mock.patch( + target="app.model.blockchain.token.IbetSecurityTokenInterface.approve_transfer", + return_value=("test_tx_hash", {"status": 1}), + ) + + # request target API + with IbetSecurityTokenContract_approve_transfer as mock_transfer: + resp = client.post( + self.base_url.format(self.test_token_address, id), + json={"operation_type": "approve"}, + headers={ + "issuer-address": issuer_address, + "eoa-password": E2EEUtils.encrypt("password"), + }, + ) + + # Assertion + assert resp.status_code == 400 + assert resp.json() == { + "meta": {"code": 101, "title": "OperationNotAllowedStateError"}, + "detail": "personal information for to_address is not registered", + } diff --git a/tests/test_app_routers_bond_transfers_{token_Address}_GET.py b/tests/test_app_routers_bond_transfers_{token_address}_GET.py similarity index 100% rename from tests/test_app_routers_bond_transfers_{token_Address}_GET.py rename to tests/test_app_routers_bond_transfers_{token_address}_GET.py diff --git a/tests/test_app_routers_share_transfer_approvals_{token_Address}_GET.py b/tests/test_app_routers_share_transfer_approvals_{token_address}_GET.py similarity index 100% rename from tests/test_app_routers_share_transfer_approvals_{token_Address}_GET.py rename to tests/test_app_routers_share_transfer_approvals_{token_address}_GET.py diff --git a/tests/test_app_routers_share_transfer_approvals_{token_Address}_{id}_GET.py b/tests/test_app_routers_share_transfer_approvals_{token_address}_{id}_GET.py similarity index 58% rename from tests/test_app_routers_share_transfer_approvals_{token_Address}_{id}_GET.py rename to tests/test_app_routers_share_transfer_approvals_{token_address}_{id}_GET.py index bfcacaa9..076a0605 100644 --- a/tests/test_app_routers_share_transfer_approvals_{token_Address}_{id}_GET.py +++ b/tests/test_app_routers_share_transfer_approvals_{token_address}_{id}_GET.py @@ -22,6 +22,7 @@ import config from app.model.db import ( + IDXPersonalInfo, IDXTransferApproval, Token, TokenType, @@ -83,9 +84,9 @@ class TestAppRoutersShareTransferApprovalsTokenAddressIdGET: # Normal Case ########################################################################### - # + # # unapproved data - def test_normal_1(self, client, db): + def test_normal_1_1(self, client, db): # prepare data: Token _token = Token() _token.type = TokenType.IBET_SHARE.value @@ -95,6 +96,38 @@ def test_normal_1(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXPersonalInfo + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = self.test_issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # latest data + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = self.test_issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # latest data + db.add(_personal_info_to) + + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -127,7 +160,88 @@ def test_normal_1(self, client, db): "exchange_address": config.ZERO_ADDRESS, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, + "amount": 200, + "application_datetime": self.test_application_datetime_str, + "application_blocktimestamp": self.test_application_blocktimestamp_str, + "approval_datetime": None, + "approval_blocktimestamp": None, + "cancellation_blocktimestamp": None, + "cancelled": False, + "escrow_finished": False, + "transfer_approved": False, + "status": 0, + "issuer_cancelable": True, + } + + # + # unapproved data (Personal information is not synchronized) + def test_normal_1_2(self, client, db): + # prepare data: Token + _token = Token() + _token.type = TokenType.IBET_SHARE.value + _token.tx_hash = self.test_transaction_hash + _token.issuer_address = self.test_issuer_address + _token.token_address = self.test_token_address + _token.abi = {} + db.add(_token) + + # prepare data: IDXTransferApproval + id = 10 + _idx_transfer_approval = IDXTransferApproval() + _idx_transfer_approval.id = id + _idx_transfer_approval.token_address = self.test_token_address + _idx_transfer_approval.exchange_address = config.ZERO_ADDRESS + _idx_transfer_approval.application_id = 100 + _idx_transfer_approval.from_address = self.test_from_address + _idx_transfer_approval.to_address = self.test_to_address + _idx_transfer_approval.amount = 200 + _idx_transfer_approval.application_datetime = self.test_application_datetime + _idx_transfer_approval.application_blocktimestamp = ( + self.test_application_blocktimestamp + ) + _idx_transfer_approval.approval_datetime = None + _idx_transfer_approval.approval_blocktimestamp = None + _idx_transfer_approval.cancellation_blocktimestamp = None + _idx_transfer_approval.cancelled = None + _idx_transfer_approval.escrow_finished = None + _idx_transfer_approval.transfer_approved = None + db.add(_idx_transfer_approval) + + # request target API + resp = client.get(self.base_url.format(self.test_token_address, id)) + + # assertion + assert resp.status_code == 200 + assert resp.json() == { + "id": 10, + "token_address": self.test_token_address, + "exchange_address": config.ZERO_ADDRESS, + "application_id": 100, + "from_address": self.test_from_address, + "from_address_personal_information": None, + "to_address": self.test_to_address, + "to_address_personal_information": None, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, @@ -154,6 +268,7 @@ def test_normal_2_1(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -169,17 +284,38 @@ def test_normal_2_1(self, client, db): ) _idx_transfer_approval.approval_datetime = None _idx_transfer_approval.approval_blocktimestamp = None - _idx_transfer_approval.cancellation_blocktimestamp = None - _idx_transfer_approval.cancelled = None + _idx_transfer_approval.cancellation_blocktimestamp = None # event synchronizing + _idx_transfer_approval.cancelled = None # event synchronizing _idx_transfer_approval.escrow_finished = None _idx_transfer_approval.transfer_approved = None db.add(_idx_transfer_approval) + # prepare data: TransferApprovalHistory _cancel_op = TransferApprovalHistory() _cancel_op.token_address = self.test_token_address _cancel_op.exchange_address = self.test_exchange_address _cancel_op.application_id = 100 _cancel_op.operation_type = TransferApprovalOperationType.CANCEL.value + _cancel_op.from_address_personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # snapshot + _cancel_op.to_address_personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # snapshot db.add(_cancel_op) # request target API @@ -193,7 +329,27 @@ def test_normal_2_1(self, client, db): "exchange_address": self.test_exchange_address, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, @@ -219,6 +375,7 @@ def test_normal_2_2(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -236,12 +393,40 @@ def test_normal_2_2(self, client, db): _idx_transfer_approval.approval_blocktimestamp = None _idx_transfer_approval.cancellation_blocktimestamp = ( self.test_cancellation_blocktimestamp - ) - _idx_transfer_approval.cancelled = True + ) # event synced + _idx_transfer_approval.cancelled = True # event synced _idx_transfer_approval.escrow_finished = None _idx_transfer_approval.transfer_approved = None db.add(_idx_transfer_approval) + # prepare data: TransferApprovalHistory + _cancel_op = TransferApprovalHistory() + _cancel_op.token_address = self.test_token_address + _cancel_op.exchange_address = self.test_exchange_address + _cancel_op.application_id = 100 + _cancel_op.operation_type = TransferApprovalOperationType.CANCEL.value + _cancel_op.from_address_personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # snapshot + _cancel_op.to_address_personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # snapshot + db.add(_cancel_op) + # request target API resp = client.get(self.base_url.format(self.test_token_address, id)) @@ -253,7 +438,27 @@ def test_normal_2_2(self, client, db): "exchange_address": self.test_exchange_address, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, @@ -279,6 +484,38 @@ def test_normal_3(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXPersonalInfo + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = self.test_issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # latest data + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = self.test_issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # latest data + db.add(_personal_info_to) + + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -296,7 +533,7 @@ def test_normal_3(self, client, db): _idx_transfer_approval.approval_blocktimestamp = None _idx_transfer_approval.cancellation_blocktimestamp = None _idx_transfer_approval.cancelled = None - _idx_transfer_approval.escrow_finished = True + _idx_transfer_approval.escrow_finished = True # escrow finished _idx_transfer_approval.transfer_approved = False db.add(_idx_transfer_approval) @@ -311,7 +548,27 @@ def test_normal_3(self, client, db): "exchange_address": self.test_exchange_address, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, @@ -338,6 +595,7 @@ def test_normal_4_1(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -358,14 +616,35 @@ def test_normal_4_1(self, client, db): _idx_transfer_approval.cancellation_blocktimestamp = None _idx_transfer_approval.cancelled = None _idx_transfer_approval.escrow_finished = True - _idx_transfer_approval.transfer_approved = None + _idx_transfer_approval.transfer_approved = None # event synchronizing db.add(_idx_transfer_approval) + # prepare data: TransferApprovalHistory _approval_op = TransferApprovalHistory() _approval_op.token_address = self.test_token_address _approval_op.exchange_address = self.test_exchange_address _approval_op.application_id = 100 _approval_op.operation_type = TransferApprovalOperationType.APPROVE.value + _approval_op.from_address_personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # snapshot + _approval_op.to_address_personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # snapshot db.add(_approval_op) # request target API @@ -379,7 +658,27 @@ def test_normal_4_1(self, client, db): "exchange_address": self.test_exchange_address, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, @@ -405,6 +704,7 @@ def test_normal_4_2(self, client, db): _token.abi = {} db.add(_token) + # prepare data: IDXTransferApproval id = 10 _idx_transfer_approval = IDXTransferApproval() _idx_transfer_approval.id = id @@ -425,9 +725,37 @@ def test_normal_4_2(self, client, db): _idx_transfer_approval.cancellation_blocktimestamp = None _idx_transfer_approval.cancelled = None _idx_transfer_approval.escrow_finished = True - _idx_transfer_approval.transfer_approved = True + _idx_transfer_approval.transfer_approved = True # synced db.add(_idx_transfer_approval) + # prepare data: TransferApprovalHistory + _approval_op = TransferApprovalHistory() + _approval_op.token_address = self.test_token_address + _approval_op.exchange_address = self.test_exchange_address + _approval_op.application_id = 100 + _approval_op.operation_type = TransferApprovalOperationType.APPROVE.value + _approval_op.from_address_personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } # snapshot + _approval_op.to_address_personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # snapshot + db.add(_approval_op) + # request target API resp = client.get(self.base_url.format(self.test_token_address, id)) @@ -439,7 +767,27 @@ def test_normal_4_2(self, client, db): "exchange_address": self.test_exchange_address, "application_id": 100, "from_address": self.test_from_address, + "from_address_personal_information": { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + }, "to_address": self.test_to_address, + "to_address_personal_information": { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + }, "amount": 200, "application_datetime": self.test_application_datetime_str, "application_blocktimestamp": self.test_application_blocktimestamp_str, diff --git a/tests/test_app_routers_share_transfer_approvals_{token_Address}_{id}_POST.py b/tests/test_app_routers_share_transfer_approvals_{token_address}_{id}_POST.py similarity index 72% rename from tests/test_app_routers_share_transfer_approvals_{token_Address}_{id}_POST.py rename to tests/test_app_routers_share_transfer_approvals_{token_address}_{id}_POST.py index 7c83a35f..ce02e0e6 100644 --- a/tests/test_app_routers_share_transfer_approvals_{token_Address}_{id}_POST.py +++ b/tests/test_app_routers_share_transfer_approvals_{token_address}_{id}_POST.py @@ -37,6 +37,7 @@ from app.model.db import ( Account, AuthToken, + IDXPersonalInfo, IDXTransferApproval, Token, TokenType, @@ -111,6 +112,36 @@ def test_normal_1_1(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenContract_approve_transfer = mock.patch( target="app.model.blockchain.token.IbetSecurityTokenInterface.approve_transfer", @@ -149,6 +180,26 @@ def test_normal_1_1(self, client, db): assert approval_op.exchange_address == config.ZERO_ADDRESS assert approval_op.application_id == 100 assert approval_op.operation_type == TransferApprovalOperationType.APPROVE.value + assert approval_op.from_address_personal_info == { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + assert approval_op.to_address_personal_info == { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # # APPROVE @@ -193,6 +244,36 @@ def test_normal_1_2(self, client, db): _idx_transfer_approval.escrow_finished = True db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenEscrow_approve_transfer = mock.patch( target="app.model.blockchain.exchange.IbetSecurityTokenEscrow.approve_transfer", @@ -231,6 +312,26 @@ def test_normal_1_2(self, client, db): assert approval_op.exchange_address == self.test_exchange_address assert approval_op.application_id == 100 assert approval_op.operation_type == TransferApprovalOperationType.APPROVE.value + assert approval_op.from_address_personal_info == { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + assert approval_op.to_address_personal_info == { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # # CANCEL @@ -274,6 +375,36 @@ def test_normal_2_1(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenContract_cancel_transfer = mock.patch( target="app.model.blockchain.token.IbetSecurityTokenInterface.cancel_transfer", @@ -312,6 +443,26 @@ def test_normal_2_1(self, client, db): assert cancel_op.exchange_address == config.ZERO_ADDRESS assert cancel_op.application_id == 100 assert cancel_op.operation_type == TransferApprovalOperationType.CANCEL.value + assert cancel_op.from_address_personal_info == { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + assert cancel_op.to_address_personal_info == { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } # # Authorization by auth-token @@ -360,6 +511,36 @@ def test_normal_3(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenContract_approve_transfer = mock.patch( target="app.model.blockchain.token.IbetSecurityTokenInterface.approve_transfer", @@ -1054,6 +1235,36 @@ def test_error_5_1(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # request target API resp = client.post( self.base_url.format(self.test_token_address, id), @@ -1114,6 +1325,36 @@ def test_error_5_2(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenContract_approve_transfer = mock.patch( target="app.model.blockchain.token.IbetSecurityTokenInterface.approve_transfer", @@ -1192,6 +1433,36 @@ def test_error_5_3(self, client, db): _idx_transfer_approval.escrow_finished = True db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # request target API resp = client.post( self.base_url.format(self.test_token_address, id), @@ -1253,6 +1524,36 @@ def test_error_5_4(self, client, db): _idx_transfer_approval.escrow_finished = True db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenEscrow_approve_transfer = mock.patch( target="app.model.blockchain.exchange.IbetSecurityTokenEscrow.approve_transfer", @@ -1324,6 +1625,36 @@ def test_error_6_1(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # request target API resp = client.post( self.base_url.format(self.test_token_address, id), @@ -1384,6 +1715,36 @@ def test_error_6_2(self, client, db): _idx_transfer_approval.cancelled = None db.add(_idx_transfer_approval) + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + _personal_info_to = IDXPersonalInfo() + _personal_info_to.account_address = self.test_to_address + _personal_info_to.issuer_address = issuer_address + _personal_info_to._personal_info = { + "key_manager": "key_manager_test2", + "name": "name_test2", + "postal_code": "postal_code_test2", + "address": "address_test2", + "email": "email_test2", + "birth": "birth_test2", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_to) + # mock IbetSecurityTokenContract_cancel_transfer = mock.patch( target="app.model.blockchain.token.IbetSecurityTokenInterface.cancel_transfer", @@ -1407,3 +1768,148 @@ def test_error_6_2(self, client, db): "meta": {"code": 110802, "title": "ContractRevertError"}, "detail": "Application is invalid.", } + + # + # InvalidParameterError + # personal information for from_address is not registered + def test_error_7_1(self, client, db): + issuer = config_eth_account("user1") + issuer_address = issuer["address"] + + # prepare data + account = Account() + account.issuer_address = issuer_address + account.keyfile = issuer["keyfile_json"] + account.eoa_password = E2EEUtils.encrypt("password") + db.add(account) + + _token = Token() + _token.type = TokenType.IBET_SHARE.value + _token.tx_hash = self.test_transaction_hash + _token.issuer_address = issuer_address + _token.token_address = self.test_token_address + _token.abi = {} + db.add(_token) + + id = 10 + _idx_transfer_approval = IDXTransferApproval() + _idx_transfer_approval.id = id + _idx_transfer_approval.token_address = self.test_token_address + _idx_transfer_approval.exchange_address = config.ZERO_ADDRESS + _idx_transfer_approval.application_id = 100 + _idx_transfer_approval.from_address = self.test_from_address + _idx_transfer_approval.to_address = self.test_to_address + _idx_transfer_approval.amount = 200 + _idx_transfer_approval.application_datetime = self.test_application_datetime + _idx_transfer_approval.application_blocktimestamp = ( + self.test_application_blocktimestamp + ) + _idx_transfer_approval.approval_datetime = None + _idx_transfer_approval.approval_blocktimestamp = None + _idx_transfer_approval.cancellation_blocktimestamp = None + _idx_transfer_approval.cancelled = None + db.add(_idx_transfer_approval) + + # mock + IbetSecurityTokenContract_approve_transfer = mock.patch( + target="app.model.blockchain.token.IbetSecurityTokenInterface.approve_transfer", + return_value=("test_tx_hash", {"status": 1}), + ) + + # request target API + with IbetSecurityTokenContract_approve_transfer as mock_transfer: + resp = client.post( + self.base_url.format(self.test_token_address, id), + json={"operation_type": "approve"}, + headers={ + "issuer-address": issuer_address, + "eoa-password": E2EEUtils.encrypt("password"), + }, + ) + + # Assertion + assert resp.status_code == 400 + assert resp.json() == { + "meta": {"code": 101, "title": "OperationNotAllowedStateError"}, + "detail": "personal information for from_address is not registered", + } + + # + # InvalidParameterError + # personal information for to_address is not registered + def test_error_7_2(self, client, db): + issuer = config_eth_account("user1") + issuer_address = issuer["address"] + + # prepare data + account = Account() + account.issuer_address = issuer_address + account.keyfile = issuer["keyfile_json"] + account.eoa_password = E2EEUtils.encrypt("password") + db.add(account) + + _token = Token() + _token.type = TokenType.IBET_SHARE.value + _token.tx_hash = self.test_transaction_hash + _token.issuer_address = issuer_address + _token.token_address = self.test_token_address + _token.abi = {} + db.add(_token) + + id = 10 + _idx_transfer_approval = IDXTransferApproval() + _idx_transfer_approval.id = id + _idx_transfer_approval.token_address = self.test_token_address + _idx_transfer_approval.exchange_address = config.ZERO_ADDRESS + _idx_transfer_approval.application_id = 100 + _idx_transfer_approval.from_address = self.test_from_address + _idx_transfer_approval.to_address = self.test_to_address + _idx_transfer_approval.amount = 200 + _idx_transfer_approval.application_datetime = self.test_application_datetime + _idx_transfer_approval.application_blocktimestamp = ( + self.test_application_blocktimestamp + ) + _idx_transfer_approval.approval_datetime = None + _idx_transfer_approval.approval_blocktimestamp = None + _idx_transfer_approval.cancellation_blocktimestamp = None + _idx_transfer_approval.cancelled = None + db.add(_idx_transfer_approval) + + _personal_info_from = IDXPersonalInfo() + _personal_info_from.account_address = self.test_from_address + _personal_info_from.issuer_address = issuer_address + _personal_info_from._personal_info = { + "key_manager": "key_manager_test1", + "name": "name_test1", + "postal_code": "postal_code_test1", + "address": "address_test1", + "email": "email_test1", + "birth": "birth_test1", + "is_corporate": False, + "tax_category": 10, + } + db.add(_personal_info_from) + + # mock + IbetSecurityTokenContract_approve_transfer = mock.patch( + target="app.model.blockchain.token.IbetSecurityTokenInterface.approve_transfer", + return_value=("test_tx_hash", {"status": 1}), + ) + + # request target API + with IbetSecurityTokenContract_approve_transfer as mock_transfer: + resp = client.post( + self.base_url.format(self.test_token_address, id), + json={"operation_type": "approve"}, + headers={ + "issuer-address": issuer_address, + "eoa-password": E2EEUtils.encrypt("password"), + }, + ) + + # Assertion + assert resp.status_code == 400 + assert resp.json() == { + "meta": {"code": 101, "title": "OperationNotAllowedStateError"}, + "detail": "personal information for to_address is not registered", + } diff --git a/tests/test_app_routers_share_transfers_{token_Address}_GET.py b/tests/test_app_routers_share_transfers_{token_address}_GET.py similarity index 100% rename from tests/test_app_routers_share_transfers_{token_Address}_GET.py rename to tests/test_app_routers_share_transfers_{token_address}_GET.py