<commit_before># encoding: utf-8

"""
.. codeauthor:: <NAME> <<EMAIL>>
"""

from __future__ import print_function, unicode_literals

import collections
import io
from decimal import Decimal
from textwrap import dedent

import pytablereader as ptr
import pytablewriter as ptw
import pytest
from path import Path
from pytablereader import InvalidTableNameError
from pytablereader.interface import TableLoader
from tabledata import TableData


Data = collections.namedtuple("Data", "value expected")

test_data_00 = Data(
    "\n".join(['"attr_a","attr_b","attr_c"', '1,4,"a"', '2,2.1,"bb"', '3,120.9,"ccc"']),
    [
        TableData(
            "tmp",
            ["attr_a", "attr_b", "attr_c"],
            [[1, 4, "a"], [2, Decimal("2.1"), "bb"], [3, Decimal("120.9"), "ccc"]],
        )
    ],
)
test_data_01 = Data(
    "\n".join(['"attr_a","attr_b","attr_c"', ' 1,4,"a"', '2, 2.1,"bb"', '3,120.9, "ccc"']),
    [
        TableData(
            "foo_bar",
            ["attr_a", "attr_b", "attr_c"],
            [
                ["attr_a", "attr_b", "attr_c"],
                [1, 4, "a"],
                [2, Decimal("2.1"), "bb"],
                [3, Decimal("120.9"), "ccc"],
            ],
        )
    ],
)
test_data_02 = Data(
    "\n".join(['3,120.9,"ccc"']),
    [TableData("foo_bar", ["attr_a", "attr_b", "attr_c"], [[3, "120.9", "ccc"]])],
)
test_data_03 = Data(
    "\n".join(['"attr_a","attr_b","attr_c"', '1,4,"a"', '2,2.1,"bb"', '3,120.9,"ccc"', "", ""]),
    [
        TableData(
            "tmp",
            ["attr_a", "attr_b", "attr_c"],
            [[1, 4, "a"], [2, Decimal("2.1"), "bb"], [3, Decimal("120.9"), "ccc"]],
        )
    ],
)
test_data_04 = Data(
    dedent(
        """\
        "attr_a","attr_b","attr_c"
        1,4,"a"
        2,2.1,"bb"
        3,120.9,"ccc"

        """
    ),
    [
        TableData(
            "tmp",
            ["attr_a", "attr_b", "attr_c"],
            [[1, 4, "a"], [2, "2.1", "bb"], [3, "120.9", "ccc"]],
        )
    ],
)
test_data_05 = Data(
    dedent(
        """\
        "姓","名","生年月日","郵便番号","住所","電話番号"
        "山田","太郎","2001/1/1","100-0002","東京都千代田区皇居外苑","03-1234-5678"
        "山田","次郎","2001/1/2","251-0036","神奈川県藤沢市江の島１丁目","03-9999-9999"
        """
    ),
    [
        TableData(
            "tmp",
            ["姓", "名", "生年月日", "郵便番号", "住所", "電話番号"],
            [
                ["山田", "太郎", "2001/1/1", "100-0002", "東京都千代田区皇居外苑", "03-1234-5678"],
                ["山田", "次郎", "2001/1/2", "251-0036", "神奈川県藤沢市江の島１丁目", "03-9999-9999"],
            ],
        )
    ],
)

