From 4f293c8338f64f1a5a9fb706d3fc5fe543f4be86 Mon Sep 17 00:00:00 2001 From: Brandon Rodriguez <brodriguez8774@gmail.com> Date: Mon, 26 Sep 2022 13:10:45 -0400 Subject: [PATCH] Correct trycatch statements to use proper database-specific error handling --- tests/connectors/core/test_database.py | 29 ++++++++++---- tests/connectors/core/test_records.py | 38 +++++++++++++++---- tests/connectors/core/test_tables.py | 25 +++++++++--- tests/connectors/mysql/test_core.py | 23 +++++++++++ tests/connectors/mysql/test_database.py | 35 +++++++++++++++++ tests/connectors/mysql/test_records.py | 26 +++++++++++++ tests/connectors/mysql/test_tables.py | 40 +++++++++++++++++++- tests/connectors/postgresql/test_database.py | 35 +++++++++++++++++ tests/connectors/postgresql/test_tables.py | 37 ++++++++++++++++++ 9 files changed, 267 insertions(+), 21 deletions(-) diff --git a/tests/connectors/core/test_database.py b/tests/connectors/core/test_database.py index 4b810ba..11c2fed 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 e507190..2322f81 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 756cb94..87e2ff1 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 68f4789..99c9c51 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 241ead5..7b1983e 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 d2d8922..ab75340 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 a1008bb..49c0677 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 9658632..67da2b7 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 b6fba69..b26b26d 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)) -- GitLab