Testing exception is thrown when queue is empty

Question

When I attempt to test the catching of an exception with a spsecific message when a Queue is empty the test fails.

Here is the Queue class:

import time
from multiprocessing import Lock, Process, Queue, Pool

class QueueFun():

    def writing_queue(self, work_tasks, name):
        while True:
            print("Writing to queue")
            work_tasks.put('1')
            time.sleep(3)

    def read_queue(self, work_tasks, name):
        while True:
            print("Reading from queue", work_tasks.empty())
            if work_tasks.empty():
                raise ValueError('Queue should not be empty')
            item = work_tasks.get()
            time.sleep(1)

if __name__ == '__main__':
    work_tasks = Queue()
    queue_fun = QueueFun()
    print('setup done')

    write_queue_process = Process(target=queue_fun.writing_queue, args=(work_tasks, "write_to_queue_process"))
    print('write_queue_process')

    write_queue_process.start()

    read_queue_process_1 = Process(target=queue_fun.read_queue, args=(work_tasks, "read_from_queue_process_1"))
    read_queue_process_1.start()
    read_queue_process_2 = Process(target=queue_fun.read_queue, args=(work_tasks, "read_from_queue_process_2"))
    read_queue_process_2.start()

    write_queue_process.join()
    read_queue_process_1.join()
    read_queue_process_2.join()

And here is the test:

import unittest
from queue import Queue

from playground.QueueFun import QueueFun


class QueueFunTests(unittest.TestCase):
    def test_empty_queue(self):
        q = QueueFun()
        work_tasks = Queue()
        self.assertRaises(ValueError('Queue should not be empty'), q.read_queue(work_tasks, "1"))

Test fails with :

    raise ValueError('Queue should not be empty')
ValueError: Queue should not be empty

Have I not implemented the test correctly ?

@efont yes, that answers my question. – Adrian Aug 14 '20 at 13:14 — Adrian, Aug 14 '20 at 13:14

score 0 · Answer 1 · answered Aug 14 '20 at 12:39

This works as expected:

class QueueFunTests(unittest.TestCase):

    def test_empty_queue(self):
        q = QueueFun()
        work_tasks = Queue()
        with self.assertRaises(ValueError) as ctx:
            q.read_queue(work_tasks, "1")
        self.assertEqual("Queue should not be empty", str(ctx.exception))

Testing exception is thrown when queue is empty

1 Answers1