Convert more manual tests to unit tests

5 files changed, 169 insertions, 1 deletions

diff --git a/tests/__init__.py b/tests/__init__.py
index d5565cd..757e47d 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -1,7 +1,16 @@
 import aiogreen
 import unittest
+try:
+    import asyncio
+except ImportError:
+    import trollius as asyncio
 
 class TestCase(unittest.TestCase):
     def setUp(self):
-        self.loop = aiogreen.EventLoop()
+        policy = aiogreen.EventLoopPolicy()
+        asyncio.set_event_loop_policy(policy)
+        self.addCleanup(asyncio.set_event_loop_policy, None)
+
+        self.loop = policy.get_event_loop()
         self.addCleanup(self.loop.close)
+        self.addCleanup(asyncio.set_event_loop, None)
diff --git a/tests/test_callback.py b/tests/test_callback.py
new file mode 100644
index 0000000..987960b
--- /dev/null
+++ b/tests/test_callback.py
@@ -0,0 +1,34 @@
+import eventlet
+import tests
+
+class CallbackTests(tests.TestCase):
+    def test_hello_world(self):
+        result = []
+
+        def hello_world(loop):
+            result.append('Hello World')
+            loop.stop()
+
+        self.loop.call_soon(hello_world, self.loop)
+        self.loop.run_forever()
+        self.assertEqual(result, ['Hello World'])
+
+    def test_soon_stop_soon(self):
+        result = []
+
+        def hello():
+            result.append("Hello")
+
+        def world():
+            result.append("World")
+            self.loop.stop()
+
+        self.loop.call_soon(hello)
+        self.loop.call_soon(self.loop.stop)
+        self.loop.call_soon(world)
+        self.loop.run_forever()
+
+        self.assertEqual(result, ["Hello"])
+
+        self.loop.run_forever()
+        self.assertEqual(result, ["Hello", "World"])
diff --git a/tests/test_coroutine.py b/tests/test_coroutine.py
new file mode 100644
index 0000000..3dbe5db
--- /dev/null
+++ b/tests/test_coroutine.py
@@ -0,0 +1,27 @@
+import tests
+
+try:
+    import asyncio
+
+    exec('''if 1:
+        def hello_world(result, delay):
+            result.append("Hello")
+            # retrieve the event loop from the policy
+            yield from asyncio.sleep(delay)
+            result.append('World')
+    ''')
+except ImportError:
+    import trollius as asyncio
+    from trollius import From
+
+    def hello_world(result, delay):
+        result.append("Hello")
+        # retrieve the event loop from the policy
+        yield From(asyncio.sleep(delay))
+        result.append('World')
+
+class CallbackTests(tests.TestCase):
+    def test_hello_world(self):
+        result = []
+        self.loop.run_until_complete(hello_world(result, 0.001))
+        self.assertEqual(result, ['Hello', 'World'])
diff --git a/tests/test_thread.py b/tests/test_thread.py
new file mode 100644
index 0000000..1f5c0e7
--- /dev/null
+++ b/tests/test_thread.py
@@ -0,0 +1,55 @@
+import aiogreen
+import eventlet
+import tests
+threading = eventlet.patcher.original('threading')
+time = eventlet.patcher.original('time')
+try:
+    import asyncio
+except ImportError:
+    import trollius as asyncio
+
+try:
+    get_ident = threading.get_ident   # Python 3
+except AttributeError:
+    get_ident = threading._get_ident   # Python 2
+
+class ThreadTests(tests.TestCase):
+    def test_ident(self):
+        result = {'ident': None}
+
+        def work():
+            result['ident'] = get_ident()
+
+        fut = self.loop.run_in_executor(None, work)
+        self.loop.run_until_complete(fut)
+
+        # ensure that work() was executed in a different thread
+        work_ident = result['ident']
+        self.assertIsNotNone(work_ident)
+        self.assertNotEqual(work_ident, get_ident())
+
+    def test_run_twice(self):
+        result = []
+
+        def work():
+            result.append("run")
+
+        fut = self.loop.run_in_executor(None, work)
+        self.loop.run_until_complete(fut)
+        self.assertEqual(result, ["run"])
+
+        # ensure that run_in_executor() can be called twice
+        fut = self.loop.run_in_executor(None, work)
+        self.loop.run_until_complete(fut)
+        self.assertEqual(result, ["run", "run"])
+
+    def test_policy(self):
+        result = {'loop': 42}   # sentinel, different than None
+
+        def work():
+            result['loop'] = asyncio.get_event_loop()
+
+        # get_event_loop() must return None in a different thread
+        fut = self.loop.run_in_executor(None, work)
+        self.loop.run_until_complete(fut)
+        self.assertIsNone(result['loop'])
diff --git a/tests/test_timer.py b/tests/test_timer.py
new file mode 100644
index 0000000..17fdd8e
--- /dev/null
+++ b/tests/test_timer.py
@@ -0,0 +1,43 @@
+import datetime
+import eventlet
+import tests
+
+class TimerTests(tests.TestCase):
+    def test_display_date(self):
+        result = []
+        delay = 0.050
+        count = 3
+
+        def display_date(end_time, loop):
+            result.append(datetime.datetime.now())
+            if (loop.time() + delay) < end_time:
+                loop.call_later(delay, display_date, end_time, loop)
+            else:
+                loop.stop()
+
+        end_time = self.loop.time() + delay * count
+        self.loop.call_soon(display_date, end_time, self.loop)
+        self.loop.run_forever()
+
+        self.assertEqual(len(result), count, result)
+
+    def test_later_stop_later(self):
+        result = []
+
+        def hello():
+            result.append("Hello")
+
+        def world(loop):
+            result.append("World")
+            loop.stop()
+
+        self.loop.call_later(0.010, hello)
+        self.loop.call_later(0.020, self.loop.stop)
+        self.loop.call_later(0.030, world, self.loop)
+        self.loop.run_forever()
+
+        eventlet.sleep(0.050)
+        self.assertEqual(result, ["Hello"])
+
+        self.loop.run_forever()
+        self.assertEqual(result, ["Hello", "World"])