'Test insertion of empty strings and null values'

def test_can_insert_empty_strings_and_nulls(self):

    s = self.session
    alpha_type_list = ['zz    int    PRIMARY    KEY']
    col_names = []
    string_types = set(('ascii', 'text', 'varchar'))
    string_columns = set('')
    non_string_types = ((PRIMITIVE_DATATYPES - string_types) - set(('blob', 'date', 'inet', 'time', 'timestamp')))
    non_string_columns = set()
    start_index = ord('a')
    for (i, datatype) in enumerate(PRIMITIVE_DATATYPES):
        col_name = chr((start_index + i))
        alpha_type_list.append('{0}    {1}'.format(col_name, datatype))
        col_names.append(col_name)
        if (datatype in non_string_types):
            non_string_columns.add(col_name)
        if (datatype in string_types):
            string_columns.add(col_name)
    execute_until_pass(s, 'CREATE    TABLE    all_empty    ({0})'.format(',    '.join(alpha_type_list)))
    columns_string = ','.join(col_names)
    s.execute('INSERT    INTO    all_empty    (zz)    VALUES    (2)')
    results = s.execute('SELECT    {0}    FROM    all_empty    WHERE    zz=2'.format(columns_string))[0]
    self.assertTrue(all(((x is None) for x in results)))
    select = s.prepare('SELECT    {0}    FROM    all_empty    WHERE    zz=?'.format(columns_string))
    results = s.execute(select.bind([2]))[0]
    self.assertTrue(all(((x is None) for x in results)))
    expected_values = dict(((col, '') for col in string_columns))
    columns_string = ','.join(string_columns)
    placeholders = ','.join((['%s'] * len(string_columns)))
    s.execute('INSERT    INTO    all_empty    (zz,    {0})    VALUES    (3,    {1})'.format(columns_string, placeholders), expected_values.values())
    results = s.execute('SELECT    {0}    FROM    all_empty    WHERE    zz=3'.format(columns_string))[0]
    for (expected, actual) in zip(expected_values.values(), results):
        self.assertEqual(actual, expected)
    results = s.execute(s.prepare('SELECT    {0}    FROM    all_empty    WHERE    zz=?'.format(columns_string)), [3])[0]
    for (expected, actual) in zip(expected_values.values(), results):
        self.assertEqual(actual, expected)
    for col in non_string_columns:
        query = 'INSERT    INTO    all_empty    (zz,    {0})    VALUES    (4,    %s)'.format(col)
        with self.assertRaises(InvalidRequest):
            s.execute(query, [''])
        insert = s.prepare('INSERT    INTO    all_empty    (zz,    {0})    VALUES    (4,    ?)'.format(col))
        with self.assertRaises(TypeError):
            s.execute(insert, [''])
    params = []
    for datatype in PRIMITIVE_DATATYPES:
        params.append(get_sample(datatype))
    columns_string = ','.join(col_names)
    placeholders = ','.join((['%s'] * len(col_names)))
    simple_insert = 'INSERT    INTO    all_empty    (zz,    {0})    VALUES    (5,    {1})'.format(columns_string, placeholders)
    s.execute(simple_insert, params)
    null_values = ([None] * len(col_names))
    s.execute(simple_insert, null_values)
    query = 'SELECT    {0}    FROM    all_empty    WHERE    zz=5'.format(columns_string)
    results = s.execute(query)[0]
    for col in results:
        self.assertEqual(None, col)
    select = s.prepare('SELECT    {0}    FROM    all_empty    WHERE    zz=?'.format(columns_string))
    results = s.execute(select.bind([5]))[0]
    for col in results:
        self.assertEqual(None, col)
    s.execute(simple_insert, params)
    placeholders = ','.join((['?'] * len(col_names)))
    insert = s.prepare('INSERT    INTO    all_empty    (zz,    {0})    VALUES    (5,    {1})'.format(columns_string, placeholders))
    s.execute(insert, null_values)
    results = s.execute(query)[0]
    for col in results:
        self.assertEqual(None, col)
    results = s.execute(select.bind([5]))[0]
    for col in results:
        self.assertEqual(None, col)
