convert handles and requests to new base module

tests/common.py

@@ -16,6 +16,7 @@
 from __future__ import print_function, unicode_literals, division, absolute_import
 
 import os
+import platform
 import unittest
 
 import uv
@@ -56,6 +57,14 @@ def reraise(exc_type, exc_value, exc_traceback):
 except: pass
 
 
+def implementation_skip(*implementations):
+    def decorator(obj):
+        return obj
+    if platform.python_implementation().lower() in implementations:
+        return unittest.skip('test is not available on the current implementation')
+    return decorator
+
+
 class TestLoop(uv.Loop):
     def __init__(self):
         super(TestLoop, self).__init__()

tests/test_fs_event.py

@@ -0,0 +1,79 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2016, Maximilian Köhl <mail@koehlma.de>
+#
+# This program is free software: you can redistribute it and/or modify it under
+# the terms of the GNU Lesser General Public License version 3 as published by
+# the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT ANY
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
+# PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with this program. If not, see <http://www.gnu.org/licenses/>.
+
+from __future__ import print_function, unicode_literals, division, absolute_import
+
+import os
+import os.path
+import tempfile
+
+from common import TestCase
+
+import uv
+
+
+class TestFSEvent(TestCase):
+    def test_fs_event_change(self):
+        def on_event(fs_event, status, name, events):
+            self.assert_equal(status, uv.error.StatusCodes.SUCCESS)
+            self.assert_equal(name, os.path.basename(self.temp_file.name))
+            self.assert_equal(events, uv.FSEvents.CHANGE)
+            fs_event.close()
+
+        def on_timeout(timer):
+            self.temp_file.write(b'x')
+            self.temp_file.flush()
+            timer.close()
+
+        self.fs_event = uv.FSEvent(on_event=on_event)
+        self.timer = uv.Timer(on_timeout=on_timeout)
+
+        with tempfile.NamedTemporaryFile() as temp_file:
+            self.temp_file = temp_file
+            self.fs_event.path = temp_file.name
+            self.fs_event.start()
+            self.timer.start(20)
+            self.loop.run()
+
+    def test_fs_event_rename(self):
+        def on_event(fs_event, status, name, events):
+            self.assert_equal(status, uv.error.StatusCodes.SUCCESS)
+            self.assert_equal(name, os.path.basename(self.temp_file.name))
+            self.assert_equal(events, uv.FSEvents.RENAME)
+            fs_event.close()
+
+        def on_timeout(timer):
+            os.rename(self.temp_file.name, self.temp_file.name + '-new-name')
+            timer.close()
+
+        self.fs_event = uv.FSEvent(on_event=on_event)
+        self.timer = uv.Timer(on_timeout=on_timeout)
+
+        with tempfile.NamedTemporaryFile() as temp_file:
+            self.temp_file = temp_file
+            self.fs_event.path = temp_file.name
+            self.fs_event.start()
+            self.timer.start(20)
+            self.loop.run()
+            os.rename(self.temp_file.name + '-new-name', self.temp_file.name)
+
+    def test_fs_event_stop(self):
+        self.fs_event = uv.FSEvent()
+
+        with tempfile.NamedTemporaryFile() as temp_file:
+            self.fs_event.path = temp_file.name
+            self.fs_event.start()
+            self.fs_event.stop()
+            self.loop.run()

tests/test_fs_poll.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2016, Maximilian Köhl <mail@koehlma.de>
+#
+# This program is free software: you can redistribute it and/or modify it under
+# the terms of the GNU Lesser General Public License version 3 as published by
+# the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT ANY
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
+# PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with this program. If not, see <http://www.gnu.org/licenses/>.
+
+from __future__ import print_function, unicode_literals, division, absolute_import
+
+import tempfile
+
+from common import TestCase
+
+import uv
+
+
+class TestFSPoll(TestCase):
+    def test_fs_poll_change(self):
+        def on_change(fs_poll, status, previous_stat, current_stat):
+            self.assert_equal(status, uv.error.StatusCodes.SUCCESS)
+            self.assert_not_equal(previous_stat.mtim, current_stat.mtim)
+            fs_poll.close()
+
+        def on_timeout(timer):
+            self.temp_file.write(b'x')
+            self.temp_file.flush()
+            timer.close()
+
+        self.fs_poll = uv.FSPoll(interval=2000, on_change=on_change)
+        self.timer = uv.Timer(on_timeout=on_timeout)
+
+        with tempfile.NamedTemporaryFile() as temp_file:
+            self.temp_file = temp_file
+            self.fs_poll.path = temp_file.name
+            self.fs_poll.start()
+            self.timer.start(1000)
+            self.loop.run()
+
+    def test_fs_poll_stop(self):
+        self.fs_poll = uv.FSPoll()
+
+        with tempfile.NamedTemporaryFile() as temp_file:
+            self.fs_poll.path = temp_file.name
+            self.fs_poll.start()
+            self.fs_poll.stop()
+            self.loop.run()

