import threading
import signal
import array
+import collections.abc
import queue
import time
import types
return self
+collections.abc.MutableSequence.register(BaseListProxy)
+
DictProxy = MakeProxyType('DictProxy', (
'__contains__', '__delitem__', '__getitem__', '__iter__', '__len__',
'__setitem__', 'clear', 'copy', 'get', 'items',
'__iter__': 'Iterator',
}
+collections.abc.MutableMapping.register(DictProxy)
ArrayProxy = MakeProxyType('ArrayProxy', (
'__len__', '__getitem__', '__setitem__'
import functools
import signal
import array
+import collections.abc
import socket
import random
import logging
a.append('hello')
self.assertEqual(f[0][:], [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'hello'])
+ def test_list_isinstance(self):
+ a = self.list()
+ self.assertIsInstance(a, collections.abc.MutableSequence)
+
def test_list_iter(self):
a = self.list(list(range(10)))
it = iter(a)
self.assertEqual(sorted(d.values()), [chr(i) for i in indices])
self.assertEqual(sorted(d.items()), [(i, chr(i)) for i in indices])
+ def test_dict_isinstance(self):
+ a = self.dict()
+ self.assertIsInstance(a, collections.abc.MutableMapping)
+
def test_dict_iter(self):
d = self.dict()
indices = list(range(65, 70))
--- /dev/null
+Register the :class:`!multiprocessing.managers.DictProxy` and :class:`!multiprocessing.managers.ListProxy` types in
+:mod:`multiprocessing.managers` to :class:`collections.abc.MutableMapping` and
+:class:`collections.abc.MutableSequence`, respectively.