Copied doc for reload() from trunk's function.rst to imp.rst
diff --git a/Lib/abc.py b/Lib/abc.py
index 54dc8e2..8751153 100644
--- a/Lib/abc.py
+++ b/Lib/abc.py
@@ -187,7 +187,7 @@
cls._abc_negative_cache.add(subclass)
return ok
# Check if it's a direct subclass
- if cls in subclass.__mro__:
+ if cls in getattr(subclass, '__mro__', ()):
cls._abc_cache.add(subclass)
return True
# Check if it's a subclass of a registered class (recursive)
diff --git a/Lib/collections.py b/Lib/collections.py
index 504ae19..884c91f 100644
--- a/Lib/collections.py
+++ b/Lib/collections.py
@@ -65,9 +65,9 @@
def __new__(cls, %(argtxt)s):
return tuple.__new__(cls, (%(argtxt)s)) \n
@classmethod
- def _make(cls, iterable):
+ def _make(cls, iterable, new=tuple.__new__, len=len):
'Make a new %(typename)s object from a sequence or iterable'
- result = tuple.__new__(cls, iterable)
+ result = new(cls, iterable)
if len(result) != %(numfields)d:
raise TypeError('Expected %(numfields)d arguments, got %%d' %% len(result))
return result \n
@@ -115,8 +115,22 @@
assert p == loads(dumps(p))
# test and demonstrate ability to override methods
- Point.__repr__ = lambda self: 'Point(%.3f, %.3f)' % self
- print(p)
+ class Point(namedtuple('Point', 'x y')):
+ @property
+ def hypot(self):
+ return (self.x ** 2 + self.y ** 2) ** 0.5
+ def __repr__(self):
+ return 'Point(x=%.3f, y=%.3f, hypot=%.3f)' % (self.x, self.y, self.hypot)
+
+ print(Point(3, 4),'\n', Point(2, 5), '\n', Point(9./7, 6))
+
+ class Point(namedtuple('Point', 'x y')):
+ 'Point class with optimized _make() and _replace() without error-checking'
+ _make = classmethod(tuple.__new__)
+ def _replace(self, _map=map, **kwds):
+ return self._make(_map(kwds.pop, ('x', 'y'), self))
+
+ print(Point(11, 22)._replace(x=100))
import doctest
TestResults = namedtuple('TestResults', 'failed attempted')
diff --git a/Lib/test/test_abc.py b/Lib/test/test_abc.py
index e6c8415..884dd32 100644
--- a/Lib/test/test_abc.py
+++ b/Lib/test/test_abc.py
@@ -56,10 +56,18 @@
self.assertEqual(F.__abstractmethods__, {"bar"})
self.assertRaises(TypeError, F) # because bar is abstract now
+ def test_subclass_oldstyle_class(self):
+ class A:
+ __metaclass__ = abc.ABCMeta
+ class OldstyleClass:
+ pass
+ self.assertFalse(issubclass(OldstyleClass, A))
+ self.assertFalse(issubclass(A, OldstyleClass))
+
def test_registration_basics(self):
class A(metaclass=abc.ABCMeta):
pass
- class B:
+ class B(object):
pass
b = B()
self.assertEqual(issubclass(B, A), False)
@@ -94,7 +102,7 @@
class A1(A):
pass
self.assertRaises(RuntimeError, A1.register, A) # cycles not allowed
- class B:
+ class B(object):
pass
A1.register(B) # ok
A1.register(B) # should pass silently
@@ -135,7 +143,7 @@
def test_all_new_methods_are_called(self):
class A(metaclass=abc.ABCMeta):
pass
- class B:
+ class B(object):
counter = 0
def __new__(cls):
B.counter += 1
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index 535f0ef..facf9fd 100644
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -1114,6 +1114,85 @@
buf = bytes(MSG)
self.serv_conn.send(buf)
+
+TIPC_STYPE = 2000
+TIPC_LOWER = 200
+TIPC_UPPER = 210
+
+def isTipcAvailable():
+ """Check if the TIPC module is loaded
+
+ The TIPC module is not loaded automatically on Ubuntu and probably
+ other Linux distros.
+ """
+ if not hasattr(socket, "AF_TIPC"):
+ return False
+ if not os.path.isfile("/proc/modules"):
+ return False
+ with open("/proc/modules") as f:
+ for line in f:
+ if line.startswith("tipc "):
+ return True
+ if test_support.debug:
+ print("TIPC module is not loaded, please 'sudo modprobe tipc'")
+ return False
+
+class TIPCTest (unittest.TestCase):
+ def testRDM(self):
+ srv = socket.socket(socket.AF_TIPC, socket.SOCK_RDM)
+ cli = socket.socket(socket.AF_TIPC, socket.SOCK_RDM)
+
+ srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ srvaddr = (socket.TIPC_ADDR_NAMESEQ, TIPC_STYPE,
+ TIPC_LOWER, TIPC_UPPER)
+ srv.bind(srvaddr)
+
+ sendaddr = (socket.TIPC_ADDR_NAME, TIPC_STYPE,
+ TIPC_LOWER + int((TIPC_UPPER - TIPC_LOWER) / 2), 0)
+ cli.sendto(MSG, sendaddr)
+
+ msg, recvaddr = srv.recvfrom(1024)
+
+ self.assertEqual(cli.getsockname(), recvaddr)
+ self.assertEqual(msg, MSG)
+
+
+class TIPCThreadableTest (unittest.TestCase, ThreadableTest):
+ def __init__(self, methodName = 'runTest'):
+ unittest.TestCase.__init__(self, methodName = methodName)
+ ThreadableTest.__init__(self)
+
+ def setUp(self):
+ self.srv = socket.socket(socket.AF_TIPC, socket.SOCK_STREAM)
+ self.srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ srvaddr = (socket.TIPC_ADDR_NAMESEQ, TIPC_STYPE,
+ TIPC_LOWER, TIPC_UPPER)
+ self.srv.bind(srvaddr)
+ self.srv.listen(5)
+ self.serverExplicitReady()
+ self.conn, self.connaddr = self.srv.accept()
+
+ def clientSetUp(self):
+ # The is a hittable race between serverExplicitReady() and the
+ # accept() call; sleep a little while to avoid it, otherwise
+ # we could get an exception
+ time.sleep(0.1)
+ self.cli = socket.socket(socket.AF_TIPC, socket.SOCK_STREAM)
+ addr = (socket.TIPC_ADDR_NAME, TIPC_STYPE,
+ TIPC_LOWER + int((TIPC_UPPER - TIPC_LOWER) / 2), 0)
+ self.cli.connect(addr)
+ self.cliaddr = self.cli.getsockname()
+
+ def testStream(self):
+ msg = self.conn.recv(1024)
+ self.assertEqual(msg, MSG)
+ self.assertEqual(self.cliaddr, self.connaddr)
+
+ def _testStream(self):
+ self.cli.send(MSG)
+ self.cli.close()
+
+
def test_main():
tests = [GeneralModuleTests, BasicTCPTest, TCPCloserTest, TCPTimeoutTest,
TestExceptions, BufferIOTest, BasicTCPTest2]
@@ -1134,6 +1213,9 @@
tests.append(BasicSocketPairTest)
if sys.platform == 'linux2':
tests.append(TestLinuxAbstractNamespace)
+ if isTipcAvailable():
+ tests.append(TIPCTest)
+ tests.append(TIPCThreadableTest)
thread_info = test_support.threading_setup()
test_support.run_unittest(*tests)