Implementation of IsReattachableCommand

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
+    )

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 *

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)))
+    })

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,
+    )