summaryrefslogtreecommitdiffstats
path: root/Lib
diff options
context:
space:
mode:
Diffstat (limited to 'Lib')
-rw-r--r--Lib/test/test_getargs2.py51
-rw-r--r--Lib/test/test_itertools.py24
-rw-r--r--Lib/test/test_smtplib.py107
-rw-r--r--Lib/trace.py25
4 files changed, 137 insertions, 70 deletions
diff --git a/Lib/test/test_getargs2.py b/Lib/test/test_getargs2.py
index 4253141..1918386 100644
--- a/Lib/test/test_getargs2.py
+++ b/Lib/test/test_getargs2.py
@@ -1,5 +1,6 @@
import unittest
from test import test_support
+from _testcapi import getargs_keywords
import warnings
warnings.filterwarnings("ignore",
@@ -248,9 +249,57 @@ class Tuple_TestCase(unittest.TestCase):
raise ValueError
self.assertRaises(TypeError, getargs_tuple, 1, seq())
+class Keywords_TestCase(unittest.TestCase):
+ def test_positional_args(self):
+ # using all positional args
+ self.assertEquals(
+ getargs_keywords((1,2), 3, (4,(5,6)), (7,8,9), 10),
+ (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+ )
+ def test_mixed_args(self):
+ # positional and keyword args
+ self.assertEquals(
+ getargs_keywords((1,2), 3, (4,(5,6)), arg4=(7,8,9), arg5=10),
+ (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+ )
+ def test_keyword_args(self):
+ # all keywords
+ self.assertEquals(
+ getargs_keywords(arg1=(1,2), arg2=3, arg3=(4,(5,6)), arg4=(7,8,9), arg5=10),
+ (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+ )
+ def test_optional_args(self):
+ # missing optional keyword args, skipping tuples
+ self.assertEquals(
+ getargs_keywords(arg1=(1,2), arg2=3, arg5=10),
+ (1, 2, 3, -1, -1, -1, -1, -1, -1, 10)
+ )
+ def test_required_args(self):
+ # required arg missing
+ try:
+ getargs_keywords(arg1=(1,2))
+ except TypeError as err:
+ self.assertEquals(str(err), "Required argument 'arg2' (pos 2) not found")
+ else:
+ self.fail('TypeError should have been raised')
+ def test_too_many_args(self):
+ try:
+ getargs_keywords((1,2),3,(4,(5,6)),(7,8,9),10,111)
+ except TypeError as err:
+ self.assertEquals(str(err), "function takes at most 5 arguments (6 given)")
+ else:
+ self.fail('TypeError should have been raised')
+ def test_invalid_keyword(self):
+ # extraneous keyword arg
+ try:
+ getargs_keywords((1,2),3,arg5=10,arg666=666)
+ except TypeError as err:
+ self.assertEquals(str(err), "'arg666' is an invalid keyword argument for this function")
+ else:
+ self.fail('TypeError should have been raised')
def test_main():
- tests = [Signed_TestCase, Unsigned_TestCase, Tuple_TestCase]
+ tests = [Signed_TestCase, Unsigned_TestCase, Tuple_TestCase, Keywords_TestCase]
try:
from _testcapi import getargs_L, getargs_K
except ImportError:
diff --git a/Lib/test/test_itertools.py b/Lib/test/test_itertools.py
index 63ca17b..ba726b5 100644
--- a/Lib/test/test_itertools.py
+++ b/Lib/test/test_itertools.py
@@ -44,6 +44,10 @@ def take(n, seq):
'Convenience function for partially consuming a long of infinite iterable'
return list(islice(seq, n))
+def fact(n):
+ 'Factorial'
+ return reduce(operator.mul, range(1, n+1), 1)
+
class TestBasicOps(unittest.TestCase):
def test_chain(self):
self.assertEqual(list(chain('abc', 'def')), list('abcdef'))
@@ -52,6 +56,26 @@ class TestBasicOps(unittest.TestCase):
self.assertEqual(take(4, chain('abc', 'def')), list('abcd'))
self.assertRaises(TypeError, chain, 2, 3)
+ def test_combinations(self):
+ self.assertRaises(TypeError, combinations, 'abc') # missing r argument
+ self.assertRaises(TypeError, combinations, 'abc', 2, 1) # too many arguments
+ self.assertRaises(ValueError, combinations, 'abc', -2) # r is negative
+ self.assertRaises(ValueError, combinations, 'abc', 32) # r is too big
+ self.assertEqual(list(combinations(range(4), 3)),
+ [(0,1,2), (0,1,3), (0,2,3), (1,2,3)])
+ for n in range(8):
+ values = [5*x-12 for x in range(n)]
+ for r in range(n+1):
+ result = list(combinations(values, r))
+ self.assertEqual(len(result), fact(n) / fact(r) / fact(n-r)) # right number of combs
+ self.assertEqual(len(result), len(set(result))) # no repeats
+ self.assertEqual(result, sorted(result)) # lexicographic order
+ for c in result:
+ self.assertEqual(len(c), r) # r-length combinations
+ self.assertEqual(len(set(c)), r) # no duplicate elements
+ self.assertEqual(list(c), sorted(c)) # keep original ordering
+ self.assert_(all(e in values for e in c)) # elements taken from input iterable
+
def test_count(self):
self.assertEqual(lzip('abc',count()), [('a', 0), ('b', 1), ('c', 2)])
self.assertEqual(lzip('abc',count(3)), [('a', 3), ('b', 4), ('c', 5)])
diff --git a/Lib/test/test_smtplib.py b/Lib/test/test_smtplib.py
index 6b00b80..5694bb5 100644
--- a/Lib/test/test_smtplib.py
+++ b/Lib/test/test_smtplib.py
@@ -18,14 +18,15 @@ HOST = "localhost"
PORT = None
def server(evt, buf):
+ serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ serv.settimeout(1)
+ serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ serv.bind(("", 0))
+ global PORT
+ PORT = serv.getsockname()[1]
+ serv.listen(5)
+ evt.set()
try:
- serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- serv.settimeout(3)
- serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- serv.bind(("", 0))
- global PORT
- PORT = serv.getsockname()[1]
- serv.listen(5)
conn, addr = serv.accept()
except socket.timeout:
pass
@@ -38,7 +39,6 @@ def server(evt, buf):
buf = buf[sent:]
n -= 1
- time.sleep(0.01)
conn.close()
finally:
@@ -52,16 +52,8 @@ class GeneralTests(TestCase):
self.evt = threading.Event()
servargs = (self.evt, b"220 Hola mundo\n")
threading.Thread(target=server, args=servargs).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)
+ self.evt.wait()
+ self.evt.clear()
def tearDown(self):
self.evt.wait()
@@ -76,29 +68,12 @@ 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(smtplib.SMTPServerDisconnected, smtp.ehlo)
- self.assertRaises(smtplib.SMTPServerDisconnected,
- smtp.send, 'test msg')
-
def testLocalHostName(self):
# check that supplied local_hostname is used
smtp = smtplib.SMTP(HOST, PORT, local_hostname="testhost")
self.assertEqual(smtp.local_hostname, "testhost")
smtp.sock.close()
- def testNonnumericPort(self):
- # 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
smtp = smtplib.SMTP(HOST, PORT)
@@ -128,6 +103,7 @@ def debugging_server(server_class, serv_evt, client_evt):
serv = server_class(("", 0), ('nowhere', -1))
global PORT
PORT = serv.getsockname()[1]
+ serv_evt.set()
try:
if hasattr(select, 'poll'):
@@ -150,12 +126,12 @@ def debugging_server(server_class, serv_evt, client_evt):
except socket.timeout:
pass
finally:
- # allow some time for the client to read the result
- time.sleep(0.5)
- serv.close()
+ if not client_evt.isSet():
+ # allow some time for the client to read the result
+ time.sleep(0.5)
+ serv.close()
asyncore.close_all()
PORT = None
- time.sleep(0.5)
serv_evt.set()
MSG_BEGIN = '---------- MESSAGE FOLLOWS ----------\n'
@@ -181,14 +157,8 @@ class DebuggingServerTests(TestCase):
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)
+ self.serv_evt.wait()
+ self.serv_evt.clear()
def tearDown(self):
# indicate that the client is finished
@@ -258,6 +228,26 @@ class DebuggingServerTests(TestCase):
self.assertEqual(self.output.getvalue(), mexpect)
+class NonConnectingTests(TestCase):
+
+ 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(smtplib.SMTPServerDisconnected, smtp.ehlo)
+ self.assertRaises(smtplib.SMTPServerDisconnected,
+ smtp.send, 'test msg')
+
+ def testNonnumericPort(self):
+ # check that non-numeric port raises socket.error
+ self.assertRaises(socket.error, smtplib.SMTP,
+ "localhost", "bogus")
+ self.assertRaises(socket.error, smtplib.SMTP,
+ "localhost:bogus")
+
+
# test response of client to a non-successful HELO message
class BadHELOServerTests(TestCase):
@@ -269,16 +259,8 @@ class BadHELOServerTests(TestCase):
self.evt = threading.Event()
servargs = (self.evt, b"199 no hello for you!\n")
threading.Thread(target=server, args=servargs).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)
+ self.evt.wait()
+ self.evt.clear()
def tearDown(self):
self.evt.wait()
@@ -355,14 +337,8 @@ class SMTPSimTests(TestCase):
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)
+ self.serv_evt.wait()
+ self.serv_evt.clear()
def tearDown(self):
# indicate that the client is finished
@@ -430,6 +406,7 @@ class SMTPSimTests(TestCase):
def test_main(verbose=None):
test_support.run_unittest(GeneralTests, DebuggingServerTests,
+ NonConnectingTests,
BadHELOServerTests, SMTPSimTests)
if __name__ == '__main__':
diff --git a/Lib/trace.py b/Lib/trace.py
index c52c8a8..06b72b7 100644
--- a/Lib/trace.py
+++ b/Lib/trace.py
@@ -53,6 +53,7 @@ import os
import re
import sys
import threading
+import time
import token
import tokenize
import types
@@ -94,6 +95,8 @@ Modifiers:
with '>>>>>> '.
-s, --summary Write a brief summary on stdout for each file.
(Can only be used with --count or --report.)
+-g, --timing Prefix each line with the time since the program started.
+ Only used while tracing.
Filters, may be repeated multiple times:
--ignore-module=<mod> Ignore the given module(s) and its submodules
@@ -431,7 +434,8 @@ def find_executable_linenos(filename):
class Trace:
def __init__(self, count=1, trace=1, countfuncs=0, countcallers=0,
- ignoremods=(), ignoredirs=(), infile=None, outfile=None):
+ ignoremods=(), ignoredirs=(), infile=None, outfile=None,
+ timing=False):
"""
@param count true iff it should count number of times each
line is executed
@@ -447,6 +451,7 @@ class Trace:
@param infile file from which to read stored counts to be
added into the results
@param outfile file in which to write the results
+ @param timing true iff timing information be displayed
"""
self.infile = infile
self.outfile = outfile
@@ -459,6 +464,9 @@ class Trace:
self._calledfuncs = {}
self._callers = {}
self._caller_cache = {}
+ self.start_time = None
+ if timing:
+ self.start_time = time.time()
if countcallers:
self.globaltrace = self.globaltrace_trackcallers
elif countfuncs:
@@ -609,6 +617,8 @@ class Trace:
key = filename, lineno
self.counts[key] = self.counts.get(key, 0) + 1
+ if self.start_time:
+ print('%.2f' % (time.time() - self.start_time), end=' ')
bname = os.path.basename(filename)
print("%s(%d): %s" % (bname, lineno,
linecache.getline(filename, lineno)), end=' ')
@@ -620,6 +630,8 @@ class Trace:
filename = frame.f_code.co_filename
lineno = frame.f_lineno
+ if self.start_time:
+ print('%.2f' % (time.time() - self.start_time), end=' ')
bname = os.path.basename(filename)
print("%s(%d): %s" % (bname, lineno,
linecache.getline(filename, lineno)), end=' ')
@@ -649,13 +661,13 @@ def main(argv=None):
if argv is None:
argv = sys.argv
try:
- opts, prog_argv = getopt.getopt(argv[1:], "tcrRf:d:msC:lT",
+ opts, prog_argv = getopt.getopt(argv[1:], "tcrRf:d:msC:lTg",
["help", "version", "trace", "count",
"report", "no-report", "summary",
"file=", "missing",
"ignore-module=", "ignore-dir=",
"coverdir=", "listfuncs",
- "trackcalls"])
+ "trackcalls", "timing"])
except getopt.error as msg:
sys.stderr.write("%s: %s\n" % (sys.argv[0], msg))
@@ -675,6 +687,7 @@ def main(argv=None):
summary = 0
listfuncs = False
countcallers = False
+ timing = False
for opt, val in opts:
if opt == "--help":
@@ -693,6 +706,10 @@ def main(argv=None):
listfuncs = True
continue
+ if opt == "-g" or opt == "--timing":
+ timing = True
+ continue
+
if opt == "-t" or opt == "--trace":
trace = 1
continue
@@ -775,7 +792,7 @@ def main(argv=None):
t = Trace(count, trace, countfuncs=listfuncs,
countcallers=countcallers, ignoremods=ignore_modules,
ignoredirs=ignore_dirs, infile=counts_file,
- outfile=counts_file)
+ outfile=counts_file, timing=timing)
try:
fp = open(progname)
try: