diff --git a/CHANGELOG.md b/CHANGELOG.md index 785f601c9..b57f867d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,8 @@ This changelog is based on [Keep a Changelog](https://keepachangelog.com/en/1.1. ### Changed - Removed duplicate import of transaction_pb2 in transaction.py +- Refactor `TokenInfo` into an immutable dataclass, remove all setters, and rewrite `_from_proto` as a pure factory for consistent parsing [#800] + ### Fixed - fixed workflow: changelog check with improved sensitivity to deletions, additions, new releases diff --git a/src/hiero_sdk_python/tokens/token_info.py b/src/hiero_sdk_python/tokens/token_info.py index 488cf9067..735414fed 100644 --- a/src/hiero_sdk_python/tokens/token_info.py +++ b/src/hiero_sdk_python/tokens/token_info.py @@ -28,7 +28,7 @@ from hiero_sdk_python.hapi.services import token_get_info_pb2 as hapi_pb -@dataclass +@dataclass(frozen=True) class TokenInfo: """Data class for basic token details: ID, name, and symbol.""" token_id: Optional[TokenId] = None @@ -70,113 +70,6 @@ class TokenInfo: ) - # === setter methods === - def set_admin_key(self, admin_key: PublicKey) -> "TokenInfo": - """Set the admin key.""" - self.admin_key = admin_key - return self - - - def set_kyc_key(self, kyc_key: PublicKey) -> "TokenInfo": - """Set the KYC key.""" - self.kyc_key = kyc_key - return self - - - def set_freeze_key(self, freeze_key: PublicKey) -> "TokenInfo": - """Set the freeze key.""" - self.freeze_key = freeze_key - return self - - - def set_wipe_key(self, wipe_key: PublicKey) -> "TokenInfo": - """Set the wipe key.""" - self.wipe_key = wipe_key - return self - - - def set_supply_key(self, supply_key: PublicKey) -> "TokenInfo": - """Set the supply key.""" - self.supply_key = supply_key - return self - - - def set_metadata_key(self, metadata_key: PublicKey) -> "TokenInfo": - """Set the metadata key.""" - self.metadata_key = metadata_key - return self - - def set_fee_schedule_key(self, fee_schedule_key: PublicKey) -> "TokenInfo": - """Set the fee schedule key.""" - self.fee_schedule_key = fee_schedule_key - return self - - def set_default_freeze_status(self, freeze_status: TokenFreezeStatus) -> "TokenInfo": - """Set the default freeze status.""" - self.default_freeze_status = freeze_status - return self - - - def set_default_kyc_status(self, kyc_status: TokenKycStatus) -> "TokenInfo": - """Set the default KYC status.""" - self.default_kyc_status = kyc_status - return self - - - def set_auto_renew_account(self, account: AccountId) -> "TokenInfo": - """Set the auto-renew account.""" - self.auto_renew_account = account - return self - - - def set_auto_renew_period(self, period: Duration) -> "TokenInfo": - """Set the auto-renew period.""" - self.auto_renew_period = period - return self - - - def set_expiry(self, expiry: Timestamp) -> "TokenInfo": - """Set the token expiry.""" - self.expiry = expiry - return self - - def set_pause_key(self, pause_key: PublicKey) -> "TokenInfo": - """Set the pause key.""" - self.pause_key = pause_key - return self - - def set_pause_status(self, pause_status: TokenPauseStatus) -> "TokenInfo": - """Set the pause status.""" - self.pause_status = pause_status - return self - - - def set_supply_type(self, supply_type: SupplyType | int) -> "TokenInfo": - """Set the supply type.""" - self.supply_type = ( - supply_type - if isinstance(supply_type, SupplyType) - else SupplyType(supply_type) - ) - return self - - - def set_metadata(self, metadata: bytes) -> "TokenInfo": - """Set the token metadata.""" - self.metadata = metadata - return self - - def set_custom_fees(self, custom_fees: List[Any]) -> "TokenInfo": - """Set the custom fees.""" - self.custom_fees = custom_fees - return self - - - # === helpers === - - - - @staticmethod def _get(proto_obj, *names): """Get the first present attribute from a list of possible names (camelCase/snake_case).""" @@ -185,6 +78,15 @@ def _get(proto_obj, *names): return getattr(proto_obj, n) return None + @staticmethod + def _public_key_from_oneof(key_msg) -> Optional[PublicKey]: + """ + Extract a PublicKey from a key oneof, or None if not present. + """ + if key_msg is not None and hasattr(key_msg, "WhichOneof") and key_msg.WhichOneof("key"): + return PublicKey._from_proto(key_msg) + return None + # === conversions === @classmethod def _from_proto(cls, proto_obj: hapi_pb.TokenInfo) -> "TokenInfo": @@ -193,60 +95,56 @@ def _from_proto(cls, proto_obj: hapi_pb.TokenInfo) -> "TokenInfo": :param proto_obj: The token_get_info_pb2.TokenInfo object. :return: An instance of TokenInfo. """ - tokenInfoObject = TokenInfo( - token_id=TokenId._from_proto(proto_obj.tokenId), - name=proto_obj.name, - symbol=proto_obj.symbol, - decimals=proto_obj.decimals, - total_supply=proto_obj.totalSupply, - treasury=AccountId._from_proto(proto_obj.treasury), - is_deleted=proto_obj.deleted, - memo=proto_obj.memo, - token_type=TokenType(proto_obj.tokenType), - max_supply=proto_obj.maxSupply, - ledger_id=proto_obj.ledger_id, - metadata=proto_obj.metadata, - ) - - tokenInfoObject.set_custom_fees(cls._parse_custom_fees(proto_obj)) + kwargs: Dict[str, Any] = { + "token_id": TokenId._from_proto(proto_obj.tokenId), + "name": proto_obj.name, + "symbol": proto_obj.symbol, + "decimals": proto_obj.decimals, + "total_supply": proto_obj.totalSupply, + "treasury": AccountId._from_proto(proto_obj.treasury), + "is_deleted": proto_obj.deleted, + "memo": proto_obj.memo, + "token_type": TokenType(proto_obj.tokenType), + "max_supply": proto_obj.maxSupply, + "ledger_id": proto_obj.ledger_id, + "metadata": proto_obj.metadata, + "custom_fees": cls._parse_custom_fees(proto_obj), + } key_sources = [ - (("adminKey",), "set_admin_key"), - (("kycKey",), "set_kyc_key"), - (("freezeKey",), "set_freeze_key"), - (("wipeKey",), "set_wipe_key"), - (("supplyKey",), "set_supply_key"), - (("metadataKey", "metadata_key"), "set_metadata_key"), - (("feeScheduleKey", "fee_schedule_key"),"set_fee_schedule_key"), - (("pauseKey", "pause_key"), "set_pause_key"), + (("adminKey",), "admin_key"), + (("kycKey",), "kyc_key"), + (("freezeKey",), "freeze_key"), + (("wipeKey",), "wipe_key"), + (("supplyKey",), "supply_key"), + (("metadataKey", "metadata_key"), "metadata_key"), + (("feeScheduleKey", "fee_schedule_key"),"fee_schedule_key"), + (("pauseKey", "pause_key"), "pause_key"), ] - for names, setter in key_sources: + for names, attr_name in key_sources: key_msg = cls._get(proto_obj, *names) - cls._copy_key_if_present(tokenInfoObject, setter, key_msg) + public_key = cls._public_key_from_oneof(key_msg) + if public_key is not None: + kwargs[attr_name] = public_key conv_map = [ - (("defaultFreezeStatus",), tokenInfoObject.set_default_freeze_status, TokenFreezeStatus._from_proto), - (("defaultKycStatus",), tokenInfoObject.set_default_kyc_status, TokenKycStatus._from_proto), - (("autoRenewAccount",), tokenInfoObject.set_auto_renew_account, AccountId._from_proto), - (("autoRenewPeriod",), tokenInfoObject.set_auto_renew_period, Duration._from_proto), - (("expiry",), tokenInfoObject.set_expiry, Timestamp._from_protobuf), - (("pauseStatus", "pause_status"), tokenInfoObject.set_pause_status, TokenPauseStatus._from_proto), - (("supplyType",), tokenInfoObject.set_supply_type, SupplyType), + (("defaultFreezeStatus",), "default_freeze_status", TokenFreezeStatus._from_proto), + (("defaultKycStatus",), "default_kyc_status", TokenKycStatus._from_proto), + (("autoRenewAccount",), "auto_renew_account", AccountId._from_proto), + (("autoRenewPeriod",), "auto_renew_period", Duration._from_proto), + (("expiry",), "expiry", Timestamp._from_protobuf), + (("pauseStatus", "pause_status"), "pause_status", TokenPauseStatus._from_proto), + (("supplyType",), "supply_type", SupplyType), ] - for names, setter, conv in conv_map: + + for names, attr_name, conv in conv_map: val = cls._get(proto_obj, *names) if val is not None: - setter(conv(val)) + kwargs[attr_name] = conv(val) - return tokenInfoObject + return cls(**kwargs) # === helpers === - @staticmethod - def _copy_key_if_present(dst: "TokenInfo", setter: str, key_msg) -> None: - # In proto3, keys are a oneof; check presence via WhichOneof - if key_msg is not None and hasattr(key_msg, "WhichOneof") and key_msg.WhichOneof("key"): - getattr(dst, setter)(PublicKey._from_proto(key_msg)) - @staticmethod def _parse_custom_fees(proto_obj) -> List[CustomFee]: out: List[CustomFee] = [] diff --git a/tests/unit/test_token_info.py b/tests/unit/test_token_info.py index 2c9cebbc8..9ab142999 100644 --- a/tests/unit/test_token_info.py +++ b/tests/unit/test_token_info.py @@ -1,5 +1,7 @@ import pytest +from dataclasses import FrozenInstanceError, replace + import hiero_sdk_python.hapi.services.basic_types_pb2 from hiero_sdk_python.tokens.token_info import TokenInfo, TokenId, AccountId, Timestamp from hiero_sdk_python.crypto.private_key import PrivateKey @@ -77,50 +79,10 @@ def test_token_info_initialization(token_info): assert token_info.expiry is None assert token_info.pause_key is None -def test_setters(token_info): - public_key = PrivateKey.generate_ed25519().public_key() - token_info.set_admin_key(public_key) - assert token_info.admin_key == public_key - - token_info.set_kyc_key(public_key) - assert token_info.kyc_key == public_key - - token_info.set_freeze_key(public_key) - assert token_info.freeze_key == public_key - - token_info.set_wipe_key(public_key) - assert token_info.wipe_key == public_key - - token_info.set_supply_key(public_key) - assert token_info.supply_key == public_key - - token_info.set_fee_schedule_key(public_key) - assert token_info.fee_schedule_key == public_key - - token_info.set_default_freeze_status(TokenFreezeStatus.FROZEN) - assert token_info.default_freeze_status == TokenFreezeStatus.FROZEN - - token_info.set_default_kyc_status(TokenKycStatus.GRANTED) - assert token_info.default_kyc_status == TokenKycStatus.GRANTED - - token_info.set_auto_renew_account(AccountId(0, 0, 300)) - assert token_info.auto_renew_account == AccountId(0, 0, 300) - - token_info.set_auto_renew_period(Duration(3600)) - assert token_info.auto_renew_period == Duration(3600) - - expiry = Timestamp(1625097600, 0) - token_info.set_expiry(expiry) - assert token_info.expiry == expiry - - token_info.set_pause_key(public_key) - assert token_info.pause_key == public_key - - token_info.set_pause_status(TokenPauseStatus.PAUSED) - assert token_info.pause_status == TokenPauseStatus.PAUSED - - token_info.set_supply_type(SupplyType.INFINITE) - assert token_info.supply_type == SupplyType.INFINITE +def test_token_info_is_immutable(token_info): + """TokenInfo deve essere immutabile (dataclass frozen).""" + with pytest.raises(FrozenInstanceError): + token_info.name = "Changed" def test_from_proto(proto_token_info): public_key = PrivateKey.generate_ed25519().public_key() @@ -168,24 +130,45 @@ def test_from_proto(proto_token_info): assert token_info.pause_status == TokenPauseStatus.PAUSED.value assert token_info.supply_type.value == SupplyType.INFINITE.value + assert token_info.pause_status == TokenPauseStatus.PAUSED + assert token_info.supply_type == SupplyType.INFINITE + def test_to_proto(token_info): public_key = PrivateKey.generate_ed25519().public_key() - token_info.set_admin_key(public_key) - token_info.set_kyc_key(public_key) - token_info.set_freeze_key(public_key) - token_info.set_wipe_key(public_key) - token_info.set_supply_key(public_key) - token_info.set_fee_schedule_key(public_key) - token_info.set_pause_key(public_key) - token_info.set_default_freeze_status(TokenFreezeStatus.FROZEN) - token_info.set_default_kyc_status(TokenKycStatus.GRANTED) - token_info.set_auto_renew_account(AccountId(0, 0, 300)) - token_info.set_auto_renew_period(Duration(3600)) - token_info.set_expiry(Timestamp(1625097600, 0)) - token_info.set_pause_status(TokenPauseStatus.PAUSED) - token_info.set_supply_type(SupplyType.INFINITE) - - proto = token_info._to_proto() + + full_token_info = replace( + token_info, + admin_key=public_key, + kyc_key=public_key, + freeze_key=public_key, + wipe_key=public_key, + supply_key=public_key, + fee_schedule_key=public_key, + pause_key=public_key, + default_freeze_status=TokenFreezeStatus.FROZEN, + default_kyc_status=TokenKycStatus.GRANTED, + auto_renew_account=AccountId(0, 0, 300), + auto_renew_period=Duration(3600), + expiry=Timestamp(1625097600, 0), + pause_status=TokenPauseStatus.PAUSED, + supply_type=SupplyType.INFINITE, + ) + # token_info.set_admin_key(public_key) + # token_info.set_kyc_key(public_key) + # token_info.set_freeze_key(public_key) + # token_info.set_wipe_key(public_key) + # token_info.set_supply_key(public_key) + # token_info.set_fee_schedule_key(public_key) + # token_info.set_pause_key(public_key) + # token_info.set_default_freeze_status(TokenFreezeStatus.FROZEN) + # token_info.set_default_kyc_status(TokenKycStatus.GRANTED) + # token_info.set_auto_renew_account(AccountId(0, 0, 300)) + # token_info.set_auto_renew_period(Duration(3600)) + # token_info.set_expiry(Timestamp(1625097600, 0)) + # token_info.set_pause_status(TokenPauseStatus.PAUSED) + # token_info.set_supply_type(SupplyType.INFINITE) + + proto = full_token_info._to_proto() assert proto.tokenId == TokenId(0, 0, 100)._to_proto() assert proto.name == "TestToken" @@ -223,4 +206,4 @@ def test_str_representation(token_info): f"token_type={token_info.token_type}, max_supply={token_info.max_supply}, " f"ledger_id={token_info.ledger_id!r}, metadata={token_info.metadata!r})" ) - assert str(token_info) == expected \ No newline at end of file + assert str(token_info) == expected