Skip to content
Merged
Show file tree
Hide file tree
Changes from all 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
41 changes: 26 additions & 15 deletions hathor/nanocontracts/runner/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -816,17 +816,27 @@ def _get_balance(
before_current_call: bool,
) -> Balance:
"""Internal implementation of get_balance."""
if contract_id is None:
contract_id = self.get_current_contract_id()
if token_uid is None:
token_uid = TokenUid(HATHOR_TOKEN_UID)

storage: NCContractStorage
if self._call_info is None:
storage = self.get_storage(contract_id)
else:
changes_tracker = self.get_current_changes_tracker(contract_id)
if contract_id is None:
# In this case we're getting the balance of the currently executing contract,
# so it's guaranteed that a changes tracker exists.
# Depending on `before_current_call`, we get the current changes tracker or its storage.

changes_tracker = self.get_current_changes_tracker()
storage = changes_tracker.storage if before_current_call else changes_tracker
else:
# In this case we're getting the balance of another contract,
# so a changes tracker may not yet exist if the contract is not in the call chain.
# We cannot retrieve the balance before the current call because we don't know whether the latest
# changes tracker was created during the current call or not.

if before_current_call:
raise NCFail('getting the balance of another contract before the current call is not supported.')

storage = self.get_current_changes_tracker_or_storage(contract_id)

return storage.get_balance(bytes(token_uid))

Expand All @@ -840,9 +850,10 @@ def get_current_contract_id(self) -> ContractId:
call_record = self.get_current_call_record()
return call_record.contract_id

