<commit_before>import os
import stat

import pytest

from topaz.objects.fileobject import W_FileObject, W_IOObject

from ..base import BaseTopazTest


class TestIO(BaseTopazTest):
    def test_constants(self, space):
        assert space.int_w(space.execute("return IO::SEEK_CUR")) == os.SEEK_CUR
        assert space.int_w(space.execute("return IO::SEEK_END")) == os.SEEK_END
        assert space.int_w(space.execute("return IO::SEEK_SET")) == os.SEEK_SET

    def test_new_from_file(self, space, tmpdir):
        contents = "foo\nbar\nbaz\n"
        f = tmpdir.join("file.txt")
        f.write(contents)

        w_res = space.execute("""
        f = File.new('%s')
        io = IO.new(f)
        return io.read
        """ % f)
        assert space.str_w(w_res) == contents

    def test_new_from_fd(self, space):
        w_res = space.execute("return IO.new(1)")
        assert isinstance(w_res, W_IOObject)

    def test_write(self, space, capfd):
        content = "foo\n"
        space.execute('return IO.new(1, "w").write("%s")' % content)
        out, err = capfd.readouterr()
        assert out == content
        content = "foo\n"

        fd = os.dup(1)
        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = IO.new(%d, "w")
            io.close
            io.write("")
            """ % fd)
        fd = os.dup(1)

    def test_push(self, space, capfd):
        space.execute('return IO.new(1, "w") << "hello" << "world"')
        out, err = capfd.readouterr()
        assert out == "helloworld"

        fd = os.dup(1)
        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = IO.new(%d, "w")
            io.close
            io << ""
            """ % fd)

    def test_read(self, space, tmpdir):
        contents = "foo\nbar\nbaz\n"
        f = tmpdir.join("file.txt")
        f.write(contents)

        w_res = space.execute("return File.new('%s').read" % f)
        assert space.str_w(w_res) == contents

        w_res = space.execute("return File.new('%s').read(4)" % f)
        assert space.str_w(w_res) == contents[:4]

        w_res = space.execute("""
        a = 'hello world'
        File.new('%s').read(10, a)
        return a
        """ % f)
        assert space.str_w(w_res) == contents[:10]

        with self.raises(space, "ArgumentError"):
            space.execute("File.new('%s').read(-1)" % f)

        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = File.new('%s')
            io.close
            io.read
            """ % f)

    def test_simple_print(self, space, capfd):
        space.execute('IO.new(1, "w").print("foo")')
        out, err = capfd.readouterr()
        assert out == "foo"
        fd = os.dup(1)
        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = IO.new(%d, "w")
            io.close
            io.print ""
            """ % fd)

    def test_multi_print(self, space, capfd):
        space.execute('IO.new(1, "w").print("This", "is", 100, "percent")')
        out, err = capfd.readouterr()
        assert out == "Thisis100percent"

    def test_print_globals(self, space, capfd):
        space.globals.set(space, "$,", space.newstr_fromstr(":"))
        space.globals.set(space, "$\\", space.newstr_fromstr("\n"))
        space.execute('IO.new(1, "w").print("foo", "bar", "baz")')
        space.globals.set(space, "$_", space.newstr_fromstr('lastprint'))
        space.execute('IO.new(1, "w").print')
        out, err = capfd.readouterr()
        assert out == "foo:bar:baz\nlastprint\n"

    def test_non_string_print_globals(self, space, capfd):
        space.globals.set(space, "$,", space.w_nil)
        space.globals.set(space, "$\\", space.w_nil)
        space.execute('IO.new(1, "w").print("foo", "bar", "baz")')
        space.globals.set(space, "$_", space.w_nil)
        space.execute('IO.new(1, "w").print')
        out, err = capfd.readouterr()
        assert out == "foobarbaz"

    def test_puts(self, space, capfd):
        space.execute("IO.new(1, 'w').puts('This', 'is\n', 100, 'percent')")
        out, err = capfd.readouterr()
        assert out == "This\nis\n100\npercent\n"
        fd = os.dup(1)
        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = IO.new(%d, "w")
            io.close
            io.puts ""
            """ % fd)

    def test_flush(self, space, capfd):
        space.execute("IO.new(1, 'w').flush.puts('String')")
        out, err = capfd.readouterr()
        assert out == "String\n"
        fd = os.dup(1)
        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = IO.new(%d, "w")
            io.close
            io.flush
            """ % fd)

    def test_globals(self, space, capfd):
        w_res = space.execute("""
        STDOUT.puts("STDOUT")
        $stdout.puts("$stdout")
        $>.puts("$>")
        STDERR.puts("STDERR")
        $stderr.puts("$stderr")
        return STDIN.read, $stdin.read
        """)
        out, err = capfd.readouterr()
        assert out == "STDOUT\n$stdout\n$>\n"
        assert err == "STDERR\n$stderr\n"
        assert self.unwrap(space, w_res) == [None, None]

    def test_rewind(self, space, tmpdir):<commit_msg>use files instead of dup'ing stdout<commit_after>import os
import stat

import pytest

from topaz.objects.fileobject import W_FileObject, W_IOObject

from ..base import BaseTopazTest


