add were_addresses_spent_from api

Author: scottbelden

iota/api.py

@@ -417,6 +417,22 @@ def store_transactions(self, trytes):
     """
     return core.StoreTransactionsCommand(self.adapter)(trytes=trytes)
 
+  def were_addresses_spent_from(self, addresses):
+    # type: (Iterable[Address]) -> dict
+    """
+    Check if a list of addresses was ever spent from, in the current
+    epoch, or in previous epochs.
+
+    :param addresses:
+      List of addresses to check.
+
+    References:
+      - https://iota.readme.io/docs/wereaddressesspentfrom
+    """
+    return core.WereAddressesSpentFromCommand(self.adapter)(
+      addresses = addresses,
+    )
+
 
 class Iota(StrictIota):
   """

iota/commands/core/__init__.py

@@ -25,3 +25,4 @@
 from .interrupt_attaching_to_tangle import *
 from .remove_neighbors import *
 from .store_transactions import *
+from .were_addresses_spent_from import *

iota/commands/core/were_addresses_spent_from.py

@@ -0,0 +1,44 @@
+# coding=utf-8
+from __future__ import absolute_import, division, print_function, \
+  unicode_literals
+
+import filters as f
+
+from iota.commands import FilterCommand, RequestFilter
+from iota.filters import AddressNoChecksum
+
+__all__ = [
+  'WereAddressesSpentFromCommand',
+]
+
+
+class WereAddressesSpentFromCommand(FilterCommand):
+  """
+  Executes `wereAddressesSpentFrom` command.
+
+  See :py:meth:`iota.api.StrictIota.were_addresses_spent_from`.
+  """
+  command = 'wereAddressesSpentFrom'
+
+  def get_request_filter(self):
+    return WereAddressesSpentFromRequestFilter()
+
+  def get_response_filter(self):
+    pass
+
+
+class WereAddressesSpentFromRequestFilter(RequestFilter):
+  def __init__(self):
+    super(WereAddressesSpentFromRequestFilter, self).__init__(
+      {
+        'addresses': (
+            f.Required
+          | f.Array
+          | f.FilterRepeater(
+                f.Required
+              | AddressNoChecksum()
+              | f.Unicode(encoding='ascii', normalize=False)
+            )
+        ),
+      }
+    )

test/commands/core/were_addresses_spent_from_test.py

@@ -0,0 +1,175 @@
+# 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 iota import Address, Iota, TryteString
+from iota.adapter import MockAdapter
+from iota.commands.core.were_addresses_spent_from import WereAddressesSpentFromCommand
+from iota.filters import Trytes
+
+
+class WereAddressesSpentFromRequestFilterTestCase(BaseFilterTestCase):
+  filter_type = WereAddressesSpentFromCommand(MockAdapter()).get_request_filter
+  skip_value_check = True
+
+  # noinspection SpellCheckingInspection
+  def setUp(self):
+    super(WereAddressesSpentFromRequestFilterTestCase, self).setUp()
+
+    # Define a few valid values that we can reuse across tests.
+    self.trytes1 = (
+      'TESTVALUE9DONTUSEINPRODUCTION99999EKJZZT'
+      'SOGJOUNVEWLDPKGTGAOIZIPMGBLHC9LMQNHLGXGYX'
+    )
+
+    self.trytes2 = (
+      'TESTVALUE9DONTUSEINPRODUCTION99999FDCDTZ'
+      'ZWLL9MYGUTLSYVSIFJ9NGALTRMCQVIIOVEQOITYTE'
+    )
+
+  def test_pass_happy_path(self):
+    """
+    Typical invocation of ``wereAddressesSpentFrom``.
+    """
+    request = {
+      # Raw trytes are extracted to match the IRI's JSON protocol.
+      'addresses': [self.trytes1, self.trytes2],
+    }
+
+    filter_ = self._filter(request)
+
+    self.assertFilterPasses(filter_)
+    self.assertDictEqual(filter_.cleaned_data, request)
+
+  def test_pass_compatible_types(self):
+    """
+    The incoming request contains values that can be converted to the
+    expected types.
+    """
+    request = {
+      'addresses': [
+        Address(self.trytes1),
+        bytearray(self.trytes2.encode('ascii')),
+      ],
+    }
+
+    filter_ = self._filter(request)
+
+    self.assertFilterPasses(filter_)
+    self.assertDictEqual(
+      filter_.cleaned_data,
+
+      {
+        'addresses': [self.trytes1, self.trytes2],
+      },
+    )
+
+  def test_fail_empty(self):
+    """
+    The incoming request is empty.
+    """
+    self.assertFilterErrors(
+      {},
+
+      {
+        'addresses': [f.FilterMapper.CODE_MISSING_KEY],
+      },
+    )
+
+  def test_fail_unexpected_parameters(self):
+    """
+    The incoming request contains unexpected parameters.
+    """
+    self.assertFilterErrors(
+      {
+        'addresses': [Address(self.trytes1)],
+
+        # I've had a perfectly wonderful evening.
+        # But this wasn't it.
+        'foo': 'bar',
+      },
+
+      {
+        'foo': [f.FilterMapper.CODE_EXTRA_KEY],
+      },
+    )
+
+  def test_fail_addresses_wrong_type(self):
+    """
+    ``addresses`` is not an array.
+    """
+    self.assertFilterErrors(
+      {
+        'addresses': Address(self.trytes1),
+      },
+
+      {
+        'addresses': [f.Type.CODE_WRONG_TYPE],
+      },
+    )
+
+  def test_fail_addresses_empty(self):
+    """
+    ``addresses`` is an array, but it's empty.
+    """
+    self.assertFilterErrors(
+      {
+        'addresses': [],
+      },
+
+      {
+        'addresses': [f.Required.CODE_EMPTY],
+      },
+    )
+
+  def test_fail_addresses_contents_invalid(self):
+    """
+    ``addresses`` is an array, but it contains invalid values.
+    """
+    self.assertFilterErrors(
+      {
+        'addresses': [
+          b'',
+          True,
+          None,
+          b'not valid trytes',
+
+          # This is actually valid; I just added it to make sure the
+          # filter isn't cheating!
+          TryteString(self.trytes2),
+
+          2130706433,
+          b'9' * 82,
+        ],
+      },
+
+      {
+        'addresses.0':  [f.Required.CODE_EMPTY],
+        'addresses.1':  [f.Type.CODE_WRONG_TYPE],
+        'addresses.2':  [f.Required.CODE_EMPTY],
+        'addresses.3':  [Trytes.CODE_NOT_TRYTES],
+        'addresses.5':  [f.Type.CODE_WRONG_TYPE],
+        'addresses.6':  [Trytes.CODE_WRONG_FORMAT],
+      },
+    )
+
+
+class WereAddressesSpentFromCommandTestCase(TestCase):
+  def setUp(self):
+    super(WereAddressesSpentFromCommandTestCase, self).setUp()
+
+    self.adapter = MockAdapter()
+
+  def test_wireup(self):
+    """
+    Verify that the command is wired up correctly.
+    """
+    self.assertIsInstance(
+      Iota(self.adapter).wereAddressesSpentFrom,
+      WereAddressesSpentFromCommand,
+    )