From 51170b89e8627aae0a3c8dd323d44768e6779148 Mon Sep 17 00:00:00 2001 From: Levente Pap Date: Tue, 10 Dec 2019 13:47:00 +0100 Subject: [PATCH] Remove Sandbox Adapter Sandbox environment not in use anymore. --- docs/adapters.rst | 1 - docs/getting_started.rst | 28 --- examples/sandbox.py | 52 ----- iota/adapter/sandbox.py | 289 ------------------------ test/adapter/sandbox_test.py | 414 ----------------------------------- test/local_pow_test.py | 19 -- 6 files changed, 803 deletions(-) delete mode 100644 examples/sandbox.py delete mode 100644 iota/adapter/sandbox.py delete mode 100644 test/adapter/sandbox_test.py diff --git a/docs/adapters.rst b/docs/adapters.rst index 83940d07..79307f30 100644 --- a/docs/adapters.rst +++ b/docs/adapters.rst @@ -13,7 +13,6 @@ the networking, communicating with a node. You can choose and configure the available adapters to be used with the API: - HttpAdapter, - - SandboxAdapter, - MockAdapter. AdapterSpec diff --git a/docs/getting_started.rst b/docs/getting_started.rst index 7eb8d1a5..574ebb55 100644 --- a/docs/getting_started.rst +++ b/docs/getting_started.rst @@ -42,7 +42,6 @@ You can: - `Run your own node.`_ - `Use a light wallet node.`_ -- `Use the sandbox node.`_ Note that light wallet nodes often disable certain features like PoW for security reasons. @@ -68,32 +67,6 @@ Test your connection to the server by sending a ``getNodeInfo`` command: You are now ready to send commands to your IOTA node! -Using the Sandbox Node ----------------------- -To connect to the sandbox node, you will need to inject a -:py:class:`SandboxAdapter` into your :py:class:`Iota` object. This will modify -your API requests so that they contain the necessary authentication metadata. - -.. code-block:: python - - from iota.adapter.sandbox import SandboxAdapter - - api = Iota( - # To use sandbox mode, inject a ``SandboxAdapter``. - adapter = SandboxAdapter( - # URI of the sandbox node. - uri = 'https://sandbox.iotatoken.com/api/v1/', - - # Access token used to authenticate requests. - # Contact the node maintainer to get an access token. - auth_token = 'auth token goes here', - ), - - # Seed used for cryptographic functions. - # If null, a random seed will be generated. - seed = b'SEED9GOES9HERE', - ) - .. _forum: https://forum.iota.org/ .. _official api: https://docs.iota.org/docs/node-software/0.1/iri/references/api-reference .. _pyota-ccurl extension: https://pypi.python.org/pypi/PyOTA-CCurl @@ -101,4 +74,3 @@ your API requests so that they contain the necessary authentication metadata. .. _run your own node.: http://iotasupport.com/headlessnode.shtml .. _slack: http://slack.iota.org/ .. _use a light wallet node.: http://iotasupport.com/lightwallet.shtml -.. _use the sandbox node.: http://dev.iota.org/sandbox diff --git a/examples/sandbox.py b/examples/sandbox.py deleted file mode 100644 index bbaac855..00000000 --- a/examples/sandbox.py +++ /dev/null @@ -1,52 +0,0 @@ -# coding=utf-8 -from __future__ import absolute_import, division, print_function, \ - unicode_literals - -from iota import Address, Iota, ProposedTransaction, Tag, TryteString -from iota.adapter.sandbox import SandboxAdapter - -# Create the API client. -api = Iota( - # To use sandbox mode, inject a ``SandboxAdapter``. - adapter=SandboxAdapter( - # URI of the sandbox node. - uri='https://sandbox.iota.org/api/v1/', - - # Access token used to authenticate requests. - # Contact the node maintainer to get an access token. - auth_token='auth token goes here', - ), - - # Seed used for cryptographic functions. - # If null, a random seed will be generated. - seed=b'SEED9GOES9HERE', -) - -# Example of sending a transfer using the sandbox. -# For more information, see :py:meth:`Iota.send_transfer`. -# noinspection SpellCheckingInspection -api.send_transfer( - depth=3, - - # One or more :py:class:`ProposedTransaction` objects to add to the - # bundle. - transfers=[ - ProposedTransaction( - # Recipient of the transfer. - address=Address( - b'TESTVALUE9DONTUSEINPRODUCTION99999FBFFTG' - b'QFWEHEL9KCAFXBJBXGE9HID9XCOHFIDABHDG9AHDR' - ), - - # Amount of IOTA to transfer. - # This value may be zero. - value=42, - - # Optional tag to attach to the transfer. - tag=Tag(b'EXAMPLE'), - - # Optional message to include with the transfer. - message=TryteString.from_string('Hello, Tangle!'), - ), - ], -) diff --git a/iota/adapter/sandbox.py b/iota/adapter/sandbox.py deleted file mode 100644 index 1df9a425..00000000 --- a/iota/adapter/sandbox.py +++ /dev/null @@ -1,289 +0,0 @@ -# coding=utf-8 -from __future__ import absolute_import, division, print_function, \ - unicode_literals - -from time import sleep -from typing import Container, Optional, Text, Union - -from requests import Response, codes -from six import moves as compat, text_type - -from iota.adapter import BadApiResponse, HttpAdapter, SplitResult -from iota.exceptions import with_context - -__all__ = [ - 'SandboxAdapter', -] - -STATUS_ABORTED = 'ABORTED' -STATUS_FAILED = 'FAILED' -STATUS_FINISHED = 'FINISHED' -STATUS_QUEUED = 'QUEUED' -STATUS_RUNNING = 'RUNNING' - - -class SandboxAdapter(HttpAdapter): - """ - HTTP adapter that sends requests to remote nodes operating in - "sandbox" mode. - - In sandbox mode, the node will only accept authenticated requests - from clients, and certain jobs are completed asynchronously. - - Note: for compatibility with Iota APIs, SandboxAdapter still - operates synchronously; it blocks until it determines that a job has - completed successfully. - - References: - - https://github.com/iotaledger/iota.py/issues/19 - - https://github.com/iotaledger/documentation/blob/sandbox/source/index.html.md - """ - DEFAULT_POLL_INTERVAL = 15 - """ - Number of seconds to wait between requests to check job status. - """ - - DEFAULT_MAX_POLLS = 8 - """ - Maximum number of times to poll for job status before giving up. - """ - - def __init__( - self, - uri, - auth_token, - poll_interval=DEFAULT_POLL_INTERVAL, - max_polls=DEFAULT_MAX_POLLS, - ): - # type: (Union[Text, SplitResult], Optional[Text], int, int) -> None - """ - :param uri: - URI of the node to connect to. - ``https://` URIs are recommended! - - Note: Make sure the URI specifies the correct path! - - Example: - - - Incorrect: ``https://sandbox.iota:14265`` - - Correct: ``https://sandbox.iota:14265/api/v1/`` - - :param auth_token: - Authorization token used to authenticate requests. - - Contact the node's maintainer to obtain a token. - - If ``None``, the adapter will not include authorization - metadata with requests. - - :param poll_interval: - Number of seconds to wait between requests to check job - status. Must be a positive integer. - - Smaller values will cause the adapter to return a result - sooner (once the node completes the job), but it increases - traffic to the node (which may trip a rate limiter and/or - incur additional costs). - - :param max_polls: - Max number of times to poll for job status before giving up. - Must be a positive integer. - - This is effectively a timeout setting for asynchronous jobs; - multiply by ``poll_interval`` to get the timeout duration. - """ - super(SandboxAdapter, self).__init__(uri) - - if not (isinstance(auth_token, text_type) or (auth_token is None)): - raise with_context( - exc=TypeError( - '``auth_token`` must be a unicode string or ``None`` ' - '(``exc.context`` has more info).' - ), - - context={ - 'auth_token': auth_token, - }, - ) - - if auth_token == '': - raise with_context( - exc=ValueError( - 'Set ``auth_token=None`` if requests do not require ' - 'authorization (``exc.context`` has more info).', - ), - - context={ - 'auth_token': auth_token, - }, - ) - - if not isinstance(poll_interval, int): - raise with_context( - exc=TypeError( - '``poll_interval`` must be an int ' - '(``exc.context`` has more info).', - ), - - context={ - 'poll_interval': poll_interval, - }, - ) - - if poll_interval < 1: - raise with_context( - exc=ValueError( - '``poll_interval`` must be > 0 ' - '(``exc.context`` has more info).', - ), - - context={ - 'poll_interval': poll_interval, - }, - ) - - if not isinstance(max_polls, int): - raise with_context( - exc=TypeError( - '``max_polls`` must be an int ' - '(``exc.context`` has more info).', - ), - - context={ - 'max_polls': max_polls, - }, - ) - - if max_polls < 1: - raise with_context( - exc=ValueError( - '``max_polls`` must be > 0 ' - '(``exc.context`` has more info).', - ), - - context={ - 'max_polls': max_polls, - }, - ) - - self.auth_token = auth_token # type: Optional[Text] - self.poll_interval = poll_interval # type: int - self.max_polls = max_polls # type: int - - @property - def node_url(self): - return compat.urllib_parse.urlunsplit(( - self.uri.scheme, - self.uri.netloc, - self.uri.path.rstrip('/') + '/commands', - self.uri.query, - self.uri.fragment, - )) - - @property - def authorization_header(self): - # type: () -> Text - """ - Returns the value to use for the ``Authorization`` header. - """ - return 'token {auth_token}'.format(auth_token=self.auth_token) - - def get_jobs_url(self, job_id): - # type: (Text) -> Text - """ - Returns the URL to check job status. - - :param job_id: - The ID of the job to check. - """ - return compat.urllib_parse.urlunsplit(( - self.uri.scheme, - self.uri.netloc, - self.uri.path.rstrip('/') + '/jobs/' + job_id, - self.uri.query, - self.uri.fragment, - )) - - def send_request(self, payload, **kwargs): - # type: (dict, dict) -> dict - if self.auth_token: - kwargs.setdefault('headers', {}) - kwargs['headers']['Authorization'] = self.authorization_header - - return super(SandboxAdapter, self).send_request(payload, **kwargs) - - def _interpret_response(self, response, payload, expected_status): - # type: (Response, dict, Container[int]) -> dict - decoded = super(SandboxAdapter, self)._interpret_response( - response=response, - payload=payload, - expected_status={codes['ok'], codes['accepted']}, - ) - - # Check to see if the request was queued for asynchronous - # execution. - if response.status_code == codes['accepted']: - poll_count = 0 - while decoded['status'] in (STATUS_QUEUED, STATUS_RUNNING): - if poll_count >= self.max_polls: - raise with_context( - exc=BadApiResponse( - '``{command}`` job timed out after ' - '{duration} seconds ' - '(``exc.context`` has more info).'.format( - command=decoded['command'], - duration=self.poll_interval * self.max_polls, - ), - ), - - context={ - 'request': payload, - 'response': decoded, - }, - ) - - self._wait_to_poll() - poll_count += 1 - - poll_response = self._send_http_request( - headers={'Authorization': self.authorization_header}, - method='get', - payload=None, - url=self.get_jobs_url(decoded['id']), - ) - - decoded = super(SandboxAdapter, self)._interpret_response( - response=poll_response, - payload=payload, - expected_status={codes['ok']}, - ) - - if decoded['status'] == STATUS_FINISHED: - return decoded[ - '{command}Response'.format(command=decoded['command'])] - - raise with_context( - exc=BadApiResponse( - decoded.get('error', {}).get('message') - or 'Command {status}: {decoded}'.format( - decoded=decoded, - status=decoded['status'].lower(), - ), - ), - - context={ - 'request': payload, - 'response': decoded, - }, - ) - - return decoded - - def _wait_to_poll(self): - """ - Waits for 1 interval (according to :py:attr:`poll_interval`). - - Implemented as a separate method so that it can be mocked during - unit tests ("Do you bite your thumb at us, sir?"). - """ - sleep(self.poll_interval) diff --git a/test/adapter/sandbox_test.py b/test/adapter/sandbox_test.py deleted file mode 100644 index 4e377915..00000000 --- a/test/adapter/sandbox_test.py +++ /dev/null @@ -1,414 +0,0 @@ -# coding=utf-8 -from __future__ import absolute_import, division, print_function, \ - unicode_literals - -import json -from collections import deque -from unittest import TestCase - -from six import text_type - -from iota import BadApiResponse -from iota.adapter import API_VERSION -from iota.adapter.sandbox import SandboxAdapter -from test import mock -from test.adapter_test import create_http_response - - -class SandboxAdapterTestCase(TestCase): - def test_regular_command(self): - """ - Sending a non-sandbox command to the node. - """ - adapter = SandboxAdapter('https://localhost', 'ACCESS-TOKEN') - - expected_result = { - 'message': 'Hello, IOTA!', - } - - mocked_response = create_http_response(json.dumps(expected_result)) - mocked_sender = mock.Mock(return_value=mocked_response) - - payload = {'command': 'helloWorld'} - - # noinspection PyUnresolvedReferences - with mock.patch.object(adapter, '_send_http_request', mocked_sender): - result = adapter.send_request(payload) - - self.assertEqual(result, expected_result) - - mocked_sender.assert_called_once_with( - payload = json.dumps(payload), - url = adapter.node_url, - - # Auth token automatically added to the HTTP request. - headers = { - 'Authorization': 'token ACCESS-TOKEN', - 'Content-type': 'application/json', - 'X-IOTA-API-Version': API_VERSION, - }, - ) - - def test_sandbox_command_succeeds(self): - """ - Sending a sandbox command to the node. - """ - adapter = SandboxAdapter('https://localhost', 'ACCESS-TOKEN') - - expected_result = { - 'message': 'Hello, IOTA!', - } - - # Simulate responses from the node. - responses =\ - deque([ - # The first request creates the job. - # Note that the response has a 202 status. - create_http_response(status=202, content=json.dumps({ - 'id': '70fef55d-6933-49fb-ae17-ec5d02bc9117', - 'status': 'QUEUED', - 'createdAt': 1483574581, - 'startedAt': None, - 'finishedAt': None, - 'command': 'helloWorld', - - 'helloWorldRequest': { - 'command': 'helloWorld', - }, - })), - - # The job is still running when we poll. - create_http_response(json.dumps({ - 'id': '70fef55d-6933-49fb-ae17-ec5d02bc9117', - 'status': 'RUNNING', - 'createdAt': 1483574581, - 'startedAt': 1483574589, - 'finishedAt': None, - 'command': 'helloWorld', - - 'helloWorldRequest': { - 'command': 'helloWorld', - }, - })), - - # The job has finished by the next polling request. - create_http_response(json.dumps({ - 'id': '70fef55d-6933-49fb-ae17-ec5d02bc9117', - 'status': 'FINISHED', - 'createdAt': 1483574581, - 'startedAt': 1483574589, - 'finishedAt': 1483574604, - 'command': 'helloWorld', - - 'helloWorldRequest': { - 'command': 'helloWorld', - }, - - 'helloWorldResponse': expected_result, - })), - ]) - - # noinspection PyUnusedLocal - def _send_http_request(*args, **kwargs): - return responses.popleft() - - mocked_sender = mock.Mock(wraps=_send_http_request) - mocked_waiter = mock.Mock() - - # noinspection PyUnresolvedReferences - with mock.patch.object(adapter, '_send_http_request', mocked_sender): - # mock.Mock ``_wait_to_poll`` so that it returns immediately, instead - # of waiting for 15 seconds. Bad for production, good for tests. - # noinspection PyUnresolvedReferences - with mock.patch.object(adapter, '_wait_to_poll', mocked_waiter): - result = adapter.send_request({'command': 'helloWorld'}) - - self.assertEqual(result, expected_result) - - def test_sandbox_command_fails(self): - """ - A sandbox command fails after an interval. - """ - adapter = SandboxAdapter('https://localhost', 'ACCESS-TOKEN') - - error_message = "You didn't say the magic word!" - - # Simulate responses from the node. - responses =\ - deque([ - # The first request creates the job. - # Note that the response has a 202 status. - create_http_response(status=202, content=json.dumps({ - 'id': '70fef55d-6933-49fb-ae17-ec5d02bc9117', - 'status': 'QUEUED', - 'createdAt': 1483574581, - 'startedAt': None, - 'finishedAt': None, - 'command': 'helloWorld', - - 'helloWorldRequest': { - 'command': 'helloWorld', - }, - })), - - # The job is still running when we poll. - create_http_response(json.dumps({ - 'id': '70fef55d-6933-49fb-ae17-ec5d02bc9117', - 'status': 'RUNNING', - 'createdAt': 1483574581, - 'startedAt': 1483574589, - 'finishedAt': None, - 'command': 'helloWorld', - - 'helloWorldRequest': { - 'command': 'helloWorld', - }, - })), - - # The job has finished by the next polling request. - create_http_response(json.dumps({ - 'id': '70fef55d-6933-49fb-ae17-ec5d02bc9117', - 'status': 'FAILED', - 'createdAt': 1483574581, - 'startedAt': 1483574589, - 'finishedAt': 1483574604, - 'command': 'helloWorld', - - 'helloWorldRequest': { - 'command': 'helloWorld', - }, - - 'error': { - 'message': error_message, - }, - })), - ]) - - # noinspection PyUnusedLocal - def _send_http_request(*args, **kwargs): - return responses.popleft() - - mocked_sender = mock.Mock(wraps=_send_http_request) - mocked_waiter = mock.Mock() - - # noinspection PyUnresolvedReferences - with mock.patch.object(adapter, '_send_http_request', mocked_sender): - # mock.Mock ``_wait_to_poll`` so that it returns immediately, instead - # of waiting for 15 seconds. Bad for production, good for tests. - # noinspection PyUnresolvedReferences - with mock.patch.object(adapter, '_wait_to_poll', mocked_waiter): - with self.assertRaises(BadApiResponse) as context: - adapter.send_request({'command': 'helloWorld'}) - - self.assertEqual(text_type(context.exception), error_message) - - def test_regular_command_null_token(self): - """ - Sending commands to a sandbox that doesn't require authorization. - - This is generally not recommended, but the sandbox node may use - other methods to control access (e.g., listen only on loopback - interface, use IP address whitelist, etc.). - """ - # No access token. - adapter = SandboxAdapter('https://localhost', None) - - expected_result = { - 'message': 'Hello, IOTA!', - } - - mocked_response = create_http_response(json.dumps(expected_result)) - mocked_sender = mock.Mock(return_value=mocked_response) - - payload = {'command': 'helloWorld'} - - # noinspection PyUnresolvedReferences - with mock.patch.object(adapter, '_send_http_request', mocked_sender): - result = adapter.send_request(payload) - - self.assertEqual(result, expected_result) - - mocked_sender.assert_called_once_with( - payload = json.dumps(payload), - url = adapter.node_url, - - headers = { - # No auth token, so no Authorization header. - # 'Authorization': 'token ACCESS-TOKEN', - 'Content-type': 'application/json', - 'X-IOTA-API-Version': API_VERSION, - }, - ) - - def test_error_job_takes_too_long(self): - """ - A job takes too long to complete, and we lose interest. - """ - adapter =\ - SandboxAdapter( - uri = 'https://localhost', - auth_token = 'token', - poll_interval = 15, - max_polls = 2, - ) - - responses =\ - deque([ - # The first request creates the job. - create_http_response(status=202, content=json.dumps({ - 'id': '70fef55d-6933-49fb-ae17-ec5d02bc9117', - 'status': 'QUEUED', - 'createdAt': 1483574581, - 'startedAt': None, - 'finishedAt': None, - 'command': 'helloWorld', - - 'helloWorldRequest': { - 'command': 'helloWorld', - }, - })), - - # The next two times we poll, the job is still in progress. - create_http_response(json.dumps({ - 'id': '70fef55d-6933-49fb-ae17-ec5d02bc9117', - 'status': 'RUNNING', - 'createdAt': 1483574581, - 'startedAt': 1483574589, - 'finishedAt': None, - 'command': 'helloWorld', - - 'helloWorldRequest': { - 'command': 'helloWorld', - }, - })), - - create_http_response(json.dumps({ - 'id': '70fef55d-6933-49fb-ae17-ec5d02bc9117', - 'status': 'RUNNING', - 'createdAt': 1483574581, - 'startedAt': 1483574589, - 'finishedAt': None, - 'command': 'helloWorld', - - 'helloWorldRequest': { - 'command': 'helloWorld', - }, - })), - ]) - - # noinspection PyUnusedLocal - def _send_http_request(*args, **kwargs): - return responses.popleft() - - mocked_sender = mock.Mock(wraps=_send_http_request) - mocked_waiter = mock.Mock() - - # noinspection PyUnresolvedReferences - with mock.patch.object(adapter, '_send_http_request', mocked_sender): - # mock.Mock ``_wait_to_poll`` so that it returns immediately, instead - # of waiting for 15 seconds. Bad for production, good for tests. - # noinspection PyUnresolvedReferences - with mock.patch.object(adapter, '_wait_to_poll', mocked_waiter): - with self.assertRaises(BadApiResponse) as context: - adapter.send_request({'command': 'helloWorld'}) - - self.assertEqual( - text_type(context.exception), - - '``helloWorld`` job timed out after 30 seconds ' - '(``exc.context`` has more info).', - ) - - def test_error_non_200_response(self): - """ - The node sends back a non-200 response. - """ - adapter = SandboxAdapter('https://localhost', 'ACCESS-TOKEN') - - decoded_response = { - 'message': 'You have reached maximum request limit.', - } - - mocked_sender = mock.Mock(return_value=create_http_response( - status = 429, - content = json.dumps(decoded_response), - )) - - # noinspection PyUnresolvedReferences - with mock.patch.object(adapter, '_send_http_request', mocked_sender): - with self.assertRaises(BadApiResponse) as context: - adapter.send_request({'command': 'helloWorld'}) - - self.assertEqual( - text_type(context.exception), - '429 response from node: {decoded}'.format(decoded=decoded_response), - ) - - def test_error_auth_token_wrong_type(self): - """ - ``auth_token`` is not a string. - """ - with self.assertRaises(TypeError): - # Nope; it has to be a unicode string. - # noinspection PyTypeChecker - SandboxAdapter('https://localhost', b'not valid') - - def test_error_auth_token_empty(self): - """ - ``auth_token`` is an empty string. - """ - with self.assertRaises(ValueError): - # If the node does not require authorization, use ``None``. - SandboxAdapter('https://localhost', '') - - def test_error_poll_interval_null(self): - """ - ``poll_interval`` is ``None``. - - The implications of allowing this are cool to think about... - but not implemented yet. - """ - with self.assertRaises(TypeError): - # noinspection PyTypeChecker - SandboxAdapter('https://localhost', 'token', None) - - def test_error_poll_interval_wrong_type(self): - """ - ``poll_interval`` is not an int. - """ - with self.assertRaises(TypeError): - # ``poll_interval`` must be an int. - # noinspection PyTypeChecker - SandboxAdapter('https://localhost', 'token', 42.0) - - def test_error_poll_interval_too_small(self): - """ - ``poll_interval`` is < 1. - """ - with self.assertRaises(ValueError): - SandboxAdapter('https://localhost', 'token', 0) - - def test_error_max_polls_null(self): - """ - ``max_polls`` is None. - """ - with self.assertRaises(TypeError): - # noinspection PyTypeChecker - SandboxAdapter('https://localhost', 'token', max_polls=None) - - def test_max_polls_wrong_type(self): - """ - ``max_polls`` is not an int. - """ - with self.assertRaises(TypeError): - # ``max_polls`` must be an int. - # noinspection PyTypeChecker - SandboxAdapter('https://localhost', 'token', max_polls=2.0) - - def test_max_polls_too_small(self): - """ - ``max_polls`` is < 1. - """ - with self.assertRaises(ValueError): - # noinspection PyTypeChecker - SandboxAdapter('https://localhost', 'token', max_polls=0) diff --git a/test/local_pow_test.py b/test/local_pow_test.py index ab494785..5b462311 100644 --- a/test/local_pow_test.py +++ b/test/local_pow_test.py @@ -4,7 +4,6 @@ from iota import Iota, TryteString, TransactionHash, TransactionTrytes, \ HttpAdapter, MockAdapter -from iota.adapter.sandbox import SandboxAdapter from iota.adapter.wrappers import RoutingWrapper from unittest import TestCase import sys @@ -109,24 +108,6 @@ def test_mock_adapter(self): self.mwm) self.assertTrue(mocked_ccurl.called) self.assertEqual(result['trytes'], self.bundle) - - def test_sandbox_adapter(self): - """ - Test if local_pow feature works with SandboxAdapter. - """ - # Note that we need correct return value to pass the - # response filter. - with patch('pow.ccurl_interface.attach_to_tangle', - MagicMock(return_value=self.bundle)) as mocked_ccurl: - api = Iota(SandboxAdapter('https://sandbox.iota:14265/api/v1/', auth_token=None), - local_pow=True) - result = api.attach_to_tangle( - self.trunk, - self.branch, - self.bundle, - self.mwm) - self.assertTrue(mocked_ccurl.called) - self.assertEqual(result['trytes'], self.bundle) def test_routing_wrapper(self): """