Make some tests pass on Windows

celery/tests/__init__.py

@@ -13,6 +13,13 @@ os.environ["EVENTLET_NOPATCH"] = "yes"
 os.environ["GEVENT_NOPATCH"] = "yes"
 
 
+try:
+    WindowsError = WindowsError
+except NameError:
+    class WindowsError(Exception):
+        pass
+
+
 def teardown():
     # Don't want SUBDEBUG log messages at finalization.
     from multiprocessing.util import get_logger
@@ -20,7 +27,10 @@ def teardown():
     import threading
     import os
     if os.path.exists("test.db"):
-        os.remove("test.db")
+        try:
+            os.remove("test.db")
+        except WindowsError:
+            pass
     remaining_threads = [thread for thread in threading.enumerate()
                             if thread.name != "MainThread"]
     if remaining_threads:

celery/tests/test_bin/test_celeryev.py

@@ -1,3 +1,5 @@
+from nose import SkipTest
+
 from celery.app import app_or_default
 from celery.bin import celeryev
 
@@ -29,19 +31,24 @@ class test_EvCommand(unittest.TestCase):
         self.assertEqual(self.ev.run(dump=True), "me dumper, you?")
         self.assertIn("celeryev:dump", proctitle.last[0])
 
-    @patch("celery.events.cursesmon", "evtop", lambda **kw: "me top, you?")
-    @patch("celery.platforms", "set_process_title", proctitle)
     def test_run_top(self):
-        self.assertEqual(self.ev.run(), "me top, you?")
-        self.assertIn("celeryev:top", proctitle.last[0])
+        if self.app.IS_WINDOWS:
+            raise SkipTest("curses monitor does not run on Windows")
+
+        @patch("celery.events.cursesmon", "evtop", lambda **kw: "me top, you?")
+        @patch("celery.platforms", "set_process_title", proctitle)
+        def _inner():
+            self.assertEqual(self.ev.run(), "me top, you?")
+            self.assertIn("celeryev:top", proctitle.last[0])
+        return _inner()
 
     @patch("celery.events.snapshot", "evcam", lambda *a, **k: (a, k))
     @patch("celery.platforms", "set_process_title", proctitle)
     def test_run_cam(self):
         a, kw = self.ev.run(camera="foo.bar.baz", logfile="logfile")
         self.assertEqual(a[0], "foo.bar.baz")
-        self.assertEqual(a[1], 1.0)
-        self.assertIsNone(a[2])
+        self.assertEqual(kw["freq"], 1.0)
+        self.assertIsNone(kw["maxrate"])
         self.assertEqual(kw["loglevel"], "INFO")
         self.assertEqual(kw["logfile"], "logfile")
         self.assertIn("celeryev:cam", proctitle.last[0])

celery/tests/test_cursesmon.py

@@ -1,6 +1,8 @@
-from celery.tests.utils import unittest
+from nose import SkipTest
+
+from celery.app import current_app
 
-from celery.events import cursesmon
+from celery.tests.utils import unittest
 
 
 class MockWindow(object):
@@ -12,6 +14,10 @@ class MockWindow(object):
 class TestCursesDisplay(unittest.TestCase):
 
     def setUp(self):
+        if current_app().IS_WINDOWS:
+            raise SkipTest("curses monitor does not run on Windows")
+
+        from celery.events import cursesmon
         self.monitor = cursesmon.CursesMonitor(object())
         self.win = MockWindow()
         self.monitor.win = self.win