Class RecordingSubscriber

tests/utils/s3transfer/__init__.py:258–277

class RecordingSubscriber(BaseSubscriber):
    def __init__(self):
        self.on_queued_calls = []
        self.on_progress_calls = []
        self.on_done_calls = []

    def on_queued(self, **kwargs):
        self.on_queued_calls.append(kwargs)

    def on_progress(self, **kwargs):
        self.on_progress_calls.append(kwargs)

    def on_done(self, **kwargs):
        self.on_done_calls.append(kwargs)

    def calculate_bytes_seen(self, **kwargs):
        amount_seen = 0
        for call in self.on_progress_calls:
            amount_seen += call['bytes_transferred']
        return amount_seen


class TransferCoordinatorWithInterrupt(TransferCoordinator):