def test_execfile(self):
        globals = {'a': 1, 'b': 2}
        locals = {'b': 200, 'c': 300}

        class M:
            "Test mapping interface versus possible calls from execfile()."
            def __init__(self):
                self.z = 10
            def __getitem__(self, key):
                if key == 'z':
                    return self.z
                raise KeyError
            def __setitem__(self, key, value):
                if key == 'z':
                    self.z = value
                    return
                raise KeyError

        locals = M()
        locals['z'] = 0
        exec(compile(open(TESTFN).read(), TESTFN, 'exec'), globals, locals)
        self.assertEqual(locals['z'], 2)

        self.assertRaises(TypeError, execfile)
        self.assertRaises(TypeError, execfile, TESTFN, {}, ())
        unlink(TESTFN)

    def try_one(self, s):
        # Write s + "\n" + s to file, then open it and ensure that successive
        # .readline()s deliver what we wrote.

        # Ensure we can open TESTFN for writing.
        support.unlink(support.TESTFN)

        # Since C doesn't guarantee we can write/read arbitrary bytes in text
        # files, use binary mode.
        f = self.open(support.TESTFN, "wb")
        try:
            # write once with \n and once without
            f.write(s)
            f.write(b"\n")
            f.write(s)
            f.close()
            f = open(support.TESTFN, "rb")
            line = f.readline()
            self.assertEqual(line, s + b"\n")
            line = f.readline()
            self.assertEqual(line, s)
            line = f.readline()
            self.assertTrue(not line)  # Must be at EOF
            f.close()
        finally:
            support.unlink(support.TESTFN)

    def test_realpath_resolve_before_normalizing(self):
        # Bug #990669: Symbolic links should be resolved before we
        # normalize the path. E.g.: if we have directories 'a', 'k' and 'y'
        # in the following hierarchy:
        # a/k/y
        #
        # and a symbolic link 'link-y' pointing to 'y' in directory 'a',
        # then realpath("link-y/..") should return 'k', not 'a'.
        try:
            os.mkdir(ABSTFN)
            os.mkdir(ABSTFN + "/k")
            os.mkdir(ABSTFN + "/k/y")
            os.symlink(ABSTFN + "/k/y", ABSTFN + "/link-y")

            # Absolute path.
            self.assertEqual(realpath(ABSTFN + "/link-y/.."), ABSTFN + "/k")
            # Relative path.
            with support.change_cwd(dirname(ABSTFN)):
                self.assertEqual(realpath(basename(ABSTFN) + "/link-y/.."),
                                 ABSTFN + "/k")
        finally:
            support.unlink(ABSTFN + "/link-y")
            safe_rmdir(ABSTFN + "/k/y")
            safe_rmdir(ABSTFN + "/k")
            safe_rmdir(ABSTFN)

    def test_builtin_max_min(self):
        self.assertEqual(max(SequenceClass(5)), 4)
        self.assertEqual(min(SequenceClass(5)), 0)
        self.assertEqual(max(8, -1), 8)
        self.assertEqual(min(8, -1), -1)

        d = {"one": 1, "two": 2, "three": 3}
        self.assertEqual(max(d), "two")
        self.assertEqual(min(d), "one")
        self.assertEqual(max(d.values()), 3)
        self.assertEqual(min(iter(d.values())), 1)

        f = open(TESTFN, "w")
        try:
            f.write("medium line\n")
            f.write("xtra large line\n")
            f.write("itty-bitty line\n")
        finally:
            f.close()
        f = open(TESTFN, "r")
        try:
            self.assertEqual(min(f), "itty-bitty line\n")
            f.seek(0, 0)
            self.assertEqual(max(f), "xtra large line\n")
        finally:
            f.close()
            try:
                unlink(TESTFN)
            except OSError:
                pass

    def test_countOf(self):
        from operator import countOf
        self.assertEqual(countOf([1,2,2,3,2,5], 2), 3)
        self.assertEqual(countOf((1,2,2,3,2,5), 2), 3)
        self.assertEqual(countOf("122325", "2"), 3)
        self.assertEqual(countOf("122325", "6"), 0)

        self.assertRaises(TypeError, countOf, 42, 1)
        self.assertRaises(TypeError, countOf, countOf, countOf)

        d = {"one": 3, "two": 3, "three": 3, 1j: 2j}
        for k in d:
            self.assertEqual(countOf(d, k), 1)
        self.assertEqual(countOf(d.values(), 3), 3)
        self.assertEqual(countOf(d.values(), 2j), 1)
        self.assertEqual(countOf(d.values(), 1j), 0)

        f = open(TESTFN, "w")
        try:
            f.write("a\n" "b\n" "c\n" "b\n")
        finally:
            f.close()
        f = open(TESTFN, "r")
        try:
            for letter, count in ("a", 1), ("b", 2), ("c", 1), ("d", 0):
                f.seek(0, 0)
                self.assertEqual(countOf(f, letter + "\n"), count)
        finally:
            f.close()
            try:
                unlink(TESTFN)
            except OSError:
                pass

    def _testBogusZipFile(self):
        support.unlink(TESTMOD)
        fp = open(TESTMOD, 'w+')
        fp.write(struct.pack('=I', 0x06054B50))
        fp.write('a' * 18)
        fp.close()
        z = zipimport.zipimporter(TESTMOD)

        try:
            self.assertRaises(TypeError, z.find_module, None)
            self.assertRaises(TypeError, z.load_module, None)
            self.assertRaises(TypeError, z.is_package, None)
            self.assertRaises(TypeError, z.get_code, None)
            self.assertRaises(TypeError, z.get_data, None)
            self.assertRaises(TypeError, z.get_source, None)

            error = zipimport.ZipImportError
            self.assertEqual(z.find_module('abc'), None)

            self.assertRaises(error, z.load_module, 'abc')
            self.assertRaises(error, z.get_code, 'abc')
            self.assertRaises(IOError, z.get_data, 'abc')
            self.assertRaises(error, z.get_source, 'abc')
            self.assertRaises(error, z.is_package, 'abc')
        finally:
            zipimport._zip_directory_cache.clear()

    def test_builtin_list(self):
        self.assertEqual(list(SequenceClass(5)), list(range(5)))
        self.assertEqual(list(SequenceClass(0)), [])
        self.assertEqual(list(()), [])

        d = {"one": 1, "two": 2, "three": 3}
        self.assertEqual(list(d), list(d.keys()))

        self.assertRaises(TypeError, list, list)
        self.assertRaises(TypeError, list, 42)

        f = open(TESTFN, "w")
        try:
            for i in range(5):
                f.write("%d\n" % i)
        finally:
            f.close()
        f = open(TESTFN, "r")
        try:
            self.assertEqual(list(f), ["0\n", "1\n", "2\n", "3\n", "4\n"])
            f.seek(0, 0)
            self.assertEqual(list(f),
                             ["0\n", "1\n", "2\n", "3\n", "4\n"])
        finally:
            f.close()
            try:
                unlink(TESTFN)
            except OSError:
                pass

        def do_test(firstlines, message, charset, lineno):
            # Raise the message in a subprocess, and catch the output
            try:
                output = open(TESTFN, "w", encoding=charset)
                output.write("""{0}if 1:
                    import traceback;
                    raise RuntimeError('{1}')
                    """.format(firstlines, message))
                output.close()
                process = subprocess.Popen([sys.executable, TESTFN],
                    stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
                stdout, stderr = process.communicate()
                stdout = stdout.decode(output_encoding).splitlines()
            finally:
                unlink(TESTFN)

            # The source lines are encoded with the 'backslashreplace' handler
            encoded_message = message.encode(output_encoding,
                                             'backslashreplace')
            # and we just decoded them with the output_encoding.
            message_ascii = encoded_message.decode(output_encoding)

            err_line = "raise RuntimeError('{0}')".format(message_ascii)
            err_msg = "RuntimeError: {0}".format(message_ascii)

            self.assertIn(("line %s" % lineno), stdout[1],
                "Invalid line number: {0!r} instead of {1}".format(
                    stdout[1], lineno))
            self.assertTrue(stdout[2].endswith(err_line),
                "Invalid traceback line: {0!r} instead of {1!r}".format(
                    stdout[2], err_line))
            self.assertTrue(stdout[3] == err_msg,
                "Invalid error message: {0!r} instead of {1!r}".format(
                    stdout[3], err_msg))

def setUpModule():
    try:
        import signal
        # The default handler for SIGXFSZ is to abort the process.
        # By ignoring it, system calls exceeding the file size resource
        # limit will raise OSError instead of crashing the interpreter.
        signal.signal(signal.SIGXFSZ, signal.SIG_IGN)
    except (ImportError, AttributeError):
        pass

    # On Windows and Mac OSX this test consumes large resources; It
    # takes a long time to build the >2 GiB file and takes >2 GiB of disk
    # space therefore the resource must be enabled to run this test.
    # If not, nothing after this line stanza will be executed.
    if sys.platform[:3] == 'win' or sys.platform == 'darwin':
        requires('largefile',
                 'test requires %s bytes and a long time to run' % str(size))
    else:
        # Only run if the current filesystem supports large files.
        # (Skip this test on Windows, since we now always support
        # large files.)
        f = open(TESTFN, 'wb', buffering=0)
        try:
            # 2**31 == 2147483648
            f.seek(2147483649)
            # Seeking is not enough of a test: you must write and flush, too!
            f.write(b'x')
            f.flush()
        except (OSError, OverflowError):
            raise unittest.SkipTest("filesystem does not have "
                                    "largefile support")
        finally:
            f.close()
            unlink(TESTFN)

 def test_codingspec(self):
     try:
         for enc in ALL_CJKENCODINGS:
             code = '# coding: {}\n'.format(enc)
             exec(code)
     finally:
         support.unlink(TESTFN)

 def test_codingspec(self):
     try:
         for enc in ALL_CJKENCODINGS:
             print('# coding:', enc, file=io.open(TESTFN, 'w'))
             exec(open(TESTFN).read())
     finally:
         support.unlink(TESTFN)

    def test_builtin_map(self):
        self.assertEqual([x+1 for x in SequenceClass(5)], list(range(1, 6)))

        d = {"one": 1, "two": 2, "three": 3}
        self.assertEqual(list(map(lambda k, d=d: (k, d[k]), d)), list(d.items()))
        dkeys = list(d.keys())
        expected = [(i < len(d) and dkeys[i] or None,
                     i,
                     i < len(d) and dkeys[i] or None)
                    for i in range(5)]

        # Deprecated map(None, ...)
        with check_py3k_warnings():
            self.assertEqual(list(SequenceClass(5)), list(range(5)))
            self.assertEqual(list(d), list(d.keys()))
            self.assertEqual(map(None, d,
                                       SequenceClass(5),
                                       iter(d.keys())),
                             expected)

        f = open(TESTFN, "w")
        try:
            for i in range(10):
                f.write("xy" * i + "\n") # line i has len 2*i+1
        finally:
            f.close()
        f = open(TESTFN, "r")
        try:
            self.assertEqual(list(map(len, f)), list(range(1, 21, 2)))
        finally:
            f.close()
            try:
                unlink(TESTFN)
            except OSError:
                pass

 def testThreads(self):
     # BufferedWriter should not raise exceptions or crash
     # when called from multiple threads.
     try:
         # We use a real file object because it allows us to
         # exercise situations where the GIL is released before
         # writing the buffer to the raw streams. This is in addition
         # to concurrency issues due to switching threads in the middle
         # of Python code.
         with io.open(support.TESTFN, "wb", buffering=0) as raw:
             bufio = io.BufferedWriter(raw, 8)
             errors = []
             def f():
                 try:
                     # Write enough bytes to flush the buffer
                     s = b"a" * 19
                     for i in range(50):
                         bufio.write(s)
                 except Exception as e:
                     errors.append(e)
                     raise
             threads = [threading.Thread(target=f) for x in range(20)]
             for t in threads:
                 t.start()
             time.sleep(0.02) # yield
             for t in threads:
                 t.join()
             self.assertFalse(errors,
                 "the following exceptions were caught: %r" % errors)
     finally:
         support.unlink(support.TESTFN)

    def test_symlink(self):
        # On Windows, the EXE needs to know where pythonXY.dll is at so we have
        # to add the directory to the path.
        env = None
        if sys.platform == "win32":
            env = {k.upper(): os.environ[k] for k in os.environ}
            env["PATH"] = "{};{}".format(
                os.path.dirname(sys.executable), env.get("PATH", ""))
            # Requires PYTHONHOME as well since we locate stdlib from the
            # EXE path and not the DLL path (which should be fixed)
            env["PYTHONHOME"] = os.path.dirname(sys.executable)
            if sysconfig.is_python_build(True):
                env["PYTHONPATH"] = os.path.dirname(os.__file__)

        # Issue 7880
        def get(python, env=None):
            cmd = [python, '-c',
                   'import sysconfig; print(sysconfig.get_platform())']
            p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE, env=env)
            out, err = p.communicate()
            if p.returncode:
                print((out, err))
                self.fail('Non-zero return code {0} (0x{0:08X})'
                            .format(p.returncode))
            return out, err
        real = os.path.realpath(sys.executable)
        link = os.path.abspath(TESTFN)
        os.symlink(real, link)
        try:
            self.assertEqual(get(real), get(link, env))
        finally:
            unlink(link)

    def test_binary_modes(self):
        uncompressed = data1 * 50

        with gzip.open(self.filename, "wb") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed)

        with gzip.open(self.filename, "rb") as f:
            self.assertEqual(f.read(), uncompressed)

        with gzip.open(self.filename, "ab") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed * 2)

        with self.assertRaises(FileExistsError):
            gzip.open(self.filename, "xb")
        support.unlink(self.filename)
        with gzip.open(self.filename, "xb") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed)

    def test_input(self):
        self.write_testfile()
        fp = open(TESTFN, 'r')
        savestdin = sys.stdin
        savestdout = sys.stdout # Eats the echo
        try:
            sys.stdin = fp
            sys.stdout = BitBucket()
            self.assertEqual(input(), "1+1")
            self.assertEqual(input(), 'The quick brown fox jumps over the lazy dog.')
            self.assertEqual(input('testing\n'), 'Dear John')

            # SF 1535165: don't segfault on closed stdin
            # sys.stdout must be a regular file for triggering
            sys.stdout = savestdout
            sys.stdin.close()
            self.assertRaises(ValueError, input)

            sys.stdout = BitBucket()
            sys.stdin = io.StringIO("NULL\0")
            self.assertRaises(TypeError, input, 42, 42)
            sys.stdin = io.StringIO("    'whitespace'")
            self.assertEqual(input(), "    'whitespace'")
            sys.stdin = io.StringIO()
            self.assertRaises(EOFError, input)

            del sys.stdout
            self.assertRaises(RuntimeError, input, 'prompt')
            del sys.stdin
            self.assertRaises(RuntimeError, input, 'prompt')
        finally:
            sys.stdin = savestdin
            sys.stdout = savestdout
            fp.close()
            unlink(TESTFN)

    def test_implicit_binary_modes(self):
        # Test implicit binary modes (no "b" or "t" in mode string).
        uncompressed = data1 * 50

        with gzip.open(self.filename, "w") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed)

        with gzip.open(self.filename, "r") as f:
            self.assertEqual(f.read(), uncompressed)

        with gzip.open(self.filename, "a") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed * 2)

        with self.assertRaises(FileExistsError):
            gzip.open(self.filename, "x")
        support.unlink(self.filename)
        with gzip.open(self.filename, "x") as f:
            f.write(uncompressed)
        with open(self.filename, "rb") as f:
            file_data = gzip.decompress(f.read())
            self.assertEqual(file_data, uncompressed)

 def test_write_long_to_file(self):
     for v in range(marshal.version + 1):
         _testcapi.pymarshal_write_long_to_file(0x12345678, support.TESTFN, v)
         with open(support.TESTFN, 'rb') as f:
             data = f.read()
         support.unlink(support.TESTFN)
         self.assertEqual(data, b'\x78\x56\x34\x12')

    def _test_open(self, do_open_close_reader, do_open_close_writer):
        filename = support.TESTFN

        # Use a fifo: until the child opens it for reading, the parent will
        # block when trying to open it for writing.
        support.unlink(filename)
        os.mkfifo(filename)
        self.addCleanup(support.unlink, filename)

        code = '\n'.join((
            'import os, time',
            '',
            'path = %a' % filename,
            'sleep_time = %r' % self.sleep_time,
            '',
            '# let the parent block',
            'time.sleep(sleep_time)',
            '',
            do_open_close_reader,
        ))

        proc = self.subprocess(code)
        with kill_on_error(proc):
            do_open_close_writer(filename)
            self.assertEqual(proc.wait(), 0)

 def restore_files(self, saved_value):
     fn = support.TESTFN
     if fn not in saved_value and (fn + '/') not in saved_value:
         if os.path.isfile(fn):
             support.unlink(fn)
         elif os.path.isdir(fn):
             support.rmtree(fn)