def receive_messages(self, queue_name, count, wait_seconds_timeout, visibility_timeout):
        """
        Attempt to retrieve visible messages from a queue.

        If a message was read by client and not deleted it is considered to be
        "inflight" and cannot be read. We make attempts to obtain ``count``
        messages but we may return less if messages are in-flight or there
        are simple not enough messages in the queue.

        :param string queue_name: The name of the queue to read from.
        :param int count: The maximum amount of messages to retrieve.
        :param int visibility_timeout: The number of seconds the message should remain invisible to other queue readers.
        """
        queue = self.get_queue(queue_name)
        result = []

        polling_end = unix_time() + wait_seconds_timeout

        # queue.messages only contains visible messages
        while True:
            for message in queue.messages:
                if not message.visible:
                    continue
                message.mark_received(
                    visibility_timeout=visibility_timeout
                )
                result.append(message)
                if len(result) >= count:
                    break

            if result or unix_time() > polling_end:
                break

        return result

    def receive_messages(self, queue_name, count, wait_seconds_timeout, visibility_timeout):
        """
        Attempt to retrieve visible messages from a queue.

        If a message was read by client and not deleted it is considered to be
        "inflight" and cannot be read. We make attempts to obtain ``count``
        messages but we may return less if messages are in-flight or there
        are simple not enough messages in the queue.

        :param string queue_name: The name of the queue to read from.
        :param int count: The maximum amount of messages to retrieve.
        :param int visibility_timeout: The number of seconds the message should remain invisible to other queue readers.
        :param int wait_seconds_timeout:  The duration (in seconds) for which the call waits for a message to arrive in
         the queue before returning. If a message is available, the call returns sooner than WaitTimeSeconds
        """
        queue = self.get_queue(queue_name)
        result = []

        polling_end = unix_time() + wait_seconds_timeout

        # queue.messages only contains visible messages
        while True:

            if result or (wait_seconds_timeout and unix_time() > polling_end):
                break

            if len(queue.messages) == 0:
                # we want to break here, otherwise it will be an infinite loop
                if wait_seconds_timeout == 0:
                    break

                import time
                time.sleep(0.001)
                continue

            messages_to_dlq = []
            for message in queue.messages:
                if not message.visible:
                    continue
                if queue.dead_letter_queue is not None and message.approximate_receive_count >= queue.redrive_policy['maxReceiveCount']:
                    messages_to_dlq.append(message)
                    continue

                message.mark_received(
                    visibility_timeout=visibility_timeout
                )
                result.append(message)
                if len(result) >= count:
                    break

            for message in messages_to_dlq:
                queue._messages.remove(message)
                queue.dead_letter_queue.add_message(message)

        return result

 def create_account_status(self):
     return {
         'CreateAccountStatus': {
             'Id': self.create_account_status_id,
             'AccountName': self.name,
             'State': 'SUCCEEDED',
             'RequestedTimestamp': unix_time(self.create_time),
             'CompletedTimestamp': unix_time(self.create_time),
             'AccountId': self.id,
         }
     }

    def __init__(self, name, region, **kwargs):
        self.name = name
        self.visibility_timeout = int(kwargs.get('VisibilityTimeout', 30))
        self.region = region

        self._messages = []

        now = unix_time()

        # kwargs can also have:
        # [Policy, RedrivePolicy]
        self.fifo_queue = kwargs.get('FifoQueue', 'false') == 'true'
        self.content_based_deduplication = kwargs.get('ContentBasedDeduplication', 'false') == 'true'
        self.kms_master_key_id = kwargs.get('KmsMasterKeyId', 'alias/aws/sqs')
        self.kms_data_key_reuse_period_seconds = int(kwargs.get('KmsDataKeyReusePeriodSeconds', 300))
        self.created_timestamp = now
        self.delay_seconds = int(kwargs.get('DelaySeconds', 0))
        self.last_modified_timestamp = now
        self.maximum_message_size = int(kwargs.get('MaximumMessageSize', 64 << 10))
        self.message_retention_period = int(kwargs.get('MessageRetentionPeriod', 86400 * 4))  # four days
        self.queue_arn = 'arn:aws:sqs:{0}:123456789012:{1}'.format(self.region, self.name)
        self.receive_message_wait_time_seconds = int(kwargs.get('ReceiveMessageWaitTimeSeconds', 0))

        # wait_time_seconds will be set to immediate return messages
        self.wait_time_seconds = int(kwargs.get('WaitTimeSeconds', 0))

        # Check some conditions
        if self.fifo_queue and not self.name.endswith('.fifo'):
            raise MessageAttributesInvalid('Queue name must end in .fifo for FIFO queues')

 def __init__(self, partition_key, data, sequence_number, explicit_hash_key):
     self.partition_key = partition_key
     self.data = data
     self.sequence_number = sequence_number
     self.explicit_hash_key = explicit_hash_key
     self.created_at_datetime = datetime.datetime.utcnow()
     self.created_at = unix_time(self.created_at_datetime)

