add load test to highlight concurrency issues

noblogsmv/test/test_state.py

@@ -73,6 +73,30 @@ class DatabaseTest(TestBase):
         self.db.set(w)
         self.assertTrue(self.db.are_we_done())
 
+    def test_loadtest(self):
+        nkeys = 100
+        nloops = 100
+        out = {'errors': 0}
+
+        for i in xrange(nkeys):
+            self.db.put(state.WorkUnit('key%d' % i, state.STATE_INIT, {i: i}))
+        self.db.conn.commit()
+
+        def loadfn(i):
+            key = 'key%d' % i
+            for j in xrange(nloops):
+                try:
+                    with state.work_transaction(self.db, key) as work:
+                        work.state = state.STATE_DONE
+                except Exception, e:
+                    out['errors'] += 1
+                    print e
+
+        threads = [threading.Thread(target=loadfn, args=(i,)) for i in xrange(nkeys)]
+        [x.start() for x in threads]
+        [x.join() for x in threads]
+        self.assertEquals(0, out['errors'])
+
 
 def _mk_state_machine(_states):
     class _MySM(state.StateMachine):