mirror of
https://github.com/wassname/catalyst.git
synced 2026-08-11 11:16:15 +08:00
BUG: fix issues with sentinel
This commit is contained in:
@@ -0,0 +1,48 @@
|
||||
from copy import copy, deepcopy
|
||||
from pickle import loads, dumps
|
||||
from unittest import TestCase
|
||||
from weakref import ref
|
||||
|
||||
from zipline.utils.sentinel import sentinel
|
||||
|
||||
|
||||
class SentinelTestCase(TestCase):
|
||||
def tearDown(self):
|
||||
sentinel._cache.clear() # don't pollute cache.
|
||||
|
||||
def test_name(self):
|
||||
self.assertEqual(sentinel('a').__name__, 'a')
|
||||
|
||||
def test_doc(self):
|
||||
self.assertEqual(sentinel('a', 'b').__doc__, 'b')
|
||||
|
||||
def test_doc_differentiates(self):
|
||||
self.assertIsNot(sentinel('a', 'b'), sentinel('a', 'c'))
|
||||
|
||||
def test_memo(self):
|
||||
self.assertIs(sentinel('a'), sentinel('a'))
|
||||
|
||||
def test_copy(self):
|
||||
a = sentinel('a')
|
||||
self.assertIs(copy(a), a)
|
||||
|
||||
def test_deepcopy(self):
|
||||
a = sentinel('a')
|
||||
self.assertIs(deepcopy(a), a)
|
||||
|
||||
def test_repr(self):
|
||||
self.assertEqual(
|
||||
repr(sentinel('a')),
|
||||
"sentinel('a')",
|
||||
)
|
||||
|
||||
def test_new(self):
|
||||
with self.assertRaises(TypeError):
|
||||
type(sentinel('a'))()
|
||||
|
||||
def test_pickle_roundtrip(self):
|
||||
a = sentinel('a')
|
||||
self.assertIs(loads(dumps(a)), a)
|
||||
|
||||
def test_weakreferencable(self):
|
||||
ref(sentinel('a'))
|
||||
+20
-14
@@ -7,19 +7,25 @@ import sys
|
||||
|
||||
|
||||
def sentinel(name, doc=None):
|
||||
@object.__new__ # bind a single instance to the name 'NotSpecified'
|
||||
class result(object):
|
||||
try:
|
||||
return sentinel._cache[name, doc] # memoized
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
@object.__new__ # bind a single instance to the name 'Sentinel'
|
||||
class Sentinel(object):
|
||||
__doc__ = doc
|
||||
__slots__ = ('__weakref__',)
|
||||
__name__ = name
|
||||
|
||||
def __new__(cls):
|
||||
raise TypeError("Can't construct new instances of %s" % name)
|
||||
raise TypeError("Can't construct new instances of %r" % name)
|
||||
|
||||
def __repr__(self):
|
||||
return name
|
||||
return 'sentinel(%r)' % name
|
||||
|
||||
def __reduce__(self):
|
||||
return name
|
||||
return sentinel, (name, doc)
|
||||
|
||||
def __deepcopy__(self, _memo):
|
||||
return self
|
||||
@@ -27,14 +33,14 @@ def sentinel(name, doc=None):
|
||||
def __copy__(self):
|
||||
return self
|
||||
|
||||
cls = type(result)
|
||||
cls.__name__ = name
|
||||
cls = type(Sentinel)
|
||||
try:
|
||||
# traverse up one frame to find the module where this is defined
|
||||
cls.__module__ = sys._getframe(1).f_globals.get(
|
||||
'__name__',
|
||||
'__main__',
|
||||
)
|
||||
except (AttributeError, ValueError):
|
||||
pass
|
||||
return result
|
||||
cls.__module__ = sys._getframe(1).f_globals['__name__']
|
||||
except (AttributeError, ValueError, KeyError):
|
||||
# Couldn't get the name from the calling scope, just use None.
|
||||
cls.__module__ = None
|
||||
|
||||
sentinel._cache[name, doc] = Sentinel # cache result
|
||||
return Sentinel
|
||||
sentinel._cache = {}
|
||||
|
||||
Reference in New Issue
Block a user