diff options
Diffstat (limited to 'Lib')
-rw-r--r-- | Lib/contextlib.py | 17 | ||||
-rw-r--r-- | Lib/distutils/command/bdist_msi.py | 5 | ||||
-rw-r--r-- | Lib/distutils/util.py | 21 | ||||
-rw-r--r-- | Lib/logging/__init__.py | 18 | ||||
-rw-r--r-- | Lib/ntpath.py | 4 | ||||
-rw-r--r-- | Lib/posixpath.py | 4 | ||||
-rwxr-xr-x | Lib/smtplib.py | 3 | ||||
-rw-r--r-- | Lib/test/test_datetime.py | 2 | ||||
-rw-r--r-- | Lib/test/test_float.py | 15 | ||||
-rw-r--r-- | Lib/test/test_runpy.py | 3 | ||||
-rw-r--r-- | Lib/test/test_smtplib.py | 205 | ||||
-rw-r--r-- | Lib/test/test_xmlrpc.py | 77 |
12 files changed, 306 insertions, 68 deletions
diff --git a/Lib/contextlib.py b/Lib/contextlib.py index 6605bea..bc810b0 100644 --- a/Lib/contextlib.py +++ b/Lib/contextlib.py @@ -105,15 +105,14 @@ def nested(*managers): vars = [] exc = (None, None, None) try: - try: - for mgr in managers: - exit = mgr.__exit__ - enter = mgr.__enter__ - vars.append(enter()) - exits.append(exit) - yield vars - except: - exc = sys.exc_info() + for mgr in managers: + exit = mgr.__exit__ + enter = mgr.__enter__ + vars.append(enter()) + exits.append(exit) + yield vars + except: + exc = sys.exc_info() finally: while exits: exit = exits.pop() diff --git a/Lib/distutils/command/bdist_msi.py b/Lib/distutils/command/bdist_msi.py index 012d06e..5225bed 100644 --- a/Lib/distutils/command/bdist_msi.py +++ b/Lib/distutils/command/bdist_msi.py @@ -633,7 +633,8 @@ class bdist_msi (Command): def get_installer_filename(self, fullname): # Factored out to allow overriding in subclasses + plat = get_platform() installer_name = os.path.join(self.dist_dir, - "%s.win32-py%s.msi" % - (fullname, self.target_version)) + "%s.%s-py%s.msi" % + (fullname, plat, self.target_version)) return installer_name diff --git a/Lib/distutils/util.py b/Lib/distutils/util.py index 6f15ce8..9aa8570 100644 --- a/Lib/distutils/util.py +++ b/Lib/distutils/util.py @@ -29,8 +29,27 @@ def get_platform (): irix-5.3 irix64-6.2 - For non-POSIX platforms, currently just returns 'sys.platform'. + Windows will return one of: + win-x86_64 (64bit Windows on x86_64 (AMD64)) + win-ia64 (64bit Windows on Itanium) + win32 (all others - specifically, sys.platform is returned) + + For other non-POSIX platforms, currently just returns 'sys.platform'. """ + if os.name == 'nt': + # sniff sys.version for architecture. + prefix = " bit (" + i = string.find(sys.version, prefix) + if i == -1: + return sys.platform + j = string.find(sys.version, ")", i) + look = sys.version[i+len(prefix):j].lower() + if look=='amd64': + return 'win-x86_64' + if look=='itanium': + return 'win-ia64' + return sys.platform + if os.name != "posix" or not hasattr(os, 'uname'): # XXX what about the architecture? NT is Intel or Alpha, # Mac OS is M68k or PPC, etc. diff --git a/Lib/logging/__init__.py b/Lib/logging/__init__.py index 229be69..ef78422 100644 --- a/Lib/logging/__init__.py +++ b/Lib/logging/__init__.py @@ -975,9 +975,7 @@ class Logger(Filterer): logger.debug("Houston, we have a %s", "thorny problem", exc_info=1) """ - if self.manager.disable >= DEBUG: - return - if DEBUG >= self.getEffectiveLevel(): + if self.isEnabledFor(DEBUG): self._log(DEBUG, msg, args, **kwargs) def info(self, msg, *args, **kwargs): @@ -989,9 +987,7 @@ class Logger(Filterer): logger.info("Houston, we have a %s", "interesting problem", exc_info=1) """ - if self.manager.disable >= INFO: - return - if INFO >= self.getEffectiveLevel(): + if self.isEnabledFor(INFO): self._log(INFO, msg, args, **kwargs) def warning(self, msg, *args, **kwargs): @@ -1003,8 +999,6 @@ class Logger(Filterer): logger.warning("Houston, we have a %s", "bit of a problem", exc_info=1) """ - if self.manager.disable >= WARNING: - return if self.isEnabledFor(WARNING): self._log(WARNING, msg, args, **kwargs) @@ -1019,8 +1013,6 @@ class Logger(Filterer): logger.error("Houston, we have a %s", "major problem", exc_info=1) """ - if self.manager.disable >= ERROR: - return if self.isEnabledFor(ERROR): self._log(ERROR, msg, args, **kwargs) @@ -1039,9 +1031,7 @@ class Logger(Filterer): logger.critical("Houston, we have a %s", "major disaster", exc_info=1) """ - if self.manager.disable >= CRITICAL: - return - if CRITICAL >= self.getEffectiveLevel(): + if self.isEnabledFor(CRITICAL): self._log(CRITICAL, msg, args, **kwargs) fatal = critical @@ -1060,8 +1050,6 @@ class Logger(Filterer): raise TypeError, "level must be an integer" else: return - if self.manager.disable >= level: - return if self.isEnabledFor(level): self._log(level, msg, args, **kwargs) diff --git a/Lib/ntpath.py b/Lib/ntpath.py index 99d7a4a..06b2293 100644 --- a/Lib/ntpath.py +++ b/Lib/ntpath.py @@ -59,7 +59,9 @@ def isabs(s): # Join two (or more) paths. def join(a, *p): - """Join two or more pathname components, inserting "\\" as needed""" + """Join two or more pathname components, inserting "\\" as needed. + If any component is an absolute path, all previous path components + will be discarded.""" path = a for b in p: b_wins = 0 # set to 1 iff b makes path irrelevant diff --git a/Lib/posixpath.py b/Lib/posixpath.py index 6f15d48..6d4a9e2 100644 --- a/Lib/posixpath.py +++ b/Lib/posixpath.py @@ -56,7 +56,9 @@ def isabs(s): # Insert a '/' unless the first part is empty or already ends in '/'. def join(a, *p): - """Join two or more pathname components, inserting '/' as needed""" + """Join two or more pathname components, inserting '/' as needed. + If any component is an absolute path, all previous path components + will be discarded.""" path = a for b in p: if b.startswith('/'): diff --git a/Lib/smtplib.py b/Lib/smtplib.py index 0d61f50..301f4fe 100755 --- a/Lib/smtplib.py +++ b/Lib/smtplib.py @@ -401,7 +401,8 @@ class SMTP: return (code,msg) self.does_esmtp=1 #parse the ehlo response -ddm - resp=self.ehlo_resp.split('\n') + assert isinstance(self.ehlo_resp, bytes), repr(self.ehlo_resp) + resp=self.ehlo_resp.decode("latin-1").split('\n') del resp[0] for each in resp: # To be able to communicate with as many SMTP servers as possible, diff --git a/Lib/test/test_datetime.py b/Lib/test/test_datetime.py index c0a96ef..b26c92e 100644 --- a/Lib/test/test_datetime.py +++ b/Lib/test/test_datetime.py @@ -1003,7 +1003,7 @@ class TestDate(HarmlessMixedComparison, unittest.TestCase): self.failUnless(self.theclass.min) self.failUnless(self.theclass.max) - def test_srftime_out_of_range(self): + def test_strftime_out_of_range(self): # For nasty technical reasons, we can't handle years before 1900. cls = self.theclass self.assertEqual(cls(1900, 1, 1).strftime("%Y"), "1900") diff --git a/Lib/test/test_float.py b/Lib/test/test_float.py index 4f4cffd..48abec9 100644 --- a/Lib/test/test_float.py +++ b/Lib/test/test_float.py @@ -81,6 +81,7 @@ class UnknownFormatTestCase(unittest.TestCase): # on an IEEE platform, all we guarantee is that bit patterns # representing infinities or NaNs do not raise an exception; all else # is accident (today). +# let's also try to guarantee that -0.0 and 0.0 don't get confused. class IEEEFormatTestCase(unittest.TestCase): if float.__getformat__("double").startswith("IEEE"): @@ -99,6 +100,20 @@ class IEEEFormatTestCase(unittest.TestCase): ('<f', LE_FLOAT_NAN)]: struct.unpack(fmt, data) + if float.__getformat__("double").startswith("IEEE"): + def test_negative_zero(self): + import math + def pos_pos(): + return 0.0, math.atan2(0.0, -1) + def pos_neg(): + return 0.0, math.atan2(-0.0, -1) + def neg_pos(): + return -0.0, math.atan2(0.0, -1) + def neg_neg(): + return -0.0, math.atan2(-0.0, -1) + self.assertEquals(pos_pos(), neg_pos()) + self.assertEquals(pos_neg(), neg_neg()) + def test_main(): test_support.run_unittest( diff --git a/Lib/test/test_runpy.py b/Lib/test/test_runpy.py index 0fcec1e..b628d77 100644 --- a/Lib/test/test_runpy.py +++ b/Lib/test/test_runpy.py @@ -4,7 +4,7 @@ import os import os.path import sys import tempfile -from test.test_support import verbose, run_unittest +from test.test_support import verbose, run_unittest, forget from runpy import _run_module_code, run_module # Set up the test code and expected results @@ -156,6 +156,7 @@ class RunModuleTest(unittest.TestCase): def _check_module(self, depth): pkg_dir, mod_fname, mod_name = ( self._make_pkg("x=1\n", depth)) + forget(mod_name) try: if verbose: print("Running from source:", mod_name) d1 = run_module(mod_name) # Read from source diff --git a/Lib/test/test_smtplib.py b/Lib/test/test_smtplib.py index cf92662..c2bd196 100644 --- a/Lib/test/test_smtplib.py +++ b/Lib/test/test_smtplib.py @@ -1,4 +1,5 @@ import asyncore +import email.utils import socket import threading import smtpd @@ -75,6 +76,15 @@ class GeneralTests(TestCase): smtp = smtplib.SMTP("%s:%s" % (HOST, PORT)) smtp.sock.close() + def testNotConnected(self): + # Test various operations on an unconnected SMTP object that + # should raise exceptions (at present the attempt in SMTP.send + # to reference the nonexistent 'sock' attribute of the SMTP object + # causes an AttributeError) + smtp = smtplib.SMTP() + self.assertRaises(AttributeError, smtp.ehlo) + self.assertRaises(AttributeError, smtp.send, 'test msg') + def testLocalHostName(self): # check that supplied local_hostname is used smtp = smtplib.SMTP(HOST, PORT, local_hostname="testhost") @@ -82,9 +92,11 @@ class GeneralTests(TestCase): smtp.sock.close() def testNonnumericPort(self): - # check that non-numeric port raises ValueError + # check that non-numeric port raises socket.error self.assertRaises(socket.error, smtplib.SMTP, "localhost", "bogus") + self.assertRaises(socket.error, smtplib.SMTP, + "localhost:bogus") def testTimeoutDefault(self): # default @@ -110,9 +122,9 @@ class GeneralTests(TestCase): smtp.sock.close() -# Test server using smtpd.DebuggingServer -def debugging_server(serv_evt, client_evt): - serv = smtpd.DebuggingServer(("", 0), ('nowhere', -1)) +# Test server thread using the specified SMTP server class +def debugging_server(server_class, serv_evt, client_evt): + serv = server_class(("", 0), ('nowhere', -1)) global PORT PORT = serv.getsockname()[1] @@ -148,11 +160,12 @@ def debugging_server(serv_evt, client_evt): MSG_BEGIN = '---------- MESSAGE FOLLOWS ----------\n' MSG_END = '------------ END MESSAGE ------------\n' -# Test behavior of smtpd.DebuggingServer -# NOTE: the SMTP objects are created with a non-default local_hostname -# argument to the constructor, since (on some systems) the FQDN lookup -# caused by the default local_hostname sometimes takes so long that the +# NOTE: Some SMTP objects in the tests below are created with a non-default +# local_hostname argument to the constructor, since (on some systems) the FQDN +# lookup caused by the default local_hostname sometimes takes so long that the # test server times out, causing the test to fail. + +# Test behavior of smtpd.DebuggingServer class DebuggingServerTests(TestCase): def setUp(self): @@ -163,7 +176,7 @@ class DebuggingServerTests(TestCase): self.serv_evt = threading.Event() self.client_evt = threading.Event() - serv_args = (self.serv_evt, self.client_evt) + serv_args = (smtpd.DebuggingServer, self.serv_evt, self.client_evt) threading.Thread(target=debugging_server, args=serv_args).start() # wait until server thread has assigned a port number @@ -189,12 +202,42 @@ class DebuggingServerTests(TestCase): smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) smtp.quit() - def testEHLO(self): + def testNOOP(self): + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + expected = (250, b'Ok') + self.assertEqual(smtp.noop(), expected) + smtp.quit() + + def testRSET(self): + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + expected = (250, b'Ok') + self.assertEqual(smtp.rset(), expected) + smtp.quit() + + def testNotImplemented(self): + # EHLO isn't implemented in DebuggingServer smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) expected = (502, b'Error: command "EHLO" not implemented') self.assertEqual(smtp.ehlo(), expected) smtp.quit() + def testVRFY(self): + # VRFY isn't implemented in DebuggingServer + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + expected = (502, b'Error: command "VRFY" not implemented') + self.assertEqual(smtp.vrfy('nobody@nowhere.com'), expected) + self.assertEqual(smtp.verify('nobody@nowhere.com'), expected) + smtp.quit() + + def testSecondHELO(self): + # check that a second HELO returns a message that it's a duplicate + # (this behavior is specific to smtpd.SMTPChannel) + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + smtp.helo() + expected = (503, b'Duplicate HELO/EHLO') + self.assertEqual(smtp.helo(), expected) + smtp.quit() + def testHELP(self): smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) self.assertEqual(smtp.help(), b'Error: command "HELP" not implemented') @@ -214,6 +257,7 @@ class DebuggingServerTests(TestCase): self.assertEqual(self.output.getvalue(), mexpect) +# test response of client to a non-successful HELO message class BadHELOServerTests(TestCase): def setUp(self): @@ -243,9 +287,148 @@ class BadHELOServerTests(TestCase): self.assertRaises(smtplib.SMTPConnectError, smtplib.SMTP, HOST, PORT, 'localhost', 3) + +sim_users = {'Mr.A@somewhere.com':'John A', + 'Ms.B@somewhere.com':'Sally B', + 'Mrs.C@somewhereesle.com':'Ruth C', + } + +sim_lists = {'list-1':['Mr.A@somewhere.com','Mrs.C@somewhereesle.com'], + 'list-2':['Ms.B@somewhere.com',], + } + +# Simulated SMTP channel & server +class SimSMTPChannel(smtpd.SMTPChannel): + def smtp_EHLO(self, arg): + resp = '250-testhost\r\n' \ + '250-EXPN\r\n' \ + '250-SIZE 20000000\r\n' \ + '250-STARTTLS\r\n' \ + '250-DELIVERBY\r\n' \ + '250 HELP' + self.push(resp) + + def smtp_VRFY(self, arg): +# print '\nsmtp_VRFY(%r)\n' % arg + + raw_addr = email.utils.parseaddr(arg)[1] + quoted_addr = smtplib.quoteaddr(arg) + if raw_addr in sim_users: + self.push('250 %s %s' % (sim_users[raw_addr], quoted_addr)) + else: + self.push('550 No such user: %s' % arg) + + def smtp_EXPN(self, arg): +# print '\nsmtp_EXPN(%r)\n' % arg + + list_name = email.utils.parseaddr(arg)[1].lower() + if list_name in sim_lists: + user_list = sim_lists[list_name] + for n, user_email in enumerate(user_list): + quoted_addr = smtplib.quoteaddr(user_email) + if n < len(user_list) - 1: + self.push('250-%s %s' % (sim_users[user_email], quoted_addr)) + else: + self.push('250 %s %s' % (sim_users[user_email], quoted_addr)) + else: + self.push('550 No access for you!') + + +class SimSMTPServer(smtpd.SMTPServer): + def handle_accept(self): + conn, addr = self.accept() + channel = SimSMTPChannel(self, conn, addr) + + def process_message(self, peer, mailfrom, rcpttos, data): + pass + + +# Test various SMTP & ESMTP commands/behaviors that require a simulated server +# (i.e., something with more features than DebuggingServer) +class SMTPSimTests(TestCase): + + def setUp(self): + self.serv_evt = threading.Event() + self.client_evt = threading.Event() + serv_args = (SimSMTPServer, self.serv_evt, self.client_evt) + threading.Thread(target=debugging_server, args=serv_args).start() + + # wait until server thread has assigned a port number + n = 500 + while PORT is None and n > 0: + time.sleep(0.01) + n -= 1 + + # wait a little longer (sometimes connections are refused + # on slow machines without this additional wait) + time.sleep(0.5) + + def tearDown(self): + # indicate that the client is finished + self.client_evt.set() + # wait for the server thread to terminate + self.serv_evt.wait() + + def testBasic(self): + # smoke test + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + smtp.quit() + + def testEHLO(self): + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + + # no features should be present before the EHLO + self.assertEqual(smtp.esmtp_features, {}) + + # features expected from the test server + expected_features = {'expn':'', + 'size': '20000000', + 'starttls': '', + 'deliverby': '', + 'help': '', + } + + smtp.ehlo() + self.assertEqual(smtp.esmtp_features, expected_features) + for k in expected_features: + self.assertTrue(smtp.has_extn(k)) + self.assertFalse(smtp.has_extn('unsupported-feature')) + smtp.quit() + + def testVRFY(self): + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + + for email, name in sim_users.items(): + expected_known = (250, bytes('%s %s' % + (name, smtplib.quoteaddr(email)))) + self.assertEqual(smtp.vrfy(email), expected_known) + + u = 'nobody@nowhere.com' + expected_unknown = (550, bytes('No such user: %s' + % smtplib.quoteaddr(u))) + self.assertEqual(smtp.vrfy(u), expected_unknown) + smtp.quit() + + def testEXPN(self): + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + + for listname, members in sim_lists.items(): + users = [] + for m in members: + users.append('%s %s' % (sim_users[m], smtplib.quoteaddr(m))) + expected_known = (250, bytes('\n'.join(users))) + self.assertEqual(smtp.expn(listname), expected_known) + + u = 'PSU-Members-List' + expected_unknown = (550, b'No access for you!') + self.assertEqual(smtp.expn(u), expected_unknown) + smtp.quit() + + + def test_main(verbose=None): test_support.run_unittest(GeneralTests, DebuggingServerTests, - BadHELOServerTests) + BadHELOServerTests, SMTPSimTests) if __name__ == '__main__': test_main() diff --git a/Lib/test/test_xmlrpc.py b/Lib/test/test_xmlrpc.py index 7dfabcf..18e130d 100644 --- a/Lib/test/test_xmlrpc.py +++ b/Lib/test/test_xmlrpc.py @@ -303,38 +303,58 @@ class SimpleServerTestCase(unittest.TestCase): SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = False def test_simple1(self): - p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) - self.assertEqual(p.pow(6,8), 6**8) + try: + p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) + self.assertEqual(p.pow(6,8), 6**8) + except xmlrpclib.ProtocolError, e: + # protocol error; provide additional information in test output + self.fail("%s\n%s" % (e, e.headers)) def test_introspection1(self): - p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) - meth = p.system.listMethods() - expected_methods = set(['pow', 'div', 'add', 'system.listMethods', - 'system.methodHelp', 'system.methodSignature', 'system.multicall']) - self.assertEqual(set(meth), expected_methods) + try: + p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) + meth = p.system.listMethods() + expected_methods = set(['pow', 'div', 'add', 'system.listMethods', + 'system.methodHelp', 'system.methodSignature', 'system.multicall']) + self.assertEqual(set(meth), expected_methods) + except xmlrpclib.ProtocolError, e: + # protocol error; provide additional information in test output + self.fail("%s\n%s" % (e, e.headers)) def test_introspection2(self): - p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) - divhelp = p.system.methodHelp('div') - self.assertEqual(divhelp, 'This is the div function') + try: + p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) + divhelp = p.system.methodHelp('div') + self.assertEqual(divhelp, 'This is the div function') + except xmlrpclib.ProtocolError, e: + # protocol error; provide additional information in test output + self.fail("%s\n%s" % (e, e.headers)) def test_introspection3(self): # the SimpleXMLRPCServer doesn't support signatures, but # at least check that we can try - p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) - divsig = p.system.methodSignature('div') - self.assertEqual(divsig, 'signatures not supported') + try: + p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) + divsig = p.system.methodSignature('div') + self.assertEqual(divsig, 'signatures not supported') + except xmlrpclib.ProtocolError, e: + # protocol error; provide additional information in test output + self.fail("%s\n%s" % (e, e.headers)) def test_multicall(self): - p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) - multicall = xmlrpclib.MultiCall(p) - multicall.add(2,3) - multicall.pow(6,8) - multicall.div(127,42) - add_result, pow_result, div_result = multicall() - self.assertEqual(add_result, 2+3) - self.assertEqual(pow_result, 6**8) - self.assertEqual(div_result, 127//42) + try: + p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) + multicall = xmlrpclib.MultiCall(p) + multicall.add(2,3) + multicall.pow(6,8) + multicall.div(127,42) + add_result, pow_result, div_result = multicall() + self.assertEqual(add_result, 2+3) + self.assertEqual(pow_result, 6**8) + self.assertEqual(div_result, 127//42) + except xmlrpclib.ProtocolError, e: + # protocol error; provide additional information in test output + self.fail("%s\n%s" % (e, e.headers)) # This is a contrived way to make a failure occur on the server side @@ -375,9 +395,16 @@ class FailingServerTestCase(unittest.TestCase): flagval = SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header self.assertEqual(flagval, False) - # test a call that won't fail just as a smoke test - p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) - self.assertEqual(p.pow(6,8), 6**8) + # enable traceback reporting + SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = True + + # test a call that shouldn't fail just as a smoke test + try: + p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT) + self.assertEqual(p.pow(6,8), 6**8) + except xmlrpclib.ProtocolError, e: + # protocol error; provide additional information in test output + self.fail("%s\n%s" % (e, e.headers)) def test_fail_no_info(self): # use the broken message class |