test_data_06 = Data(
    dedent(
        """\
        smokey,Linux 3.0-ARCH,x86
        12345678901,12345 1234567890123,123
        12345678901,1234567890123456789,12345
        11 bytes,19 bytes,5 byt
        test line:,"Some \"\"comma, quote\"\"",foo
        skylight,Linux 3.0-ARCH,x86
        polaris,Linux 3.0-ARCH,amd64
        asgard,Windows 6.1.7600,amd64
        galileo,Windows 6.2.8102,x86
        kepler,Windows 6.2.8123,amd64
        wrfbox,Windows 6.2.8133,amd64
        """
    ),
    [
        TableData(
            "tmp",
            ["smokey", "Linux 3.0-ARCH", "x86"],
            [
                [12345678901, "12345 1234567890123", 123],
                [12345678901, 1234567890123456789, 12345],
                ["11 bytes", "19 bytes", "5 byt"],
                ["test line:", 'Some "comma, quote"', "foo"],
                ["skylight", "Linux 3.0-ARCH", "x86"],
                ["polaris", "Linux 3.0-ARCH", "amd64"],
                ["asgard", "Windows 6.1.7600", "amd64"],
                ["galileo", "Windows 6.2.8102", "x86"],
                ["kepler", "Windows 6.2.8123", "amd64"],
                ["wrfbox", "Windows 6.2.8133", "amd64"],
            ],
        )
    ],
)
test_data_multibyte = Data(
    dedent(
        """\
        "姓","名","生年月日","郵便番号","住所","電話番号"
        "山田","太郎","2001/1/1","100-0002","東京都千代田区皇居外苑","03-1234-5678"
        "山田","次郎","2001/1/2","251-0036","神奈川県藤沢市江の島１丁目","03-9999-9999"
        """
    ),
    [
        TableData(
            "multibyte",
            ["姓", "名", "生年月日", "郵便番号", "住所", "電話番号"],
            [
                ["山田", "太郎", "2001/1/1", "100-0002", "東京都千代田区皇居外苑", "03-1234-5678"],
                ["山田", "次郎", "2001/1/2", "251-0036", "神奈川県藤沢市江の島１丁目", "03-9999-9999"],
            ],
        )
    ],
)


class Test_CsvTableFileLoader_make_table_name(object):
    def setup_method(self, method):
        TableLoader.clear_table_count()

    @pytest.mark.parametrize(
        ["value", "source", "expected"],
        [
            ["%(default)s", "/path/to/data.csv", "data"],
            ["%(filename)s", "/path/to/data.csv", "data"],
            ["prefix_%(filename)s", "/path/to/data.csv", "prefix_data"],
            ["%(filename)s_suffix", "/path/to/data.csv", "data_suffix"],
            ["prefix_%(filename)s_suffix", "/path/to/data.csv", "prefix_data_suffix"],
            ["%(filename)s%(filename)s", "/path/to/data.csv", "datadata"],
            ["%(format_name)s%(format_id)s_%(filename)s", "/path/to/data.csv", "csv0_data"],
            ["%(%(filename)s)", "/path/to/data.csv", "%(data)"],
        ],
    )
    def test_normal(self, value, source, expected):
        loader = ptr.CsvTableFileLoader(source)
        loader.table_name = value

        assert loader.make_table_name() == expected

    @pytest.mark.parametrize(
        ["value", "source", "expected"],
        [
            [None, "/path/to/data.csv", ValueError],
            ["", "/path/to/data.csv", ValueError],
            ["%(filename)s", None, InvalidTableNameError],
            ["%(filename)s", "", InvalidTableNameError],
        ],
    )
    def test_exception(self, value, source, expected):
        loader = ptr.CsvTableFileLoader(source)
        loader.table_name = value

        with pytest.raises(expected):
            loader.make_table_name()


