def setUp(self):
# WeakSet is Python 2.7+
if not hasattr(weakref, 'WeakSet'):
self.skipTest('No weakref.WeakSet available')
BaseTestCase.setUp(self)
python类WeakSet()的实例源码
def getReference(self, obj):
return weakref.WeakSet(obj)
def __init__(self, context, device_p, can_load_configuration=True):
"""
You should not instanciate this class directly.
Call USBContext methods to receive instances of this class.
"""
self.__context = context
self.__close_set = WeakSet()
libusb1.libusb_ref_device(device_p)
self.device_p = device_p
# Fetch device descriptor
device_descriptor = libusb1.libusb_device_descriptor()
result = libusb1.libusb_get_device_descriptor(
device_p, byref(device_descriptor))
mayRaiseUSBError(result)
self.device_descriptor = device_descriptor
if can_load_configuration:
self.__configuration_descriptor_list = descriptor_list = []
append = descriptor_list.append
device_p = self.device_p
for configuration_id in xrange(
self.device_descriptor.bNumConfigurations):
config = libusb1.libusb_config_descriptor_p()
result = libusb1.libusb_get_config_descriptor(
device_p, configuration_id, byref(config))
# pylint: disable=undefined-variable
if result == ERROR_NOT_FOUND:
# pylint: enable=undefined-variable
# Some devices (ex windows' root hubs) tell they have
# one configuration, but they have no configuration
# descriptor.
continue
mayRaiseUSBError(result)
append(config.contents)
def __init__(self):
"""
Create a new USB context.
"""
# Used to prevent an exit to cause a segfault if a concurrent thread
# is still in libusb.
self.__context_refcount = 0
self.__context_cond = threading.Condition()
self.__context_p = libusb1.libusb_context_p()
self.__hotplug_callback_dict = {}
self.__close_set = WeakSet()
def __init__(self, address):
self._address = address
self._free_instances = weakref.WeakSet()
# initialize the pipe attribute before calling _server_pipe_handle()
# because this function can raise an exception and the destructor calls
# the close() method
self._pipe = None
self._accept_pipe_future = None
self._pipe = self._server_pipe_handle(True)
def __init__(self, concurrency=0xffffffff):
self._loop = None
self._results = []
self._iocp = _overlapped.CreateIoCompletionPort(
_overlapped.INVALID_HANDLE_VALUE, NULL, 0, concurrency)
self._cache = {}
self._registered = weakref.WeakSet()
self._unregistered = []
self._stopped_serving = weakref.WeakSet()
def __init__(self, protocol):
self.protocol = protocol
self.calls = WeakSet()
self.loops = WeakSet()
def reset(self):
for call in self.calls:
if call.active():
call.cancel()
for loop in self.loops:
if loop.running:
loop.stop()
self.calls = WeakSet()
self.loops = WeakSet()
def __init__(self):
self._functions = WeakSet()
self._methods = WeakKeyDictionary()
def __init__(self, curveCls, numpoints=10000, **kwargs):
super(DataManager, self).__init__()
self._numpoints = numpoints
self._lostCurveDatas = weakref.WeakSet()
self._curveCls = curveCls
self._lastSamples = collections.OrderedDict()
self._sources = collections.OrderedDict()
def __init__(self, parent_hash=ZERO_HASH, unlocked_block_storage={}, did_lock_to_index_f=None):
self.parent_hash = parent_hash
self.hash_to_index_lookup = {}
self.weight_lookup = {}
self.chain_finder = ChainFinder()
self.change_callbacks = weakref.WeakSet()
self._longest_chain_cache = None
self.did_lock_to_index_f = did_lock_to_index_f
self.unlocked_block_storage = unlocked_block_storage
self._locked_chain = []
def _on_loaded(self):
self._instances = WeakSet()
def setUp(self):
# need to keep references to them
self.items = [ustr(c) for c in ('a', 'b', 'c')]
self.items2 = [ustr(c) for c in ('x', 'y', 'z')]
self.letters = [ustr(c) for c in string.ascii_letters]
self.s = WeakSet(self.items)
self.d = dict.fromkeys(self.items)
self.obj = ustr('F')
self.fs = WeakSet([self.obj])
def test_methods(self):
weaksetmethods = dir(WeakSet)
for method in dir(set):
if method == 'test_c_api' or method.startswith('_'):
continue
self.assertIn(method, weaksetmethods,
"WeakSet missing method " + method)
def test_new_or_init(self):
self.assertRaises(TypeError, WeakSet, [], 2)
def test_union(self):
u = self.s.union(self.items2)
for c in self.letters:
self.assertEqual(c in u, c in self.d or c in self.items2)
self.assertEqual(self.s, WeakSet(self.items))
self.assertEqual(type(u), WeakSet)
self.assertRaises(TypeError, self.s.union, [[]])
for C in set, frozenset, dict.fromkeys, list, tuple:
x = WeakSet(self.items + self.items2)
c = C(self.items2)
self.assertEqual(self.s.union(c), x)
def test_intersection(self):
i = self.s.intersection(self.items2)
for c in self.letters:
self.assertEqual(c in i, c in self.d and c in self.items2)
self.assertEqual(self.s, WeakSet(self.items))
self.assertEqual(type(i), WeakSet)
for C in set, frozenset, dict.fromkeys, list, tuple:
x = WeakSet([])
self.assertEqual(self.s.intersection(C(self.items2)), x)
def test_difference(self):
i = self.s.difference(self.items2)
for c in self.letters:
self.assertEqual(c in i, c in self.d and c not in self.items2)
self.assertEqual(self.s, WeakSet(self.items))
self.assertEqual(type(i), WeakSet)
self.assertRaises(TypeError, self.s.difference, [[]])
def test_symmetric_difference(self):
i = self.s.symmetric_difference(self.items2)
for c in self.letters:
self.assertEqual(c in i, (c in self.d) ^ (c in self.items2))
self.assertEqual(self.s, WeakSet(self.items))
self.assertEqual(type(i), WeakSet)
self.assertRaises(TypeError, self.s.symmetric_difference, [[]])
def test_gc(self):
# Create a nest of cycles to exercise overall ref count check
s = WeakSet(Foo() for i in range(1000))
for elem in s:
elem.cycle = s
elem.sub = elem
elem.set = WeakSet([elem])