class TestIO(BaseTopazTest):
    def test_constants(self, space):
        assert space.int_w(space.execute("return IO::SEEK_CUR")) == os.SEEK_CUR
        assert space.int_w(space.execute("return IO::SEEK_END")) == os.SEEK_END
        assert space.int_w(space.execute("return IO::SEEK_SET")) == os.SEEK_SET

    def test_new_from_file(self, space, tmpdir):
        contents = "foo\nbar\nbaz\n"
        f = tmpdir.join("file.txt")
        f.write(contents)

        w_res = space.execute("""
        f = File.new('%s')
        io = IO.new(f)
        return io.read
        """ % f)
        assert space.str_w(w_res) == contents

    def test_new_from_fd(self, space):
        w_res = space.execute("return IO.new(1)")
        assert isinstance(w_res, W_IOObject)

    def test_write(self, space, capfd, tmpdir):
        content = "foo\n"
        space.execute('return IO.new(1, "w").write("%s")' % content)
        out, err = capfd.readouterr()
        assert out == content
        content = "foo\n"

        f = tmpdir.join("file.txt")
        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = File.new('%s', "w")
            io.close
            io.write("")
            """ % f)

    def test_push(self, space, capfd, tmpdir):
        space.execute('return IO.new(1, "w") << "hello" << "world"')
        out, err = capfd.readouterr()
        assert out == "helloworld"

        f = tmpdir.join("file.txt")
        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = File.new('%s', "w")
            io.close
            io << ""
            """ % f)

    def test_read(self, space, tmpdir):
        contents = "foo\nbar\nbaz\n"
        f = tmpdir.join("file.txt")
        f.write(contents)

        w_res = space.execute("return File.new('%s').read" % f)
        assert space.str_w(w_res) == contents

        w_res = space.execute("return File.new('%s').read(4)" % f)
        assert space.str_w(w_res) == contents[:4]

        w_res = space.execute("""
        a = 'hello world'
        File.new('%s').read(10, a)
        return a
        """ % f)
        assert space.str_w(w_res) == contents[:10]

        with self.raises(space, "ArgumentError"):
            space.execute("File.new('%s').read(-1)" % f)

        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = File.new('%s')
            io.close
            io.read
            """ % f)

    def test_simple_print(self, space, capfd, tmpdir):
        space.execute('IO.new(1, "w").print("foo")')
        out, err = capfd.readouterr()
        assert out == "foo"

        f = tmpdir.join("file.txt")
        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = File.new('%s', "w")
            io.close
            io.print ""
            """ % f)

    def test_multi_print(self, space, capfd):
        space.execute('IO.new(1, "w").print("This", "is", 100, "percent")')
        out, err = capfd.readouterr()
        assert out == "Thisis100percent"

    def test_print_globals(self, space, capfd):
        space.globals.set(space, "$,", space.newstr_fromstr(":"))
        space.globals.set(space, "$\\", space.newstr_fromstr("\n"))
        space.execute('IO.new(1, "w").print("foo", "bar", "baz")')
        space.globals.set(space, "$_", space.newstr_fromstr('lastprint'))
        space.execute('IO.new(1, "w").print')
        out, err = capfd.readouterr()
        assert out == "foo:bar:baz\nlastprint\n"

    def test_non_string_print_globals(self, space, capfd):
        space.globals.set(space, "$,", space.w_nil)
        space.globals.set(space, "$\\", space.w_nil)
        space.execute('IO.new(1, "w").print("foo", "bar", "baz")')
        space.globals.set(space, "$_", space.w_nil)
        space.execute('IO.new(1, "w").print')
        out, err = capfd.readouterr()
        assert out == "foobarbaz"

    def test_puts(self, space, capfd, tmpdir):
        space.execute("IO.new(1, 'w').puts('This', 'is\n', 100, 'percent')")
        out, err = capfd.readouterr()
        assert out == "This\nis\n100\npercent\n"

        f = tmpdir.join("file.txt")
        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = File.new('%s', "w")
            io.close
            io.puts ""
            """ % f)

    def test_flush(self, space, capfd, tmpdir):
        space.execute("IO.new(1, 'w').flush.puts('String')")
        out, err = capfd.readouterr()
        assert out == "String\n"

        f = tmpdir.join("file.txt")
        with self.raises(space, "IOError", "closed stream"):
            space.execute("""
            io = File.new('%s', "w")
            io.close
            io.flush
            """ % f)

    def test_globals(self, space, capfd):
        w_res = space.execute("""
        STDOUT.puts("STDOUT")
        $stdout.puts("$stdout")
        $>.puts("$>")
        STDERR.puts("STDERR")
        $stderr.puts("$stderr")
        return STDIN.read, $stdin.read
        """)
        out, err = capfd.readouterr()
        assert out == "STDOUT\n$stdout\n$>\n"
        assert err == "STDERR\n$stderr\n"
        assert self.unwrap(space, w_res) == [None, None]

    def test_rewind(self, space, tmpdir):
{'fraction_non_alphanumeric': 0.17396504139834407, 'fraction_numerical': 0.0060717571297148115, 'mean_word_length': 2.1051128249071693, 'pattern_counts': {'":': 2, '<': 15, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5303366', 'n_tokens_mistral': 3755, 'n_tokens_neox': 3685, 'n_words': 705}