diff --git a/iota/api.py b/iota/api.py index aeb6f831..4c8d56a1 100644 --- a/iota/api.py +++ b/iota/api.py @@ -886,3 +886,27 @@ def send_trytes(self, trytes, depth, min_weight_magnitude=None): depth = depth, minWeightMagnitude = min_weight_magnitude, ) + + def is_reattachable(self, addresses): + # type: (Iterable[Address]) -> dict + """ + This API function helps you to determine whether you should replay a + transaction or make a completely new transaction with a different seed. + What this function does, is it takes one or more input addresses (i.e. from spent transactions) + as input and then checks whether any transactions with a value transferred are confirmed. + If yes, it means that this input address has already been successfully used in a different + transaction and as such you should no longer replay the transaction. + + :param addresses: + List of addresses. + + :return: + Dict containing the following values:: + { + 'reattachable': List[bool], + Always a list, even if only one address was queried. + } + """ + return extended.IsReattachableCommand(self.adapter)( + addresses=addresses + ) diff --git a/iota/commands/extended/__init__.py b/iota/commands/extended/__init__.py index f7a5bdf9..af596c9e 100644 --- a/iota/commands/extended/__init__.py +++ b/iota/commands/extended/__init__.py @@ -18,6 +18,7 @@ from .get_latest_inclusion import * from .get_new_addresses import * from .get_transfers import * +from .is_reattachable import * from .prepare_transfer import * from .replay_bundle import * from .send_transfer import * diff --git a/iota/commands/extended/is_reattachable.py b/iota/commands/extended/is_reattachable.py new file mode 100644 index 00000000..b3085678 --- /dev/null +++ b/iota/commands/extended/is_reattachable.py @@ -0,0 +1,75 @@ +# coding=utf-8 +from __future__ import absolute_import, division, print_function, unicode_literals +from typing import List + +import filters as f + +from iota import Address +from iota.commands import FilterCommand, RequestFilter, ResponseFilter +from iota.commands.extended import GetLatestInclusionCommand +from iota.commands.extended.utils import find_transaction_objects +from iota.filters import Trytes + +__all__ = [ + 'IsReattachableCommand', +] + + +class IsReattachableCommand(FilterCommand): + """ + Executes ``isReattachable`` extended API command. + """ + command = 'isReattachable' + + def get_request_filter(self): + return IsReattachableRequestFilter() + + def get_response_filter(self): + return IsReattachableResponseFilter() + + def _execute(self, request): + addresses = request['addresses'] # type: List[Address] + + # fetch full transaction objects + transactions = find_transaction_objects(adapter=self.adapter, **{'addresses': addresses}) + + # map and filter transactions, which have zero value. + # If multiple transactions for the same address are returned the one with the + # highest attachment_timestamp is selected + transactions = sorted(transactions, key=lambda t: t.attachment_timestamp) + transaction_map = {t.address: t.hash for t in transactions if t.value > 0} + + # fetch inclusion states + inclusion_states = GetLatestInclusionCommand(adapter=self.adapter)(hashes=list(transaction_map.values())) + inclusion_states = inclusion_states['states'] + + return { + 'reattachable': [not inclusion_states[transaction_map[address]] for address in addresses] + } + + +class IsReattachableRequestFilter(RequestFilter): + def __init__(self): + super(IsReattachableRequestFilter, self).__init__( + { + 'addresses': ( + f.Required + | f.Array + | f.FilterRepeater( + f.Required + | Trytes(result_type=Address) + | f.Unicode(encoding='ascii', normalize=False) + ) + ) + } + ) + + +class IsReattachableResponseFilter(ResponseFilter): + def __init__(self): + super(IsReattachableResponseFilter, self).__init__({ + 'reattachable': ( + f.Required + | f.Array + | f.FilterRepeater(f.Type(bool))) + }) diff --git a/test/commands/extended/is_reattachable_test.py b/test/commands/extended/is_reattachable_test.py new file mode 100644 index 00000000..edef5fd5 --- /dev/null +++ b/test/commands/extended/is_reattachable_test.py @@ -0,0 +1,206 @@ +# coding=utf-8 +from __future__ import absolute_import, division, print_function, \ + unicode_literals + +from unittest import TestCase + +import filters as f +from filters.test import BaseFilterTestCase +from six import text_type + +from iota import Address, Iota +from iota.adapter import MockAdapter +from iota.commands.extended.is_reattachable import IsReattachableCommand + + +class IsReattachableRequestFilterTestCase(BaseFilterTestCase): + filter_type = IsReattachableCommand(MockAdapter()).get_request_filter + skip_value_check = True + + # noinspection SpellCheckingInspection + def setUp(self): + super(IsReattachableRequestFilterTestCase, self).setUp() + + # Define a few valid values that we can reuse across tests. + self.address_1 = ( + 'TESTVALUE9DONTUSEINPRODUCTION99999EKJZZT' + 'SOGJOUNVEWLDPKGTGAOIZIPMGBLHC9LMQNHLGXGYX' + ) + + self.address_2 = ( + 'TESTVALUE9DONTUSEINPRODUCTION99999FDCDTZ' + 'ZWLL9MYGUTLSYVSIFJ9NGALTRMCQVIIOVEQOITYTE' + ) + + def test_pass_happy_path(self): + """ + Filter for list of valid string addresses + """ + + request = { + # Raw trytes are extracted to match the IRI's JSON protocol. + 'addresses': [ + self.address_1, + Address(self.address_2) + ], + } + + filter_ = self._filter(request) + + self.assertFilterPasses(filter_) + + self.assertDictEqual( + filter_.cleaned_data, + { + 'addresses': [ + text_type(Address(self.address_1)), + text_type(Address(self.address_2)) + ], + }, + ) + + def test_pass_compatible_types(self): + """ + The incoming request contains values that can be converted to the + expected types. + """ + request = { + 'addresses': [ + Address(self.address_1), + bytearray(self.address_2.encode('ascii')), + ], + } + + filter_ = self._filter(request) + + self.assertFilterPasses(filter_) + self.assertDictEqual( + filter_.cleaned_data, + { + 'addresses': [self.address_1, self.address_2], + }, + ) + + def test_pass_incompatible_types(self): + """ + The incoming request contains values that can NOT be converted to the + expected types. + """ + request = { + 'addresses': [ + 1234234, + False + ], + } + + self.assertFilterErrors( + request, + { + 'addresses.0': [f.Type.CODE_WRONG_TYPE], + 'addresses.1': [f.Type.CODE_WRONG_TYPE] + }, + ) + + def test_fail_empty(self): + """ + The incoming request is empty. + """ + self.assertFilterErrors( + {}, + { + 'addresses': [f.FilterMapper.CODE_MISSING_KEY], + }, + ) + + def test_fail_single_address(self): + """ + The incoming request contains a single address + """ + request = { + 'addresses': Address(self.address_1) + } + + self.assertFilterErrors( + request, + { + 'addresses': [f.Type.CODE_WRONG_TYPE], + } + ) + + +# noinspection SpellCheckingInspection +class IsReattachableResponseFilterTestCase(BaseFilterTestCase): + filter_type = IsReattachableCommand(MockAdapter()).get_response_filter + skip_value_check = True + + # noinspection SpellCheckingInspection + def setUp(self): + super(IsReattachableResponseFilterTestCase, self).setUp() + + # Define a few valid values that we can reuse across tests. + self.addresses_1 = ( + 'TESTVALUE9DONTUSEINPRODUCTION99999EKJZZT' + 'SOGJOUNVEWLDPKGTGAOIZIPMGBLHC9LMQNHLGXGYX' + ) + + def test_pass_happy_path(self): + """ + Typical ``IsReattachable`` request. + """ + response = { + 'reattachable': [True, False] + } + + filter_ = self._filter(response) + + self.assertFilterPasses(filter_) + self.assertDictEqual(filter_.cleaned_data, response) + + def test_fail_empty(self): + """ + The incoming response is empty. + """ + self.assertFilterErrors( + {}, + { + 'reattachable': [f.Required.CODE_EMPTY], + }, + ) + + def test_pass_incompatible_types(self): + """ + The response contains values that can NOT be converted to the + expected types. + """ + request = { + 'reattachable': [ + 1234234, + b'', + 'test' + ], + } + + self.assertFilterErrors( + request, + { + 'reattachable.0': [f.Type.CODE_WRONG_TYPE], + 'reattachable.1': [f.Type.CODE_WRONG_TYPE], + 'reattachable.2': [f.Type.CODE_WRONG_TYPE] + }, + ) + + +class IsReattachableCommandTestCase(TestCase): + def setUp(self): + super(IsReattachableCommandTestCase, self).setUp() + + self.adapter = MockAdapter() + + def test_wireup(self): + """ + Verify that the command is wired up correctly. + """ + self.assertIsInstance( + Iota(self.adapter).isReattachable, + IsReattachableCommand, + )