Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
self.subscriber = RecordingSubscriber()
self.second_subscriber = RecordingSubscriber()
self.call_args = CallArgs(subscribers=[
self.subscriber, self.second_subscriber]
)
self.transfer_meta = TransferMeta(self.call_args)
self.transfer_future = TransferFuture(self.transfer_meta)
def setUp(self):
self.transfer_future = mock.Mock(spec=TransferFuture)
self.transfer_meta = TransferMeta()
self.transfer_future.meta = self.transfer_meta
def test_returns_future_with_meta(self):
self._setup_default_stubbed_responses()
future = self.method(**self.create_call_kwargs())
# The result is called so we ensure that the entire process executes
# before we try to clean up resources in the tearDown.
future.result()
# Assert the return value is a future with metadata associated to it.
self.assertIsInstance(future, TransferFuture)
self.assertIsInstance(future.meta, TransferMeta)
def _get_future_with_components(self, call_args):
transfer_id = self._id_counter
# Creates a new transfer future along with its components
transfer_coordinator = TransferCoordinator(transfer_id=transfer_id)
# Track the transfer coordinator for transfers to manage.
self._coordinator_controller.add_transfer_coordinator(
transfer_coordinator)
# Also make sure that the transfer coordinator is removed once
# the transfer completes so it does not stick around in memory.
transfer_coordinator.add_done_callback(
self._coordinator_controller.remove_transfer_coordinator,
transfer_coordinator)
components = {
'meta': TransferMeta(call_args, transfer_id=transfer_id),
'coordinator': transfer_coordinator
}
transfer_future = TransferFuture(**components)
return transfer_future, components
def _get_future_with_components(self, call_args):
transfer_id = self._id_counter
# Creates a new transfer future along with its components
transfer_coordinator = TransferCoordinator(transfer_id=transfer_id)
# Track the transfer coordinator for transfers to manage.
self._coordinator_controller.add_transfer_coordinator(
transfer_coordinator)
# Also make sure that the transfer coordinator is removed once
# the transfer completes so it does not stick around in memory.
transfer_coordinator.add_done_callback(
self._coordinator_controller.remove_transfer_coordinator,
transfer_coordinator)
components = {
'meta': TransferMeta(call_args, transfer_id=transfer_id),
'coordinator': transfer_coordinator
}
transfer_future = TransferFuture(**components)
return transfer_future, components
def _get_future_with_components(self, call_args):
transfer_id = self._id_counter
# Creates a new transfer future along with its components
transfer_coordinator = TransferCoordinator(transfer_id=transfer_id)
# Track the transfer coordinator for transfers to manage.
self._coordinator_controller.add_transfer_coordinator(
transfer_coordinator)
# Also make sure that the transfer coordinator is removed once
# the transfer completes so it does not stick around in memory.
transfer_coordinator.add_done_callback(
self._coordinator_controller.remove_transfer_coordinator,
transfer_coordinator)
components = {
'meta': TransferMeta(call_args, transfer_id=transfer_id),
'coordinator': transfer_coordinator
}
transfer_future = TransferFuture(**components)
return transfer_future, components
def __init__(self, meta=None, coordinator=None):
"""The future associated to a submitted transfer request
:type meta: TransferMeta
:param meta: The metadata associated to the request. This object
is visible to the requester.
:type coordinator: TransferCoordinator
:param coordinator: The coordinator associated to the request. This
object is not visible to the requester.
"""
self._meta = meta
if meta is None:
self._meta = TransferMeta()
self._coordinator = coordinator
if coordinator is None:
self._coordinator = TransferCoordinator()