summaryrefslogtreecommitdiffstats
path: root/Lib/test
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test')
-rw-r--r--Lib/test/test_concurrent_futures.py7
-rw-r--r--Lib/test/test_csv.py8
-rw-r--r--Lib/test/test_marshal.py24
-rw-r--r--Lib/test/test_shutil.py10
-rw-r--r--Lib/test/test_smtplib.py111
5 files changed, 150 insertions, 10 deletions
diff --git a/Lib/test/test_concurrent_futures.py b/Lib/test/test_concurrent_futures.py
index 5968980..fda6f5b 100644
--- a/Lib/test/test_concurrent_futures.py
+++ b/Lib/test/test_concurrent_futures.py
@@ -367,6 +367,13 @@ class ExecutorTest(unittest.TestCase):
self.assertEqual([None, None], results)
+ def test_shutdown_race_issue12456(self):
+ # Issue #12456: race condition at shutdown where trying to post a
+ # sentinel in the call queue blocks (the queue is full while processes
+ # have exited).
+ self.executor.map(str, [2] * (self.worker_count + 1))
+ self.executor.shutdown()
+
class ThreadPoolExecutorTest(ThreadPoolMixin, ExecutorTest):
def test_map_submits_without_iteration(self):
diff --git a/Lib/test/test_csv.py b/Lib/test/test_csv.py
index 8ca1e62..2cca0e9 100644
--- a/Lib/test/test_csv.py
+++ b/Lib/test/test_csv.py
@@ -459,20 +459,20 @@ class TestDialectExcel(TestCsvBase):
'5', '6']])
def test_quoted_quote(self):
- self.readerAssertEqual('1,2,3,"""I see,"" said the blind man","as he picked up his hammer and saw"',
+ self.readerAssertEqual('1,2,3,"""I see,"" said the happy man","as he picked up his hammer and saw"',
[['1', '2', '3',
- '"I see," said the blind man',
+ '"I see," said the happy man',
'as he picked up his hammer and saw']])
def test_quoted_nl(self):
input = '''\
1,2,3,"""I see,""
-said the blind man","as he picked up his
+said the happy man","as he picked up his
hammer and saw"
9,8,7,6'''
self.readerAssertEqual(input,
[['1', '2', '3',
- '"I see,"\nsaid the blind man',
+ '"I see,"\nsaid the happy man',
'as he picked up his\nhammer and saw'],
['9','8','7','6']])
diff --git a/Lib/test/test_marshal.py b/Lib/test/test_marshal.py
index 8a5590a..dec8129 100644
--- a/Lib/test/test_marshal.py
+++ b/Lib/test/test_marshal.py
@@ -228,6 +228,30 @@ class BugsTestCase(unittest.TestCase):
invalid_string = b'l\x02\x00\x00\x00\x00\x00\x00\x00'
self.assertRaises(ValueError, marshal.loads, invalid_string)
+ def test_multiple_dumps_and_loads(self):
+ # Issue 12291: marshal.load() should be callable multiple times
+ # with interleaved data written by non-marshal code
+ # Adapted from a patch by Engelbert Gruber.
+ data = (1, 'abc', b'def', 1.0, (2, 'a', ['b', b'c']))
+ for interleaved in (b'', b'0123'):
+ ilen = len(interleaved)
+ positions = []
+ try:
+ with open(support.TESTFN, 'wb') as f:
+ for d in data:
+ marshal.dump(d, f)
+ if ilen:
+ f.write(interleaved)
+ positions.append(f.tell())
+ with open(support.TESTFN, 'rb') as f:
+ for i, d in enumerate(data):
+ self.assertEqual(d, marshal.load(f))
+ if ilen:
+ f.read(ilen)
+ self.assertEqual(positions[i], f.tell())
+ finally:
+ support.unlink(support.TESTFN)
+
def test_main():
support.run_unittest(IntTestCase,
diff --git a/Lib/test/test_shutil.py b/Lib/test/test_shutil.py
index 20e9412..b17ff3e 100644
--- a/Lib/test/test_shutil.py
+++ b/Lib/test/test_shutil.py
@@ -732,11 +732,11 @@ class TestShutil(unittest.TestCase):
"disk_usage not available on this platform")
def test_disk_usage(self):
usage = shutil.disk_usage(os.getcwd())
- self.assertTrue(usage.total > 0)
- self.assertTrue(usage.used > 0)
- self.assertTrue(usage.free >= 0)
- self.assertTrue(usage.total >= usage.used)
- self.assertTrue(usage.total > usage.free)
+ self.assertGreater(usage.total, 0)
+ self.assertGreater(usage.used, 0)
+ self.assertGreaterEqual(usage.free, 0)
+ self.assertGreaterEqual(usage.total, usage.used)
+ self.assertGreater(usage.total, usage.free)
class TestMove(unittest.TestCase):
diff --git a/Lib/test/test_smtplib.py b/Lib/test/test_smtplib.py
index 1204707..95c1bfc 100644
--- a/Lib/test/test_smtplib.py
+++ b/Lib/test/test_smtplib.py
@@ -320,13 +320,16 @@ class DebuggingServerTests(unittest.TestCase):
# XXX (see comment in testSend)
time.sleep(0.01)
smtp.quit()
+ # make sure the Bcc header is still in the message.
+ self.assertEqual(m['Bcc'], 'John Root <root@localhost>, "Dinsdale" '
+ '<warped@silly.walks.com>')
self.client_evt.set()
self.serv_evt.wait()
self.output.flush()
# Add the X-Peer header that DebuggingServer adds
m['X-Peer'] = socket.gethostbyname('localhost')
- # The Bcc header is deleted before serialization.
+ # The Bcc header should not be transmitted.
del m['Bcc']
mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
self.assertEqual(self.output.getvalue(), mexpect)
@@ -365,6 +368,112 @@ class DebuggingServerTests(unittest.TestCase):
re.MULTILINE)
self.assertRegex(debugout, to_addr)
+ def testSendMessageWithSpecifiedAddresses(self):
+ # Make sure addresses specified in call override those in message.
+ m = email.mime.text.MIMEText('A test message')
+ m['From'] = 'foo@bar.com'
+ m['To'] = 'John, Dinsdale'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.send_message(m, from_addr='joe@example.com', to_addrs='foo@example.net')
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ # Add the X-Peer header that DebuggingServer adds
+ m['X-Peer'] = socket.gethostbyname('localhost')
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+ debugout = smtpd.DEBUGSTREAM.getvalue()
+ sender = re.compile("^sender: joe@example.com$", re.MULTILINE)
+ self.assertRegex(debugout, sender)
+ for addr in ('John', 'Dinsdale'):
+ to_addr = re.compile(r"^recips: .*'{}'.*$".format(addr),
+ re.MULTILINE)
+ self.assertNotRegex(debugout, to_addr)
+ recip = re.compile(r"^recips: .*'foo@example.net'.*$", re.MULTILINE)
+ self.assertRegex(debugout, recip)
+
+ def testSendMessageWithMultipleFrom(self):
+ # Sender overrides To
+ m = email.mime.text.MIMEText('A test message')
+ m['From'] = 'Bernard, Bianca'
+ m['Sender'] = 'the_rescuers@Rescue-Aid-Society.com'
+ m['To'] = 'John, Dinsdale'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.send_message(m)
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ # Add the X-Peer header that DebuggingServer adds
+ m['X-Peer'] = socket.gethostbyname('localhost')
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+ debugout = smtpd.DEBUGSTREAM.getvalue()
+ sender = re.compile("^sender: the_rescuers@Rescue-Aid-Society.com$", re.MULTILINE)
+ self.assertRegex(debugout, sender)
+ for addr in ('John', 'Dinsdale'):
+ to_addr = re.compile(r"^recips: .*'{}'.*$".format(addr),
+ re.MULTILINE)
+ self.assertRegex(debugout, to_addr)
+
+ def testSendMessageResent(self):
+ m = email.mime.text.MIMEText('A test message')
+ m['From'] = 'foo@bar.com'
+ m['To'] = 'John'
+ m['CC'] = 'Sally, Fred'
+ m['Bcc'] = 'John Root <root@localhost>, "Dinsdale" <warped@silly.walks.com>'
+ m['Resent-Date'] = 'Thu, 1 Jan 1970 17:42:00 +0000'
+ m['Resent-From'] = 'holy@grail.net'
+ m['Resent-To'] = 'Martha <my_mom@great.cooker.com>, Jeff'
+ m['Resent-Bcc'] = 'doe@losthope.net'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.send_message(m)
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ # The Resent-Bcc headers are deleted before serialization.
+ del m['Bcc']
+ del m['Resent-Bcc']
+ # Add the X-Peer header that DebuggingServer adds
+ m['X-Peer'] = socket.gethostbyname('localhost')
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+ debugout = smtpd.DEBUGSTREAM.getvalue()
+ sender = re.compile("^sender: holy@grail.net$", re.MULTILINE)
+ self.assertRegex(debugout, sender)
+ for addr in ('my_mom@great.cooker.com', 'Jeff', 'doe@losthope.net'):
+ to_addr = re.compile(r"^recips: .*'{}'.*$".format(addr),
+ re.MULTILINE)
+ self.assertRegex(debugout, to_addr)
+
+ def testSendMessageMultipleResentRaises(self):
+ m = email.mime.text.MIMEText('A test message')
+ m['From'] = 'foo@bar.com'
+ m['To'] = 'John'
+ m['CC'] = 'Sally, Fred'
+ m['Bcc'] = 'John Root <root@localhost>, "Dinsdale" <warped@silly.walks.com>'
+ m['Resent-Date'] = 'Thu, 1 Jan 1970 17:42:00 +0000'
+ m['Resent-From'] = 'holy@grail.net'
+ m['Resent-To'] = 'Martha <my_mom@great.cooker.com>, Jeff'
+ m['Resent-Bcc'] = 'doe@losthope.net'
+ m['Resent-Date'] = 'Thu, 2 Jan 1970 17:42:00 +0000'
+ m['Resent-To'] = 'holy@grail.net'
+ m['Resent-From'] = 'Martha <my_mom@great.cooker.com>, Jeff'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ with self.assertRaises(ValueError):
+ smtp.send_message(m)
+ smtp.close()
class NonConnectingTests(unittest.TestCase):
generated by cgit v0.12 at 2025-12-13 07:20:01 (GMT)