Add next_response lock test.
authorJack Miller <jack@codezen.org>
Tue, 22 Jun 2010 21:18:27 +0000 (16:18 -0500)
committerJack Miller <jack@codezen.org>
Tue, 22 Jun 2010 21:28:16 +0000 (16:28 -0500)
Crappy race condition tests suck.

Signed-off-by: Jack Miller <jack@codezen.org>
canto_curses/gui.py
test.py
tests/gui.py [new file with mode: 0644]

index 75bf821..18bfc86 100755 (executable)
@@ -213,8 +213,11 @@ class CantoCursesGui():
         if self.backend.responses:
             log.debug("DISCARD: %s" % (self.backend.responses[0],))
             self.backend.response_lock.acquire()
+            r = self.backend.responses[0]
             self.backend.responses = self.backend.responses[1:]
             self.backend.response_lock.release()
+            return r
+        return None
 
     def wait_response(self, cmd):
         log.debug("waiting on %s" % cmd)
diff --git a/test.py b/test.py
index 0d16a33..b05f0e7 100755 (executable)
--- a/test.py
+++ b/test.py
@@ -9,6 +9,7 @@
 #   published by the Free Software Foundation.
 
 import tests.main
+import tests.gui
 
 import logging
 
@@ -23,12 +24,15 @@ import unittest
 import sys
 
 all_modules = {
-        "main" : tests.main.Tests }
+        "main" : tests.main.Tests ,
+        "gui" : tests.gui.Tests }
 
 all_tests = {
         "test_args" : tests.main.Tests,
         "test_start_daemon" : tests.main.Tests,
-        "test_ensure_files" : tests.main.Tests
+        "test_ensure_files" : tests.main.Tests,
+        "test_next_response_goodlock" : tests.gui.Tests,
+        "test_next_response_badlock" : tests.gui.Tests,
 }
 
 if __name__ == "__main__":
diff --git a/tests/gui.py b/tests/gui.py
new file mode 100644 (file)
index 0000000..b2551ff
--- /dev/null
@@ -0,0 +1,87 @@
+# -*- coding: utf-8 -*-
+#Canto-curses - ncurses RSS reader
+#   Copyright (C) 2010 Jack Miller <jack@codezen.org>
+#
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License version 2 as 
+#   published by the Free Software Foundation.
+
+from canto_curses.main import CantoCurses
+from canto_curses.gui import CantoCursesGui
+
+from threading import Thread, Lock
+import unittest
+import time
+import os
+
+class FakeLock():
+    def acquire(self):
+        pass
+
+    def release(self):
+        pass
+
+class Tests(unittest.TestCase):
+
+    def hold_lock(self, ccurses):
+        for i in xrange(1000):
+            ccurses.response_lock.acquire()
+            ccurses.responses.append(("ITEMS", "%d" % i))
+            ccurses.response_lock.release()
+
+    # This might a pretty poor test, as all race condition tests
+    # end up being (i.e. not immune to false positives by the nature of the
+    # problem). I can say that with a FakeLock() I generally get corrupted
+    # output anywhere between 0 and 110 tries, so I made 150 iterations
+    # my test.
+
+    def __test_next_response(self, lock):
+        c = CantoCurses()
+        g = CantoCursesGui()
+        c.response_lock = lock
+        g.backend = c
+
+        # Make sure empty list doesn't matter
+        g.backend.responses = []
+        g.next_response()
+
+        t = Thread(target=self.hold_lock, args=(c,))
+        t.start()
+
+        discards = []
+
+        while t.isAlive():
+            r = g.next_response()
+            if r:
+                print r
+                discards.append(r)
+
+        t.join()
+
+        # Ensure that we didn't lose anything
+        self.assertTrue(len(discards) + len(g.backend.responses) == 1000)
+
+        # Ensure that we discarded in order
+        for i, d in enumerate(discards):
+            self.assertTrue(d == ("ITEMS", "%d" % i))
+
+        # Ensure remaining responses are in order
+        for i, d in enumerate(g.backend.responses):
+            i += len(discards)
+            self.assertTrue(d == ("ITEMS", "%d" % i))
+
+    def test_next_response_goodlock(self):
+        for i in xrange(150):
+            self.__test_next_response(Lock())
+
+    def test_next_response_badlock(self):
+        gen = 0;
+        try:
+            for i in xrange(1000):
+                gen = i
+                self.__test_next_response(FakeLock())
+        except:
+            print "Got race in gen %d" % gen
+            return
+
+        print "WARNING: No race with badlock"