class Test_CsvTableFileLoader_load(object):
    def setup_method(self, method):
        TableLoader.clear_table_count()

    @pytest.mark.parametrize(
        ["test_id", "table_text", "filename", "header_list", "expected"],
        [
            [0, test_data_00.value, "tmp.csv", [], test_data_00.expected],
            [
                1,
                test_data_01.value,
                "hoge/foo_bar.csv",
                ["attr_a", "attr_b", "attr_c"],
                test_data_01.expected,
            ],
            [
                2,
                test_data_02.value,
                "hoge/foo_bar.csv",
                ["attr_a", "attr_b", "attr_c"],
                test_data_02.expected,
            ],
            [3, test_data_03.value, "tmp.csv", [], test_data_03.expected],
            [4, test_data_04.value, "tmp.csv", [], test_data_04.expected],
            [5, test_data_05.value, "tmp.csv", [], test_data_05.expected],
            [6, test_data_06.value, "tmp.csv", [], test_data_06.expected],
        ],
    )
    def test_normal(self, tmpdir, test_id, table_text, filename, header_list, expected):
        file_path = Path(str(tmpdir.join(filename)))
        file_path.parent.makedirs_p()

        with io.open(file_path, "w", encoding="utf-8") as f:
            f.write(table_text)

        loader = ptr.CsvTableFileLoader(file_path)
        loader.header_list = header_list

        for tabledata in loader.load():
            print("test-id={}".format(test_id))
            print(ptw.dump_tabledata(tabledata))

            assert tabledata.in_tabledata_list(expected)

    @pytest.mark.parametrize(
        ["test_id", "table_text", "filename", "encoding", "header_list", "expected"],
        [
            [
                7,
                test_data_multibyte.value,
                "multibyte.csv",
                "utf16",
                [],
                test_data_multibyte.expected,
            ]
        ],
    )
    def test_normal_multibyte(
        self, tmpdir, test_id, table_text, filename, encoding, header_list, expected
    ):
        file_path = Path(str(tmpdir.join(filename)))
        file_path.parent.makedirs_p()

        with io.open(file_path, "w", encoding=encoding) as f:
            f.write(table_text)

        loader = ptr.CsvTableFileLoader(file_path)
        loader.header_list = header_list

        for tabledata in loader.load():
            print("test-id={}".format(test_id))
            print(ptw.dump_tabledata(tabledata))

            assert tabledata in expected

    @pytest.mark.parametrize(
        ["table_text", "filename", "header_list", "expected"],
        [
            ["", "hoge.csv", [], ptr.DataError],
            ["\n".join(['"attr_a","attr_b","attr_c"']), "hoge.csv", [], ptr.DataError],
            ["\n".join([]), "hoge.csv", ["attr_a", "attr_b", "attr_c"], ptr.DataError],
        ],
    )
    def test_exception(self, tmpdir, table_text, filename, header_list, expected):
        p_csv = tmpdir.join(filename)<commit_msg>Modify an assert<commit_after># encoding: utf-8

"""
.. codeauthor:: <NAME> <<EMAIL>>
"""

from __future__ import print_function, unicode_literals

import collections
import io
from decimal import Decimal
from textwrap import dedent

import pytablereader as ptr
import pytablewriter as ptw
import pytest
from path import Path
from pytablereader import InvalidTableNameError
from pytablereader.interface import TableLoader
from tabledata import TableData


Data = collections.namedtuple("Data", "value expected")

test_data_00 = Data(
    "\n".join(['"attr_a","attr_b","attr_c"', '1,4,"a"', '2,2.1,"bb"', '3,120.9,"ccc"']),
    [
        TableData(
            "tmp",
            ["attr_a", "attr_b", "attr_c"],
            [[1, 4, "a"], [2, Decimal("2.1"), "bb"], [3, Decimal("120.9"), "ccc"]],
        )
    ],
)
test_data_01 = Data(
    "\n".join(['"attr_a","attr_b","attr_c"', ' 1,4,"a"', '2, 2.1,"bb"', '3,120.9, "ccc"']),
    [
        TableData(
            "foo_bar",
            ["attr_a", "attr_b", "attr_c"],
            [
                ["attr_a", "attr_b", "attr_c"],
                [1, 4, "a"],
                [2, Decimal("2.1"), "bb"],
                [3, Decimal("120.9"), "ccc"],
            ],
        )
    ],
)
test_data_02 = Data(
    "\n".join(['3,120.9,"ccc"']),
    [TableData("foo_bar", ["attr_a", "attr_b", "attr_c"], [[3, "120.9", "ccc"]])],
)
test_data_03 = Data(
    "\n".join(['"attr_a","attr_b","attr_c"', '1,4,"a"', '2,2.1,"bb"', '3,120.9,"ccc"', "", ""]),
    [
        TableData(
            "tmp",
            ["attr_a", "attr_b", "attr_c"],
            [[1, 4, "a"], [2, Decimal("2.1"), "bb"], [3, Decimal("120.9"), "ccc"]],
        )
    ],
)
test_data_04 = Data(
    dedent(
        """\
        "attr_a","attr_b","attr_c"
        1,4,"a"
        2,2.1,"bb"
        3,120.9,"ccc"

        """
    ),
    [
        TableData(
            "tmp",
            ["attr_a", "attr_b", "attr_c"],
            [[1, 4, "a"], [2, "2.1", "bb"], [3, "120.9", "ccc"]],
        )
    ],
)
test_data_05 = Data(
    dedent(
        """\
        "姓","名","生年月日","郵便番号","住所","電話番号"
        "山田","太郎","2001/1/1","100-0002","東京都千代田区皇居外苑","03-1234-5678"
        "山田","次郎","2001/1/2","251-0036","神奈川県藤沢市江の島１丁目","03-9999-9999"
        """
    ),
    [
        TableData(
            "tmp",
            ["姓", "名", "生年月日", "郵便番号", "住所", "電話番号"],
            [
                ["山田", "太郎", "2001/1/1", "100-0002", "東京都千代田区皇居外苑", "03-1234-5678"],
                ["山田", "次郎", "2001/1/2", "251-0036", "神奈川県藤沢市江の島１丁目", "03-9999-9999"],
            ],
        )
    ],
)

