Unit testing multiprocessing code on Windows

[Matt Chaput]

On 18/02/2011 2:54 AM, Terry Reedy wrote:
> On 2/17/2011 6:31 PM, Matt Chaput wrote:
>> Does anyone know the "right" way to write a unit test for code that uses
>> multiprocessing on Windows?
>
> I would start with Lib/test/test_multiprocessing.

Good idea, but on the one hand it doesn't seem to be doing anything 
special, and on the other hand it seems to do it's own things like not 
having its test cases inherit from unittest.TestCase. I also don't know 
if the Python devs start it with distutils or nosetests, which are the 
ones I'm having a problem with. For example, starting my test suite 
inside PyDev doesn't show the bug.

My test code isn't doing anything unusual... this is pretty much all I 
do to trigger the bug. (None of the imported code has anything to do 
with processes.)


from __future__ import with_statement
import unittest
import random

from whoosh import fields, query
from whoosh.support.testing import TempIndex

try:
     import multiprocessing
except ImportError:
     multiprocessing = None


if multiprocessing:
     class MPFCTask(multiprocessing.Process):
         def __init__(self, storage, indexname):
             multiprocessing.Process.__init__(self)
             self.storage = storage
             self.indexname = indexname

         def run(self):
             ix = self.storage.open_index(self.indexname)
             with ix.searcher() as s:
                 r = s.search(query.Every(), sortedby="key", limit=None)
                 result = "".join([h["key"] for h in r])
                 assert result == 
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz", result

class TestSorting(unittest.TestCase):
     def test_mp_fieldcache(self):
         if not multiprocessing:
             return

         schema = fields.Schema(key=fields.KEYWORD(stored=True))
         with TempIndex(schema, "mpfieldcache") as ix:
             domain = 
list(u"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
             random.shuffle(domain)
             w = ix.writer()
             for char in domain:
                 w.add_document(key=char)
             w.commit()

             tasks = [MPFCTask(ix.storage, ix.indexname) for _ in xrange(4)]
             for task in tasks:
                 task.start()
             for task in tasks:
                 task.join()


if __name__ == '__main__':
     unittest.main()