import sys

import os

import unittest

import itertools

import time

import threading

from array import array

from weakref import proxy

from test import test_support

from test.test_support import TESTFN, findfile, run_unittest

from UserList import UserList

class AutoFileTests(unittest.TestCase):

    # file tests for which a test file is automatically set up

    def setUp(self):

        self.f = open(TESTFN, 'wb')

    def tearDown(self):

        if self.f:

            self.f.close()

        os.remove(TESTFN)

    def testWeakRefs(self):

        # verify weak references

        p = proxy(self.f)

        p.write('teststring')

        self.assertEquals(self.f.tell(), p.tell())

        self.f.close()

        self.f = None

        self.assertRaises(ReferenceError, getattr, p, 'tell')

    def testAttributes(self):

        # verify expected attributes exist

        f = self.f

        softspace = f.softspace

        f.name     # merely shouldn't blow up

        f.mode     # ditto

        f.closed   # ditto

        # verify softspace is writable

        f.softspace = softspace    # merely shouldn't blow up

        # verify the others aren't

        for attr in 'name', 'mode', 'closed':

            self.assertRaises((AttributeError, TypeError), setattr, f, attr, 'oops')

    def testReadinto(self):

        # verify readinto

        self.f.write('12')

        self.f.close()

        a = array('c', 'x'*10)

        self.f = open(TESTFN, 'rb')

        n = self.f.readinto(a)

        self.assertEquals('12', a.tostring()[:n])

    def testWritelinesUserList(self):

        # verify writelines with instance sequence

        l = UserList(['1', '2'])

        self.f.writelines(l)

        self.f.close()

        self.f = open(TESTFN, 'rb')

        buf = self.f.read()

        self.assertEquals(buf, '12')

    def testWritelinesIntegers(self):

        # verify writelines with integers

        self.assertRaises(TypeError, self.f.writelines, [1, 2, 3])

    def testWritelinesIntegersUserList(self):

        # verify writelines with integers in UserList

        l = UserList([1,2,3])

        self.assertRaises(TypeError, self.f.writelines, l)

    def testWritelinesNonString(self):

        # verify writelines with non-string object

        class NonString:

            pass

        self.assertRaises(TypeError, self.f.writelines,

                          [NonString(), NonString()])

    def testRepr(self):

        # verify repr works

        self.assert_(repr(self.f).startswith("<open file '" + TESTFN))

    def testErrors(self):

        f = self.f

        self.assertEquals(f.name, TESTFN)

        self.assert_(not f.isatty())

        self.assert_(not f.closed)

        self.assertRaises(TypeError, f.readinto, "")

        f.close()

        self.assert_(f.closed)

    def testMethods(self):

        methods = ['fileno', 'flush', 'isatty', 'next', 'read', 'readinto',

                   'readline', 'readlines', 'seek', 'tell', 'truncate',

                   'write', 'xreadlines', '__iter__']

        if sys.platform.startswith('atheos'):

            methods.remove('truncate')

        # __exit__ should close the file

        self.f.__exit__(None, None, None)

        self.assert_(self.f.closed)

        for methodname in methods:

            method = getattr(self.f, methodname)

            # should raise on closed file

            self.assertRaises(ValueError, method)

        self.assertRaises(ValueError, self.f.writelines, [])

        # file is closed, __exit__ shouldn't do anything

        self.assertEquals(self.f.__exit__(None, None, None), None)

        # it must also return None if an exception was given

        try:

            1/0

        except:

            self.assertEquals(self.f.__exit__(*sys.exc_info()), None)

