diff options
Diffstat (limited to 'Lib')
| -rw-r--r-- | Lib/abc.py | 2 | ||||
| -rw-r--r-- | Lib/collections.py | 22 | ||||
| -rw-r--r-- | Lib/test/test_abc.py | 14 | ||||
| -rw-r--r-- | Lib/test/test_socket.py | 82 |
4 files changed, 112 insertions, 8 deletions
@@ -187,7 +187,7 @@ class ABCMeta(type): 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 namedtuple(typename, field_names, verbose=False): 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 @@ if __name__ == '__main__': 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 @@ class TestABC(unittest.TestCase): 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 TestABC(unittest.TestCase): 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 @@ class TestABC(unittest.TestCase): 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 @@ class BufferIOTest(SocketConnectedTest): 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 @@ def test_main(): 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) |