This repository has been archived by the owner on Jan 13, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 124
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #108 from jinnerbichler/38-is_reattachable
Implementation of IsReattachableCommand
- Loading branch information
Showing
4 changed files
with
306 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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))) | ||
}) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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, | ||
) |