[TIP] How do you unit test threading classes?

[Eric Larson]

albert <albertpython at 126.com> writes:

> Hi all, I'm a little afraid to test threading classes, and don't know
> how, How do you guys unittest threading classes? Any good articles to
> recommend?
>

There are two aspects of testing classes that utilize threading (and to
a slightly lesser extent multiprocessing.Process).

The first, which I'm assuming you're concerned with, is testing whether
some object or action on some object is threadsafe. It is *really*
difficult to confirm an object is absolutely threadsafe. That doesn't
mean it is impossible, but it is very difficult.

Since asserting threadsafety can be a bit challenging, the best tactic
I've seen for testing within the context of threads is to be sure all
functionality that is not associated with the communication between
threads is in separate, easily testable functions.

For example, if I had a bit list of URLs that I wanted to download, I
might start a thread for each download and perform whatever action. Here
is what the thread class might look like:

  class URLWorker(Thread):
      def __init__(self, url_queue):
          self.url_queue = url_queue
          super(URLWorker, self).__init__()

      def run(self):
          while True:
              url = self.url_queue.get()
              if url == SENTINAL:
                  break
              self.download_and_process(url)

      def download_and_process(self, url):
          url_proc = URLProcessor(url)
          url_proc.download_and_updatedb()