Get started with blueprint SDK — part 2
Introduction
This article is part 2 of the 2-part tutorial to assist developers in getting started with the blueprint SDK. By the end of the tutorial, you will have developed your first blueprint using the SDK. In this part, you will implement (code) and test your first blueprint.
If you haven't completed part 1 of this tutorial yet, proceed to Get started with blueprint SDK — part 1.
Prerequisite
To execute this part of the tutorial, first install Hathor core from source code. See:
How to install Hathor core from source code
Overview of the task
In part 1 of this tutorial, we conceived and designed the blueprint. In this part, we implement (code) and test the designed blueprint. Let's continue using the simple swap contract as an example for our blueprint.
Using the blueprint SDK, each blueprint should be implemented as a single module. In addition, a second module containing the unit tests for the blueprint should be written.
Thus, during this tutorial, we will code the following two new Python modules in hathor-core
:
hathor-core/tests/nanocontracts/test_blueprints/swap_demo.py
hathor-core/tests/nanocontracts/blueprints/test_swap_demo.py
We suggest you expand the following pane and perform an inspectional reading of the entire blueprint's code that we will develop. This will give you the initial context needed to understand the step-by-step we will perform throughout the tutorial:
hathor-core/tests/nanocontracts/test_blueprints/swap_demo.py
from hathor.nanocontracts.blueprint import Blueprint
from hathor.nanocontracts.context import Context
from hathor.nanocontracts.exception import NCFail
from hathor.nanocontracts.types import NCDepositAction, NCWithdrawalAction, TokenUid, public, view
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
# TokenA identifier and quantity multiplier
token_a: TokenUid
multiplier_a: int
# TokenB identifier and quantity multiplier
token_b: TokenUid
multiplier_b: int
# Count number of swaps executed
swaps_counter: int
@public(allow_deposit=True)
def initialize(
self,
ctx: Context,
token_a: TokenUid,
token_b: TokenUid,
multiplier_a: int,
multiplier_b: int
) -> None:
"""Initialize the contract."""
if token_a == token_b:
raise NCFail
if set(ctx.actions.keys()) != {token_a, token_b}:
raise InvalidTokens
self.token_a = token_a
self.token_b = token_b
self.multiplier_a = multiplier_a
self.multiplier_b = multiplier_b
self.swaps_counter = 0
@public(allow_deposit=True, allow_withdrawal=True)
def swap(self, ctx: Context) -> None:
"""Execute a token swap."""
if set(ctx.actions.keys()) != {self.token_a, self.token_b}:
raise InvalidTokens
action_a = ctx.get_single_action(self.token_a)
action_b = ctx.get_single_action(self.token_b)
if not (
(isinstance(action_a, NCDepositAction) and isinstance(action_b, NCWithdrawalAction))
or (isinstance(action_a, NCWithdrawalAction) and isinstance(action_b, NCDepositAction))
):
raise InvalidActions
if not self.is_ratio_valid(action_a.amount, action_b.amount):
raise InvalidRatio
# All good! Let's accept the transaction.
self.swaps_counter += 1
@view
def is_ratio_valid(self, qty_a: int, qty_b: int) -> bool:
"""Check if the swap quantities are valid."""
return (self.multiplier_a * qty_a == self.multiplier_b * qty_b)
class InvalidTokens(NCFail):
pass
class InvalidActions(NCFail):
pass
class InvalidRatio(NCFail):
pass
Sequence of steps
We will represent the process of implementing and testing our blueprint's code through a hands-on walkthrough divided into this sequence of steps:
- Create the blueprint module.
- Write the
initialize
method. - Write the
swap
method. - Write the
is_ratio_valid
method. - Review the blueprint module.
- Create the unit test module.
- Write the
test_lifecycle
method. - Write the
_initialize
method. - Write the
_swap
method. - Review the blueprint unit test module.
Task execution
Now, it's time to get your hands dirty. In this section, we will describe the steps in detail.
To begin, let's suppose you have already opened Hathor core source code in your code editor.
Step 1: create the blueprint module
In the project directory tree, navigate to /hathor/nanocontracts/blueprints/
. There, you will find the source code of all built-in blueprints. Each blueprint is implemented as a single module.
- Let's create the
swap.py
file for our blueprint, and add the standard imports that are necessary for (if not all) almost all blueprints:
from hathor.nanocontracts.blueprint import Blueprint
from hathor.nanocontracts.context import Context
from hathor.nanocontracts.exception import NCFail
from hathor.nanocontracts.types import NCDepositAction, NCWithdrawalAction, TokenUid, public, view
Let's understand the utility of each of these imports:
Blueprint
is the base class for defining blueprints.Context
is the class that models the context in which the call is taking place.NCFail
is the base class that models failures in contract execution as exceptions.NCDepositAction
is the class that models deposits.NCWithdrawalAction
is the class that models withdrawals.TokenUid
is the type used for token UUIDs.public
is a decorator used to mark a blueprint method as public.view
is a decorator used to mark a blueprint methods as view.
Note that only names strictly allowed by the blueprint SDK can be imported. For which are these names, see Imports at Blueprint SDK — guidelines.
- We define our
Swap
class that inherits fromBlueprint
, and that will model swap contracts:
from hathor.nanocontracts.blueprint import Blueprint
from hathor.nanocontracts.context import Context
from hathor.nanocontracts.exception import NCFail
from hathor.nanocontracts.types import NCDepositAction, NCWithdrawalAction, TokenUid, public, view
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
- Our class will then have five instance attributes:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
# TokenA identifier and quantity multiplier.
token_a: TokenUid
multiplier_a: int
# TokenB identifier and quantity multiplier.
token_b: TokenUid
multiplier_b: int
# Count number of swaps executed.
swaps_counter: int
Note that the Blueprint SDK mandates specific rules for using these type annotations. To know what these rules are, see Type annotations at Blueprint development guidelines. Furthermore, only value types strictly allowed can be used for attributes. To know which are these types, see Allowed value types at Blueprint development guidelines.
Step 2: write the initialize
method
- We define the following signature and docstring for our
initialize
method:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public(allow_deposit=True)
def initialize(
self,
ctx: Context,
token_a: TokenUid,
token_b: TokenUid,
multiplier_a: int,
multiplier_b: int
) -> None:
"""Initialize the contract."""
For the correct initialization of the contract, we must ensure that the arguments passed in the method call are correct. Hathor engine guarantees the implementation of default checks required for all method calls. For example, during initialization, the balance of any contract will always be zero. Therefore, if one calls the initialize
method containing withdrawal requests, the method will not even be invoked. Another situation that generates invalidity is a deposit or withdrawal request for a token with an nonexistent id i.e., not registered on the blockchain.
- Thus, we only need to worry about the checks related to our business rules. First, let's check if the contract creator has passed two distinct tokens for the contract to operate:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public(allow_deposit=True)
def initialize(
self,
ctx: Context,
token_a: TokenUid,
token_b: TokenUid,
multiplier_a: int,
multiplier_b: int
) -> None:
"""Initialize the contract."""
if token_a == token_b:
raise NCFail
- As specified during the blueprint design, the contract creator must provide initial liquidity for the contract to operate. Let's then implement a check to ensure that the initial deposits of tokens made by the contract creator are only of the two tokens that they defined for the contract's operation:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public(allow_deposit=True)
def initialize(
self,
ctx: Context,
token_a: TokenUid,
token_b: TokenUid,
multiplier_a: int,
multiplier_b: int
) -> None:
"""Initialize the contract."""
if token_a == token_b:
raise NCFail
if set(ctx.actions.keys()) != {token_a, token_b}:
raise InvalidTokens
class InvalidTokens(NCFail):
pass
actions
from ctx
is a dictionary in which the keys are token_uid
and the values are action
themselves. Also, note that we have defined a subclass to specify the NCFail
exception for cases where an attempt is made to deposit tokens that the contract does not operate with.
- Finally, we can assign the appropriate values to the contract's attributes:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public(allow_deposit=True)
def initialize(
self,
ctx: Context,
token_a: TokenUid,
token_b: TokenUid,
multiplier_a: int,
multiplier_b: int
) -> None:
"""Initialize the contract."""
if token_a == token_b:
raise NCFail
if set(ctx.actions.keys()) != {token_a, token_b}:
raise InvalidTokens
self.token_a = token_a
self.token_b = token_b
self.multiplier_a = multiplier_a
self.multiplier_b = multiplier_b
self.swaps_counter = 0
class InvalidTokens(NCFail):
pass
Step 3: write the swap
method
To fulfill the functionality defined in our blueprint design, we need to code just one public method to execute token swaps. The contract user should call the swap
method tied to precisely two actions: the deposit of one token and the proportional withdrawal of the other.
- Thus, calling the
swap
method does not require the contract user to pass any additional arguments. All necessary information will be found within theContext
created by the protocol during the transaction processing:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public(allow_deposit=True)
def initialize(
self,
ctx: Context,
token_a: TokenUid,
token_b: TokenUid,
multiplier_a: int,
multiplier_b: int
) -> None:
"""Initialize the contract."""
...
@public(allow_deposit=True, allow_withdrawal=True)
def swap(self, ctx: Context) -> None:
"""Execute a token swap."""
class InvalidTokens(NCFail):
pass
Like the initialize
method, we will code our swap
method as follows: we perform all the checks to ensure that the contract user's request is correct and executable, and then we update the contract's state. Remember that actions
encapsulated in ctx
is a dictionary whose keys are token_uid
. Thus, we know that each action refers to a distinct token.
- Let's start by checking if the contract user has sent exactly two actions, and if each of them refers to one of the tokens with which the contract operates:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public(allow_deposit=True, allow_withdrawal=True)
def swap(self, ctx: Context) -> None:
"""Execute a token swap."""
if set(ctx.actions.keys()) != {self.token_a, self.token_b}:
raise InvalidTokens
class InvalidTokens(NCFail):
pass
- Once confirmed that we have received two actions, one for each of the tokens the contract operates, we will store them in local variables:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public(allow_deposit=True, allow_withdrawal=True)
def swap(self, ctx: Context) -> None:
"""Execute a token swap."""
if set(ctx.actions.keys()) != {self.token_a, self.token_b}:
raise InvalidTokens
action_a = ctx.get_single_action(self.token_a)
action_b = ctx.get_single_action(self.token_b)
class InvalidTokens(NCFail):
pass
- Now let's check if the two actions correspond respectively to a deposit and a withdrawal:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public(allow_deposit=True, allow_withdrawal=True)
def swap(self, ctx: Context) -> None:
"""Execute a token swap."""
if set(ctx.actions.keys()) != {self.token_a, self.token_b}:
raise InvalidTokens
action_a = ctx.get_single_action(self.token_a)
action_b = ctx.get_single_action(self.token_b)
if not (
(isinstance(action_a, NCDepositAction) and isinstance(action_b, NCWithdrawalAction))
or (isinstance(action_a, NCWithdrawalAction) and isinstance(action_b, NCDepositAction))
):
raise InvalidActions
class InvalidTokens(NCFail):
pass
class InvalidActions(NCFail):
pass
Note that we have defined a subclass of NCFail
to specialize the exception that occurs when the method does not receive exactly one DEPOSIT
action and one WITHDRAWAL
action.
- Now, we just need to check if the quantities of tokens requested by the contract user for withdrawal and deposit comply with the conversion ratio defined by the contract. Define a helper method to perform this check:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public(allow_deposit=True, allow_withdrawal=True)
def swap(self, ctx: Context) -> None:
"""Execute a token swap."""
if set(ctx.actions.keys()) != {self.token_a, self.token_b}:
raise InvalidTokens
action_a = ctx.get_single_action(self.token_a)
action_b = ctx.get_single_action(self.token_b)
if not (
(isinstance(action_a, NCDepositAction) and isinstance(action_b, NCWithdrawalAction))
or (isinstance(action_a, NCWithdrawalAction) and isinstance(action_b, NCDepositAction))
):
raise InvalidActions
if not self.is_ratio_valid(action_a.amount, action_b.amount):
raise InvalidRatio
class InvalidTokens(NCFail):
pass
class InvalidActions(NCFail):
pass
class InvalidRatio(NCFail):
pass
Note that we have defined a subclass of NCFail
to specialize the exception that occurs when the conversion requested by the contract user is not compatible with that defined by the contract.
- At this point, the swap request has passed all checks. We update the contract's state and conclude the method execution:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public(allow_deposit=True, allow_withdrawal=True)
def swap(self, ctx: Context) -> None:
"""Execute a token swap."""
if set(ctx.actions.keys()) != {self.token_a, self.token_b}:
raise InvalidTokens
action_a = ctx.get_single_action(self.token_a)
action_b = ctx.get_single_action(self.token_b)
if not (
(isinstance(action_a, NCDepositAction) and isinstance(action_b, NCWithdrawalAction))
or (isinstance(action_a, NCWithdrawalAction) and isinstance(action_b, NCDepositAction))
):
raise InvalidActions
if not self.is_ratio_valid(action_a.amount, action_b.amount):
raise InvalidRatio
# All good! Let's accept the transaction.
self.swaps_counter += 1
class InvalidTokens(NCFail):
pass
class InvalidActions(NCFail):
pass
class InvalidRatio(NCFail):
pass
Just like any public method, if no exception is raised during execution, the method will have executed successfully, it will end returning nothing (None
), the contract's state will have been updated, and after that, Hathor engine — not the contract — will take care of updating the multi-token balance of the contract, according to the set of actions that was approved. This characterizes that all actions are then executed simultaneously on the blockchain.
Step 4: write the is_ratio_valid
method
We have already implemented the two public methods necessary for the creation and execution of the contract respectively. To complete the module of our blueprint, we only need to write the is_ratio_valid
method we defined in the previous step. is_ratio_valid
is a helper method, existing solely to make the code of our blueprint more readable, and thus it is internal.
is_ratio_valid
should check if the quantities of tokens the user wishes to swap are proportional to the ratio with which the contract operates. For this, it will receive as parameters (in addition to the contract instance itself) the two quantities of tokens from the two actions, and should returnTrue
if the ratio is valid:
...
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
...
@public
def initialize(
self,
ctx: Context,
token_a: TokenUid,
token_b: TokenUid,
multiplier_a: int,
multiplier_b: int
) -> None:
"""Initialize the contract."""
...
@public
def swap(self, ctx: Context) -> None:
"""Execute a token swap."""
...
@view
def is_ratio_valid(self, qty_a: int, qty_b: int) -> bool:
"""Check if the swap quantities are valid."""
return (self.multiplier_a * qty_a == self.multiplier_b * qty_b)
class InvalidTokens(NCFail):
pass
class InvalidActions(NCFail):
pass
class InvalidRatio(NCFail):
pass
Step 5: review the blueprint module
With this, we have completed the code for our blueprint. Let's review the complete code of the module we just implemented before moving on to the testing phase:
hathor-core/tests/nanocontracts/test_blueprints/swap_demo.py
from hathor.nanocontracts.blueprint import Blueprint
from hathor.nanocontracts.context import Context
from hathor.nanocontracts.exception import NCFail
from hathor.nanocontracts.types import NCDepositAction, NCWithdrawalAction, TokenUid, public, view
class SwapDemo(Blueprint):
"""Blueprint to execute swaps between tokens."""
# TokenA identifier and quantity multiplier
token_a: TokenUid
multiplier_a: int
# TokenB identifier and quantity multiplier
token_b: TokenUid
multiplier_b: int
# Count number of swaps executed
swaps_counter: int
@public(allow_deposit=True)
def initialize(
self,
ctx: Context,
token_a: TokenUid,
token_b: TokenUid,
multiplier_a: int,
multiplier_b: int
) -> None:
"""Initialize the contract."""
if token_a == token_b:
raise NCFail
if set(ctx.actions.keys()) != {token_a, token_b}:
raise InvalidTokens
self.token_a = token_a
self.token_b = token_b
self.multiplier_a = multiplier_a
self.multiplier_b = multiplier_b
self.swaps_counter = 0
@public(allow_deposit=True, allow_withdrawal=True)
def swap(self, ctx: Context) -> None:
"""Execute a token swap."""
if set(ctx.actions.keys()) != {self.token_a, self.token_b}:
raise InvalidTokens
action_a = ctx.get_single_action(self.token_a)
action_b = ctx.get_single_action(self.token_b)
if not (
(isinstance(action_a, NCDepositAction) and isinstance(action_b, NCWithdrawalAction))
or (isinstance(action_a, NCWithdrawalAction) and isinstance(action_b, NCDepositAction))
):
raise InvalidActions
if not self.is_ratio_valid(action_a.amount, action_b.amount):
raise InvalidRatio
# All good! Let's accept the transaction.
self.swaps_counter += 1
@view
def is_ratio_valid(self, qty_a: int, qty_b: int) -> bool:
"""Check if the swap quantities are valid."""
return (self.multiplier_a * qty_a == self.multiplier_b * qty_b)
class InvalidTokens(NCFail):
pass
class InvalidActions(NCFail):
pass
class InvalidRatio(NCFail):
pass
Step 6: create the unit test module
With our blueprint code ready, it's time to verify that it behaves as intended. The blueprint SDK provides a framework based on the Python package pytest
to facilitate the execution of automated unit tests. Let's use it to code our test suite.
As we did previously with the blueprint's code, we suggest that you expand the following pane and perform an inspectional reading of the entire unit test module that we will develop. This will provide the necessary context to understand the steps we will undertake from this point forward:
hathor-core/tests/nanocontracts/blueprints/test_swap.py
from hathor.nanocontracts.context import Context
from hathor.nanocontracts.nc_types import make_nc_type_for_arg_type as make_nc_type
from hathor.nanocontracts.storage.contract_storage import Balance
from hathor.nanocontracts.types import NCDepositAction, NCWithdrawalAction, TokenUid
from tests.nanocontracts.blueprints.unittest import BlueprintTestCase
from tests.nanocontracts.test_blueprints.swap_demo import InvalidActions, InvalidRatio, InvalidTokens, SwapDemo
SWAP_NC_TYPE = make_nc_type(int)
class SwapDemoTestCase(BlueprintTestCase):
def setUp(self):
super().setUp()
self.blueprint_id = self.gen_random_blueprint_id()
self.contract_id = self.gen_random_contract_id()
self.nc_catalog.blueprints[self.blueprint_id] = SwapDemo
# Test doubles:
self.token_a = self.gen_random_token_uid()
self.token_b = self.gen_random_token_uid()
self.token_c = self.gen_random_token_uid()
self.address = self.gen_random_address()
self.tx = self.get_genesis_tx()
def _initialize(
self,
init_token_a: tuple[TokenUid, int, int],
init_token_b: tuple[TokenUid, int, int]
) -> None:
# Arrange:
token_a, multiplier_a, amount_a = init_token_a
token_b, multiplier_b, amount_b = init_token_b
deposit_a = NCDepositAction(token_uid=token_a, amount=amount_a)
deposit_b = NCDepositAction(token_uid=token_b, amount=amount_b)
context = Context(
actions=[deposit_a, deposit_b],
vertex=self.tx,
address=self.address,
timestamp=self.now
)
# Act:
self.runner.create_contract(
self.contract_id,
self.blueprint_id,
context,
token_a,
token_b,
multiplier_a,
multiplier_b,
)
self.nc_storage = self.runner.get_storage(self.contract_id)
def _swap(
self,
amount_a: tuple[int, TokenUid],
amount_b: tuple[int, TokenUid]
) -> None:
# Arrange:
value_a, token_a = amount_a
value_b, token_b = amount_b
action_a_type = self.get_action_type(value_a)
action_b_type = self.get_action_type(value_b)
swap_a = action_a_type(token_uid=token_a, amount=abs(value_a))
swap_b = action_b_type(token_uid=token_b, amount=abs(value_b))
context = Context(
actions=[swap_a, swap_b],
vertex=self.tx,
address=self.address,
timestamp=self.now
)
# Act:
self.runner.call_public_method(self.contract_id, 'swap', context)
def test_lifecycle(self) -> None:
# Create a contract.
# Arrange and act within:
self._initialize((self.token_a, 1, 100_00), (self.token_b, 1, 100_00))
# Assert:
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(0, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make a valid swap.
# Arrange and act within:
self._swap((20_00, self.token_a), (-20_00, self.token_b))
# Assert:
self.assertEqual(
Balance(value=120_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=80_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(1, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make multiple invalid swaps raising all possible exceptions.
with self.assertRaises(InvalidTokens):
self._swap((-20_00, self.token_a), (20_00, self.token_c))
with self.assertRaises(InvalidActions):
self._swap((20_00, self.token_a), (40_00, self.token_b))
with self.assertRaises(InvalidRatio):
self._swap((20_00, self.token_a), (-40_00, self.token_b))
def get_action_type(self, amount: int) -> type[NCDepositAction] | type[NCWithdrawalAction]:
if amount >= 0:
return NCDepositAction
else:
return NCWithdrawalAction
In the project directory tree, navigate to /tests/nanocontracts/blueprints/
. The unit test suite for each blueprint is implemented as a single module.
- Let’s create the
test_swap_demo.py
file, and add the standard imports that are necessary for (if not all) almost all blueprints unit test suites:
from hathor.nanocontracts.context import Context
from hathor.nanocontracts.nc_types import make_nc_type_for_arg_type as make_nc_type
from hathor.nanocontracts.storage.contract_storage import Balance
from hathor.nanocontracts.types import NCDepositAction, NCWithdrawalAction, TokenUid
from tests.nanocontracts.blueprints.unittest import BlueprintTestCase
from tests.nanocontracts.test_blueprints.swap_demo import InvalidActions, InvalidRatio, InvalidTokens, SwapDemo
To test our blueprint, we need the public method calls to be made by emulating the conditions under which they are invoked by Hathor engine. Fortunately, the blueprint SDK handles this for us through the use of the superclass BlueprintTestCase
and the runner
object. The superclass BlueprintTestCase
creates a fake blockchain system, and the runner
object is able to perform the role it normally does in the Hathor core, of invoking the execution of methods from a blueprint.
- We define our
SwapTestCase
class that inherits fromBlueprintTestCase
:
from hathor.nanocontracts.context import Context
from hathor.nanocontracts.nc_types import make_nc_type_for_arg_type as make_nc_type
from hathor.nanocontracts.storage.contract_storage import Balance
from hathor.nanocontracts.types import NCDepositAction, NCWithdrawalAction, TokenUid
from tests.nanocontracts.blueprints.unittest import BlueprintTestCase
from tests.nanocontracts.test_blueprints.swap_demo import InvalidActions, InvalidRatio, InvalidTokens, SwapDemo
class SwapDemoTestCase(BlueprintTestCase):
With this in mind, we can think about what needs to be tested in our blueprint. We need to test the following blueprint interactions:
- Contract creation
- Contract execution
In other words, we need to test the public methods of the blueprint. We need to ensure that the two public methods initialize
and swap
execute successfully when given appropriate inputs and fail as expected when given inappropriate inputs.
Let's then follow this sequence in our test case: (1) create a valid contract; (2) execute a valid swap; (3) execute all the swaps that should fail, one for each exception we defined.
- For this, we will implement a single test method
test_lifecycle
that performs all these tests, simulating the usual lifecycle of a contract on the blockchain:
...
class SwapDemoTestCase(BlueprintTestCase):
def test_lifecycle(self) -> None:
# Create a contract.
# TO DO
# Make a valid swap.
# TO DO
# Make multiple invalid swaps raising all possible exceptions.
# TO DO
- To make it possible to create as many new test flows as we want beyond
test_lifecycle
, and to avoid code repetition in multiple calls to the public methodsinitialize
andswap
, we will create the helper methods_initialize
and_swap
. These should prepare the necessary objects to be sent to the blueprint (arrange) and call the respective method using therunner
(act):
...
class SwapDemoTestCase(BlueprintTestCase):
# def _initialize(self, TO DO) -> None:
# TO DO
# def _swap(self, TO DO) -> None:
# TO DO
def test_lifecycle(self) -> None:
# Create a contract.
# TO DO
# Make a valid swap.
# TO DO
# Make multiple invalid swaps raising all possible exceptions.
# TO DO
- Now, we just need to add the
setUp
method to our class, standard to every test case usingpytest
:
...
class SwapDemoTestCase(BlueprintTestCase):
def setUp(self):
super().setUp()
# TO DO
# def _initialize(self, TO DO) -> None:
# TO DO
# def _swap(self, TO DO) -> None:
# TO DO
def test_lifecycle(self) -> None:
# Create a contract.
# TO DO
# Make a valid swap.
# TO DO
# Make multiple invalid swaps raising all possible exceptions.
# TO DO
After finishing the coding of the four methods we defined, we will complete the implementation of our class, as well as the entire test module. Let's code them in the following sequence: (1) first the test_lifecycle
method, defining the test flow; (2) then the _initialize
method; and (3) finally the _swap
method. (4) The setUp
method will be used to create all necessary test doubles, and therefore will be written ad hoc throughout the process.
Our test execution will start with unittest
invoking the setUp
method for the necessary preliminary preparation for the tests. The statement super().setUp()
carries out the standard preparation required for testing any blueprint.
- Add to the
setUp
method the creation of a dummy contract and its registration, along with the blueprint, on chain:
...
class SwapDemoTestCase(BlueprintTestCase):
def setUp(self):
super().setUp()
self.blueprint_id = self.gen_random_blueprint_id()
self.contract_id = self.gen_random_contract_id()
self.nc_catalog.blueprints[self.blueprint_id] = SwapDemo
# TO DO
...
Step 7: write the test_lifecycle
method
Our flow starts with the creation of a swap contract. To create a swap contract, a user make a call to the initialize
method of our blueprint. Like any call, it can have multiple actions, and for our blueprint, it must have four mandatory arguments: (1) the two distinct tokens with which the contract will operate; and (2) the respective multipliers for each token.
- The creation call of a swap contract requires sending two valid tokens on the blockchain as arguments. Let's add to
setUp
the creation of two dummy tokens, usinggen_random_token_uid()
already implemented in the superclass:
...
class SwapDemoTestCase(BlueprintTestCase):
def setUp(self):
super().setUp()
self.contract_id = self.gen_random_nanocontract_id()
self.runner.register_contract(Swap, self.contract_id)
self.nc_storage = self.runner.get_storage(self.contract_id)
# Test doubles:
self.token_a = self.gen_random_token_uid()
self.token_b = self.gen_random_token_uid()
# TO DO
...
- All other arguments for creating a contract can be arbitrary literals, and with that, we can already create a valid contract. Note that this defines the signature of the
_initialize
method:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def test_lifecycle(self) -> None:
# Create a contract.
# Arrange and act within:
self._initialize((self.token_a, 1, 100_00), (self.token_b, 1, 100_00))
# Make a valid swap.
# TO DO
# Make multiple invalid swaps raising all possible exceptions.
# TO DO
- Our
_initialize
method receives two identical tuples: (token, token's multiplicative factor, and token amount to be deposited). Let's check if the contract's state was correctly initialized:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def test_lifecycle(self) -> None:
# Create a contract.
# Arrange and act within:
self._initialize((self.token_a, 1, 100_00), (self.token_b, 1, 100_00))
# Assert:
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(0, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make a valid swap.
# TO DO
# Make multiple invalid swaps raising all possible exceptions.
# TO DO
Note that assertEqual
and nc_storage
are implemented in the superclass of our test class.
With our swap contract created, let's test a valid swap case.
- To execute the contract, i.e., to perform a swap, the user should make a call to the
swap
method without passing any additional arguments. In this case, we only need to test the actions. In our contract, there must be exactly two actions, one deposit of token A and one withdrawal of token B, in amounts determined by the ratio:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def test_lifecycle(self) -> None:
# Create a contract.
# Arrange and act within:
self._initialize((self.token_a, 1, 100_00), (self.token_b, 1, 100_00))
# Assert:
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(0, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make a valid swap.
# Arrange and act within:
self._swap((20_00, self.token_a), (-20_00, self.token_b))
# Make multiple invalid swaps raising all possible exceptions.
# TO DO
Note that we defined the signature of the _swap
method with two tuples as parameters, one for the action of each token. The blueprint SDK allows us to use negative numbers to encapsulate in the amounts the type of action: positive token amounts denote a deposit, and negative a withdrawal. This facilitates coding and readability of tests.
- We now need to ensure that the state of the contract — multi-token balance and attribute values — was updated correctly after contract execution:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def test_lifecycle(self) -> None:
# Create a contract.
# Arrange and act within:
self._initialize((self.token_a, 1, 100_00), (self.token_b, 1, 100_00))
# Assert:
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(0, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make a valid swap.
# Arrange and act within:
self._swap((20_00, self.token_a), (-20_00, self.token_b))
# Assert:
self.assertEqual(
Balance(value=120_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=80_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(1, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make multiple invalid swaps raising all possible exceptions.
# TO DO
Now let's make swap attempts that should fail. Recall that during the implementation of our blueprint, we created classes specifying the three exceptions that can make a contract execution fail. This facilitates checking the failure cases.
- The first case to check is requests for swaps of tokens with which the contract does not operate with. Recall that before invoking a contract's execution, Hathor engine verifies if the tokens from deposit and withdrawal actions exist on blockchain. Therefore, let's create a dummy token C, which exists on blockchain but is not operated by the contract:
...
class SwapDemoTestCase(BlueprintTestCase):
def setUp(self):
super().setUp()
self.runner = self.create_runner(Swap)
# Test doubles:
self.token_a = self.gen_random_token_uid()
self.token_b = self.gen_random_token_uid()
self.token_c = self.gen_random_token_uid()
# TO DO
...
- Check if the expected execution failure occurs when a user requests a swap of at least one token that the contract does not operate with. It should return the
InvalidTokens
exception:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def test_lifecycle(self) -> None:
# Create a contract.
# Arrange and act within:
self._initialize((self.token_a, 1, 100_00), (self.token_b, 1, 100_00))
# Assert:
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(0, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make a valid swap.
# Arrange and act within:
self._swap((20_00, self.token_a), (-20_00, self.token_b))
# Assert:
self.assertEqual(
Balance(value=120_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=80_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(1, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make multiple invalid swaps raising all possible exceptions.
with self.assertRaises(InvalidTokens):
self._swap((-20_00, self.token_a), (20_00, self.token_c))
- Now, check if the user sent actions related to the two tokens with which the contract operates with, but they were not one deposit and one withdrawal. It should return the
InvalidActions
exception:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def test_lifecycle(self) -> None:
# Create a contract.
# Arrange and act within:
self._initialize((self.token_a, 1, 100_00), (self.token_b, 1, 100_00))
# Assert:
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(0, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make a valid swap.
# Arrange and act within:
self._swap((20_00, self.token_a), (-20_00, self.token_b))
# Assert:
self.assertEqual(
Balance(value=120_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=80_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(1, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make multiple invalid swaps raising all possible exceptions.
with self.assertRaises(InvalidTokens):
self._swap((-20_00, self.token_a), (20_00, self.token_c))
with self.assertRaises(InvalidActions):
self._swap((20_00, self.token_a), (40_00, self.token_b))
- Finally, check if the tokens and actions are correct, but the deposit and withdrawal quantities do not correspond to the contract's conversion ratio. It should return the
InvalidRatio
exception:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def test_lifecycle(self) -> None:
# Create a contract.
# Arrange and act within:
self._initialize((self.token_a, 1, 100_00), (self.token_b, 1, 100_00))
# Assert:
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(0, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make a valid swap.
# Arrange and act within:
self._swap((20_00, self.token_a), (-20_00, self.token_b))
# Assert:
self.assertEqual(
Balance(value=120_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=80_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(1, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make multiple invalid swaps raising all possible exceptions.
with self.assertRaises(InvalidTokens):
self._swap((-20_00, self.token_a), (20_00, self.token_c))
with self.assertRaises(InvalidActions):
self._swap((20_00, self.token_a), (40_00, self.token_b))
with self.assertRaises(InvalidRatio):
self._swap((20_00, self.token_a), (-40_00, self.token_b))
Step 8: write the _initialize
method
Our helper method _initialize
will prepare all objects that our blueprint's initialize
needs to receive (arrange); then, it will use the runner
to call initialize
from swap
(act).
_initialize
already receives within its parameters the four arguments that must be passed toinitialize
:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def _initialize(
self,
init_token_a: tuple[TokenUid, int, int],
init_token_b: tuple[TokenUid, int, int]
) -> None:
...
- Now we need to create the
context
object. To createcontext
, we need dummy values fortx
andaddress
. For this, we will proceed similarly to what we did with the tokens, adding dummy creation insetUp
:
...
class SwapDemoTestCase(BlueprintTestCase):
def setUp(self):
super().setUp()
self.contract_id = self.gen_random_nanocontract_id()
self.runner.register_contract(Swap, self.contract_id)
self.nc_storage = self.runner.get_storage(self.contract_id)
# Test doubles:
self.token_a = self.gen_random_token_uid()
self.token_b = self.gen_random_token_uid()
self.token_c = self.gen_random_token_uid()
self.address = self.gen_random_address()
self.tx = self.get_genesis_tx()
...
- Unpack the received parameters into local variables:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def _initialize(
self,
init_token_a: tuple[TokenUid, int, int],
init_token_b: tuple[TokenUid, int, int]
) -> None:
# Arrange:
token_a, multiplier_a, amount_a = init_token_a
token_b, multiplier_b, amount_b = init_token_b
...
- Create the two actions needed to form the
context
object:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def _initialize(
self,
init_token_a: tuple[TokenUid, int, int],
init_token_b: tuple[TokenUid, int, int]
) -> None:
# Arrange:
token_a, multiplier_a, amount_a = init_token_a
token_b, multiplier_b, amount_b = init_token_b
deposit_a = NCDepositAction(token_uid=token_a, amount=amount_a)
deposit_b = NCDepositAction(token_uid=token_b, amount=amount_b)
...
- And then, create the
context
object:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def _initialize(self, init_token_a: tuple[TokenUid, int, int], init_token_b: tuple[TokenUid, int, int]) -> None:
# Arrange:
token_a, multiplier_a, amount_a = init_token_a
token_b, multiplier_b, amount_b = init_token_b
deposit_a = NCAction(NCActionType.DEPOSIT, token_a, amount_a)
deposit_b = NCAction(NCActionType.DEPOSIT, token_b, amount_b)
context = Context(
actions=[deposit_a, deposit_b],
vertex=self.tx,
address=self.address,
timestamp=self.now
)
...
Note that now
is implemented in the superclass to generate an int
type timestamp.
- Finally, it's time to use the runner to call the method under test, as it's done in the Hathor core:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def _initialize(self, init_token_a: tuple[TokenUid, int, int], init_token_b: tuple[TokenUid, int, int]) -> None:
# Arrange:
token_a, multiplier_a, amount_a = init_token_a
token_b, multiplier_b, amount_b = init_token_b
deposit_a = NCAction(NCActionType.DEPOSIT, token_a, amount_a)
deposit_b = NCAction(NCActionType.DEPOSIT, token_b, amount_b)
context = Context(
actions = [deposit_a, deposit_b],
tx = self.tx,
address = self.address,
timestamp = self.now
)
# Act:
self.runner.create_contract(
self.contract_id,
self.blueprint_id,
context,
token_a,
token_b,
multiplier_a,
multiplier_b,
)
self.nc_storage = self.runner.get_storage(self.contract_id)
...
Step 9: write the _swap
method
Our helper method _swap
will function similarly to _initialize
, but it will handle the execution of contracts by calling the swap
method.
- In
test_lifecycle
, we already defined the signature for_initialize
as having two tuples as parameters. Each tuple contains a value (amount plus type of action) and the token involved in the conversion:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def _swap(
self,
amount_a: tuple[int, TokenUid],
amount_b: tuple[int, TokenUid]
) -> None:
...
- Unpack the received parameters into local variables:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def _swap(
self,
amount_a: tuple[int, TokenUid],
amount_b: tuple[int, TokenUid]
) -> None:
# Arrange:
value_a, token_a = amount_a
value_b, token_b = amount_b
...
- Now we need to create the
context
object containing the two actions that make up a swap. Remember that the type of action is encapsulated in the sign of the amount passed as a parameter. We will also write the_get_action_type
helper method to define the type of action:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def get_action_type(self, amount: int) -> type[NCDepositAction] | type[NCWithdrawalAction]:
if amount >= 0:
return NCDepositAction
else:
return NCWithdrawalAction
...
def _swap(
self,
amount_a: tuple[int, TokenUid],
amount_b: tuple[int, TokenUid]
) -> None:
# Arrange:
value_a, token_a = amount_a
value_b, token_b = amount_b
action_a_type = self.get_action_type(value_a)
action_b_type = self.get_action_type(value_b)
swap_a = action_a_type(token_uid=token_a, amount=abs(value_a))
swap_b = action_b_type(token_uid=token_b, amount=abs(value_b))
...
- And then, create the
context
object:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def _swap(
self,
amount_a: tuple[int, TokenUid],
amount_b: tuple[int, TokenUid]
) -> None:
# Arrange:
value_a, token_a = amount_a
value_b, token_b = amount_b
action_a_type = self.get_action_type(value_a)
action_b_type = self.get_action_type(value_b)
swap_a = action_a_type(token_uid=token_a, amount=abs(value_a))
swap_b = action_b_type(token_uid=token_b, amount=abs(value_b))
context = Context(
actions=[swap_a, swap_b],
vertex=self.tx,
address=self.address,
timestamp=self.now
)
...
- Finally, it's time to use the
runner
to call the method under test, as it's done in the Hathor core:
...
class SwapDemoTestCase(BlueprintTestCase):
...
def _swap(
self,
amount_a: tuple[int, TokenUid],
amount_b: tuple[int, TokenUid]
) -> None:
# Arrange:
value_a, token_a = amount_a
value_b, token_b = amount_b
action_a_type = self.get_action_type(value_a)
action_b_type = self.get_action_type(value_b)
swap_a = action_a_type(token_uid=token_a, amount=abs(value_a))
swap_b = action_b_type(token_uid=token_b, amount=abs(value_b))
context = Context(
actions=[swap_a, swap_b],
vertex=self.tx,
address=self.address,
timestamp=self.now
)
# Act:
self.runner.call_public_method(self.contract_id, 'swap', context)
...
Step 10: review the blueprint unit test module
With this, we have completed the code for our test suite. Let's review the complete code of the module we just implemented before starting running the tests:
hathor-core/tests/nanocontracts/blueprints/test_swap.py
from hathor.nanocontracts.context import Context
from hathor.nanocontracts.nc_types import make_nc_type_for_arg_type as make_nc_type
from hathor.nanocontracts.storage.contract_storage import Balance
from hathor.nanocontracts.types import NCDepositAction, NCWithdrawalAction, TokenUid
from tests.nanocontracts.blueprints.unittest import BlueprintTestCase
from tests.nanocontracts.test_blueprints.swap_demo import InvalidActions, InvalidRatio, InvalidTokens, SwapDemo
SWAP_NC_TYPE = make_nc_type(int)
class SwapDemoTestCase(BlueprintTestCase):
def setUp(self):
super().setUp()
self.blueprint_id = self.gen_random_blueprint_id()
self.contract_id = self.gen_random_contract_id()
self.nc_catalog.blueprints[self.blueprint_id] = SwapDemo
# Test doubles:
self.token_a = self.gen_random_token_uid()
self.token_b = self.gen_random_token_uid()
self.token_c = self.gen_random_token_uid()
self.address = self.gen_random_address()
self.tx = self.get_genesis_tx()
def _initialize(
self,
init_token_a: tuple[TokenUid, int, int],
init_token_b: tuple[TokenUid, int, int]
) -> None:
# Arrange:
token_a, multiplier_a, amount_a = init_token_a
token_b, multiplier_b, amount_b = init_token_b
deposit_a = NCDepositAction(token_uid=token_a, amount=amount_a)
deposit_b = NCDepositAction(token_uid=token_b, amount=amount_b)
context = Context(
actions=[deposit_a, deposit_b],
vertex=self.tx,
address=self.address,
timestamp=self.now
)
# Act:
self.runner.create_contract(
self.contract_id,
self.blueprint_id,
context,
token_a,
token_b,
multiplier_a,
multiplier_b,
)
self.nc_storage = self.runner.get_storage(self.contract_id)
def _swap(
self,
amount_a: tuple[int, TokenUid],
amount_b: tuple[int, TokenUid]
) -> None:
# Arrange:
value_a, token_a = amount_a
value_b, token_b = amount_b
action_a_type = self.get_action_type(value_a)
action_b_type = self.get_action_type(value_b)
swap_a = action_a_type(token_uid=token_a, amount=abs(value_a))
swap_b = action_b_type(token_uid=token_b, amount=abs(value_b))
context = Context(
actions=[swap_a, swap_b],
vertex=self.tx,
address=self.address,
timestamp=self.now
)
# Act:
self.runner.call_public_method(self.contract_id, 'swap', context)
def test_lifecycle(self) -> None:
# Create a contract.
# Arrange and act within:
self._initialize((self.token_a, 1, 100_00), (self.token_b, 1, 100_00))
# Assert:
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=100_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(0, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make a valid swap.
# Arrange and act within:
self._swap((20_00, self.token_a), (-20_00, self.token_b))
# Assert:
self.assertEqual(
Balance(value=120_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_a)
)
self.assertEqual(
Balance(value=80_00, can_mint=False, can_melt=False), self.nc_storage.get_balance(self.token_b)
)
self.assertEqual(1, self.nc_storage.get_obj(b'swaps_counter', SWAP_NC_TYPE))
# Make multiple invalid swaps raising all possible exceptions.
with self.assertRaises(InvalidTokens):
self._swap((-20_00, self.token_a), (20_00, self.token_c))
with self.assertRaises(InvalidActions):
self._swap((20_00, self.token_a), (40_00, self.token_b))
with self.assertRaises(InvalidRatio):
self._swap((20_00, self.token_a), (-40_00, self.token_b))
def get_action_type(self, amount: int) -> type[NCDepositAction] | type[NCWithdrawalAction]:
if amount >= 0:
return NCDepositAction
else:
return NCWithdrawalAction
Task completed
At this point, you have completed the source code for your first blueprint using the blueprint SDK. Now, it's time to run your suite of automated unit tests. Start a shell session, navigate to the root of the hathor-core
project and perform the tests using pytest
:
poetry run pytest -v -n0 tests/nanocontracts/blueprints/test_swap_demo.py \
-W ignore::DeprecationWarning \
-W ignore::PendingDeprecationWarning \
-W ignore::FutureWarning
Key takeaways
Keep in mind that the automated unit testing approach presented in this article is the quickest, easiest, and most practical way to test your blueprint. After these tests, the blueprint SDK also provides tools for integration testing, in which a complete blockchain test system is run, and it is then possible to instantiate nano contracts on chain from the developed blueprint.