tests/test_gc.py

@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2016, Maximilian Köhl <mail@koehlma.de>
+#
+# This program is free software: you can redistribute it and/or modify it under
+# the terms of the GNU Lesser General Public License version 3 as published by
+# the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT ANY
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
+# PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with this program. If not, see <http://www.gnu.org/licenses/>.
+
+from __future__ import print_function, unicode_literals, division, absolute_import
+
+import gc
+import weakref
+
+from common import TestCase, implementation_skip
+
+import uv
+
+
+@implementation_skip('pypy')
+class TestGC(TestCase):
+    def test_gc_loop(self):
+        loop = uv.Loop()
+        weak_loop = weakref.ref(loop)
+        base_loop = weakref.ref(loop.base_loop)
+        del loop
+        gc.collect()
+        self.assert_not_equal(weak_loop(), None)
+        self.assert_not_equal(base_loop(), None)
+        uv.Loop._thread_locals.loop = None
+        gc.collect()
+        self.assert_equal(weak_loop(), None)
+        self.assert_equal(base_loop(), None)
+
+    def test_gc_handle(self):
+        weak_handle = weakref.ref(uv.Prepare())
+        gc.collect()
+        self.assert_equal(weak_handle(), None)
+        prepare = uv.Prepare()
+        prepare.start()
+        weak_handle = weakref.ref(prepare)
+        del prepare
+        gc.collect()
+        self.assert_not_equal(weak_handle(), None)
+

tests/test_loop.py

@@ -13,6 +13,7 @@
 # You should have received a copy of the GNU Lesser General Public License along
 # with this program. If not, see <http://www.gnu.org/licenses/>.
 
+import threading
 import time
 
 from common import TestCase
@@ -155,20 +156,35 @@ def on_prepare(prepare):
 
         self.loop.run()
 
+    def test_call_later(self):
+        self.callback_called = False
+
+        def callback():
+            self.callback_called = True
+
+        def on_prepare(prepare):
+            prepare.close()
+
+        self.prepare = uv.Prepare(on_prepare=on_prepare)
+        self.prepare.start()
+
+        self.loop.call_later(callback)
+        self.loop.run()
+
+        self.assert_true(self.callback_called)
 
-    '''
     def test_current_loop(self):
-        self.assertEqual(uv.Loop.default_loop(), uv.Loop.current_loop())
+        self.assertEqual(uv.Loop.get_default(), uv.Loop.get_current())
 
         self.loop1 = uv.Loop()
         self.loop2 = uv.Loop()
 
         def on_prepare1(prepare):
-            self.assertEqual(self.loop1, uv.Loop.current_loop())
+            self.assertEqual(self.loop1, uv.Loop.get_current())
             prepare.close()
 
         def on_prepare2(prepare):
-            self.assertEqual(self.loop2, uv.Loop.current_loop())
+            self.assertEqual(self.loop2, uv.Loop.get_current())
             prepare.close()
 
         def main1():
@@ -190,6 +206,7 @@ def main2():
 
         thread1.join()
 
-        self.assertEqual(uv.Loop.default_loop(), uv.Loop.current_loop())
-    '''
+        self.assertEqual(self.loop2, uv.Loop.get_current())
 
+        self.loop1.close()
+        self.loop2.close()

uv/common.py

@@ -18,18 +18,16 @@
 import os
 import platform
 import sys
-import threading
-import weakref
 
 from collections import OrderedDict
 
 from . import library
 from .library import ffi
 
+
 is_py2 = sys.version_info[0] == 2
 is_py3 = sys.version_info[0] == 3
 
-
 is_pypy = platform.python_implementation().lower() == 'pypy'
 is_cpython = platform.python_implementation().lower() == 'cpython'