Track living NativeResources. Unit tests assert they're all gone. (#67)

awscrt/__init__.py

@@ -12,6 +12,7 @@
 # permissions and limitations under the License.
 
 from sys import version_info
+from weakref import WeakSet
 
 __all__ = ['io', 'mqtt', 'crypto', 'http']
 
@@ -24,8 +25,19 @@ class NativeResource(object):
     have B's native code Py_INCREF/DECREF A's python class. This ensures that A will not be destroyed before B.
     If we simply had python class B referencing A, and the GC decided to clean up both, it might destroy A before B.
     """
+
+    # For tracking live NativeResources in tests/debug.
+    # Note that WeakSet can accurately report if 0 objects exist, but iteration isn't 100% thread-safe.
+    _track_lifetime = False
+    _living = WeakSet()
+
     __slots__ = ('_binding', '__weakref__')
 
+    def __init__(self):
+        if NativeResource._track_lifetime:
+            NativeResource._living.add(self)
+
+
 def isinstance_str(x):
     """
     Python 2/3 compatible way to check isinstance(x, str).

test/__init__.py

@@ -10,3 +10,18 @@
 # on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 # express or implied. See the License for the specific language governing
 # permissions and limitations under the License.
+
+from awscrt import NativeResource
+import unittest
+
+
+class NativeResourceTest(unittest.TestCase):
+    """
+    Test fixture asserts there are no living NativeResources when a test completes.
+    """
+
+    def setUp(self):
+        NativeResource._track_lifetime = True
+
+    def tearDown(self):
+        self.assertEqual(0, len(NativeResource._living))

test/test_io.py

@@ -11,30 +11,36 @@
 # express or implied. See the License for the specific language governing
 # permissions and limitations under the License.
 
-import awscrt.io
+from __future__ import absolute_import
+from awscrt.io import ClientBootstrap, DefaultHostResolver, EventLoopGroup
+from test import NativeResourceTest
 import unittest
 
 
-class TestEventLoopGroup(unittest.TestCase):
+class EventLoopGroupTest(NativeResourceTest):
     def test_init_defaults(self):
-        event_loop_group = awscrt.io.EventLoopGroup()
+        event_loop_group = EventLoopGroup()
 
     def test_1_thread(self):
-        event_loop_group = awscrt.io.EventLoopGroup(1)
+        event_loop_group = EventLoopGroup(1)
 
 
-class TestDefaultHostResolver(unittest.TestCase):
+class DefaultHostResolverTest(NativeResourceTest):
     def test_init(self):
-        event_loop_group = awscrt.io.EventLoopGroup()
-        host_resolver = awscrt.io.DefaultHostResolver(event_loop_group)
+        event_loop_group = EventLoopGroup()
+        host_resolver = DefaultHostResolver(event_loop_group)
 
 
-class TestClientBootstrap(unittest.TestCase):
+class ClientBootstrapTest(NativeResourceTest):
     def test_init_defaults(self):
-        event_loop_group = awscrt.io.EventLoopGroup()
-        bootstrap = awscrt.io.ClientBootstrap(event_loop_group)
+        event_loop_group = EventLoopGroup()
+        bootstrap = ClientBootstrap(event_loop_group)
 
     def test_init(self):
-        event_loop_group = awscrt.io.EventLoopGroup()
-        host_resolver = awscrt.io.DefaultHostResolver(event_loop_group)
-        bootstrap = awscrt.io.ClientBootstrap(event_loop_group, host_resolver)
+        event_loop_group = EventLoopGroup()
+        host_resolver = DefaultHostResolver(event_loop_group)
+        bootstrap = ClientBootstrap(event_loop_group, host_resolver)
+
+
+if __name__ == '__main__':
+    unittest.main()