def test_list_closed_workflow_executions():
    conn = setup_swf_environment()
    # Leave one workflow execution open to make sure it isn't displayed
    conn.start_workflow_execution(
        'test-domain', 'uid-abcd1234', 'test-workflow', 'v1.0'
    )
    # One closed workflow execution
    run_id = conn.start_workflow_execution(
        'test-domain', 'uid-abcd12345', 'test-workflow', 'v1.0'
    )['runId']
    conn.terminate_workflow_execution('test-domain', 'uid-abcd12345',
                                      details='some details',
                                      reason='a more complete reason',
                                      run_id=run_id)

    yesterday = datetime.now() - timedelta(days=1)
    oldest_date = unix_time(yesterday)
    response = conn.list_closed_workflow_executions(
        'test-domain',
        start_oldest_date=oldest_date,
        workflow_id='test-workflow')
    execution_infos = response['executionInfos']
    len(execution_infos).should.equal(1)
    open_workflow = execution_infos[0]
    open_workflow['workflowType'].should.equal({'version': 'v1.0',
                                                'name': 'test-workflow'})
    open_workflow.should.contain('startTimestamp')
    open_workflow['execution']['workflowId'].should.equal('uid-abcd12345')
    open_workflow['execution'].should.contain('runId')
    open_workflow['cancelRequested'].should.be(False)
    open_workflow['executionStatus'].should.equal('CLOSED')

    def __init__(self, name, region, **kwargs):
        self.name = name
        self.region = region
        self.tags = {}
        self.permissions = {}

        self._messages = []

        now = unix_time()
        self.created_timestamp = now
        self.queue_arn = 'arn:aws:sqs:{0}:123456789012:{1}'.format(self.region,
                                                                   self.name)
        self.dead_letter_queue = None

        # default settings for a non fifo queue
        defaults = {
            'ContentBasedDeduplication': 'false',
            'DelaySeconds': 0,
            'FifoQueue': 'false',
            'KmsDataKeyReusePeriodSeconds': 300,  # five minutes
            'KmsMasterKeyId': None,
            'MaximumMessageSize': int(64 << 10),
            'MessageRetentionPeriod': 86400 * 4,  # four days
            'Policy': None,
            'ReceiveMessageWaitTimeSeconds': 0,
            'RedrivePolicy': None,
            'VisibilityTimeout': 30,
        }

        defaults.update(kwargs)
        self._set_attributes(defaults, now)

        # Check some conditions
        if self.fifo_queue and not self.name.endswith('.fifo'):
            raise MessageAttributesInvalid('Queue name must end in .fifo for FIFO queues')

 def describe(self):
     results = {
         "Table": {
             "CreationDateTime": unix_time(self.created_at),
             "KeySchema": {
                 "HashKeyElement": {
                     "AttributeName": self.hash_key_attr,
                     "AttributeType": self.hash_key_type
                 },
             },
             "ProvisionedThroughput": {
                 "ReadCapacityUnits": self.read_capacity,
                 "WriteCapacityUnits": self.write_capacity
             },
             "TableName": self.name,
             "TableStatus": "ACTIVE",
             "ItemCount": len(self),
             "TableSizeBytes": 0,
         }
     }
     if self.has_range_key:
         results["Table"]["KeySchema"]["RangeKeyElement"] = {
             "AttributeName": self.range_key_attr,
             "AttributeType": self.range_key_type
         }
     return results

 def complete(self, event_id, result=None):
     self.execution_status = "CLOSED"
     self.close_status = "COMPLETED"
     self.close_timestamp = unix_time()
     self._add_event(
         "WorkflowExecutionCompleted",
         decision_task_completed_event_id=event_id,
         result=result,
     )

 def fail(self, event_id, details=None, reason=None):
     # TODO: implement lenght constraints on details/reason
     self.execution_status = "CLOSED"
     self.close_status = "FAILED"
     self.close_timestamp = unix_time()
     self._add_event(
         "WorkflowExecutionFailed",
         decision_task_completed_event_id=event_id,
         details=details,
         reason=reason,
     )

 def describe(self):
     return {
         'Account': {
             'Id': self.id,
             'Arn': self.arn,
             'Email': self.email,
             'Name': self.name,
             'Status': self.status,
             'JoinedMethod': self.joined_method,
             'JoinedTimestamp': unix_time(self.create_time),
         }
     }

 def start(self):
     self.start_timestamp = unix_time()
     self._add_event(
         "WorkflowExecutionStarted",
         child_policy=self.child_policy,
         execution_start_to_close_timeout=self.execution_start_to_close_timeout,
         # TODO: fix this hardcoded value
         parent_initiated_event_id=0,
         task_list=self.task_list,
         task_start_to_close_timeout=self.task_start_to_close_timeout,
         workflow_type=self.workflow_type,
     )
     self.schedule_decision_task()

 def describe(self):
     results = {
         'Table': {
             'AttributeDefinitions': self.attr,
             'ProvisionedThroughput': self.throughput,
             'TableSizeBytes': 0,
             'TableName': self.name,
             'TableStatus': 'ACTIVE',
             'KeySchema': self.schema,
             'ItemCount': len(self),
             'CreationDateTime': unix_time(self.created_at),
             'GlobalSecondaryIndexes': [index for index in self.global_indexes],
         }
     }
     return results

 def to_dict(self):
     key_dict = {
         "KeyMetadata": {
             "AWSAccountId": self.account_id,
             "Arn": self.arn,
             "CreationDate": "%d" % unix_time(),
             "Description": self.description,
             "Enabled": self.enabled,
             "KeyId": self.id,
             "KeyUsage": self.key_usage,
             "KeyState": self.key_state,
         }
     }
     if self.key_state == 'PendingDeletion':
         key_dict['KeyMetadata']['DeletionDate'] = iso_8601_datetime_without_milliseconds(self.deletion_date)
     return key_dict

 def describe(self):
     results = {
         "Table": {
             "AttributeDefinitions": self.attr,
             "ProvisionedThroughput": self.throughput,
             "TableSizeBytes": 0,
             "TableName": self.name,
             "TableStatus": "ACTIVE",
             "KeySchema": self.schema,
             "ItemCount": len(self),
             "CreationDateTime": unix_time(self.created_at),
             "GlobalSecondaryIndexes": [index for index in self.global_indexes],
             "LocalSecondaryIndexes": [index for index in self.indexes],
         }
     }
     return results

 def __init__(self, activity_id, activity_type, scheduled_event_id,
              workflow_execution, timeouts, input=None):
     self.activity_id = activity_id
     self.activity_type = activity_type
     self.details = None
     self.input = input
     self.last_heartbeat_timestamp = unix_time()
     self.scheduled_event_id = scheduled_event_id
     self.started_event_id = None
     self.state = "SCHEDULED"
     self.task_token = str(uuid.uuid4())
     self.timeouts = timeouts
     self.timeout_type = None
     self.workflow_execution = workflow_execution
     # this is *not* necessarily coherent with workflow execution history,
     # but that shouldn't be a problem for tests
     self.scheduled_at = datetime.utcnow()

    def __init__(self, name, visibility_timeout, wait_time_seconds, region):
        self.name = name
        self.visibility_timeout = visibility_timeout or 30
        self.region = region

        # wait_time_seconds will be set to immediate return messages
        self.wait_time_seconds = wait_time_seconds or 0
        self._messages = []

        now = unix_time()

        self.created_timestamp = now
        self.delay_seconds = 0
        self.last_modified_timestamp = now
        self.maximum_message_size = 64 << 10
        self.message_retention_period = 86400 * 4  # four days
        self.queue_arn = 'arn:aws:sqs:{0}:123456789012:{1}'.format(self.region, self.name)
        self.receive_message_wait_time_seconds = 0

 def describe(self, base_key='TableDescription'):
     results = {
         base_key: {
             'AttributeDefinitions': self.attr,
             'ProvisionedThroughput': self.throughput,
             'TableSizeBytes': 0,
             'TableName': self.name,
             'TableStatus': 'ACTIVE',
             'TableArn': self.table_arn,
             'KeySchema': self.schema,
             'ItemCount': len(self),
             'CreationDateTime': unix_time(self.created_at),
             'GlobalSecondaryIndexes': [index for index in self.global_indexes],
             'LocalSecondaryIndexes': [index for index in self.indexes],
         }
     }
     if self.stream_specification and self.stream_specification['StreamEnabled']:
         results[base_key]['StreamSpecification'] = self.stream_specification
         if self.latest_stream_label:
             results[base_key]['LatestStreamLabel'] = self.latest_stream_label
             results[base_key]['LatestStreamArn'] = self.table_arn + '/stream/' + self.latest_stream_label
     return results

 def __init__(self, event_id, event_type, event_timestamp=None, **kwargs):
     if event_type not in SUPPORTED_HISTORY_EVENT_TYPES:
         raise NotImplementedError(
             "HistoryEvent does not implement attributes for type '{0}'".format(event_type)
         )
     self.event_id = event_id
     self.event_type = event_type
     if event_timestamp:
         self.event_timestamp = event_timestamp
     else:
         self.event_timestamp = unix_time()
     # pre-populate a dict: {"camelCaseKey": value}
     self.event_attributes = {}
     for key, value in kwargs.items():
         if value:
             camel_key = underscores_to_camelcase(key)
             if key == "task_list":
                 value = {"name": value}
             elif key == "workflow_type":
                 value = {"name": value.name, "version": value.version}
             elif key == "activity_type":
                 value = value.to_short_dict()
             self.event_attributes[camel_key] = value

    def _set_attributes(self, attributes, now=None):
        if not now:
            now = unix_time()

        integer_fields = ('DelaySeconds', 'KmsDataKeyreusePeriodSeconds',
                          'MaximumMessageSize', 'MessageRetentionPeriod',
                          'ReceiveMessageWaitTime', 'VisibilityTimeout')
        bool_fields = ('ContentBasedDeduplication', 'FifoQueue')

        for key, value in six.iteritems(attributes):
            if key in integer_fields:
                value = int(value)
            if key in bool_fields:
                value = value == "true"

            if key == 'RedrivePolicy' and value is not None:
                continue

            setattr(self, camelcase_to_underscores(key), value)

        if attributes.get('RedrivePolicy', None):
            self._setup_dlq(attributes['RedrivePolicy'])

        self.last_modified_timestamp = now