feat: Add AssessedCustomFee model with protobuf support and tests (hiero-ledger#1660)

Signed-off-by: Siddhartha Ganguly <gangulysiddhartha22@gmail.com>
Signed-off-by: gangulysiddhartha22-cmyk <gangulysiddhartha22@gmail.com>

Author: gangulysiddhartha22-cmyk

CHANGELOG.md

@@ -31,6 +31,7 @@ This changelog is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.
 - Format `tests/unit/custom_fee_test.py` with black for code style consistency. (#1525)
 
 ### Added
+- Added `AssessedCustomFee` domain model to represent assessed custom fees. (`#1637`)
 - Add __repr__ method for ContractId class to improve debugging (#1714)
 - Added Protobuf Training guide to enhance developer understanding of proto serialization
   and deserialization (#1645)

src/hiero_sdk_python/__init__.py

@@ -51,6 +51,7 @@
 from .tokens.token_unpause_transaction import TokenUnpauseTransaction
 from .tokens.token_pause_transaction import TokenPauseTransaction
 from .tokens.token_airdrop_claim import TokenClaimAirdropTransaction
+from .tokens.assessed_custom_fee import AssessedCustomFee
 
 # Transaction
 from .transaction.transaction import Transaction
@@ -202,6 +203,7 @@
     "HbarTransfer",
     "TokenPauseTransaction",
     "TokenUnpauseTransaction",
+    "AssessedCustomFee",
 
     # Transaction
     "Transaction",

src/hiero_sdk_python/tokens/assessed_custom_fee.py

@@ -0,0 +1,98 @@
+from dataclasses import dataclass, field
+from typing import Optional
+
+from hiero_sdk_python.account.account_id import AccountId
+from hiero_sdk_python.hapi.services.custom_fees_pb2 import (
+    AssessedCustomFee as AssessedCustomFeeProto,
+)
+from hiero_sdk_python.tokens.token_id import TokenId
+
+
+@dataclass
+class AssessedCustomFee:
+    """Assessed custom fee information included in transaction records.
+
+    This class represents fees assessed due to custom fee schedules on tokens or
+    topics. It appears in `TransactionRecord.assessed_custom_fees` (repeated field).
+
+    Example:
+        Suppose you have a TransactionRecord from getTransactionRecord():
+
+        record = client.get_transaction_record(tx_id)
+
+        for fee in record.assessed_custom_fees:
+            if fee.token_id is None:
+                print(f"HBAR fee of {fee.amount} tinybars collected by {fee.fee_collector_account_id}")
+            else:
+                print(f"Token fee of {fee.amount} units of {fee.token_id} "
+                    f"collected by {fee.fee_collector_account_id}, "
+                    f"paid by {', '.join(str(p) for p in fee.effective_payer_account_ids)}")
+    """
+
+    amount: int
+    """The amount of the fee assessed, in the smallest units of the token (or tinybars for HBAR)."""
+
+    token_id: Optional[TokenId] = None
+    """The ID of the token used to pay the fee; None if paid in HBAR."""
+
+    fee_collector_account_id: Optional[AccountId] = None
+    """The account ID that collects/receives this assessed custom fee (required field)."""
+
+    effective_payer_account_ids: list[AccountId] = field(default_factory=list)
+    """The list of accounts that effectively paid this assessed fee (repeated field)."""
+
+    def __post_init__(self) -> None:
+        if self.fee_collector_account_id is None:
+            raise ValueError(
+                "fee_collector_account_id is required for AssessedCustomFee"
+            )
+
+    @classmethod
+    def _from_proto(cls, proto: AssessedCustomFeeProto) -> "AssessedCustomFee":
+        """Create an AssessedCustomFee instance from the protobuf message."""
+        token_id = (
+            TokenId._from_proto(proto.token_id) if proto.HasField("token_id") else None
+        )
+
+        if not proto.HasField("fee_collector_account_id"):
+            raise ValueError(
+                "fee_collector_account_id is required in AssessedCustomFee proto"
+            )
+
+        return cls(
+            amount=proto.amount,
+            token_id=token_id,
+            fee_collector_account_id=AccountId._from_proto(
+                proto.fee_collector_account_id
+            ),
+            effective_payer_account_ids=[
+                AccountId._from_proto(payer_proto)
+                for payer_proto in proto.effective_payer_account_id
+            ],
+        )
+
+    def _to_proto(self) -> AssessedCustomFeeProto:
+        """Convert this AssessedCustomFee instance back to a protobuf message."""
+        proto = AssessedCustomFeeProto(
+            amount=self.amount,
+            fee_collector_account_id=self.fee_collector_account_id._to_proto(),
+        )
+
+        if self.token_id is not None:
+            proto.token_id.CopyFrom(self.token_id._to_proto())
+
+        for payer in self.effective_payer_account_ids:
+            proto.effective_payer_account_id.append(payer._to_proto())
+
+        return proto
+
+    def __str__(self) -> str:
+        """Returns a human-readable string representation."""
+        return (
+            f"AssessedCustomFee("
+            f"amount={self.amount}, "
+            f"token_id={self.token_id}, "
+            f"fee_collector_account_id={self.fee_collector_account_id}, "
+            f"effective_payer_account_ids={self.effective_payer_account_ids}"
+            f")"
+        )

tests/integration/account_id_population_e2e_test.py

@@ -49,7 +49,7 @@ def test_populate_account_id_num(env, evm_address):
     ), "Expected child transaction for auto-account creation"
 
     created_account_id = transfer_receipt.children[0].account_id
-    assert created_account_id is not None
+    assert created_account_id is not None, f"AccountId not found in child transaction: {transfer_receipt.children[0]}"
 
     mirror_account_id = AccountId.from_evm_address(evm_address, 0, 0)
     assert mirror_account_id.num == 0
@@ -92,7 +92,8 @@ def test_populate_account_id_evm_address(env, evm_address):
     ), "Expected child transaction for auto-account creation"
 
     created_account_id = transfer_receipt.children[0].account_id
-    assert created_account_id is not None
+    assert created_account_id is not None, f"AccountId not found in child transaction: {transfer_receipt.children[0]}"
+
     assert created_account_id.evm_address is None
 
     # Wait for mirror_node to update

tests/unit/assessed_custom_fee_test.py

@@ -0,0 +1,199 @@
+import pytest
+
+from hiero_sdk_python.account.account_id import AccountId
+from hiero_sdk_python.hapi.services.custom_fees_pb2 import AssessedCustomFee as AssessedCustomFeeProto
+from hiero_sdk_python.tokens.assessed_custom_fee import AssessedCustomFee
+from hiero_sdk_python.tokens.token_id import TokenId
+
+
+pytestmark = pytest.mark.unit
+
+
+# If conftest.py has fixtures like sample_account_id or sample_token_id, use them.
+# Otherwise, define simple ones here (adjust shard/realm/num as needed for realism).
+@pytest.fixture
+def sample_account_id() -> AccountId:
+    return AccountId(shard=0, realm=0, num=123456)
+
+
+@pytest.fixture
+def sample_token_id() -> TokenId:
+    return TokenId(shard=0, realm=0, num=789012)
+
+
+@pytest.fixture
+def another_account_id() -> AccountId:
+    return AccountId(shard=0, realm=0, num=999999)
+
+
+def test_constructor_all_fields(
+    sample_account_id: AccountId,
+    sample_token_id: TokenId,
+    another_account_id: AccountId,
+):
+    payers = [sample_account_id, another_account_id]
+    fee = AssessedCustomFee(
+        amount=1_500_000_000,
+        token_id=sample_token_id,
+        fee_collector_account_id=sample_account_id,
+        effective_payer_account_ids=payers,
+    )
+    assert fee.amount == 1_500_000_000
+    assert fee.token_id == sample_token_id
+    assert fee.fee_collector_account_id == sample_account_id
+    assert fee.effective_payer_account_ids == payers
+    # Protect against breaking changes
+    assert hasattr(fee, 'amount')
+    assert hasattr(fee, 'token_id')
+    assert hasattr(fee, 'fee_collector_account_id')
+    assert hasattr(fee, 'effective_payer_account_ids')
+
+
+def test_constructor_hbar_case(sample_account_id: AccountId):
+    fee = AssessedCustomFee(
+        amount=100_000_000,
+        token_id=None,
+        fee_collector_account_id=sample_account_id,
+    )
+    assert fee.amount == 100_000_000
+    assert fee.token_id is None
+    assert fee.fee_collector_account_id == sample_account_id
+    assert fee.effective_payer_account_ids == []
+
+
+def test_constructor_empty_payers(sample_account_id: AccountId, sample_token_id: TokenId):
+    fee = AssessedCustomFee(
+        amount=420,
+        token_id=sample_token_id,
+        fee_collector_account_id=sample_account_id,
+        effective_payer_account_ids=[],
+    )
+    assert fee.effective_payer_account_ids == []
+    assert fee.token_id == sample_token_id
+
+def test_constructor_missing_fee_collector_raises():
+    """Verify that omitting fee_collector_account_id raises ValueError."""
+    with pytest.raises(ValueError, match="fee_collector_account_id is required"):
+        AssessedCustomFee(
+            amount=100,
+            token_id=None,
+            fee_collector_account_id=None,
+        )
+
+def test_from_proto_missing_token_id(sample_account_id: AccountId):
+    """Verify that absence of token_id in protobuf correctly maps to None."""
+    proto = AssessedCustomFeeProto(
+        amount=750_000,
+        fee_collector_account_id=sample_account_id._to_proto(),
+        # intentionally no token_id → proto.HasField("token_id") should be False
+    )
+
+    fee = AssessedCustomFee._from_proto(proto)
+
+    assert fee.amount == 750_000
+    assert fee.token_id is None, "token_id should be None when not present in proto"
+    assert fee.fee_collector_account_id == sample_account_id
+    assert fee.effective_payer_account_ids == [], "effective payers should default to empty list"
+
+def test_from_proto_with_token_id(sample_account_id: AccountId, sample_token_id: TokenId):
+    """Verify that token_id is correctly deserialized when present in proto."""
+    proto = AssessedCustomFeeProto(
+        amount=500_000,
+        token_id=sample_token_id._to_proto(),
+        fee_collector_account_id=sample_account_id._to_proto(),
+    )
+    proto.effective_payer_account_id.append(sample_account_id._to_proto())
+
+    fee = AssessedCustomFee._from_proto(proto)
+
+    assert fee.amount == 500_000
+    assert fee.token_id is not None
+    assert fee.token_id == sample_token_id
+    assert fee.fee_collector_account_id == sample_account_id
+    assert len(fee.effective_payer_account_ids) == 1
+
+def test_from_proto_missing_fee_collector_raises():
+    """Verify that missing fee_collector_account_id in proto raises ValueError."""
+    proto = AssessedCustomFeeProto(amount=750_000)
+    with pytest.raises(ValueError, match="fee_collector_account_id is required"):
+        AssessedCustomFee._from_proto(proto)
+
+def test_to_proto_basic_fields(
+    sample_account_id: AccountId,
+    sample_token_id: TokenId,
+    another_account_id: AccountId,
+):
+    """Verify that all basic fields are correctly serialized to protobuf."""
+    payers = [sample_account_id, another_account_id]
+
+    fee = AssessedCustomFee(
+        amount=2_000_000,
+        token_id=sample_token_id,
+        fee_collector_account_id=sample_account_id,
+        effective_payer_account_ids=payers,
+    )
+
+    proto = fee._to_proto()
+
+    # Core presence and value checks
+    assert proto.amount == 2_000_000
+    assert proto.HasField("token_id"), "token_id should be set when present"
+    assert proto.HasField("fee_collector_account_id")
+    assert len(proto.effective_payer_account_id) == 2, "should serialize both effective payers"
+
+    # Deeper structural checks (helps catch broken _to_proto implementations)
+    assert proto.token_id.shardNum == sample_token_id.shard
+    assert proto.token_id.realmNum == sample_token_id.realm
+    assert proto.token_id.tokenNum == sample_token_id.num
+
+    # Optional: verify collector (often useful when debugging)
+    assert proto.fee_collector_account_id.shardNum == sample_account_id.shard
+    assert proto.fee_collector_account_id.realmNum == sample_account_id.realm
+    assert proto.fee_collector_account_id.accountNum == sample_account_id.num
+
+
+def test_round_trip_conversion(
+    sample_account_id: AccountId,
+    sample_token_id: TokenId,
+):
+    original = AssessedCustomFee(
+        amount=987_654_321,
+        token_id=sample_token_id,
+        fee_collector_account_id=sample_account_id,
+        effective_payer_account_ids=[sample_account_id],
+    )
+
+    proto = original._to_proto()
+    reconstructed = AssessedCustomFee._from_proto(proto)
+
+    assert reconstructed.amount == original.amount
+    assert reconstructed.token_id == original.token_id
+    assert reconstructed.fee_collector_account_id == original.fee_collector_account_id
+    assert reconstructed.effective_payer_account_ids == original.effective_payer_account_ids
+
+
+def test_str_contains_expected_fields(
+    sample_account_id: AccountId,
+    sample_token_id: TokenId,
+):
+    fee = AssessedCustomFee(
+        amount=5_000_000,
+        token_id=sample_token_id,
+        fee_collector_account_id=sample_account_id,
+        effective_payer_account_ids=[sample_account_id],
+    )
+
+    s = str(fee)
+    assert "AssessedCustomFee" in s
+    assert "amount=5000000" in s
+    assert str(sample_token_id) in s
+    assert str(sample_account_id) in s
+    assert "effective_payer_account_ids" in s
+
+    # HBAR case
+    hbar_fee = AssessedCustomFee(
+        amount=123_456,
+        fee_collector_account_id=sample_account_id,
+    )
+    hbar_str = str(hbar_fee)
+    assert "token_id=None" in hbar_str