test_data_06 = Data(
    dedent(
        """\
        smokey,Linux 3.0-ARCH,x86
        12345678901,12345 1234567890123,123
        12345678901,1234567890123456789,12345
        11 bytes,19 bytes,5 byt
        test line:,"Some \"\"comma, quote\"\"",foo
        skylight,Linux 3.0-ARCH,x86
        polaris,Linux 3.0-ARCH,amd64
        asgard,Windows 6.1.7600,amd64
        galileo,Windows 6.2.8102,x86
        kepler,Windows 6.2.8123,amd64
        wrfbox,Windows 6.2.8133,amd64
        """
    ),
    [
        TableData(
            "tmp",
            ["smokey", "Linux 3.0-ARCH", "x86"],
            [
                [12345678901, "12345 1234567890123", 123],
                [12345678901, 1234567890123456789, 12345],
                ["11 bytes", "19 bytes", "5 byt"],
                ["test line:", 'Some "comma, quote"', "foo"],
                ["skylight", "Linux 3.0-ARCH", "x86"],
                ["polaris", "Linux 3.0-ARCH", "amd64"],
                ["asgard", "Windows 6.1.7600", "amd64"],
                ["galileo", "Windows 6.2.8102", "x86"],
                ["kepler", "Windows 6.2.8123", "amd64"],
                ["wrfbox", "Windows 6.2.8133", "amd64"],
            ],
        )
    ],
)
test_data_multibyte = Data(
    dedent(
        """\
        "姓","名","生年月日","郵便番号","住所","電話番号"
        "山田","太郎","2001/1/1","100-0002","東京都千代田区皇居外苑","03-1234-5678"
        "山田","次郎","2001/1/2","251-0036","神奈川県藤沢市江の島１丁目","03-9999-9999"
        """
    ),
    [
        TableData(
            "multibyte",
            ["姓", "名", "生年月日", "郵便番号", "住所", "電話番号"],
            [
                ["山田", "太郎", "2001/1/1", "100-0002", "東京都千代田区皇居外苑", "03-1234-5678"],
                ["山田", "次郎", "2001/1/2", "251-0036", "神奈川県藤沢市江の島１丁目", "03-9999-9999"],
            ],
        )
    ],
)


class Test_CsvTableFileLoader_make_table_name(object):
    def setup_method(self, method):
        TableLoader.clear_table_count()

    @pytest.mark.parametrize(
        ["value", "source", "expected"],
        [
            ["%(default)s", "/path/to/data.csv", "data"],
            ["%(filename)s", "/path/to/data.csv", "data"],
            ["prefix_%(filename)s", "/path/to/data.csv", "prefix_data"],
            ["%(filename)s_suffix", "/path/to/data.csv", "data_suffix"],
            ["prefix_%(filename)s_suffix", "/path/to/data.csv", "prefix_data_suffix"],
            ["%(filename)s%(filename)s", "/path/to/data.csv", "datadata"],
            ["%(format_name)s%(format_id)s_%(filename)s", "/path/to/data.csv", "csv0_data"],
            ["%(%(filename)s)", "/path/to/data.csv", "%(data)"],
        ],
    )
    def test_normal(self, value, source, expected):
        loader = ptr.CsvTableFileLoader(source)
        loader.table_name = value

        assert loader.make_table_name() == expected

    @pytest.mark.parametrize(
        ["value", "source", "expected"],
        [
            [None, "/path/to/data.csv", ValueError],
            ["", "/path/to/data.csv", ValueError],
            ["%(filename)s", None, InvalidTableNameError],
            ["%(filename)s", "", InvalidTableNameError],
        ],
    )
    def test_exception(self, value, source, expected):
        loader = ptr.CsvTableFileLoader(source)
        loader.table_name = value

        with pytest.raises(expected):
            loader.make_table_name()


class Test_CsvTableFileLoader_load(object):
    def setup_method(self, method):
        TableLoader.clear_table_count()

    @pytest.mark.parametrize(
        ["test_id", "table_text", "filename", "header_list", "expected"],
        [
            [0, test_data_00.value, "tmp.csv", [], test_data_00.expected],
            [
                1,
                test_data_01.value,
                "hoge/foo_bar.csv",
                ["attr_a", "attr_b", "attr_c"],
                test_data_01.expected,
            ],
            [
                2,
                test_data_02.value,
                "hoge/foo_bar.csv",
                ["attr_a", "attr_b", "attr_c"],
                test_data_02.expected,
            ],
            [3, test_data_03.value, "tmp.csv", [], test_data_03.expected],
            [4, test_data_04.value, "tmp.csv", [], test_data_04.expected],
            [5, test_data_05.value, "tmp.csv", [], test_data_05.expected],
            [6, test_data_06.value, "tmp.csv", [], test_data_06.expected],
        ],
    )
    def test_normal(self, tmpdir, test_id, table_text, filename, header_list, expected):
        file_path = Path(str(tmpdir.join(filename)))
        file_path.parent.makedirs_p()

        with io.open(file_path, "w", encoding="utf-8") as f:
            f.write(table_text)

        loader = ptr.CsvTableFileLoader(file_path)
        loader.header_list = header_list

        for tabledata in loader.load():
            print("test-id={}".format(test_id))
            print(ptw.dump_tabledata(tabledata))

            assert tabledata.in_tabledata_list(expected)

    @pytest.mark.parametrize(
        ["test_id", "table_text", "filename", "encoding", "header_list", "expected"],
        [
            [
                7,
                test_data_multibyte.value,
                "multibyte.csv",
                "utf16",
                [],
                test_data_multibyte.expected,
            ]
        ],
    )
    def test_normal_multibyte(
        self, tmpdir, test_id, table_text, filename, encoding, header_list, expected
    ):
        file_path = Path(str(tmpdir.join(filename)))
        file_path.parent.makedirs_p()

        with io.open(file_path, "w", encoding=encoding) as f:
            f.write(table_text)

        loader = ptr.CsvTableFileLoader(file_path)
        loader.header_list = header_list

        for tabledata in loader.load():
            print("test-id={}".format(test_id))
            print(ptw.dump_tabledata(tabledata))

            assert tabledata.in_tabledata_list(expected)

    @pytest.mark.parametrize(
        ["table_text", "filename", "header_list", "expected"],
        [
            ["", "hoge.csv", [], ptr.DataError],
            ["\n".join(['"attr_a","attr_b","attr_c"']), "hoge.csv", [], ptr.DataError],
            ["\n".join([]), "hoge.csv", ["attr_a", "attr_b", "attr_c"], ptr.DataError],
        ],
    )
    def test_exception(self, tmpdir, table_text, filename, header_list, expected):
        p_csv = tmpdir.join(filename)
{'fraction_non_alphanumeric': 0.1911838932196736, 'fraction_numerical': 0.06337277774662105, 'mean_word_length': 2.027504244482173, 'pattern_counts': {'":': 0, '<': 9, '<?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': '4704402', 'n_tokens_mistral': 7225, 'n_tokens_neox': 6649, 'n_words': 993}