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()
--
http://mail.python.org/mailman/listinfo/python-list
