Skip to content
This repository was archived by the owner on Jan 13, 2023. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions iota/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,12 @@ def __getattr__(self, command):
# noinspection PyTypeChecker
return None

# Fix an error when invoking dunder methods.
# https://github.com/iotaledger/iota.lib.py/issues/206
if command.startswith("__"):
# noinspection PyUnresolvedReferences
return super(StrictIota, self).__getattr__(command)

try:
command_class = self.commands[command]
except KeyError:
Expand Down
221 changes: 119 additions & 102 deletions test/api_test.py
Original file line number Diff line number Diff line change
@@ -1,137 +1,154 @@
# coding=utf-8
from __future__ import absolute_import, division, print_function, \
unicode_literals
unicode_literals

from abc import ABCMeta
from unittest import TestCase

from iota import InvalidCommand, Iota, StrictIota
from six import with_metaclass

from iota import InvalidCommand, StrictIota
from iota.adapter import MockAdapter
from iota.commands import CustomCommand
from iota.commands.core.get_node_info import GetNodeInfoCommand


class CustomCommandTestCase(TestCase):
def setUp(self):
super(CustomCommandTestCase, self).setUp()
def setUp(self):
super(CustomCommandTestCase, self).setUp()

self.name = 'helloWorld'
self.adapter = MockAdapter()
self.command = CustomCommand(self.adapter, self.name)
self.name = 'helloWorld'
self.adapter = MockAdapter()
self.command = CustomCommand(self.adapter, self.name)

def test_call(self):
"""
Sending a custom command.
"""
expected_response = {'message': 'Hello, IOTA!'}
def test_call(self):
"""
Sending a custom command.
"""
expected_response = {'message': 'Hello, IOTA!'}

self.adapter.seed_response('helloWorld', expected_response)
self.adapter.seed_response('helloWorld', expected_response)

response = self.command()
response = self.command()

self.assertEqual(response, expected_response)
self.assertTrue(self.command.called)
self.assertEqual(response, expected_response)
self.assertTrue(self.command.called)

self.assertListEqual(
self.adapter.requests,
[{'command': 'helloWorld'}],
)
self.assertListEqual(
self.adapter.requests,
[{'command': 'helloWorld'}],
)

def test_call_with_parameters(self):
"""
Sending a custom command with parameters.
"""
expected_response = {'message': 'Hello, IOTA!'}
def test_call_with_parameters(self):
"""
Sending a custom command with parameters.
"""
expected_response = {'message': 'Hello, IOTA!'}

self.adapter.seed_response('helloWorld', expected_response)
self.adapter.seed_response('helloWorld', expected_response)

response = self.command(foo='bar', baz='luhrmann')
response = self.command(foo='bar', baz='luhrmann')

self.assertEqual(response, expected_response)
self.assertTrue(self.command.called)
self.assertEqual(response, expected_response)
self.assertTrue(self.command.called)

self.assertListEqual(
self.adapter.requests,
[{'command': 'helloWorld', 'foo': 'bar', 'baz': 'luhrmann'}],
)
self.assertListEqual(
self.adapter.requests,
[{'command': 'helloWorld', 'foo': 'bar', 'baz': 'luhrmann'}],
)

def test_call_error_already_called(self):
"""
A command can only be called once.
"""
self.adapter.seed_response('helloWorld', {})
self.command()
def test_call_error_already_called(self):
"""
A command can only be called once.
"""
self.adapter.seed_response('helloWorld', {})
self.command()

with self.assertRaises(RuntimeError):
self.command(extra='params')
with self.assertRaises(RuntimeError):
self.command(extra='params')

self.assertDictEqual(self.command.request, {'command': 'helloWorld'})
self.assertDictEqual(self.command.request, {'command': 'helloWorld'})

def test_call_reset(self):
"""
Resetting a command allows it to be called more than once.
"""
self.adapter.seed_response('helloWorld', {'message': 'Hello, IOTA!'})
self.command()
def test_call_reset(self):
"""
Resetting a command allows it to be called more than once.
"""
self.adapter.seed_response('helloWorld', {'message': 'Hello, IOTA!'})
self.command()

self.command.reset()
self.command.reset()

self.assertFalse(self.command.called)
self.assertIsNone(self.command.request)
self.assertIsNone(self.command.response)
self.assertFalse(self.command.called)
self.assertIsNone(self.command.request)
self.assertIsNone(self.command.response)

expected_response = {'message': 'Welcome back!'}
self.adapter.seed_response('helloWorld', expected_response)
response = self.command(foo='bar')
expected_response = {'message': 'Welcome back!'}
self.adapter.seed_response('helloWorld', expected_response)
response = self.command(foo='bar')

self.assertDictEqual(response, expected_response)
self.assertDictEqual(self.command.response, expected_response)
self.assertDictEqual(response, expected_response)
self.assertDictEqual(self.command.response, expected_response)

self.assertDictEqual(
self.command.request,
self.assertDictEqual(
self.command.request,

{
'command': 'helloWorld',
'foo': 'bar',
},
)
{
'command': 'helloWorld',
'foo': 'bar',
},
)


class IotaApiTestCase(TestCase):
def test_init_with_uri(self):
"""
Passing a URI to the initializer instead of an adapter instance.
"""
api = StrictIota('mock://')
self.assertIsInstance(api.adapter, MockAdapter)

def test_registered_command(self):
"""
Preparing a documented command.
"""
api = StrictIota(MockAdapter())

# We just need to make sure the correct command type is
# instantiated; individual commands have their own unit tests.
command = api.getNodeInfo
self.assertIsInstance(command, GetNodeInfoCommand)

def test_unregistered_command(self):
"""
Attempting to create an unsupported command.
"""
api = StrictIota(MockAdapter())

with self.assertRaises(InvalidCommand):
# noinspection PyStatementEffect
api.helloWorld

def test_create_command(self):
"""
Preparing an experimental/undocumented command.
"""
api = StrictIota(MockAdapter())

custom_command = api.create_command('helloWorld')

self.assertIsInstance(custom_command, CustomCommand)
self.assertEqual(custom_command.command, 'helloWorld')
def test_init_with_uri(self):
"""
Passing a URI to the initializer instead of an adapter instance.
"""
api = StrictIota('mock://')
self.assertIsInstance(api.adapter, MockAdapter)

def test_registered_command(self):
"""
Preparing a documented command.
"""
api = StrictIota(MockAdapter())

# We just need to make sure the correct command type is
# instantiated; individual commands have their own unit tests.
command = api.getNodeInfo
self.assertIsInstance(command, GetNodeInfoCommand)

def test_unregistered_command(self):
"""
Attempting to create an unsupported command.
"""
api = StrictIota(MockAdapter())

with self.assertRaises(InvalidCommand):
# noinspection PyStatementEffect
api.helloWorld

def test_create_command(self):
"""
Preparing an experimental/undocumented command.
"""
api = StrictIota(MockAdapter())

custom_command = api.create_command('helloWorld')

self.assertIsInstance(custom_command, CustomCommand)
self.assertEqual(custom_command.command, 'helloWorld')

def test_use_in_abstract_class(self):
"""
Tests a regression where adding a client instance to an abstract
class definition raises an exception.

References:
- https://github.com/iotaledger/iota.lib.py/issues/206
"""
# This statement will raise an exception if the regression is
# present; no assertions necessary.
# noinspection PyUnusedLocal
class CustomClient(with_metaclass(ABCMeta)):
client = StrictIota(MockAdapter())