浏览代码

Test PingTask + ExecuteRemoteTask

Ask Solem 16 年之前
父节点
当前提交
f4d65cf874
共有 1 个文件被更改,包括 21 次插入0 次删除
  1. 21 0
      celery/tests/test_task_builtins.py

+ 21 - 0
celery/tests/test_task_builtins.py

@@ -0,0 +1,21 @@
+import unittest
+from celery.task.builtins import PingTask, ExecuteRemoteTask
+import pickle
+
+
+def some_func(i):
+    return i * i
+
+
+class TestPingTask(unittest.TestCase):
+
+    def test_ping(self):
+        self.assertEquals(PingTask.apply().get(), 'pong')
+
+
+class TestRemoteExecuteTask(unittest.TestCase):
+
+    def test_execute_remote(self):
+        self.assertEquals(ExecuteRemoteTask.apply(
+                            args=[pickle.dumps(some_func), [10], {}]).get(),
+                          100)