PY3: port scrapy.utils.trackref

scrapy/utils/trackref.py

@@ -10,15 +10,17 @@ alias to object in that case).
 """
 
 from __future__ import print_function
-import weakref, os, six
-from collections import defaultdict
+import weakref
 from time import time
 from operator import itemgetter
+from collections import defaultdict
+import six
+
 
 NoneType = type(None)
-
 live_refs = defaultdict(weakref.WeakKeyDictionary)
 
+
 class object_ref(object):
     """Inherit from this class (instead of object) to a keep a record of live
     instances"""
@@ -30,27 +32,34 @@ class object_ref(object):
         live_refs[cls][obj] = time()
         return obj
 
+
 def format_live_refs(ignore=NoneType):
-    s = "Live References" + os.linesep + os.linesep
+    s = "Live References\n\n"
     now = time()
-    for cls, wdict in six.iteritems(live_refs):
+    for cls, wdict in sorted(six.iteritems(live_refs),
+                             key=lambda x: x[0].__name__):
         if not wdict:
             continue
         if issubclass(cls, ignore):
             continue
-        oldest = min(wdict.itervalues())
-        s += "%-30s %6d   oldest: %ds ago" % (cls.__name__, len(wdict), \
-            now-oldest) + os.linesep
+        oldest = min(six.itervalues(wdict))
+        s += "%-30s %6d   oldest: %ds ago\n" % (
+            cls.__name__, len(wdict), now - oldest
+        )
     return s
 
+
 def print_live_refs(*a, **kw):
     print(format_live_refs(*a, **kw))
 
+
 def get_oldest(class_name):
     for cls, wdict in six.iteritems(live_refs):
         if cls.__name__ == class_name:
-            if wdict:
-                return min(six.iteritems(wdict), key=itemgetter(1))[0]
+            if not wdict:
+                break
+            return min(six.iteritems(wdict), key=itemgetter(1))[0]
+
 
 def iter_all(class_name):
     for cls, wdict in six.iteritems(live_refs):

tests/test_utils_trackref.py

@@ -0,0 +1,70 @@
+import six
+import unittest
+from scrapy.utils import trackref
+from tests import mock
+
+
+class Foo(trackref.object_ref):
+    pass
+
+
+class Bar(trackref.object_ref):
+    pass
+
+
+class TrackrefTestCase(unittest.TestCase):
+
+    def setUp(self):
+        trackref.live_refs.clear()
+
+    def test_format_live_refs(self):
+        o1 = Foo()  # NOQA
+        o2 = Bar()  # NOQA
+        o3 = Foo()  # NOQA
+        self.assertEqual(
+            trackref.format_live_refs(),
+            '''\
+Live References
+
+Bar                                 1   oldest: 0s ago
+Foo                                 2   oldest: 0s ago
+''')
+
+        self.assertEqual(
+            trackref.format_live_refs(ignore=Foo),
+            '''\
+Live References
+
+Bar                                 1   oldest: 0s ago
+''')
+
+    @mock.patch('sys.stdout', new_callable=six.StringIO)
+    def test_print_live_refs_empty(self, stdout):
+        trackref.print_live_refs()
+        self.assertEqual(stdout.getvalue(), 'Live References\n\n\n')
+
+    @mock.patch('sys.stdout', new_callable=six.StringIO)
+    def test_print_live_refs_with_objects(self, stdout):
+        o1 = Foo()  # NOQA
+        trackref.print_live_refs()
+        self.assertEqual(stdout.getvalue(), '''\
+Live References
+
+Foo                                 1   oldest: 0s ago\n\n''')
+
+    def test_get_oldest(self):
+        o1 = Foo()  # NOQA
+        o2 = Bar()  # NOQA
+        o3 = Foo()  # NOQA
+        self.assertIs(trackref.get_oldest('Foo'), o1)
+        self.assertIs(trackref.get_oldest('Bar'), o2)
+        self.assertIsNone(trackref.get_oldest('XXX'))
+
+    def test_iter_all(self):
+        o1 = Foo()  # NOQA
+        o2 = Bar()  # NOQA
+        o3 = Foo()  # NOQA
+        self.assertEqual(
+            set(trackref.iter_all('Foo')),
+            {o1, o3},
+        )