class OtherFileTests(unittest.TestCase):

    def testModeStrings(self):

        # check invalid mode strings

        for mode in ("", "aU", "wU+"):

            try:

                f = open(TESTFN, mode)

            except ValueError:

                pass

            else:

                f.close()

                self.fail('%r is an invalid file mode' % mode)

        # Some invalid modes fail on Windows, but pass on Unix

        # Issue3965: avoid a crash on Windows when filename is unicode

        for name in (TESTFN, unicode(TESTFN), unicode(TESTFN + '\t')):

            try:

                f = open(name, "rr")

            except IOError:

                pass

            else:

                f.close()

    def testStdin(self):

        # This causes the interpreter to exit on OSF1 v5.1.

        if sys.platform != 'osf1V5':

            self.assertRaises(IOError, sys.stdin.seek, -1)

        else:

            print >>sys.__stdout__, (

                '  Skipping sys.stdin.seek(-1), it may crash the interpreter.'

                ' Test manually.')

        self.assertRaises(IOError, sys.stdin.truncate)

    def testUnicodeOpen(self):

        # verify repr works for unicode too

        f = open(unicode(TESTFN), "w")

        self.assert_(repr(f).startswith("<open file u'" + TESTFN))

        f.close()

        os.unlink(TESTFN)

    def testBadModeArgument(self):

        # verify that we get a sensible error message for bad mode argument

        bad_mode = "qwerty"

        try:

            f = open(TESTFN, bad_mode)

        except ValueError, msg:

            if msg[0] != 0:

                s = str(msg)

                if s.find(TESTFN) != -1 or s.find(bad_mode) == -1:

                    self.fail("bad error message for invalid mode: %s" % s)

            # if msg[0] == 0, we're probably on Windows where there may be

            # no obvious way to discover why open() failed.

        else:

            f.close()

            self.fail("no error for invalid mode: %s" % bad_mode)

    def testSetBufferSize(self):

        # make sure that explicitly setting the buffer size doesn't cause

        # misbehaviour especially with repeated close() calls

        for s in (-1, 0, 1, 512):

            try:

                f = open(TESTFN, 'w', s)

                f.write(str(s))

                f.close()

                f.close()

                f = open(TESTFN, 'r', s)

                d = int(f.read())

                f.close()

                f.close()

            except IOError, msg:

                self.fail('error setting buffer size %d: %s' % (s, str(msg)))

            self.assertEquals(d, s)

    def testTruncateOnWindows(self):

        os.unlink(TESTFN)

        def bug801631():

            # SF bug <http://www.python.org/sf/801631>

            # "file.truncate fault on windows"

            f = open(TESTFN, 'wb')

            f.write('12345678901')   # 11 bytes

            f.close()

            f = open(TESTFN,'rb+')

            data = f.read(5)

            if data != '12345':

                self.fail("Read on file opened for update failed %r" % data)

            if f.tell() != 5:

                self.fail("File pos after read wrong %d" % f.tell())

            f.truncate()

            if f.tell() != 5:

                self.fail("File pos after ftruncate wrong %d" % f.tell())

            f.close()

            size = os.path.getsize(TESTFN)

            if size != 5:

                self.fail("File size after ftruncate wrong %d" % size)

        try:

            bug801631()

        finally:

            os.unlink(TESTFN)

    def testIteration(self):

        # Test the complex interaction when mixing file-iteration and the

        # various read* methods. Ostensibly, the mixture could just be tested

        # to work when it should work according to the Python language,

        # instead of fail when it should fail according to the current CPython

        # implementation.  People don't always program Python the way they

        # should, though, and the implemenation might change in subtle ways,

        # so we explicitly test for errors, too; the test will just have to

        # be updated when the implementation changes.

        dataoffset = 16384

        filler = "ham\n"

        assert not dataoffset % len(filler), \

            "dataoffset must be multiple of len(filler)"

        nchunks = dataoffset // len(filler)

        testlines = [

            "spam, spam and eggs\n",

            "eggs, spam, ham and spam\n",

            "saussages, spam, spam and eggs\n",

            "spam, ham, spam and eggs\n",

            "spam, spam, spam, spam, spam, ham, spam\n",

            "wonderful spaaaaaam.\n"

        ]

        methods = [("readline", ()), ("read", ()), ("readlines", ()),

                   ("readinto", (array("c", " "*100),))]

        try:

            # Prepare the testfile

            bag = open(TESTFN, "w")

            bag.write(filler * nchunks)

            bag.writelines(testlines)

            bag.close()

            # Test for appropriate errors mixing read* and iteration

            for methodname, args in methods:

                f = open(TESTFN)

                if f.next() != filler:

                    self.fail, "Broken testfile"

                meth = getattr(f, methodname)

                try:

                    meth(*args)

                except ValueError:

                    pass

                else:

                    self.fail("%s%r after next() didn't raise ValueError" %

                                     (methodname, args))

                f.close()

            # Test to see if harmless (by accident) mixing of read* and

            # iteration still works. This depends on the size of the internal

            # iteration buffer (currently 8192,) but we can test it in a

            # flexible manner.  Each line in the bag o' ham is 4 bytes

            # ("h", "a", "m", "\n"), so 4096 lines of that should get us

            # exactly on the buffer boundary for any power-of-2 buffersize

            # between 4 and 16384 (inclusive).

            f = open(TESTFN)

            for i in range(nchunks):

                f.next()

            testline = testlines.pop(0)

            try:

                line = f.readline()

            except ValueError:

                self.fail("readline() after next() with supposedly empty "

                          "iteration-buffer failed anyway")

            if line != testline:

                self.fail("readline() after next() with empty buffer "

                          "failed. Got %r, expected %r" % (line, testline))

            testline = testlines.pop(0)

            buf = array("c", "\x00" * len(testline))

            try:

                f.readinto(buf)

            except ValueError:

                self.fail("readinto() after next() with supposedly empty "

                          "iteration-buffer failed anyway")

            line = buf.tostring()

            if line != testline:

                self.fail("readinto() after next() with empty buffer "

                          "failed. Got %r, expected %r" % (line, testline))

            testline = testlines.pop(0)

            try:

                line = f.read(len(testline))

            except ValueError:

                self.fail("read() after next() with supposedly empty "

                          "iteration-buffer failed anyway")

            if line != testline:

                self.fail("read() after next() with empty buffer "

                          "failed. Got %r, expected %r" % (line, testline))

            try:

                lines = f.readlines()

            except ValueError:

                self.fail("readlines() after next() with supposedly empty "

                          "iteration-buffer failed anyway")

            if lines != testlines:

                self.fail("readlines() after next() with empty buffer "

                          "failed. Got %r, expected %r" % (line, testline))

            # Reading after iteration hit EOF shouldn't hurt either

            f = open(TESTFN)

            try:

                for line in f:

                    pass

                try:

                    f.readline()

                    f.readinto(buf)

                    f.read()

                    f.readlines()

                except ValueError:

                    self.fail("read* failed after next() consumed file")

            finally:

                f.close()

        finally:

            os.unlink(TESTFN)

