7

I'm pretty new at unit testing using unit test module and i need to test a class which using the threading module. I didn't really find any solution on what is the best way to unit a class using threads. I start to test if the thread was called (i.e. alive) but then I saw there is no method to kill a thread. I need as week to be sure the code in thread in doing what it's supposed to do. Do i need to use the thread for that or should I test only the function by itself?

Thanks for your help!

Here is my code (the interesting part):

class VideoLoader(Node):
   # some code

   def execute(self):
     self.thread1 = Thread(target = run_resizer, args = (self,))

def run_resizer(node):
    while True:
        while not len(node.frames_to_resize):
            time.sleep(0.001)      
        node.frames.append(resize({'height': 400, 'width': 500))
I159
  • 29,741
  • 31
  • 97
  • 132
Romanzo Criminale
  • 1,289
  • 2
  • 14
  • 21
  • Related/duplicate? http://stackoverflow.com/questions/12159/how-should-i-unit-test-threaded-code – Iguananaut Dec 18 '13 at 00:43
  • Thanks I actually haven't seen this post. But the only stuff I understand is that it is hard to test multi threading even if my code is not really complex. – Romanzo Criminale Dec 18 '13 at 01:26

1 Answers1

0

I had the same problem, but I couldn't find a good solution, so I did a workaround. I completed your code to be independently executable. I did make one change: I'm using composition instead of inheritance (as recommended by the Gang of Four in the Design Patterns book).

# loader.py
from threading import Thread
import time

class VideoLoader:
    def __init__(self, node):
        self.thread1 = None
        self.node = node
        self.is_resizer_active = False

    def execute(self):
        self.thread1 = Thread(target=self.run_resizer)
        self.thread1.start()

    def run_resizer(self):
        node = self.node
        self.is_resizer_active = True
        while self.is_resizer_active: # need some way of stopping it eventually
            while not len(node.frames_to_resize):
                time.sleep(0.001)
            node.frames.append(resize({'height': 400, 'width': 500}))

def resize(*a, **kw):pass # just to have an executable test


import unittest
from unittest.mock import MagicMock
from loader import VideoLoader

class TestVideoLoader(unittest.TestCase):
    def test_ensure_frame_resized(self):
        fake_node = MagicMock()
        fake_node.frames_to_resize = [1] # to make sure you fall through the 2nd while loop
        loader = VideoLoader(fake_node)
        loader.execute()
        loader.is_resizer_active = False
        loader.thread1.join()
        fake_node.frames.append.assert_called()

if __name__ == '__main__':
    unittest.main()
PotatoPeeler
  • 43
  • 3