def get_current_changes_tracker(self, contract_id: ContractId) -> NCChangesTracker:
def get_current_changes_tracker(self) -> NCChangesTracker:
"""Return the NCChangesTracker for the current method being executed."""
assert self._call_info is not None
contract_id = self.get_current_contract_id()
change_trackers = self._call_info.change_trackers[contract_id]
assert len(change_trackers) > 0
return change_trackers[-1]
Expand Down Expand Up @@ -950,7 +961,7 @@ def syscall_revoke_authorities(self, token_uid: TokenUid, *, revoke_mint: bool,
if token_uid == HATHOR_TOKEN_UID:
raise NCInvalidSyscall(f'contract {contract_id.hex()} cannot revoke authorities from HTR token')

changes_tracker = self.get_current_changes_tracker(contract_id)
changes_tracker = self.get_current_changes_tracker()
assert changes_tracker.nc_id == call_record.contract_id
balance = changes_tracker.get_balance(token_uid)

Expand Down Expand Up @@ -993,7 +1004,7 @@ def syscall_mint_tokens(
if token_uid == HATHOR_TOKEN_UID:
raise NCInvalidSyscall(f'contract {call_record.contract_id.hex()} cannot mint HTR tokens')

changes_tracker = self.get_current_changes_tracker(call_record.contract_id)
changes_tracker = self.get_current_changes_tracker()
assert changes_tracker.nc_id == call_record.contract_id

balance = changes_tracker.get_balance(token_uid)
Expand Down Expand Up @@ -1027,7 +1038,7 @@ def syscall_melt_tokens(
if token_uid == HATHOR_TOKEN_UID:
raise NCInvalidSyscall(f'contract {call_record.contract_id.hex()} cannot melt HTR tokens')

changes_tracker = self.get_current_changes_tracker(call_record.contract_id)
changes_tracker = self.get_current_changes_tracker()
assert changes_tracker.nc_id == call_record.contract_id

balance = changes_tracker.get_balance(token_uid)
Expand Down Expand Up @@ -1108,7 +1119,7 @@ def syscall_create_child_deposit_token(
token_id = derive_child_token_id(parent_id, cleaned_token_symbol, salt=salt)
token_version = TokenVersion.DEPOSIT

changes_tracker = self.get_current_changes_tracker(parent_id)
changes_tracker = self.get_current_changes_tracker()
changes_tracker.create_token(
token_id=token_id,
token_name=token_name,
Expand Down Expand Up @@ -1165,7 +1176,7 @@ def syscall_create_child_fee_token(
token_id = derive_child_token_id(parent_id, cleaned_token_symbol, salt=salt)
token_version = TokenVersion.FEE

changes_tracker = self.get_current_changes_tracker(parent_id)
changes_tracker = self.get_current_changes_tracker()
changes_tracker.create_token(
token_id=token_id,
token_name=token_name,
Expand Down Expand Up @@ -1209,7 +1220,7 @@ def syscall_change_blueprint(self, blueprint_id: BlueprintId) -> None:
# exception.
self.tx_storage.get_blueprint_class(blueprint_id)

nc_storage = self.get_current_changes_tracker(last_call_record.contract_id)
nc_storage = self.get_current_changes_tracker()
nc_storage.set_blueprint_id(blueprint_id)

def _get_token(self, token_uid: TokenUid) -> TokenDescription:
Expand All @@ -1220,7 +1231,7 @@ def _get_token(self, token_uid: TokenUid) -> TokenDescription:
NCInvalidSyscall when the token isn't found.
"""
call_record = self.get_current_call_record()
changes_tracker = self.get_current_changes_tracker(call_record.contract_id)
changes_tracker = self.get_current_changes_tracker()
assert call_record.contract_id == changes_tracker.nc_id

if changes_tracker.has_token(token_uid):
Expand Down Expand Up @@ -1275,7 +1286,7 @@ def _update_tokens_amount(
AssertionError: If call_record.index_updates is None
"""
call_record = self.get_current_call_record()
changes_tracker = self.get_current_changes_tracker(call_record.contract_id)
changes_tracker = self.get_current_changes_tracker()

assert changes_tracker.nc_id == call_record.contract_id
assert call_record.index_updates is not None
Expand Down
109 changes: 109 additions & 0 deletions tests/nanocontracts/test_balance_syscalls.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
# Copyright 2025 Hathor Labs
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import pytest

from hathor import Blueprint, Context, ContractId, NCFail, TokenUid, public
from tests.nanocontracts.blueprints.unittest import BlueprintTestCase


class MyBlueprint(Blueprint):
token_uid: TokenUid

@public
def initialize(self, ctx: Context, token_uid: TokenUid) -> None:
self.token_uid = token_uid

@public
def test_balance_syscalls_on_self(self, ctx: Context) -> None:
self.syscall.get_current_balance()
self.syscall.get_balance_before_current_call()

self.syscall.can_mint(self.token_uid)
self.syscall.can_mint_before_current_call(self.token_uid)

self.syscall.can_melt(self.token_uid)
self.syscall.can_melt_before_current_call(self.token_uid)

@public
def test_current_balance_syscalls_on_another(self, ctx: Context, other_id: ContractId) -> None:
self.syscall.get_current_balance(contract_id=other_id)
self.syscall.can_mint(self.token_uid, contract_id=other_id)
self.syscall.can_melt(self.token_uid, contract_id=other_id)

@public
def test_get_balance_before_current_call_on_another(self, ctx: Context, other_id: ContractId) -> None:
self.syscall.get_balance_before_current_call(contract_id=other_id)

@public
def test_can_mint_before_current_call_on_another(self, ctx: Context, other_id: ContractId) -> None:
self.syscall.can_mint_before_current_call(self.token_uid, contract_id=other_id)

@public
def test_can_melt_before_current_call_on_another(self, ctx: Context, other_id: ContractId) -> None:
self.syscall.can_melt_before_current_call(self.token_uid, contract_id=other_id)


class BalanceSyscallsTestCase(BlueprintTestCase):
def setUp(self) -> None:
super().setUp()

self.blueprint_id = self._register_blueprint_class(MyBlueprint)
self.contract_id1 = self.gen_random_contract_id()
self.contract_id2 = self.gen_random_contract_id()

token_uid = self.gen_random_token_uid()
self.runner.create_contract(self.contract_id1, self.blueprint_id, self.create_context(), token_uid)
self.runner.create_contract(self.contract_id2, self.blueprint_id, self.create_context(), token_uid)

def test_balance_syscalls_on_self(self) -> None:
self.runner.call_public_method(self.contract_id1, 'test_balance_syscalls_on_self', self.create_context())

def test_current_balance_syscalls_on_another(self) -> None:
self.runner.call_public_method(
self.contract_id1,
'test_current_balance_syscalls_on_another',
self.create_context(),
self.contract_id2,
)

def test_get_balance_before_current_call_on_another(self) -> None:
msg = 'getting the balance of another contract before the current call is not supported.'
with pytest.raises(NCFail, match=msg):
self.runner.call_public_method(
self.contract_id1,
'test_get_balance_before_current_call_on_another',
self.create_context(),
self.contract_id2,
)

def test_can_mint_before_current_call_on_another(self) -> None:
msg = 'getting the balance of another contract before the current call is not supported.'
with pytest.raises(NCFail, match=msg):
self.runner.call_public_method(
self.contract_id1,
'test_can_mint_before_current_call_on_another',
self.create_context(),
self.contract_id2,
)

def test_can_melt_before_current_call_on_another(self) -> None:
msg = 'getting the balance of another contract before the current call is not supported.'
with pytest.raises(NCFail, match=msg):
self.runner.call_public_method(
self.contract_id1,
'test_can_melt_before_current_call_on_another',
self.create_context(),
self.contract_id2,
)
Loading