class FileSubclassTests(unittest.TestCase):

    def testExit(self):

        # test that exiting with context calls subclass' close

        class C(file):

            def __init__(self, *args):

                self.subclass_closed = False

                file.__init__(self, *args)

            def close(self):

                self.subclass_closed = True

                file.close(self)

        with C(TESTFN, 'w') as f:

            pass

        self.failUnless(f.subclass_closed)

class FileThreadingTests(unittest.TestCase):

    # These tests check the ability to call various methods of file objects

    # (including close()) concurrently without crashing the Python interpreter.

    # See #815646, #595601

    def setUp(self):

        self.f = None

        self.filename = TESTFN

        with open(self.filename, "w") as f:

            f.write("\n".join("0123456789"))

        self._count_lock = threading.Lock()

        self.close_count = 0

        self.close_success_count = 0

    def tearDown(self):

        if self.f:

            try:

                self.f.close()

            except (EnvironmentError, ValueError):

                pass

        try:

            os.remove(self.filename)

        except EnvironmentError:

            pass

    def _create_file(self):

        self.f = open(self.filename, "w+")

    def _close_file(self):

        with self._count_lock:

            self.close_count += 1

        self.f.close()

        with self._count_lock:

            self.close_success_count += 1

    def _close_and_reopen_file(self):

        self._close_file()

        # if close raises an exception thats fine, self.f remains valid so

        # we don't need to reopen.

        self._create_file()

    def _run_workers(self, func, nb_workers, duration=0.2):

        with self._count_lock:

            self.close_count = 0

            self.close_success_count = 0

        self.do_continue = True

        threads = []

        try:

            for i in range(nb_workers):

                t = threading.Thread(target=func)

                t.start()

                threads.append(t)

            for _ in xrange(100):

                time.sleep(duration/100)

                with self._count_lock:

                    if self.close_count-self.close_success_count > nb_workers+1:

                        if test_support.verbose:

                            print 'Q',

                        break

            time.sleep(duration)

        finally:

            self.do_continue = False

            for t in threads:

                t.join()

    def _test_close_open_io(self, io_func, nb_workers=5):

        def worker():

            self._create_file()

            funcs = itertools.cycle((

                lambda: io_func(),

                lambda: self._close_and_reopen_file(),

            ))

            for f in funcs:

                if not self.do_continue:

                    break

                try:

                    f()

                except (IOError, ValueError):

                    pass

        self._run_workers(worker, nb_workers)

        if test_support.verbose:

            # Useful verbose statistics when tuning this test to take

            # less time to run but still ensuring that its still useful.

            #

            # the percent of close calls that raised an error

            percent = 100. - 100.*self.close_success_count/self.close_count

            print self.close_count, ('%.4f ' % percent),

    def test_close_open(self):

        def io_func():

            pass

        self._test_close_open_io(io_func)

    def test_close_open_flush(self):

        def io_func():

            self.f.flush()

        self._test_close_open_io(io_func)

    def test_close_open_iter(self):

        def io_func():

            list(iter(self.f))

        self._test_close_open_io(io_func)

    def test_close_open_isatty(self):

        def io_func():

            self.f.isatty()

        self._test_close_open_io(io_func)

    def test_close_open_print(self):

        def io_func():

            print >> self.f, ''

        self._test_close_open_io(io_func)

    def test_close_open_read(self):

        def io_func():

            self.f.read(0)

        self._test_close_open_io(io_func)

    def test_close_open_readinto(self):

        def io_func():

            a = array('c', 'xxxxx')

            self.f.readinto(a)

        self._test_close_open_io(io_func)

    def test_close_open_readline(self):

        def io_func():

            self.f.readline()

        self._test_close_open_io(io_func)

    def test_close_open_readlines(self):

        def io_func():

            self.f.readlines()

        self._test_close_open_io(io_func)

    def test_close_open_seek(self):

        def io_func():

            self.f.seek(0, 0)

        self._test_close_open_io(io_func)

    def test_close_open_tell(self):

        def io_func():

            self.f.tell()

        self._test_close_open_io(io_func)

    def test_close_open_truncate(self):

        def io_func():

            self.f.truncate()

        self._test_close_open_io(io_func)

    def test_close_open_write(self):

        def io_func():

            self.f.write('')

        self._test_close_open_io(io_func)

    def test_close_open_writelines(self):

        def io_func():

            self.f.writelines('')

        self._test_close_open_io(io_func)

class StdoutTests(unittest.TestCase):

    def test_move_stdout_on_write(self):

        # Issue 3242: sys.stdout can be replaced (and freed) during a

        # print statement; prevent a segfault in this case

        save_stdout = sys.stdout

        class File:

            def write(self, data):

                if '\n' in data:

                    sys.stdout = save_stdout

        try: