diff --git a/tests/connectors/core/test_database.py b/tests/connectors/core/test_database.py index 4b810ba5de56783823ba28fe067b272e3304aa44..11c2fed45f770b4b19687e7b24ae0408c0fb3538 100644 --- a/tests/connectors/core/test_database.py +++ b/tests/connectors/core/test_database.py @@ -25,6 +25,21 @@ class CoreDatabaseTestMixin: """ cls.test_db_name_start = cls.test_db_name_start.format(cls.db_type) + cls.error_handler__database_does_not_exist = None + cls.error_handler__database_already_exists = None + + def test_error_catch_types(self): + """Tests to ensure database ERROR types are properly caught. + + For example, MySQL and PostgreSQL interfaces do not catch "Database Does Not Exist" errors the same way. + These tests make sure this error (and others) are properly caught, regardless of what database is being called. + """ + # Ensure error types are first defined. + if not self.error_handler__database_does_not_exist: + raise ValueError('Please define error handler for "Database Does Not Exist" error type.') + if not self.error_handler__database_already_exists: + raise ValueError('Please define error handler for "Database Already Exists" error type.') + def test__select(self): """ Test logic for `SELECT;` query. @@ -44,7 +59,7 @@ class CoreDatabaseTestMixin: # Verify database exists. try: self.connector.database.create(db_name) - except self.db_error_handler.ProgrammingError: + except self.error_handler__database_already_exists: # Database already exists, as we want. pass @@ -63,7 +78,7 @@ class CoreDatabaseTestMixin: # Verify database exists. try: self.connector.database.create(db_name) - except self.db_error_handler.ProgrammingError: + except self.error_handler__database_already_exists: # Database already exists, as we want. pass @@ -86,7 +101,7 @@ class CoreDatabaseTestMixin: # Verify database exists. try: self.connector.database.create(db_name) - except self.db_error_handler.ProgrammingError: + except self.error_handler__database_already_exists: # Database already exists, as we want. pass @@ -104,7 +119,7 @@ class CoreDatabaseTestMixin: # Verify database does not exist. try: self.connector.database.drop(db_name) - except self.db_error_handler.OperationalError: + except self.error_handler__database_does_not_exist: # Database does not yet exist, as we want. pass @@ -148,7 +163,7 @@ class CoreDatabaseTestMixin: # Verify database does not yet exist. try: self.connector.database.create(db_name) - except self.db_error_handler.OperationalError: + except self.error_handler__database_already_exists: # Database already exists, as we want. pass @@ -175,7 +190,7 @@ class CoreDatabaseTestMixin: # Verify database exists. try: self.connector.database.create(db_name) - except self.db_error_handler.OperationalError: + except self.error_handler__database_already_exists: # Database already exists, as we want. pass @@ -224,7 +239,7 @@ class CoreDatabaseTestMixin: # Verify database does not yet exist. try: self.connector.database.create(db_name) - except self.db_error_handler.OperationalError: + except self.error_handler__database_already_exists: # Database already exists, as we want. pass diff --git a/tests/connectors/core/test_records.py b/tests/connectors/core/test_records.py index e507190a03dca88a0de9167f6e673c1f2e053f12..2322f815b43475374c4fcce0d9f62615b39e9f3c 100644 --- a/tests/connectors/core/test_records.py +++ b/tests/connectors/core/test_records.py @@ -8,6 +8,7 @@ various specific database test classes. This ensures that all databases types ru # System Imports. import datetime +import textwrap # Internal Imports. @@ -26,6 +27,29 @@ class CoreRecordsTestMixin: """ cls.test_db_name_start = cls.test_db_name_start.format(cls.db_type) + # Define database-specific query values. + cls._basic_table_columns = textwrap.dedent( + """ + ( + id INT(11) NOT NULL AUTO_INCREMENT, + PRIMARY KEY (id) + ) + """ + ).strip() + + cls.error_handler__table_does_not_exist = None + cls.error_handler__table_already_exists = None + + def test_error_catch_types(self): + """Tests to ensure database ERROR types are properly caught. + + For example, MySQL and PostgreSQL interfaces do not catch "Database Does Not Exist" errors the same way. + These tests make sure this error (and others) are properly caught, regardless of what database is being called. + """ + # Ensure error types are first defined. + if not self.error_handler__table_already_exists: + raise ValueError('Please define error handler for "Table Already Exists" error type.') + def test__select__success(self): """ Test `SELECT` query. @@ -35,7 +59,7 @@ class CoreRecordsTestMixin: # Verify table exists. try: self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query__basic)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_already_exists: # Table already exists, as we want. pass @@ -110,7 +134,7 @@ class CoreRecordsTestMixin: # Verify table exists. try: self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query__basic)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_already_exists: # Table already exists, as we want. pass @@ -183,7 +207,7 @@ class CoreRecordsTestMixin: # Verify table exists. try: self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query__basic)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_already_exists: # Table already exists, as we want. pass @@ -220,7 +244,7 @@ class CoreRecordsTestMixin: # Verify table exists. try: self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query__datetime)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_already_exists: # Table already exists, as we want. pass @@ -282,7 +306,7 @@ class CoreRecordsTestMixin: # Verify table exists. try: self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query__basic)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_already_exists: # Table already exists, as we want. pass @@ -370,7 +394,7 @@ class CoreRecordsTestMixin: # Verify table exists. try: self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query__datetime)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_already_exists: # Table already exists, as we want. pass @@ -483,7 +507,7 @@ class CoreRecordsTestMixin: # Verify table exists. try: self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query__basic)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_already_exists: # Table already exists, as we want. pass diff --git a/tests/connectors/core/test_tables.py b/tests/connectors/core/test_tables.py index 756cb9401e84b3a22845112596f097e403e0dfce..87e2ff1f112ccd1573ede67216dfae28071e2bdd 100644 --- a/tests/connectors/core/test_tables.py +++ b/tests/connectors/core/test_tables.py @@ -25,6 +25,21 @@ class CoreTablesTestMixin: """ cls.test_db_name_start = cls.test_db_name_start.format(cls.db_type) + cls.error_handler__table_does_not_exist = None + cls.error_handler__table_already_exists = None + + def test_error_catch_types(self): + """Tests to ensure database ERROR types are properly caught. + + For example, MySQL and PostgreSQL interfaces do not catch "Database Does Not Exist" errors the same way. + These tests make sure this error (and others) are properly caught, regardless of what database is being called. + """ + # Ensure error types are first defined. + if not self.error_handler__table_does_not_exist: + raise ValueError('Please define error handler for "Table Does Not Exist" error type.') + if not self.error_handler__table_already_exists: + raise ValueError('Please define error handler for "Table Already Exists" error type.') + def test__create_table___col_str(self): """ Tests that connector object properly creates new tables, via str of column data. @@ -93,7 +108,7 @@ class CoreTablesTestMixin: # Verify table exists. try: self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_already_exists: # Table already exists, as we want. pass @@ -109,7 +124,7 @@ class CoreTablesTestMixin: # Verify table does not exist. try: self.connector.query.execute('DROP TABLE {0};'.format(table_name)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_does_not_exist: # Table does not yet exist, as we want. pass @@ -153,7 +168,7 @@ class CoreTablesTestMixin: # Verify table exists. try: self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_already_exists: # Table already exists, as we want. pass @@ -178,7 +193,7 @@ class CoreTablesTestMixin: # # Verify table exists. # try: # self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query)) - # except self.db_error_handler.OperationalError: + # except self.error_handler__table_already_exists: # # Table already exists, as we want. # pass # @@ -265,7 +280,7 @@ class CoreTablesTestMixin: # Verify table exists. try: self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_query)) - except self.db_error_handler.OperationalError: + except self.error_handler__table_already_exists: # Table already exists, as we want. pass diff --git a/tests/connectors/mysql/test_core.py b/tests/connectors/mysql/test_core.py index 68f4789b6a5cd7aae2bbfbef39d68cfd9d82d31d..99c9c512241561987bf4f8f7db801342c976bbf9 100644 --- a/tests/connectors/mysql/test_core.py +++ b/tests/connectors/mysql/test_core.py @@ -45,3 +45,26 @@ class TestMysqlDatabaseParent(CoreTestParent): cls.db_type = cls.connector._config.db_type cls._implemented_db_types = cls.connector._config._implemented_db_types cls.db_error_handler = MySQLdb + +class TestMysqlErrorHandlers(TestMysqlDatabaseParent): + """""" + @classmethod + def setUpClass(cls): + # Run parent setup logic. + super().setUpClass() + + # Also call CoreTestMixin setup logic. + cls.set_up_class() + + # Define database name to use in tests. + cls.test_db_name = '{0}test_error_handlers'.format(cls.test_db_name_start) + + @classmethod + def set_up_class(cls): + """ + Acts as the equivalent of the UnitTesting "setUpClass()" function. + + However, since this is not inheriting from a given TestCase, + calling the literal function here would override instead. + """ + cls.test_db_name_start = cls.test_db_name_start.format(cls.db_type) diff --git a/tests/connectors/mysql/test_database.py b/tests/connectors/mysql/test_database.py index 241ead518c74d4a7f3708a9901ccea03e94a6098..7b1983e158b0a811a8e835cd7e4103f8e835e180 100644 --- a/tests/connectors/mysql/test_database.py +++ b/tests/connectors/mysql/test_database.py @@ -33,3 +33,38 @@ class TestMysqlDatabase(TestMysqlDatabaseParent, CoreDatabaseTestMixin): if len(results) > 0: for result in results: cls.connector.tables.drop(result) + + cls.error_handler__database_does_not_exist = cls.db_error_handler.OperationalError + cls.error_handler__database_already_exists = cls.db_error_handler.ProgrammingError + + def test_error_catch_types(self): + """Tests to ensure database ERROR types are properly caught. + + Ex: MySQL and PostgreSQL interfaces do not catch "Database does not exist" errors the same. + These tests make sure this error (and others) are properly caught, regardless of what database is + being called. + """ + # Call parent logic. + super().test_error_catch_types() + + with self.subTest('Verify handling when database does not exist'): + # Make sure we're using a database name that is not yet created. + db_name = 'NewDatabaseName' + results = self.connector.database.show() + if db_name in results: + raise AssertionError('Database already present. Incorrect name provided.') + + # Check that we use the correct handler. + with self.assertRaises(self.error_handler__database_does_not_exist): + self.connector.query.execute('DROP DATABASE {0};'.format(db_name)) + + with self.subTest('Verify handling when database already exists'): + # Make sure we're using a database name that is already created. + db_name = 'test_database' + results = self.connector.database.show() + if db_name not in results: + raise AssertionError('Database not yet present. Incorrect name provided.') + + # Check that we use the correct handler. + with self.assertRaises(self.error_handler__database_already_exists): + self.connector.query.execute('CREATE DATABASE {0};'.format(db_name)) diff --git a/tests/connectors/mysql/test_records.py b/tests/connectors/mysql/test_records.py index d2d8922d22dc87ded326bdc5bf85022720fd12a1..ab753402bf9694d53fe52000b51c89ed385cb7e9 100644 --- a/tests/connectors/mysql/test_records.py +++ b/tests/connectors/mysql/test_records.py @@ -56,3 +56,29 @@ class TestMysqlRecords(TestMysqlDatabaseParent, CoreRecordsTestMixin): ) """ ).strip() + + cls.error_handler__table_does_not_exist = cls.db_error_handler.OperationalError + cls.error_handler__table_already_exists = cls.db_error_handler.OperationalError + + def test_error_catch_types(self): + """Tests to ensure database ERROR types are properly caught. + + Ex: MySQL and PostgreSQL interfaces do not catch "Database does not exist" errors the same. + These tests make sure this error (and others) are properly caught, regardless of what database is + being called. + """ + # Call parent logic. + super().test_error_catch_types() + + with self.subTest('Verify handling when database already exists'): + # Make sure we're using a table name that is not already created. + table_name = 'test_table' + self.connector.tables.create(table_name, self._basic_table_columns) + + results = self.connector.tables.show() + if table_name not in results: + raise AssertionError('Table not yet present. Incorrect name provided.') + + # Check that we use the correct handler. + with self.assertRaises(self.error_handler__table_already_exists): + self.connector.query.execute('CREATE TABLE {0} {1};'.format(table_name, self._basic_table_columns)) diff --git a/tests/connectors/mysql/test_tables.py b/tests/connectors/mysql/test_tables.py index a1008bbe6097d99dc3e3606e3421d4b2f4d18755..49c0677c7d987d55170df818cb42f0f124d0aaa4 100644 --- a/tests/connectors/mysql/test_tables.py +++ b/tests/connectors/mysql/test_tables.py @@ -3,10 +3,9 @@ Tests for "tables" logic of "MySQL" DB Connector class. """ # System Imports. - -# Internal Imports. import textwrap +# Internal Imports. from .test_core import TestMysqlDatabaseParent from tests.connectors.core.test_tables import CoreTablesTestMixin @@ -55,3 +54,40 @@ class TestMysqlTables(TestMysqlDatabaseParent, CoreTablesTestMixin): ) """ ).strip() + + cls.error_handler__table_does_not_exist = cls.db_error_handler.OperationalError + cls.error_handler__table_already_exists = cls.db_error_handler.OperationalError + + def test_error_catch_types(self): + """Tests to ensure database ERROR types are properly caught. + + Ex: MySQL and PostgreSQL interfaces do not catch "Database does not exist" errors the same. + These tests make sure this error (and others) are properly caught, regardless of what database is + being called. + """ + # Call parent logic. + super().test_error_catch_types() + + with self.subTest('Verify handling when database does not exist'): + # Make sure we're using a table name that is not yet created. + table_name = 'NewTableName' + results = self.connector.tables.show() + if table_name in results: + raise AssertionError('Table already present. Incorrect name provided.') + + # Check that we use the correct handler. + with self.assertRaises(self.error_handler__table_does_not_exist): + self.connector.query.execute('DROP TABLE {0};'.format(table_name)) + + with self.subTest('Verify handling when database already exists'): + # Make sure we're using a table name that is not already created. + table_name = 'test_table' + self.connector.tables.create(table_name, self._basic_table_columns) + + results = self.connector.tables.show() + if table_name not in results: + raise AssertionError('Table not yet present. Incorrect name provided.') + + # Check that we use the correct handler. + with self.assertRaises(self.error_handler__table_already_exists): + self.connector.query.execute('CREATE TABLE {0} {1};'.format(table_name, self._basic_table_columns)) diff --git a/tests/connectors/postgresql/test_database.py b/tests/connectors/postgresql/test_database.py index 9658632d007184ab84588a69c49091588b1f70b6..67da2b70780e72518586d222b9550398d11db1d3 100644 --- a/tests/connectors/postgresql/test_database.py +++ b/tests/connectors/postgresql/test_database.py @@ -33,3 +33,38 @@ class TestPostgresqlDatabase(TestPostgresqlDatabaseParent, CoreDatabaseTestMixin if len(results) > 0: for result in results: cls.connector.tables.drop(result) + + cls.error_handler__database_does_not_exist = cls.db_error_handler.errors.InvalidCatalogName + cls.error_handler__database_already_exists = cls.db_error_handler.errors.DuplicateDatabase + + def test_error_catch_types(self): + """Tests to ensure database ERROR types are properly caught. + + Ex: MySQL and PostgreSQL interfaces do not catch "Database does not exist" errors the same. + These tests make sure this error (and others) are properly caught, regardless of what database is + being called. + """ + # Call parent logic. + super().test_error_catch_types() + + with self.subTest('Verify handling when database does not exist'): + # Make sure we're using a database name that is not yet created. + db_name = 'NewDatabaseName' + results = self.connector.database.show() + if db_name in results: + raise AssertionError('Database already present. Incorrect name provided.') + + # Check that we use the correct handler. + with self.assertRaises(self.error_handler__database_does_not_exist): + self.connector.query.execute('DROP DATABASE {0};'.format(db_name)) + + with self.subTest('Verify handling when database already exists'): + # Make sure we're using a database name that is not already created. + db_name = 'test_database' + results = self.connector.database.show() + if db_name not in results: + raise AssertionError('Database not yet present. Incorrect name provided.') + + # Check that we use the correct handler. + with self.assertRaises(self.error_handler__database_already_exists): + self.connector.query.execute('CREATE DATABASE {0};'.format(db_name)) diff --git a/tests/connectors/postgresql/test_tables.py b/tests/connectors/postgresql/test_tables.py index b6fba6947efc7f7b8bcbfbc26d4e2b6a4262489d..b26b26d0b7bc4651f6396f173e06b60a88a563b0 100644 --- a/tests/connectors/postgresql/test_tables.py +++ b/tests/connectors/postgresql/test_tables.py @@ -52,3 +52,40 @@ class TestPostgresqlTables(TestPostgresqlDatabaseParent, CoreTablesTestMixin): ) """ ).strip() + + cls.error_handler__table_does_not_exist = cls.db_error_handler.errors.UndefinedTable + cls.error_handler__table_already_exists = cls.db_error_handler.errors.DuplicateTable + + def test_error_catch_types(self): + """Tests to ensure database ERROR types are properly caught. + + Ex: MySQL and PostgreSQL interfaces do not catch "Database does not exist" errors the same. + These tests make sure this error (and others) are properly caught, regardless of what database is + being called. + """ + # Call parent logic. + super().test_error_catch_types() + + with self.subTest('Verify handling when database does not exist'): + # Make sure we're using a table name that is not yet created. + table_name = 'NewTableName' + results = self.connector.tables.show() + if table_name in results: + raise AssertionError('Table already present. Incorrect name provided.') + + # Check that we use the correct handler. + with self.assertRaises(self.error_handler__table_does_not_exist): + self.connector.query.execute('DROP TABLE {0};'.format(table_name)) + + with self.subTest('Verify handling when database already exists'): + # Make sure we're using a table name that is not already created. + table_name = 'test_table' + self.connector.tables.create(table_name, self._basic_table_columns) + + results = self.connector.tables.show() + if table_name not in results: + raise AssertionError('Table not yet present. Incorrect name provided.') + + # Check that we use the correct handler. + with self.assertRaises(self.error_handler__table_already_exists): + self.connector.query.execute('CREATE TABLE {0} {1};'.format(table_name, self._basic_table_columns))