<commit_before>import asyncio
from unittest.mock import patch, MagicMock
from pymysql.err import OperationalError

import os
import builtins
from tests._testutils import run_until_complete
from tests.base import AIOPyMySQLTestCase


class TestLoadLocal(AIOPyMySQLTestCase):
    def setUp(self):
        super().setUp()
        self.loop.run_until_complete(self._prepare_table())

    def tearDown(self):
        self.loop.run_until_complete(self._drop_table())
        super().setUp()

    @asyncio.coroutine
    def _prepare_table(self):
        conn = self.connections[2]
        c = yield from conn.cursor()
        yield from c.execute("DROP TABLE IF EXISTS test_load_local;")
        yield from c.execute("CREATE TABLE test_load_local "
                             "(a INTEGER, b INTEGER)")
        yield from c.close()

    def _drop_table(self):
        conn = self.connections[2]
        c = yield from conn.cursor()
        yield from c.execute("DROP TABLE test_load_local")
        yield from c.close()

    @run_until_complete
    def test_no_file(self):
        # Test load local infile when the file does not exist
        conn = self.connections[2]
        c = yield from conn.cursor()
        with self.assertRaises(OperationalError):
            yield from c.execute("LOAD DATA LOCAL INFILE 'no_data.txt'"
                                 " INTO TABLE test_load_local fields "
                                 "terminated by ','")
        yield from c.close()

    @run_until_complete
    def test_error_on_file_read(self):
        # Test exception while reading file
        conn = self.connections[2]
        c = yield from conn.cursor()

        with patch.object(builtins, 'open') as open_mocked:
            m = MagicMock()
            m.read.side_effect = OperationalError(1024, 'Error reading file')
            m.close.return_value = None
            open_mocked.return_value = m

            with self.assertRaises(OperationalError):
                yield from c.execute("LOAD DATA LOCAL INFILE 'some.txt'"
                                     " INTO TABLE test_load_local fields "
                                     "terminated by ','")
        yield from c.close()

    @run_until_complete
    def test_load_file(self):
        # Test load local infile with a valid file
        conn = self.connections[2]
        c = yield from conn.cursor()
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'fixtures',
                                'load_local_data.txt')
        yield from c.execute(
            ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
             "test_load_local FIELDS TERMINATED BY ','").format(filename)
        )
        yield from c.execute("SELECT COUNT(*) FROM test_load_local")
        resp = yield from c.fetchone()
        yield from c.close()
        self.assertEqual(22749, resp[0])

    @run_until_complete
    def test_load_warnings(self):
        # Test load local infile produces the appropriate warnings
        import warnings

        conn = self.connections[2]
        c = yield from conn.cursor()
        filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                'fixtures',
                                'load_local_warn_data.txt')

        with warnings.catch_warnings(record=True) as w:
            yield from c.execute(
                ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
                 "test_load_local FIELDS TERMINATED BY ','").format(
                    filename)
            )
            self.assertEqual(True, "Incorrect integer value"
                             in str(w[-1].message))<commit_msg>move local file tests to fixtures<commit_after>import asyncio
import builtins
import os
from unittest.mock import patch, MagicMock

import pytest
from pymysql.err import OperationalError


@pytest.yield_fixture
def table_local_file(connection, loop):

    @asyncio.coroutine
    def prepare_table(conn):
        c = yield from conn.cursor()
        yield from c.execute("DROP TABLE IF EXISTS test_load_local;")
        yield from c.execute("CREATE TABLE test_load_local "
                             "(a INTEGER, b INTEGER)")
        yield from c.close()

    @asyncio.coroutine
    def drop_table(conn):
        c = yield from conn.cursor()
        yield from c.execute("DROP TABLE test_load_local")
        yield from c.close()

    loop.run_until_complete(prepare_table(connection))
    yield
    loop.run_until_complete(drop_table(connection))


@pytest.mark.run_loop
def test_no_file(connection, table_local_file):
    # Test load local infile when the file does not exist
    c = yield from connection.cursor()
    sql = "LOAD DATA LOCAL INFILE 'no_data.txt'" + \
          " INTO TABLE test_load_local fields " + \
          "terminated by ','"
    with pytest.raises(OperationalError):
        yield from c.execute(sql)
    yield from c.close()


@pytest.mark.run_loop
def test_error_on_file_read(cursor, table_local_file):

    with patch.object(builtins, 'open') as open_mocked:
        m = MagicMock()
        m.read.side_effect = OperationalError(1024, 'Error reading file')
        m.close.return_value = None
        open_mocked.return_value = m

        with pytest.raises(OperationalError):
            yield from cursor.execute("LOAD DATA LOCAL INFILE 'some.txt'"
                                      " INTO TABLE test_load_local fields "
                                      "terminated by ','")
    yield from cursor.close()


@pytest.mark.run_loop
def test_load_file(connection, table_local_file):
    # Test load local infile with a valid file
    c = yield from connection.cursor()
    filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                            'fixtures',
                            'load_local_data.txt')
    yield from c.execute(
        ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
         "test_load_local FIELDS TERMINATED BY ','").format(filename)
    )
    yield from c.execute("SELECT COUNT(*) FROM test_load_local")
    resp = yield from c.fetchone()
    yield from c.close()
    assert 22749 == resp[0]


@pytest.mark.run_loop
def test_load_warnings(connection, table_local_file):
    # Test load local infile produces the appropriate warnings
    import warnings

    c = yield from connection.cursor()

    # TODO: Move to pathlib
    filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                            'fixtures', 'load_local_warn_data.txt')

    sql = ("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
           "test_load_local FIELDS TERMINATED BY ','").format(filename)

    with warnings.catch_warnings(record=True) as w:
        yield from c.execute(sql)
    assert "Incorrect integer value" in str(w[-1].message)
{'fraction_non_alphanumeric': 0.0918157817541801, 'fraction_numerical': 0.004693458492226459, 'mean_word_length': 2.205923836389281, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1518888', 'n_tokens_mistral': 1988, 'n_tokens_neox': 1915, 'n_words': 510}