Class TestMssqlCliClientMultipleStatement

tests/test_mssqlcliclient.py:226–261

class TestMssqlCliClientMultipleStatement(MssqlCliClient):
    test_data = [
        # random strings used to test for errors with non-existing tables
        (u"select 'Morning' as [Name] UNION ALL select 'Evening'; " +
         u"select 1;", [2, 1]),
        (u"select 1; select 'foo' from %s;" % random_str(), [1, 0]),
        (u"select 'foo' from %s; select 2;" % random_str(), [0, 1])
    ]

    @staticmethod
    @pytest.mark.parametrize("query_str, rows_outputted", test_data)
    def test_mssqlcliclient_multiple_statement(client, query_str, rows_outputted):
        """
            Verify correct execution of queries separated by semi-colon
        """

        for (i, (rows, _, _, query_executed, is_error)) in \
            enumerate(client.execute_query(query_str)):

            queries = ["{};".format(query) for query in query_str.split(';')]
            assert query_executed == queries[i].strip()
            assert len(rows) == rows_outputted[i]
            assert (is_error and len(rows) == 0) or (not is_error)

    @staticmethod
    def test_mssqlcliclient_multiple_merge(client_with_db):
        """
        Tests query with multiple merges. Requires creation of temp db.
        """
        file_input, _ = get_io_paths('multiple_merge.txt')
        query_input = get_file_contents(file_input)

        for rows, _, status, _, is_error in client_with_db.execute_query(query_input):
            if is_error:
                raise AssertionError("Query execution failed: {}".format(status))
            assert () == rows