# Copyright (c) 2008-2012 testtools developers. See LICENSE for details.

import io
import os
import tempfile
import unittest

from testtools import TestCase
from testtools.compat import (
    _b,
)
from testtools.content import (
    attach_file,
    Content,
    content_from_file,
    content_from_stream,
    JSON,
    json_content,
    StackLinesContent,
    StacktraceContent,
    TracebackContent,
    text_content,
)
from testtools.content_type import (
    ContentType,
    UTF8_TEXT,
)
from testtools.matchers import (
    Equals,
    MatchesException,
    Raises,
    raises,
)
from testtools.tests.helpers import an_exc_info


raises_value_error = Raises(MatchesException(ValueError))


class TestContent(TestCase):
    def test___init___None_errors(self):
        self.assertThat(lambda: Content(None, None), raises_value_error)
        self.assertThat(
            lambda: Content(None, lambda: ["traceback"]), raises_value_error
        )
        self.assertThat(
            lambda: Content(ContentType("text", "traceback"), None), raises_value_error
        )

    def test___init___sets_ivars(self):
        content_type = ContentType("foo", "bar")
        content = Content(content_type, lambda: ["bytes"])
        self.assertEqual(content_type, content.content_type)
        self.assertEqual(["bytes"], list(content.iter_bytes()))

    def test___eq__(self):
        content_type = ContentType("foo", "bar")

        def one_chunk():
            return [_b("bytes")]

        def two_chunk():
            return [_b("by"), _b("tes")]

        content1 = Content(content_type, one_chunk)
        content2 = Content(content_type, one_chunk)
        content3 = Content(content_type, two_chunk)
        content4 = Content(content_type, lambda: [_b("by"), _b("te")])
        content5 = Content(ContentType("f", "b"), two_chunk)
        self.assertEqual(content1, content2)
        self.assertEqual(content1, content3)
        self.assertNotEqual(content1, content4)
        self.assertNotEqual(content1, content5)

    def test___repr__(self):
        content = Content(
            ContentType("application", "octet-stream"),
            lambda: [_b("\x00bin"), _b("ary\xff")],
        )
        self.assertIn("\\x00binary\\xff", repr(content))

    def test_iter_text_not_text_errors(self):
        content_type = ContentType("foo", "bar")
        content = Content(content_type, lambda: ["bytes"])
        self.assertThat(content.iter_text, raises_value_error)

    def test_iter_text_decodes(self):
        content_type = ContentType("text", "strange", {"charset": "utf8"})
        content = Content(content_type, lambda: ["bytes\xea".encode()])
        self.assertEqual(["bytes\xea"], list(content.iter_text()))

    def test_iter_text_default_charset_iso_8859_1(self):
        content_type = ContentType("text", "strange")
        text = "bytes\xea"
        iso_version = text.encode("ISO-8859-1")
        content = Content(content_type, lambda: [iso_version])
        self.assertEqual([text], list(content.iter_text()))

    def test_as_text(self):
        content_type = ContentType("text", "strange", {"charset": "utf8"})
        content = Content(content_type, lambda: ["bytes\xea".encode()])
        self.assertEqual("bytes\xea", content.as_text())

    def test_from_file(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        os.write(fd, _b("some data"))
        os.close(fd)
        content = content_from_file(path, UTF8_TEXT, chunk_size=2)
        self.assertThat(
            list(content.iter_bytes()),
            Equals([_b("so"), _b("me"), _b(" d"), _b("at"), _b("a")]),
        )

    def test_from_nonexistent_file(self):
        directory = tempfile.mkdtemp()
        nonexistent = os.path.join(directory, "nonexistent-file")
        content = content_from_file(nonexistent)
        self.assertThat(content.iter_bytes, raises(IOError))

    def test_from_file_default_type(self):
        content = content_from_file("/nonexistent/path")
        self.assertThat(content.content_type, Equals(UTF8_TEXT))

    def test_from_file_eager_loading(self):
        fd, path = tempfile.mkstemp()
        os.write(fd, _b("some data"))
        os.close(fd)
        content = content_from_file(path, UTF8_TEXT, buffer_now=True)
        os.remove(path)
        self.assertThat("".join(content.iter_text()), Equals("some data"))

    def test_from_file_with_simple_seek(self):
        f = tempfile.NamedTemporaryFile()
        f.write(_b("some data"))
        f.flush()
        self.addCleanup(f.close)
        content = content_from_file(f.name, UTF8_TEXT, chunk_size=50, seek_offset=5)
        self.assertThat(list(content.iter_bytes()), Equals([_b("data")]))

    def test_from_file_with_whence_seek(self):
        f = tempfile.NamedTemporaryFile()
        f.write(_b("some data"))
        f.flush()
        self.addCleanup(f.close)
        content = content_from_file(
            f.name, UTF8_TEXT, chunk_size=50, seek_offset=-4, seek_whence=2
        )
        self.assertThat(list(content.iter_bytes()), Equals([_b("data")]))

    def test_from_stream(self):
        data = io.StringIO("some data")
        content = content_from_stream(data, UTF8_TEXT, chunk_size=2)
        self.assertThat(
            list(content.iter_bytes()), Equals(["so", "me", " d", "at", "a"])
        )

    def test_from_stream_default_type(self):
        data = io.StringIO("some data")
        content = content_from_stream(data)
        self.assertThat(content.content_type, Equals(UTF8_TEXT))

    def test_from_stream_eager_loading(self):
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        self.addCleanup(os.close, fd)
        os.write(fd, _b("some data"))
        stream = open(path, "rb")
        self.addCleanup(stream.close)
        content = content_from_stream(stream, UTF8_TEXT, buffer_now=True)
        os.write(fd, _b("more data"))
        self.assertThat("".join(content.iter_text()), Equals("some data"))

    def test_from_stream_with_simple_seek(self):
        data = io.BytesIO(_b("some data"))
        content = content_from_stream(data, UTF8_TEXT, chunk_size=50, seek_offset=5)
        self.assertThat(list(content.iter_bytes()), Equals([_b("data")]))

    def test_from_stream_with_whence_seek(self):
        data = io.BytesIO(_b("some data"))
        content = content_from_stream(
            data, UTF8_TEXT, chunk_size=50, seek_offset=-4, seek_whence=2
        )
        self.assertThat(list(content.iter_bytes()), Equals([_b("data")]))

    def test_from_text(self):
        data = "some data"
        expected = Content(UTF8_TEXT, lambda: [data.encode("utf8")])
        self.assertEqual(expected, text_content(data))

    def test_text_content_raises_TypeError_when_passed_bytes(self):
        data = _b("Some Bytes")
        self.assertRaises(TypeError, text_content, data)

    def test_text_content_raises_TypeError_when_passed_non_text(self):
        bad_values = (None, list(), dict(), 42, 1.23)
        for value in bad_values:
            self.assertThat(
                lambda: text_content(value),
                raises(
                    TypeError(
                        "text_content must be given text, not '%s'."
                        % type(value).__name__
                    )
                ),
            )

    def test_json_content(self):
        data = {"foo": "bar"}
        expected = Content(JSON, lambda: [_b('{"foo": "bar"}')])
        self.assertEqual(expected, json_content(data))


class TestStackLinesContent(TestCase):
    def _get_stack_line_and_expected_output(self):
        stack_lines = [
            ("/path/to/file", 42, "some_function", 'print("Hello World")'),
        ]
        expected = (
            '  File "/path/to/file", line 42, in some_function\n'
            '    print("Hello World")\n'
        )
        return stack_lines, expected

    def test_single_stack_line(self):
        stack_lines, expected = self._get_stack_line_and_expected_output()
        actual = StackLinesContent(stack_lines).as_text()

        self.assertEqual(expected, actual)

    def test_prefix_content(self):
        stack_lines, expected = self._get_stack_line_and_expected_output()
        prefix = self.getUniqueString() + "\n"
        content = StackLinesContent(stack_lines, prefix_content=prefix)
        actual = content.as_text()
        expected = prefix + expected

        self.assertEqual(expected, actual)

    def test_postfix_content(self):
        stack_lines, expected = self._get_stack_line_and_expected_output()
        postfix = "\n" + self.getUniqueString()
        content = StackLinesContent(stack_lines, postfix_content=postfix)
        actual = content.as_text()
        expected = expected + postfix

        self.assertEqual(expected, actual)

    def test___init___sets_content_type(self):
        stack_lines, expected = self._get_stack_line_and_expected_output()
        content = StackLinesContent(stack_lines)
        expected_content_type = ContentType(
            "text", "x-traceback", {"language": "python", "charset": "utf8"}
        )

        self.assertEqual(expected_content_type, content.content_type)


class TestTracebackContent(TestCase):
    def test___init___None_errors(self):
        self.assertThat(lambda: TracebackContent(None, None), raises_value_error)

    def test___init___sets_ivars(self):
        content = TracebackContent(an_exc_info, self)
        content_type = ContentType(
            "text", "x-traceback", {"language": "python", "charset": "utf8"}
        )
        self.assertEqual(content_type, content.content_type)
        result = unittest.TestResult()
        expected = result._exc_info_to_string(an_exc_info, self)
        self.assertEqual(expected, "".join(list(content.iter_text())))


class TestStacktraceContent(TestCase):
    def test___init___sets_ivars(self):
        content = StacktraceContent()
        content_type = ContentType(
            "text", "x-traceback", {"language": "python", "charset": "utf8"}
        )
        self.assertEqual(content_type, content.content_type)

    def test_prefix_is_used(self):
        prefix = self.getUniqueString()
        actual = StacktraceContent(prefix_content=prefix).as_text()
        self.assertTrue(actual.startswith(prefix))

    def test_postfix_is_used(self):
        postfix = self.getUniqueString()
        actual = StacktraceContent(postfix_content=postfix).as_text()
        self.assertTrue(actual.endswith(postfix))

    def test_top_frame_is_skipped_when_no_stack_is_specified(self):
        actual = StacktraceContent().as_text()
        self.assertNotIn("testtools/content.py", actual)


class TestAttachFile(TestCase):
    def make_file(self, data):
        # GZ 2011-04-21: This helper could be useful for methods above trying
        #                to use mkstemp, but should handle write failures and
        #                always close the fd. There must be a better way.
        fd, path = tempfile.mkstemp()
        self.addCleanup(os.remove, path)
        os.write(fd, _b(data))
        os.close(fd)
        return path

    def test_simple(self):
        class SomeTest(TestCase):
            def test_foo(self):
                pass

        test = SomeTest("test_foo")
        data = "some data"
        path = self.make_file(data)
        my_content = text_content(data)
        attach_file(test, path, name="foo")
        self.assertEqual({"foo": my_content}, test.getDetails())

    def test_optional_name(self):
        # If no name is provided, attach_file just uses the base name of the
        # file.
        class SomeTest(TestCase):
            def test_foo(self):
                pass

        test = SomeTest("test_foo")
        path = self.make_file("some data")
        base_path = os.path.basename(path)
        attach_file(test, path)
        self.assertEqual([base_path], list(test.getDetails()))

    def test_lazy_read(self):
        class SomeTest(TestCase):
            def test_foo(self):
                pass

        test = SomeTest("test_foo")
        path = self.make_file("some data")
        attach_file(test, path, name="foo", buffer_now=False)
        content = test.getDetails()["foo"]
        content_file = open(path, "w")
        content_file.write("new data")
        content_file.close()
        self.assertEqual("".join(content.iter_text()), "new data")

    def test_eager_read_by_default(self):
        class SomeTest(TestCase):
            def test_foo(self):
                pass

        test = SomeTest("test_foo")
        path = self.make_file("some data")
        attach_file(test, path, name="foo")
        content = test.getDetails()["foo"]
        content_file = open(path, "w")
        content_file.write("new data")
        content_file.close()
        self.assertEqual("".join(content.iter_text()), "some data")


def test_suite():
    from unittest import TestLoader

    return TestLoader().loadTestsFromName(__name__)