How to use pyqs - 10 common examples

def test_process_worker_with_parent_process_alive_and_should_exit(os):
    """
    Test worker processes exit when parent is alive and shutdown is set
    """
    # Setup PPID
    os.getppid.return_value = 1234

    # When I have a parent process, and shutdown is set
    worker = ProcessWorker("foo", INTERVAL, parent_id=1)
    worker.process_message = Mock()
    worker.shutdown()

    # Then I return from run()
    worker.run().should.be.none

def test_process_worker_with_parent_process_dead_and_should_not_exit(os):
    """
    Test worker processes exit when parent is dead and shutdown is not set
    """
    # Setup PPID
    os.getppid.return_value = 1

    # When I have no parent process, and shutdown is not set
    worker = ProcessWorker("foo", INTERVAL, parent_id=1)
    worker.process_message = Mock()

    # Then I return from run()
    worker.run().should.be.none

def test_process_worker_with_parent_process_alive_and_should_not_exit(os):
    """
    Test worker processes do not exit when parent is alive and shutdown
    is not set
    """
    # Setup PPID
    os.getppid.return_value = 1

    # Setup dummy read_message
    def process_message():
        raise Exception("Called")

    # When I have a parent process, and shutdown is not set
    worker = ProcessWorker("foo", INTERVAL, parent_id=1)
    worker.process_message = process_message

    # Then process_message() is reached
    worker.run.when.called_with().should.throw(Exception, "Called")

def test_read_worker_with_parent_process_dead_and_should_not_exit(os):
    """
    Test read workers exit when parent is dead and shutdown is not set
    """
    # Setup SQS Queue
    conn = boto3.client('sqs', region_name='us-east-1')
    queue_url = conn.create_queue(QueueName="tester")['QueueUrl']

    # Setup PPID
    os.getppid.return_value = 123

    # Setup internal queue
    q = Queue(1)

    # When I have no parent process, and shutdown is not set
    worker = ReadWorker(queue_url, q, BATCHSIZE, parent_id=1)
    worker.read_message = Mock()

    # Then I return from run()
    worker.run().should.be.none

"ReceiptHandle": "receipt-1234",
    }

    # Add message to internal queue
    internal_queue = Queue()
    internal_queue.put(
        {
            "queue": queue_url,
            "message": message,
            "start_time": time.time(),
            "timeout": 30,
        }
    )

    # Process message
    worker = ProcessWorker(internal_queue, INTERVAL, parent_id=1)
    worker.process_message()

    # Check output
    kwargs = json.loads(message['Body'])['kwargs']
    expected_result = (
        u"Processed task tests.tasks.index_incrementer in 0.0000 seconds "
        "with args: [] and kwargs: {}".format(kwargs)
    )
    logger.handlers[0].messages['info'].should.equal([expected_result])

"message": message,
            "start_time": time.time(),
            "timeout": 30,
        }
    )
    internal_queue.put(
        {
            "queue": queue_url,
            "message": message,
            "start_time": time.time(),
            "timeout": 30,
        }
    )

    # When I Process messages
    worker = ProcessWorker(internal_queue, INTERVAL, parent_id=1)
    worker._messages_to_process_before_shutdown = 2

    # Then I return from run()
    worker.run().should.be.none

    # With messages still on the queue
    internal_queue.empty().should.be.false
    internal_queue.full().should.be.false

"ReceiptHandle": "receipt-1234",
    }

    # Add message to internal queue with timeout of 0 that started long ago
    internal_queue = Queue()
    internal_queue.put(
        {
            "queue": queue_url,
            "message": message,
            "start_time": 0,
            "timeout": 0,
        }
    )

    # When I process the message
    worker = ProcessWorker(internal_queue, INTERVAL, parent_id=1)
    worker.process_message()

    # Then I get an error about exceeding the visibility timeout
    kwargs = json.loads(message['Body'])['kwargs']
    msg1 = (
        "Discarding task tests.tasks.index_incrementer with args: [] "
        "and kwargs: {} due to exceeding "
        "visibility timeout"
    ).format(kwargs)  # noqa
    logger.handlers[0].messages['warning'][0].lower().should.contain(
        msg1.lower())

def test_manager_start_and_stop():
    """
    Test managing process can start and stop child processes
    """
    conn = boto3.client('sqs', region_name='us-east-1')
    conn.create_queue(QueueName="email")

    manager = ManagerWorker(
        queue_prefixes=['email'], worker_concurrency=2, interval=1,
        batchsize=10,
    )

    len(manager.worker_children).should.equal(2)

    manager.worker_children[0].is_alive().should.equal(False)
    manager.worker_children[1].is_alive().should.equal(False)

    manager.start()

    manager.worker_children[0].is_alive().should.equal(True)
    manager.worker_children[1].is_alive().should.equal(True)

    manager.stop()

def test_worker_to_large_batch_size():
    """
    Test workers with too large of a batch size
    """
    BATCHSIZE = 10000
    CONCURRENCY = 1
    QUEUE_PREFIX = "tester"
    INTERVAL = 0.0
    conn = boto3.client('sqs', region_name='us-east-1')
    conn.create_queue(QueueName="tester")['QueueUrl']

    worker = ManagerWorker(QUEUE_PREFIX, CONCURRENCY, INTERVAL, BATCHSIZE)
    worker.batchsize.should.equal(MESSAGE_DOWNLOAD_BATCH_SIZE)

Test managing process handles OS signals
    """

    # Setup SQS Queue
    conn = boto3.client('sqs', region_name='us-east-1')
    conn.create_queue(QueueName="tester")

    # Mock out sys.exit
    sys.exit = Mock()

    # Have our inner method send our signal
    def process_counts():
        os.kill(os.getpid(), signal.SIGTERM)

    # Setup Manager
    manager = ManagerWorker(
        queue_prefixes=["tester"], worker_concurrency=1, interval=1,
        batchsize=10,
    )
    manager.process_counts = process_counts
    manager._graceful_shutdown = MagicMock()

    # When we start and trigger a signal
    manager.start()
    manager.sleep()

    # Then we exit
    sys.exit.assert_called_once_with(0)