ansible/test/units/TestModuleUtilsDatabase.py

import collections
import mock
import os
import re

from nose.tools import eq_
try:
    from nose.tools import assert_raises_regexp
except ImportError:
    # Python < 2.7
    def assert_raises_regexp(expected, regexp, callable, *a, **kw):
        try:
            callable(*a, **kw)
        except expected as e:
            if isinstance(regexp, basestring):
                regexp = re.compile(regexp)
            if not regexp.search(str(e)):
                raise Exception('"%s" does not match "%s"' %
                                (regexp.pattern, str(e)))
        else:
            if hasattr(expected,'__name__'): excName = expected.__name__
            else: excName = str(expected)
            raise AssertionError("%s not raised" % excName)

from ansible.module_utils.database import (
    pg_quote_identifier,
    SQLParseError,
)


# Note: Using nose's generator test cases here so we can't inherit from
# unittest.TestCase
class TestQuotePgIdentifier(object):

    # These are all valid strings
    # The results are based on interpreting the identifier as a table name
    valid = {
        # User quoted
        '"public.table"': '"public.table"',
        '"public"."table"': '"public"."table"',
        '"schema test"."table test"': '"schema test"."table test"',

        # We quote part
        'public.table': '"public"."table"',
        '"public".table': '"public"."table"',
        'public."table"': '"public"."table"',
        'schema test.table test': '"schema test"."table test"',
        '"schema test".table test': '"schema test"."table test"',
        'schema test."table test"': '"schema test"."table test"',

        # Embedded double quotes
        'table "test"': '"table ""test"""',
        'public."table ""test"""': '"public"."table ""test"""',
        'public.table "test"': '"public"."table ""test"""',
        'schema "test".table': '"schema ""test"""."table"',
        '"schema ""test""".table': '"schema ""test"""."table"',
        '"""wat"""."""test"""': '"""wat"""."""test"""',
        # Sigh, handle these as well:
        '"no end quote': '"""no end quote"',
        'schema."table': '"schema"."""table"',
        '"schema.table': '"""schema"."table"',
        'schema."table.something': '"schema"."""table"."something"',

        # Embedded dots
        '"schema.test"."table.test"': '"schema.test"."table.test"',
        '"schema.".table': '"schema."."table"',
        '"schema."."table"': '"schema."."table"',
        'schema.".table"': '"schema".".table"',
        '"schema".".table"': '"schema".".table"',
        '"schema.".".table"': '"schema.".".table"',
        # These are valid but maybe not what the user intended
        '."table"': '".""table"""',
        'table.': '"table."',
    }

    invalid = {
        ('test.too.many.dots', 'table'): 'PostgreSQL does not support table with more than 3 dots',
        ('"test.too".many.dots', 'database'): 'PostgreSQL does not support database with more than 1 dots',
        ('test.too."many.dots"', 'database'): 'PostgreSQL does not support database with more than 1 dots',
        ('"test"."too"."many"."dots"', 'database'): "PostgreSQL does not support database with more than 1 dots",
        ('"test"."too"."many"."dots"', 'schema'): "PostgreSQL does not support schema with more than 2 dots",
        ('"test"."too"."many"."dots"', 'table'): "PostgreSQL does not support table with more than 3 dots",
        ('"test"."too"."many"."dots"."for"."column"', 'column'): "PostgreSQL does not support column with more than 4 dots",
        ('"table "invalid" double quote"', 'table'): 'User escaped identifiers must escape extra quotes',
        ('"schema "invalid"""."table "invalid"', 'table'): 'User escaped identifiers must escape extra quotes',
        ('"schema."table"','table'): 'User escaped identifiers must escape extra quotes',
        ('"schema".', 'table'): 'Identifier name unspecified or unquoted trailing dot',
    }

    def check_valid_quotes(self, identifier, quoted_identifier):
        eq_(pg_quote_identifier(identifier, 'table'), quoted_identifier)

    def test_valid_quotes(self):
        for identifier in self.valid:
            yield self.check_valid_quotes, identifier, self.valid[identifier]

    def check_invalid_quotes(self, identifier, id_type, msg):
        assert_raises_regexp(SQLParseError, msg, pg_quote_identifier, *(identifier, id_type))

    def test_invalid_quotes(self):
        for test in self.invalid:
            yield self.check_invalid_quotes, test[0], test[1], self.invalid[test]

    def test_how_many_dots(self):
        eq_(pg_quote_identifier('role', 'role'), '"role"')
        assert_raises_regexp(SQLParseError, "PostgreSQL does not support role with more than 1 dots", pg_quote_identifier, *('role.more', 'role'))

        eq_(pg_quote_identifier('db', 'database'), '"db"')
        assert_raises_regexp(SQLParseError, "PostgreSQL does not support database with more than 1 dots", pg_quote_identifier, *('db.more', 'database'))

        eq_(pg_quote_identifier('db.schema', 'schema'), '"db"."schema"')
        assert_raises_regexp(SQLParseError, "PostgreSQL does not support schema with more than 2 dots", pg_quote_identifier, *('db.schema.more', 'schema'))

        eq_(pg_quote_identifier('db.schema.table', 'table'), '"db"."schema"."table"')
        assert_raises_regexp(SQLParseError, "PostgreSQL does not support table with more than 3 dots", pg_quote_identifier, *('db.schema.table.more', 'table'))

        eq_(pg_quote_identifier('db.schema.table.column', 'column'), '"db"."schema"."table"."column"')
        assert_raises_regexp(SQLParseError, "PostgreSQL does not support column with more than 4 dots", pg_quote_identifier, *('db.schema.table.column.more', 'column'))