Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,7 @@ This changelog is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.
- Renamed `pr-missing-linked-issue.yml` and `pr_missing_linked_issue.js` to `bot-pr-missing-linked-issue.yml` and `bot-pr-missing-linked-issue.js` respectively. Enhanced LinkBot PR comment with clickable hyperlinks to documentation for linking issues and creating issues. (#1264)
- Enhance assignment bot to guard against users with spam PRs `.github/scripts/bot-assignment-check.sh`
- Add CodeRabbit documentation review prompts for docs, sdk_users, and sdk_developers with priorities, philosophy, and edge case checks. ([#1236](https://github.com/hiero-ledger/hiero-sdk-python/issues/1236))
- Enhance NodeAddress tests with additional coverage for proto conversion `tests/unit/node_address_test.py`

### Fixed
- GFI workflow casing
Expand Down
5 changes: 1 addition & 4 deletions src/hiero_sdk_python/address_book/node_address.py
Original file line number Diff line number Diff line change
Expand Up @@ -101,11 +101,8 @@ def _to_proto(self):
if self._account_id:
node_address_proto.nodeAccountId.CopyFrom(self._account_id._to_proto())

service_endpoints: List[Endpoint] = []
for endpoint in self._addresses:
service_endpoints.append(endpoint._to_proto())

node_address_proto.serviceEndpoint = service_endpoints
node_address_proto.serviceEndpoint.append(endpoint._to_proto())

return node_address_proto

Expand Down
183 changes: 181 additions & 2 deletions tests/unit/node_address_test.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
import binascii
import pytest
from hiero_sdk_python.account.account_id import AccountId
from hiero_sdk_python.address_book.endpoint import Endpoint
from hiero_sdk_python.address_book.node_address import NodeAddress

from hiero_sdk_python.hapi.services.basic_types_pb2 import NodeAddress as NodeAddressProto
pytestmark = pytest.mark.unit

def test_init():
Expand Down Expand Up @@ -55,4 +56,182 @@ def test_string_representation():
assert "NodeAccountId: 0.0.123" in result
assert "CertHash: 73616d706c652d636572742d68617368" in result # hex representation of sample-cert-hash
assert "NodeId: 1234" in result
assert "PubKey: sample-public-key" in result
assert "PubKey: sample-public-key" in result


def test_to_proto():
"""Test conversion of NodeAddress to protobuf with endpoints."""
account_id = AccountId(0, 0, 123)
endpoints = [
Endpoint(address=bytes("192.168.1.1", "utf-8"), port=8080, domain_name="example1.com"),
Endpoint(address=bytes("192.168.1.2", "utf-8"), port=8081, domain_name="example2.com"),
Endpoint(address=bytes("192.168.1.3", "utf-8"), port=8082, domain_name="example3.com"),
]
node_address = NodeAddress(
public_key="sample-public-key",
account_id=account_id,
node_id=1234,
cert_hash=b"sample-cert-hash",
addresses=endpoints,
description="Sample Node"
)

node_address_proto = node_address._to_proto()
# Protect against breaking changes - verify return type
assert isinstance(node_address_proto, NodeAddressProto)
# Scalars
assert node_address_proto.RSA_PubKey == "sample-public-key"
assert node_address_proto.nodeId == 1234
assert node_address_proto.nodeCertHash == b"sample-cert-hash"
assert node_address_proto.description == "Sample Node"

# AccountId
assert node_address_proto.nodeAccountId.shardNum == 0
assert node_address_proto.nodeAccountId.realmNum == 0
assert node_address_proto.nodeAccountId.accountNum == 123

# ServiceEndpoint
# Verify all endpoints are serialized
assert len(node_address_proto.serviceEndpoint) == 3
for i, ep_proto in enumerate(node_address_proto.serviceEndpoint):
assert ep_proto.ipAddressV4 == endpoints[i]._address
assert ep_proto.port == endpoints[i]._port
assert ep_proto.domain_name == endpoints[i]._domain_name


def test_from_dict():
"""Test creation of NodeAddress from a dictionary with hex cert hash."""
node_dict = {
"public_key": "sample-public-key",
"node_account_id": "0.0.123",
"node_id": 1234,
"node_cert_hash": binascii.hexlify(b"sample-cert-hash").decode("utf-8"),
"description": "Sample Node",
"service_endpoints": [
{"ip_address_v4": "192.168.1.1", "port": 8080, "domain_name": "example.com"}
],
}

# Create NodeAddress from dict
node_address = NodeAddress._from_dict(node_dict)

assert node_address._public_key == "sample-public-key"
assert node_address._account_id == AccountId.from_string("0.0.123")
assert node_address._node_id == 1234
assert node_address._cert_hash == b"sample-cert-hash"
assert node_address._description == "Sample Node"
assert len(node_address._addresses) == 1


def test_from_dict_with_0x_prefix():
"""Test _from_dict handles cert hash with 0x prefix."""
node_dict = {
"public_key": "sample-public-key",
"node_account_id": "0.0.123",
"node_id": 1234,
"node_cert_hash": "0x" + binascii.hexlify(b"sample-cert-hash").decode("utf-8"),
"description": "Sample Node",
"service_endpoints": [],
}

node_address = NodeAddress._from_dict(node_dict)

assert node_address._cert_hash == b"sample-cert-hash"


def test_from_proto():
"""Test creation of NodeAddress from protobuf with endpoint."""
account_id_proto = AccountId(0, 0, 123)._to_proto()
endpoint_proto = Endpoint(
address=bytes("192.168.1.1", "utf-8"),
port=8080,
domain_name="example.com"
)._to_proto()

# Create NodeAddressProto
node_address_proto = NodeAddressProto(
RSA_PubKey="sample-public-key",
nodeAccountId=account_id_proto,
nodeId=1234,
nodeCertHash=b"sample-cert-hash",
description="Sample Node",
)
node_address_proto.serviceEndpoint.append(endpoint_proto)

node_address = NodeAddress._from_proto(node_address_proto)

assert node_address._public_key == "sample-public-key"
assert node_address._account_id == AccountId(0, 0, 123)
assert node_address._node_id == 1234
assert node_address._cert_hash == b"sample-cert-hash"
assert node_address._description == "Sample Node"
assert len(node_address._addresses) == 1


def test_round_trip():
"""Ensure NodeAddress → Proto → NodeAddress round trip works."""
account_id = AccountId(0, 0, 123)
endpoint = Endpoint(
address=bytes("192.168.1.1", "utf-8"),
port=8080,
domain_name="example.com"
)

# Create NodeAddress
node_address = NodeAddress(
public_key="sample-public-key",
account_id=account_id,
node_id=1234,
cert_hash=b"sample-cert-hash",
addresses=[endpoint],
description="Sample Node"
)

# Convert to proto
proto = node_address._to_proto()
# Convert back from proto
node_address2 = NodeAddress._from_proto(proto)

# Assert all fields are equal
assert node_address._public_key == node_address2._public_key
assert node_address._account_id == node_address2._account_id
assert node_address._node_id == node_address2._node_id
assert node_address._cert_hash == node_address2._cert_hash
assert node_address._description == node_address2._description
# Verify addresses match
assert len(node_address._addresses) == len(node_address2._addresses)
for i, endpoint in enumerate(node_address._addresses):
assert endpoint._address == node_address2._addresses[i]._address
assert endpoint._port == node_address2._addresses[i]._port
assert endpoint._domain_name == node_address2._addresses[i]._domain_name


def test_empty_addresses():
"""Test NodeAddress with no endpoints produces empty serviceEndpoint."""
node_address = NodeAddress(
public_key="sample-public-key",
account_id=AccountId(0, 0, 123),
node_id=1234,
cert_hash=b"sample-cert-hash",
addresses=[],
description="No endpoints"
)

proto = node_address._to_proto()
assert len(proto.serviceEndpoint) == 0

def test_to_proto_none_account_id():
"""Test _to_proto handles None account_id gracefully."""
node_address = NodeAddress(
public_key="sample-public-key",
account_id=None,
node_id=1234,
cert_hash=b"sample-cert-hash",
addresses=[],
description="No account"
)

proto = node_address._to_proto()

# Should not have nodeAccountId set
assert not proto.HasField('nodeAccountId')