Skip to content
Snippets Groups Projects
Commit 77692175 authored by Brandon Rodriguez's avatar Brandon Rodriguez
Browse files

Add basic "identifier" validation logic

parent 8a42ac57
Branches
Tags
No related merge requests found
......@@ -9,6 +9,8 @@ Should be inherited by language-specific connectors.
import copy
# User Imports.
import re
from py_dbcn.logging import init_logging
......@@ -32,16 +34,68 @@ class BaseValidate:
# Define provided direct parent object.
self._parent = parent
# region Name Validation
def _identifier(self, name):
"""Generalized validation for "identifier naming conventions".
All other "identifiers" should probably be run through this function.
See https://dev.mysql.com/doc/refman/8.0/en/identifiers.html
"""
# Check if value is quoted.
is_quoted = False
if len(name) > 1 and name[0] == name[-1] and name[0] in ['`', '"', "'"]:
is_quoted = True
max_len = 66
else:
max_len = 64
print('name: {0}'.format(name))
print('is_quoted: {0}'.format(is_quoted))
# Check against max possible length.
if len(name) > max_len:
return (False, 'is longer than 64 characters.')
# Check acceptable patterns.
if is_quoted is False:
# Check against "unquoted patterns".
'0-9a-zA-Z$_'
# Check against "quoted patterns".
pattern = re.compile('^([0-9a-zA-Z$_])+$')
if not re.match(pattern, name):
return (False, 'does not match acceptable characters.')
else:
# Check against "quoted patterns".
pattern = re.compile(u'^([\u0001-\u007F])+$', flags=re.UNICODE)
if not re.match(pattern, name):
return (False, 'does not match acceptable characters.')
# Passed all tests.
return (True, '')
def database_name(self, name):
"""
Validates that provided database name uses set of acceptable characters.
:param name: Potential name of database to validate.
:return: True if valid | False otherwise.
"""
# For now, always return as valid.
return True
# Check if value is quoted.
is_quoted = False
if len(name) > 1 and name[0] == name[-1] and name[0] in ['`', '"', "'"]:
is_quoted = True
# region Name Validation
# Validate using "general identifier" logic.
results = self._identifier(name)
if results[0] is False:
if is_quoted:
raise ValueError(u'Invalid database name of {0}. Name {1}'.format(str(name), results[1]))
else:
raise ValueError(u'Invalid database name of "{0}". Name {1}'.format(str(name), results[1]))
# Passed checks.
return True
def table_name(self, name):
"""
......@@ -49,9 +103,48 @@ class BaseValidate:
:param name: Potential name of table to validate.
:return: True if valid | False otherwise.
"""
# For now, always return as valid.
# Check if value is quoted.
is_quoted = False
if len(name) > 1 and name[0] == name[-1] and name[0] in ['`', '"', "'"]:
is_quoted = True
# Validate using "general identifier" logic.
results = self._identifier(name)
if results[0] is False:
if is_quoted:
raise ValueError(u'Invalid table name of {0}. Name {1}'.format(str(name), results[1]))
else:
raise ValueError(u'Invalid table name of "{0}". Name {1}'.format(str(name), results[1]))
# Passed checks.
return True
def table_column(self, name):
"""
Validates that provided table name uses set of acceptable characters.
:param name: Potential name of table to validate.
:return: True if valid | False otherwise.
"""
# Check if value is quoted.
is_quoted = False
if len(name) > 1 and name[0] == name[-1] and name[0] in ['`', '"', "'"]:
is_quoted = True
# Validate using "general identifier" logic.
results = self._identifier(name)
if results[0] is False:
if is_quoted:
raise ValueError(u'Invalid column name of {0}. Name {1}'.format(str(name), results[1]))
else:
raise ValueError(u'Invalid column name of "{0}". Name {1}'.format(str(name), results[1]))
# Passed checks.
return True
# endregion Name Validation
def table_columns(self, columns):
"""
Validates that provided column values match expected syntax.
......@@ -110,8 +203,6 @@ class BaseValidate:
# For now, always return as valid.
return columns
# endregion Name Validation
# region Clause Validation
def select_clause(self, clause):
......
......@@ -3,15 +3,13 @@ Tests for "validate" logic of "MySQL" DB Connector class.
"""
# System Imports.
import unittest
import re
# User Imports.
from config import mysql_config, sqlite_config
from py_dbcn.connectors import MysqlDbConnector, PostgresqlDbConnector, SqliteDbConnector
from py_dbcn.connectors.core.validate import BaseValidate
from .test_core import TestMysqlDatabaseParent
class TestMysqlValidate(unittest.TestCase):
class TestMysqlValidate(TestMysqlDatabaseParent):
"""
Tests "MySQL" DB Connector class validation logic.
"""
......@@ -19,3 +17,465 @@ class TestMysqlValidate(unittest.TestCase):
def setUpClass(cls):
# Run parent setup logic.
super().setUpClass()
def test__identifier__success(self):
"""
Test "general identifier" validation, when it should succeed.
"""
with self.subTest('"Permitted characters in unquoted Identifiers"'):
# Ensure capital letters validate.
result = self.connector.validate._identifier('ABCDEFGHIJKLMNOPQRSTUVWXYZ')
self.assertTrue(result[0])
self.assertEqual(result[1], '')
# Ensure lowercase characters validate.
result = self.connector.validate._identifier('abcdefghijklmnopqrstuvwxyz')
self.assertTrue(result[0])
self.assertEqual(result[1], '')
# Ensure integer characters validate.
result = self.connector.validate._identifier('0123456789')
self.assertTrue(result[0])
self.assertEqual(result[1], '')
# Ensure dollar and underscore validate.
result = self.connector.validate._identifier('_$')
self.assertTrue(result[0])
self.assertEqual(result[1], '')
with self.subTest('At max length - unquoted'):
test_str = 'Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
self.assertEqual(len(test_str), 64)
result = self.connector.validate._identifier(test_str)
self.assertTrue(result[0])
self.assertEqual(result[1], '')
with self.subTest('At max length - quoted'):
test_str = '`Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa`'
self.assertEqual(len(test_str), 66)
result = self.connector.validate._identifier(test_str)
self.assertTrue(result[0])
self.assertEqual(result[1], '')
with self.subTest(
'"Permitted characters in quoted identifiers include the full Unicode Basic Multilingual Plane (BMP), '
'except U+0000"'
):
test_str = u''
for index in range(127):
# Check len of str with new value added.
new_test_str = test_str + chr(index + 1)
if len(new_test_str) > 64:
# At max acceptable length. Test current value and then reset string.
test_str = u'`' + test_str + u'`'
result = self.connector.validate._identifier(test_str)
self.assertTrue(result[0])
self.assertEqual(result[1], '')
test_str = u''
# Update str with new value.
test_str += chr(index + 1)
test_str = u'`' + test_str + u'`'
result = self.connector.validate._identifier(test_str)
self.assertTrue(result[0])
self.assertEqual(result[1], '')
def test__identifier__failure(self):
"""
Test "general identifier" validation, when it should fail.
"""
with self.subTest('Identifier too long - unquoted'):
test_str = 'Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
self.assertEqual(len(test_str), 65)
result = self.connector.validate._identifier(test_str)
self.assertFalse(result[0])
self.assertEqual(result[1], 'is longer than 64 characters.')
with self.subTest('Identifier too long - quoted'):
test_str = '`Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa`'
self.assertEqual(len(test_str), 67)
result = self.connector.validate._identifier(test_str)
self.assertFalse(result[0])
self.assertEqual(result[1], 'is longer than 64 characters.')
with self.subTest('Invalid characters - unquoted'):
test_str = '!@#%^&*()-+=~\'"[]{}<>|\\/:;,.?'
for item in test_str:
result = self.connector.validate._identifier(item)
self.assertFalse(result[0])
self.assertEqual(result[1], 'does not match acceptable characters.')
# For now, "extended" range is considered invalid.
# Not sure if we'll want to enable this at some point?
for index in range(65535):
# Skip "acceptable" values.
if index + 1 <= 127:
continue
# Check len of str with new value added.
test_str = u'{0}'.format(chr(index + 1))
result = self.connector.validate._identifier(test_str)
self.assertFalse(result[0])
self.assertEqual(result[1], 'does not match acceptable characters.')
with self.subTest('Invalid characters - quoted'):
# Check that hex 0 is invalid.
result = self.connector.validate._identifier(u'`' + chr(0) + u'`')
self.assertFalse(result[0])
self.assertEqual(result[1], 'does not match acceptable characters.')
# For now, "extended" range is considered invalid.
# Not sure if we'll want to enable this at some point?
for index in range(65535):
# Skip "acceptable" values.
if index + 1 <= 127:
continue
# Check len of str with new value added.
test_str = u'`' + chr(index + 1) + u'`'
result = self.connector.validate._identifier(test_str)
self.assertFalse(result[0])
self.assertEqual(result[1], 'does not match acceptable characters.')
def test__database_name__success(self):
"""
Test "database name" validation, when it should succeed.
"""
with self.subTest('"Permitted characters in unquoted Identifiers"'):
# Ensure capital letters validate.
self.assertTrue(self.connector.validate.database_name('ABCDEFGHIJKLMNOPQRSTUVWXYZ'))
# Ensure lowercase characters validate.
self.assertTrue(self.connector.validate.database_name('abcdefghijklmnopqrstuvwxyz'))
# Ensure integer characters validate.
self.assertTrue(self.connector.validate.database_name('0123456789'))
# Ensure dollar and underscore validate.
self.assertTrue(self.connector.validate.database_name('_$'))
with self.subTest('At max length - unquoted'):
test_str = 'Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
self.assertEqual(len(test_str), 64)
self.assertTrue(self.connector.validate.database_name(test_str))
with self.subTest('At max length - quoted'):
test_str = '`Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa`'
self.assertEqual(len(test_str), 66)
self.assertTrue(self.connector.validate.database_name(test_str))
with self.subTest(
'"Permitted characters in quoted identifiers include the full Unicode Basic Multilingual Plane (BMP), '
'except U+0000"'
):
test_str = u''
for index in range(127):
# Check len of str with new value added.
new_test_str = test_str + chr(index + 1)
if len(new_test_str) > 64:
# At max acceptable length. Test current value and then reset string.
test_str = u'`' + test_str + u'`'
self.assertTrue(self.connector.validate.database_name(test_str))
test_str = u''
# Update str with new value.
test_str += chr(index + 1)
test_str = u'`' + test_str + u'`'
self.assertTrue(self.connector.validate.database_name(test_str))
def test__database_name__failure(self):
"""
Test "database name" validation, when it should fail.
"""
with self.subTest('Identifier too long - unquoted'):
test_str = 'Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
self.assertEqual(len(test_str), 65)
with self.assertRaises(ValueError) as err:
self.connector.validate.database_name(test_str)
self.assertIn('Invalid database name of "', str(err.exception))
self.assertIn('". Name is longer than 64 characters.', str(err.exception))
with self.subTest('Identifier too long - quoted'):
test_str = '`Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa`'
self.assertEqual(len(test_str), 67)
with self.assertRaises(ValueError) as err:
self.connector.validate.database_name(test_str)
self.assertIn('Invalid database name of ', str(err.exception))
self.assertIn('. Name is longer than 64 characters.', str(err.exception))
with self.subTest('Invalid characters - unquoted'):
test_str = '!@#%^&*()-+=~\'"[]{}<>|\\/:;,.?'
for item in test_str:
with self.assertRaises(ValueError) as err:
self.connector.validate.database_name(item)
self.assertIn('Invalid database name of "', str(err.exception))
self.assertIn('". Name does not match acceptable characters.', str(err.exception))
# For now, "extended" range is considered invalid.
# Not sure if we'll want to enable this at some point?
for index in range(65535):
# Skip "acceptable" values.
if index + 1 <= 127:
continue
# Check len of str with new value added.
test_str = u'{0}'.format(chr(index + 1))
with self.assertRaises(ValueError) as err:
self.connector.validate.database_name(test_str)
self.assertIn('Invalid database name of "', str(err.exception))
self.assertIn('". Name does not match acceptable characters.', str(err.exception))
with self.subTest('Invalid characters - quoted'):
# Check that hex 0 is invalid.
with self.assertRaises(ValueError) as err:
self.connector.validate.database_name(u'`' + chr(0) + u'`')
self.assertIn('Invalid database name of ', str(err.exception))
self.assertIn('. Name does not match acceptable characters.', str(err.exception))
# For now, "extended" range is considered invalid.
# Not sure if we'll want to enable this at some point?
for index in range(65535):
# Skip "acceptable" values.
if index + 1 <= 127:
continue
# Check len of str with new value added.
test_str = u'`' + chr(index + 1) + u'`'
with self.assertRaises(ValueError) as err:
self.connector.validate.database_name(test_str)
self.assertIn('Invalid database name of ', str(err.exception))
self.assertIn('. Name does not match acceptable characters.', str(err.exception))
def test__table_name__success(self):
"""
Test "table name" validation, when it should succeed.
"""
with self.subTest('"Permitted characters in unquoted Identifiers"'):
# Ensure capital letters validate.
self.assertTrue(self.connector.validate.table_name('ABCDEFGHIJKLMNOPQRSTUVWXYZ'))
# Ensure lowercase characters validate.
self.assertTrue(self.connector.validate.table_name('abcdefghijklmnopqrstuvwxyz'))
# Ensure integer characters validate.
self.assertTrue(self.connector.validate.table_name('0123456789'))
# Ensure dollar and underscore validate.
self.assertTrue(self.connector.validate.table_name('_$'))
with self.subTest('At max length - unquoted'):
test_str = 'Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
self.assertEqual(len(test_str), 64)
self.assertTrue(self.connector.validate.table_name(test_str))
with self.subTest('At max length - quoted'):
test_str = '`Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa`'
self.assertEqual(len(test_str), 66)
self.assertTrue(self.connector.validate.table_name(test_str))
with self.subTest(
'"Permitted characters in quoted identifiers include the full Unicode Basic Multilingual Plane (BMP), '
'except U+0000"'
):
test_str = u''
for index in range(127):
# Check len of str with new value added.
new_test_str = test_str + chr(index + 1)
if len(new_test_str) > 64:
# At max acceptable length. Test current value and then reset string.
test_str = u'`' + test_str + u'`'
self.assertTrue(self.connector.validate.table_name(test_str))
test_str = u''
# Update str with new value.
test_str += chr(index + 1)
test_str = u'`' + test_str + u'`'
self.assertTrue(self.connector.validate.table_name(test_str))
def test__table_name__failure(self):
"""
Test "table name" validation, when it should fail.
"""
with self.subTest('Identifier too long - unquoted'):
test_str = 'Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
self.assertEqual(len(test_str), 65)
with self.assertRaises(ValueError) as err:
self.connector.validate.table_name(test_str)
self.assertIn('Invalid table name of "', str(err.exception))
self.assertIn('". Name is longer than 64 characters.', str(err.exception))
with self.subTest('Identifier too long - quoted'):
test_str = '`Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa`'
self.assertEqual(len(test_str), 67)
with self.assertRaises(ValueError) as err:
self.connector.validate.table_name(test_str)
self.assertIn('Invalid table name of ', str(err.exception))
self.assertIn('. Name is longer than 64 characters.', str(err.exception))
with self.subTest('Invalid characters - unquoted'):
test_str = '!@#%^&*()-+=~\'"[]{}<>|\\/:;,.?'
for item in test_str:
with self.assertRaises(ValueError) as err:
self.connector.validate.table_name(item)
self.assertIn('Invalid table name of "', str(err.exception))
self.assertIn('". Name does not match acceptable characters.', str(err.exception))
# For now, "extended" range is considered invalid.
# Not sure if we'll want to enable this at some point?
for index in range(65535):
# Skip "acceptable" values.
if index + 1 <= 127:
continue
# Check len of str with new value added.
test_str = u'{0}'.format(chr(index + 1))
with self.assertRaises(ValueError) as err:
self.connector.validate.table_name(test_str)
self.assertIn('Invalid table name of "', str(err.exception))
self.assertIn('". Name does not match acceptable characters.', str(err.exception))
with self.subTest('Invalid characters - quoted'):
# Check that hex 0 is invalid.
with self.assertRaises(ValueError) as err:
self.connector.validate.table_name(u'`' + chr(0) + u'`')
self.assertIn('Invalid table name of ', str(err.exception))
self.assertIn('. Name does not match acceptable characters.', str(err.exception))
# For now, "extended" range is considered invalid.
# Not sure if we'll want to enable this at some point?
for index in range(65535):
# Skip "acceptable" values.
if index + 1 <= 127:
continue
# Check len of str with new value added.
test_str = u'`' + chr(index + 1) + u'`'
with self.assertRaises(ValueError) as err:
self.connector.validate.table_name(test_str)
self.assertIn('Invalid table name of ', str(err.exception))
self.assertIn('. Name does not match acceptable characters.', str(err.exception))
def test__table_column__success(self):
"""
Test "column name" validation, when it should succeed.
"""
with self.subTest('"Permitted characters in unquoted Identifiers"'):
# Ensure capital letters validate.
self.assertTrue(self.connector.validate.table_column('ABCDEFGHIJKLMNOPQRSTUVWXYZ'))
# Ensure lowercase characters validate.
self.assertTrue(self.connector.validate.table_column('abcdefghijklmnopqrstuvwxyz'))
# Ensure integer characters validate.
self.assertTrue(self.connector.validate.table_column('0123456789'))
# Ensure dollar and underscore validate.
self.assertTrue(self.connector.validate.table_column('_$'))
with self.subTest('At max length - unquoted'):
test_str = 'Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
self.assertEqual(len(test_str), 64)
self.assertTrue(self.connector.validate.table_column(test_str))
with self.subTest('At max length - quoted'):
test_str = '`Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa`'
self.assertEqual(len(test_str), 66)
self.assertTrue(self.connector.validate.table_column(test_str))
with self.subTest(
'"Permitted characters in quoted identifiers include the full Unicode Basic Multilingual Plane (BMP), '
'except U+0000"'
):
test_str = u''
for index in range(127):
# Check len of str with new value added.
new_test_str = test_str + chr(index + 1)
if len(new_test_str) > 64:
# At max acceptable length. Test current value and then reset string.
test_str = u'`' + test_str + u'`'
self.assertTrue(self.connector.validate.table_column(test_str))
test_str = u''
# Update str with new value.
test_str += chr(index + 1)
test_str = u'`' + test_str + u'`'
self.assertTrue(self.connector.validate.table_column(test_str))
def test__table_column__failure(self):
"""
Test "column name" validation, when it should fail.
"""
with self.subTest('Identifier too long - unquoted'):
test_str = 'Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
self.assertEqual(len(test_str), 65)
with self.assertRaises(ValueError) as err:
self.connector.validate.table_column(test_str)
self.assertIn('Invalid column name of "', str(err.exception))
self.assertIn('". Name is longer than 64 characters.', str(err.exception))
with self.subTest('Identifier too long - quoted'):
test_str = '`Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa`'
self.assertEqual(len(test_str), 67)
with self.assertRaises(ValueError) as err:
self.connector.validate.table_column(test_str)
self.assertIn('Invalid column name of ', str(err.exception))
self.assertIn('. Name is longer than 64 characters.', str(err.exception))
with self.subTest('Invalid characters - unquoted'):
test_str = '!@#%^&*()-+=~\'"[]{}<>|\\/:;,.?'
for item in test_str:
with self.assertRaises(ValueError) as err:
self.connector.validate.table_column(item)
self.assertIn('Invalid column name of "', str(err.exception))
self.assertIn('". Name does not match acceptable characters.', str(err.exception))
# For now, "extended" range is considered invalid.
# Not sure if we'll want to enable this at some point?
for index in range(65535):
# Skip "acceptable" values.
if index + 1 <= 127:
continue
# Check len of str with new value added.
test_str = u'{0}'.format(chr(index + 1))
with self.assertRaises(ValueError) as err:
self.connector.validate.table_column(test_str)
self.assertIn('Invalid column name of "', str(err.exception))
self.assertIn('". Name does not match acceptable characters.', str(err.exception))
with self.subTest('Invalid characters - quoted'):
# Check that hex 0 is invalid.
with self.assertRaises(ValueError) as err:
self.connector.validate.table_column(u'`' + chr(0) + u'`')
self.assertIn('Invalid column name of ', str(err.exception))
self.assertIn('. Name does not match acceptable characters.', str(err.exception))
# For now, "extended" range is considered invalid.
# Not sure if we'll want to enable this at some point?
for index in range(65535):
# Skip "acceptable" values.
if index + 1 <= 127:
continue
# Check len of str with new value added.
test_str = u'`' + chr(index + 1) + u'`'
with self.assertRaises(ValueError) as err:
self.connector.validate.table_column(test_str)
self.assertIn('Invalid column name of ', str(err.exception))
self.assertIn('. Name does not match acceptable characters.', str(err.exception))
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment