Time.sleep () in Python unit test when using threads
I am learning about threads in Python and unit testing thinking will suit my needs.
Using this http://www.tutorialspoint.com/python/python_multithreading.htm as a starting point. However, when I use time.sleep () in my function, the test seems to return from that code block.
import unittest
import thread
import time
class Test(unittest.TestCase):
def test_threads(self):
thread.start_new_thread(self.print_time, ("Thread 1", 1))
def print_time(self, threadName, delay):
count = 0
while count < 5:
time.sleep(delay)
count += 1
print "%s: %s" % (threadName, time.ctime(time.time()))
if __name__ == "__main__":
unittest.main()
Using this code will output nothing. Getting rid of the time.sleep (delay) outputs:
Thread 1: Mon Aug 03 11:36:56 2015
Thread 1: Mon Aug 03 11:36:56 2015
Thread 1: Mon Aug 03 11:36:56 2015
Thread 1: Mon Aug 03 11:36:56 2015
Thread 1: Mon Aug 03 11:36:56 2015
How can I use time.sleep () when I am using a unit test in Python?
source to share
The behavior you are seeing is due to the fact that your script ended before the stream had time to print its output. As per the docs in the Caveats section:
When the main thread exits, the system determines if other threads survive
Thus, it is quite possible that your thread will be killed during the first sleep (). Even if the thread survives, I doubt you will see the output in the same document:
When the main thread exits [..], standard I / O files are not flushed
Without time.sleep(delay)
, apparently print
ready before the script finishes.
A naive way to fix this is to have your test method hibernate () for the time it takes for the subprocess to complete, in order to replace your test method with
def test_threads(self):
Thread(target = self.print_time, args = ("Thread 1", 1)).start()
time.sleep(6)
Usually if you are using a module Threading
for this type of work. When you use it to start threads in non-daemon mode (the default), the calling thread only stops after all running threads have finished.
source to share
Hmm. I get what you get.
You can use threading.Thread()
?
Code:
import unittest
from threading import Thread
import time
class Test(unittest.TestCase):
def test_threads(self):
Thread(target = self.print_time, args = ("Thread 1", 1)).start()
def print_time(self, threadName, delay):
count = 0
while count < 5:
time.sleep(delay)
count += 1
print "%s: %s" % (threadName, time.ctime(time.time()))
if __name__ == "__main__":
unittest.main()
Output:
----------------------------------------------------------------------
Ran 1 test in 0.001s
OK
Thread 1: Mon Aug 03 13:07:46 2015
Thread 1: Mon Aug 03 13:07:47 2015
Thread 1: Mon Aug 03 13:07:48 2015
Thread 1: Mon Aug 03 13:07:49 2015
Thread 1: Mon Aug 03 13:07:50 2015
Note, though, that the test completes before the thread completes.
source to share