From 82c58310e92f3aaad2b944e8a431082ca958a5bb Mon Sep 17 00:00:00 2001
From: Brandon Rodriguez <brodriguez8774@gmail.com>
Date: Fri, 19 Aug 2022 15:29:07 -0400
Subject: [PATCH] Add ability to hide query display for all functions
---
py_dbcn/connectors/core/database.py | 76 ++++++++------
py_dbcn/connectors/core/query.py | 6 +-
py_dbcn/connectors/core/records.py | 46 +++++---
py_dbcn/connectors/core/tables.py | 121 ++++++++++++++--------
py_dbcn/connectors/postgresql/database.py | 20 ++--
tests/connectors/core/test_display.py | 117 +++++++++++----------
6 files changed, 233 insertions(+), 153 deletions(-)
diff --git a/py_dbcn/connectors/core/database.py b/py_dbcn/connectors/core/database.py
index 7ae2211..4b600de 100644
--- a/py_dbcn/connectors/core/database.py
+++ b/py_dbcn/connectors/core/database.py
@@ -35,33 +35,40 @@ class BaseDatabase:
self._show_databases_query = None
self._current_database_query = None
- def select(self):
- """Returns name of currently selected database."""
+ def select(self, display_query=True):
+ """Returns name of currently selected database.
+
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
+ """
if not self._current_database_query:
raise ValueError('SELECT CURRENT DATABASE query is not defined.')
- return (self._base.query.execute(
+ results = self._base.query.execute(
self._current_database_query,
- display_query=False,
- )[0][0]).strip()
+ display_query=display_query,
+ )[0][0].strip()
+
+ return results
- def current(self):
+ def current(self, display_query=True):
"""Returns name of currently selected database.
Alias for select().
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
- return self.select()
+ return self.select(display_query=display_query)
- def _get(self, show=False):
- """
- Gets list of all currently-available databases.
+ def _get(self, display_query=False, show=False):
+ """Gets list of all currently-available databases.
+
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to False.
:param show: Bool indicating if results should be printed to console or not. Used for "SHOW DATABASES" query.
"""
if not self._show_databases_query:
raise ValueError('SHOW DATABASES query is not defined.')
# Generate and execute query.
- results = self._base.query.execute(self._show_databases_query)
+ results = self._base.query.execute(self._show_databases_query, display_query=display_query)
# Convert to more friendly format.
formatted_results = []
@@ -75,15 +82,18 @@ class BaseDatabase:
# Return data.
return results
- def show(self):
- """
- Displays all databases available for selection.
- """
- return self._get(show=True)
+ def show(self, display_query=True):
+ """Displays all databases available for selection.
- def use(self, db_name):
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
- Selects given database for use.
+ return self._get(display_query=display_query, show=True)
+
+ def use(self, db_name, display_query=True):
+ """Selects given database for use.
+
+ :param db_name: Name of db to use.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
# First, check that provided name is valid format.
if not self._base.validate.database_name(db_name):
@@ -101,13 +111,14 @@ class BaseDatabase:
# Switch active database.
query = 'USE {0};'.format(db_name)
- self._base.query.execute(query)
+ self._base.query.execute(query, display_query=display_query)
logger.results('Database changed to "{0}".'.format(db_name))
- def create(self, db_name):
- """
- Creates new database with provided name.
+ def create(self, db_name, display_query=True):
+ """Creates new database with provided name.
+
:param db_name: Desired name of new database.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
# First, check that provided name is valid format.
if not self._base.validate.database_name(db_name):
@@ -125,13 +136,14 @@ class BaseDatabase:
# Create new database.
query = 'CREATE DATABASE {0};'.format(db_name)
- self._base.query.execute(query)
+ self._base.query.execute(query, display_query=display_query)
logger.results('Created database "{0}".'.format(db_name))
- def drop(self, db_name):
- """
- Deletes database with provided name.
+ def drop(self, db_name, display_query=True):
+ """Deletes database with provided name.
+
:param db_name: Name of database to delete.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
# First, check that provided name is valid format.
if not self._base.validate.database_name(db_name):
@@ -147,11 +159,13 @@ class BaseDatabase:
# Remove database.
query = 'DROP DATABASE {0};'.format(db_name)
- self._base.query.execute(query)
+ self._base.query.execute(query, display_query=display_query)
logger.results('Dropped database "{0}".'.format(db_name))
- def delete(self, db_name):
- """
- Alias for database "drop" function.
+ def delete(self, db_name, display_query=True):
+ """Alias for database "drop" function.
+
+ :param db_name: Name of database to delete.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
- self.drop(db_name)
+ self.drop(db_name, display_query=display_query)
diff --git a/py_dbcn/connectors/core/query.py b/py_dbcn/connectors/core/query.py
index b12fff0..a8cef8e 100644
--- a/py_dbcn/connectors/core/query.py
+++ b/py_dbcn/connectors/core/query.py
@@ -32,7 +32,11 @@ class BaseQuery:
self._parent = parent
def execute(self, query, display_query=True):
- """Core function to execute database queries."""
+ """Core function to execute database queries.
+
+ :param query: Query to execute.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
+ """
if display_query:
logger.query(query)
diff --git a/py_dbcn/connectors/core/records.py b/py_dbcn/connectors/core/records.py
index e76712a..c8b3c16 100644
--- a/py_dbcn/connectors/core/records.py
+++ b/py_dbcn/connectors/core/records.py
@@ -32,8 +32,13 @@ class BaseRecords:
# Define provided direct parent object.
self._parent = parent
- def select(self, table_name, select_clause=None):
- """"""
+ def select(self, table_name, select_clause=None, display_query=True):
+ """Selects records from provided table.
+
+ :param table_name: Name of table to select from.
+ :param select_clause: Clause to choose selected columns.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
+ """
# Check that provided table name is valid format.
if not self._base.validate.table_name(table_name):
raise ValueError('Invalid table name of "{0}".'.format(table_name))
@@ -43,14 +48,20 @@ class BaseRecords:
# Select record.
query = 'SELECT {0} FROM {1};'.format(select_clause, table_name)
- results = self._base.query.execute(query)
+ results = self._base.query.execute(query, display_query=display_query)
logger.query('{0}'.format(query))
self._base.display.records.select(results, logger, table_name, select_clause)
return results
- def insert(self, table_name, values_clause, columns_clause=None):
- """"""
+ def insert(self, table_name, values_clause, columns_clause=None, display_query=True):
+ """Inserts record(s) into provided table.
+
+ :param table_name: Name of table to insert into.
+ :param values_clause: Clause to specify values to insert.
+ :param columns_clause: Clause to specify columns to insert into.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
+ """
# Check that provided table name is valid format.
if not self._base.validate.table_name(table_name):
raise ValueError('Invalid table name of "{0}".'.format(table_name))
@@ -90,14 +101,20 @@ class BaseRecords:
INSERT INTO {0}{1}
VALUES {2};
""".format(table_name, columns_clause, values_clause)
- results = self._base.query.execute(query)
+ results = self._base.query.execute(query, display_query=display_query)
logger.query('{0}'.format(query))
logger.results('{0}'.format(results))
return results
- def update(self, table_name, values_clause, where_clause):
- """"""
+ def update(self, table_name, values_clause, where_clause, display_query=True):
+ """Updates record in provided table.
+
+ :param table_name: Name of table to insert into.
+ :param values_clause: Clause to specify values to insert.
+ :param where_clause: Clause to limit update scope.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
+ """
# Check that provided table name is valid format.
if not self._base.validate.table_name(table_name):
raise ValueError('Invalid table name of "{0}".'.format(table_name))
@@ -135,14 +152,19 @@ class BaseRecords:
SET {1}
WHERE {2};
""".format(table_name, values_clause, where_clause)
- results = self._base.query.execute(query)
+ results = self._base.query.execute(query, display_query=display_query)
logger.query('{0}'.format(query))
logger.results('{0}'.format(results))
return results
- def delete(self, table_name, where_clause):
- """"""
+ def delete(self, table_name, where_clause, display_query=True):
+ """Deletes record(s) in given table.
+
+ :param table_name: Name of table to insert into.
+ :param where_clause: Clause to limit delete scope.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
+ """
# Check that provided table name is valid format.
if not self._base.validate.table_name(table_name):
raise ValueError('Invalid table name of "{0}".'.format(table_name))
@@ -153,7 +175,7 @@ class BaseRecords:
# Delete record.
query = 'DELETE FROM {0} WHERE {1};'.format(table_name, where_clause)
- results = self._base.query.execute(query)
+ results = self._base.query.execute(query, display_query=display_query)
logger.query('{0}'.format(query))
logger.results('{0}'.format(results))
diff --git a/py_dbcn/connectors/core/tables.py b/py_dbcn/connectors/core/tables.py
index 61614a5..e6c912f 100644
--- a/py_dbcn/connectors/core/tables.py
+++ b/py_dbcn/connectors/core/tables.py
@@ -35,16 +35,17 @@ class BaseTables:
self._show_tables_query = None
self._describe_table_query = None
- def _get(self, show=False):
- """
- Gets list of all currently-available tables in database.
+ def _get(self, display_query=False, show=False):
+ """Gets list of all currently-available tables in database.
+
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to False.
:param show: Bool indicating if results should be printed to console or not. Used for "SHOW TABLES" query.
"""
if not self._show_tables_query:
raise ValueError('SHOW TABLES query is not defined.')
# Generate and execute query.
- results = self._base.query.execute(self._show_tables_query)
+ results = self._base.query.execute(self._show_tables_query, display_query=display_query)
# Convert to more friendly format.
formatted_results = []
@@ -58,15 +59,17 @@ class BaseTables:
# Return data.
return results
- def show(self):
- """
- Displays all tables available in database.
- """
- return self._get(show=True)
+ def show(self, display_query=True):
+ """Displays all tables available in database.
- def describe(self, table_name):
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
- Describes given table in database.
+ return self._get(display_query=display_query, show=True)
+
+ def describe(self, table_name, display_query=True):
+ """Describes given table in database.
+
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
if not self._describe_table_query:
raise ValueError('DESCRIBE TABLE query is not defined.')
@@ -82,20 +85,17 @@ class BaseTables:
# Generate and execute query.
query = self._describe_table_query.format(table_name)
- results = self._base.query.execute(query)
+ results = self._base.query.execute(query, display_query=display_query)
self._base.display.tables.describe(results, logger)
return results
- # pydbcn__postgresql_unittest__test_tables,public,test_tables__count,
- # column_name, data_type, is_nullable, column_default,
- # (field),
+ def create(self, table_name, table_columns, display_query=True):
+ """Creates new table with provided name.
- def create(self, table_name, table_columns):
- """
- Creates new table with provided name.
:param table_name: Desired name of new table.
:param table_columns: Column values for new table.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
# First, check that provided name is valid format.
if not self._base.validate.table_name(table_name):
@@ -118,11 +118,17 @@ class BaseTables:
# Create new table.
# raise NotImplemented('Function needs column-definition handling.')
query = 'CREATE TABLE {0} {1};'.format(table_name, table_columns)
- self._base.query.execute(query)
+ self._base.query.execute(query, display_query=display_query)
logger.results('Created table "{0}".'.format(table_name))
- def modify(self, table_name, modify_clause, column_clause):
- """Modifies table column with provided name."""
+ def modify(self, table_name, modify_clause, column_clause, display_query=True):
+ """Modifies table column with provided name.
+
+ :param table_name: Name of table to modify.
+ :param modify_clause: Clause of values to apply.
+ :param column_clause: Clause of columns to update.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
+ """
if str(modify_clause).upper() == 'ADD':
modify_clause = 'ADD'
elif str(modify_clause).upper() == 'DROP':
@@ -146,29 +152,51 @@ class BaseTables:
ALTER TABLE {0}
{1} {2};
""".format(table_name, modify_clause, column_clause)
- self._base.query.execute(query)
+ self._base.query.execute(query, display_query=display_query)
logger.results('Created table "{0}".'.format(table_name))
- def update(self, table_name, modify_clause, column_clause):
- """Alias for modify()."""
- return self.modify(table_name, modify_clause, column_clause)
+ def update(self, table_name, modify_clause, column_clause, display_query=True):
+ """Alias for modify().
+
+ :param table_name: Name of table to modify.
+ :param modify_clause: Clause of values to apply.
+ :param column_clause: Clause of columns to update.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
+ """
+ return self.modify(table_name, modify_clause, column_clause, display_query=display_query)
+
+ def add_column(self, table_name, column_clause, display_query=True):
+ """Adds column to provided table.
+
+ :param table_name: Name of table to modify.
+ :param column_clause: Clause of columns to add.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
+ """
+ return self.modify(table_name, 'ADD', column_clause, display_query=display_query)
- def add_column(self, table_name, column_clause):
- """Adds column to provided table."""
- return self.modify(table_name, 'ADD', column_clause)
+ def drop_column(self, table_name, column_clause, display_query=True):
+ """Drops column from provided table.
- def drop_column(self, table_name, column_clause):
- """Drops column from provided table."""
- return self.modify(table_name, 'DROP', column_clause)
+ :param table_name: Name of table to modify.
+ :param column_clause: Clause of columns to drop.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
+ """
+ return self.modify(table_name, 'DROP', column_clause, display_query=display_query)
- def modify_column(self, table_name, column_clause):
- """Modifies column in provided table."""
- return self.modify(table_name, 'MODIFY', column_clause)
+ def modify_column(self, table_name, column_clause, display_query=True):
+ """Modifies column in provided table.
- def drop(self, table_name):
+ :param table_name: Name of table to modify.
+ :param column_clause: Clause of columns to update.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
- Deletes table with provided name.
+ return self.modify(table_name, 'MODIFY', column_clause, display_query=display_query)
+
+ def drop(self, table_name, display_query=True):
+ """Deletes table with provided name.
+
:param table_name: Name of table to delete.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
# First, check that provided name is valid format.
if not self._base.validate.table_name(table_name):
@@ -184,19 +212,22 @@ class BaseTables:
# Remove table.
query = 'DROP TABLE {0};'.format(table_name)
- self._base.query.execute(query)
+ self._base.query.execute(query, display_query=display_query)
logger.results('Dropped table "{0}".'.format(table_name))
- def delete(self, table_name):
- """
- Alias for table "drop" function.
- """
- self.drop(table_name)
+ def delete(self, table_name, display_query=True):
+ """Alias for table "drop" function.
- def count(self, table_name):
+ :param table_name: Name of table to delete.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
- Returns number of all records present in provided table.
+ self.drop(table_name, display_query=display_query)
+
+ def count(self, table_name, display_query=True):
+ """Returns number of all records present in provided table.
+
:param table_name: Name of table to count.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
# Get list of valid tables.
available_tables = self._get()
@@ -208,7 +239,7 @@ class BaseTables:
)
# Count records in table.
- result = self._base.records.select(table_name, 'COUNT(*)')
+ result = self._base.records.select(table_name, 'COUNT(*)', display_query=display_query)
result = result[0][0]
return result
diff --git a/py_dbcn/connectors/postgresql/database.py b/py_dbcn/connectors/postgresql/database.py
index 4919b7b..2f076bb 100644
--- a/py_dbcn/connectors/postgresql/database.py
+++ b/py_dbcn/connectors/postgresql/database.py
@@ -30,9 +30,11 @@ class PostgresqlDatabase(BaseDatabase):
self._show_databases_query = 'SELECT datname FROM pg_database;'
self._current_database_query = 'SELECT current_database();'
- def use(self, db_name):
- """
- Selects given database for use.
+ def use(self, db_name, display_query=True):
+ """Selects given database for use.
+
+ :param db_name: Name of db to use.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
# First, check that provided name is valid format.
if not self._base.validate.database_name(db_name):
@@ -59,6 +61,9 @@ class PostgresqlDatabase(BaseDatabase):
'Could not find database "{0}". Valid options are {1}.'.format(db_name, available_databases)
)
+ if display_query:
+ logger.query('Switching databases. No query to display. Recreating connection.')
+
# Switch active database.
# PostgreSQL is annoying in that it doesn't seem to have a friendly way to switch databases.
# The only method seems to be by destroying the current connection and recreating a new one, this time
@@ -67,10 +72,11 @@ class PostgresqlDatabase(BaseDatabase):
self._base.create_connection(db_name=db_name)
logger.results('Database changed to "{0}".'.format(db_name))
- def drop(self, db_name):
- """
- Deletes database with provided name.
+ def drop(self, db_name, display_query=True):
+ """Deletes database with provided name.
+
:param db_name: Name of database to delete.
+ :param display_query: Optional bool indicating if query should output to console or not. Defaults to True.
"""
# First, check that provided name is valid format.
if not self._base.validate.database_name(db_name):
@@ -110,7 +116,7 @@ class PostgresqlDatabase(BaseDatabase):
switched_db = True
query = 'DROP DATABASE {0};'.format(db_name)
- self._base.query.execute(query)
+ self._base.query.execute(query, display_query=display_query)
logger.results('Dropped database "{0}".'.format(db_name))
# If we switched database, then immediately close connection at this point,
diff --git a/tests/connectors/core/test_display.py b/tests/connectors/core/test_display.py
index cdd6e89..30691bc 100644
--- a/tests/connectors/core/test_display.py
+++ b/tests/connectors/core/test_display.py
@@ -195,94 +195,97 @@ class CoreDisplayTablesTestMixin:
# Since this directly tests display of tables, ensure we use a fresh database.
db_name = '{0}d__tables__show'.format(self.test_db_name[0:-15])
- self.connector.database.create(db_name)
- self.connector.database.use(db_name)
+ self.connector.database.create(db_name, display_query=False)
+ self.connector.database.use(db_name, display_query=False)
with self.subTest('With no tables present'):
# Capture logging output.
- with self.assertLogs(None, 'INFO') as ilog:
+ with self.assertLogs(None, 'QUERY') as qlog:
self.connector.tables.show()
- self.assertText(self.get_logging_output(ilog, 0), 'SHOW TABLES;')
- self.assertText(self.get_logging_output(ilog, 1), 'Empty Set')
+ self.assertText(self.get_logging_output(qlog, 0), 'SHOW TABLES;')
+ self.assertText(self.get_logging_output(qlog, 1), 'Empty Set')
with self.subTest('Db name longer - Pt 1'):
# Create table.
- self.connector.tables.create('category', columns)
+ self.connector.tables.create('category', columns, display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.tables.show()
self.assertText(self.get_logging_output(ilog, 0), 'SHOW TABLES;')
- self.assertText(self.get_logging_output(ilog, 1), self.expected_output.tables.SHOW__DB_LONGER__PT_1)
+ self.assertText(self.get_logging_output(ilog, 3), self.expected_output.tables.SHOW__DB_LONGER__PT_1)
with self.subTest('Db name longer - Pt 2'):
# Create table.
- self.connector.tables.create('priority', columns)
+ self.connector.tables.create('priority', columns, display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.tables.show()
self.assertText(self.get_logging_output(ilog, 0), 'SHOW TABLES;')
- self.assertText(self.get_logging_output(ilog, 1), self.expected_output.tables.SHOW__DB_LONGER__PT_2)
+ self.assertText(self.get_logging_output(ilog, 3), self.expected_output.tables.SHOW__DB_LONGER__PT_2)
with self.subTest('Db name longer - Pt 3'):
# Create table.
- self.connector.tables.create('a', columns)
+ self.connector.tables.create('a', columns, display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.tables.show()
self.assertText(self.get_logging_output(ilog, 0), 'SHOW TABLES;')
- self.assertText(self.get_logging_output(ilog, 1), self.expected_output.tables.SHOW__DB_LONGER__PT_3)
+ self.assertText(self.get_logging_output(ilog, 3), self.expected_output.tables.SHOW__DB_LONGER__PT_3)
with self.subTest('Db name and table name equal length'):
# Create table.
self.connector.tables.create(
'test__testing__this_is_a_really_long_table_name__test_',
columns,
+ display_query=False,
)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.tables.show()
self.assertText(self.get_logging_output(ilog, 0), 'SHOW TABLES;')
- self.assertText(self.get_logging_output(ilog, 1), self.expected_output.tables.SHOW__EQUAL_LENGTH)
+ self.assertText(self.get_logging_output(ilog, 3), self.expected_output.tables.SHOW__EQUAL_LENGTH)
with self.subTest('Table name longer - Pt 1'):
# Create table.
self.connector.tables.create(
'test__testing__this_is_a_really_long_table_name__test__',
columns,
+ display_query=False,
)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.tables.show()
self.assertText(self.get_logging_output(ilog, 0), 'SHOW TABLES;')
- self.assertText(self.get_logging_output(ilog, 1), self.expected_output.tables.SHOW__TABLE_LONGER__PT_1)
+ self.assertText(self.get_logging_output(ilog, 3), self.expected_output.tables.SHOW__TABLE_LONGER__PT_1)
with self.subTest('Table name longer - Pt 2'):
# Create table.
- self.connector.tables.create('zzz', columns)
+ self.connector.tables.create('zzz', columns, display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.tables.show()
self.assertText(self.get_logging_output(ilog, 0), 'SHOW TABLES;')
- self.assertText(self.get_logging_output(ilog, 1), self.expected_output.tables.SHOW__TABLE_LONGER__PT_2)
+ self.assertText(self.get_logging_output(ilog, 3), self.expected_output.tables.SHOW__TABLE_LONGER__PT_2)
with self.subTest('Table name longer - Pt 3'):
# Create table.
self.connector.tables.create(
'test__testing__this_is_a_really_long_table_name__testing__',
columns,
+ display_query=False,
)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.tables.show()
self.assertText(self.get_logging_output(ilog, 0), 'SHOW TABLES;')
- self.assertText(self.get_logging_output(ilog, 1), self.expected_output.tables.SHOW__TABLE_LONGER__PT_3)
+ self.assertText(self.get_logging_output(ilog, 3), self.expected_output.tables.SHOW__TABLE_LONGER__PT_3)
def test__display__describe_tables(self):
""""""
@@ -293,38 +296,38 @@ class CoreDisplayTablesTestMixin:
# Since this directly tests display of tables, ensure we use a fresh database.
db_name = '{0}d__tables__desc'.format(self.test_db_name[0:-15])
- self.connector.database.create(db_name)
- self.connector.database.use(db_name)
+ self.connector.database.create(db_name, display_query=False)
+ self.connector.database.use(db_name, display_query=False)
# Create initial table to describe.
- self.connector.tables.create('category', columns)
+ self.connector.tables.create('category', columns, display_query=False)
with self.subTest('With only id'):
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.tables.describe('category')
- self.assertText(self.get_logging_output(ilog, 1), 'DESCRIBE category;')
- self.assertText(self.get_logging_output(ilog, 2), self.expected_output.tables.DESCRIBE__COLS_ID)
+ self.assertText(self.get_logging_output(ilog, 0), 'DESCRIBE category;')
+ self.assertText(self.get_logging_output(ilog, 1), self.expected_output.tables.DESCRIBE__COLS_ID)
with self.subTest('With id, name'):
# Add new table column.
- self.connector.tables.add_column('category', 'name VARCHAR(100)')
+ self.connector.tables.add_column('category', 'name VARCHAR(100)', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.tables.describe('category')
- self.assertText(self.get_logging_output(ilog, 1), 'DESCRIBE category;')
- self.assertText(self.get_logging_output(ilog, 2), self.expected_output.tables.DESCRIBE__COLS_ID_NAME)
+ self.assertText(self.get_logging_output(ilog, 0), 'DESCRIBE category;')
+ self.assertText(self.get_logging_output(ilog, 1), self.expected_output.tables.DESCRIBE__COLS_ID_NAME)
with self.subTest('With id, name, desc'):
# Add new table column.
- self.connector.tables.add_column('category', 'description VARCHAR(100)')
+ self.connector.tables.add_column('category', 'description VARCHAR(100)', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.tables.describe('category')
- self.assertText(self.get_logging_output(ilog, 1), 'DESCRIBE category;')
- self.assertText(self.get_logging_output(ilog, 2), self.expected_output.tables.DESCRIBE__COLS_ID_NAME_DESC)
+ self.assertText(self.get_logging_output(ilog, 0), 'DESCRIBE category;')
+ self.assertText(self.get_logging_output(ilog, 1), self.expected_output.tables.DESCRIBE__COLS_ID_NAME_DESC)
class CoreDisplayRecordsMixin:
@@ -351,7 +354,7 @@ class CoreDisplayRecordsMixin:
PRIMARY KEY ( id )
)"""
- self.connector.tables.create('category', columns)
+ self.connector.tables.create('category', columns, display_query=False)
with self.subTest('With no records present'):
# Capture logging output.
@@ -362,113 +365,113 @@ class CoreDisplayRecordsMixin:
with self.subTest('With 1 record present'):
# Create record.
- self.connector.records.insert('category', '(1, "tn", "td")')
+ self.connector.records.insert('category', '(1, "tn", "td")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_1)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_1)
with self.subTest('With 2 records present'):
# Create record.
- self.connector.records.insert('category', '(2, "t n", "t d")')
+ self.connector.records.insert('category', '(2, "t n", "t d")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_2)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_2)
with self.subTest('With 3 records present'):
# Create record.
- self.connector.records.insert('category', '(3, "te n", "te d")')
+ self.connector.records.insert('category', '(3, "te n", "te d")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_3)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_3)
with self.subTest('With 4 records present'):
# Create record.
- self.connector.records.insert('category', '(4, "tes n", "tes d")')
+ self.connector.records.insert('category', '(4, "tes n", "tes d")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_4)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_4)
with self.subTest('With 5 records present'):
# Create record.
- self.connector.records.insert('category', '(5, "test n", "test d")')
+ self.connector.records.insert('category', '(5, "test n", "test d")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_5)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_5)
with self.subTest('With 6 records present'):
# Create record.
- self.connector.records.insert('category', '(6, "test na", "test de")')
+ self.connector.records.insert('category', '(6, "test na", "test de")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_6)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_6)
with self.subTest('With 7 records present'):
# Create record.
- self.connector.records.insert('category', '(7, "test nam", "test des")')
+ self.connector.records.insert('category', '(7, "test nam", "test des")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_7)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_7)
with self.subTest('With 8 records present'):
# Create record.
- self.connector.records.insert('category', '(8, "test name", "test desc")')
+ self.connector.records.insert('category', '(8, "test name", "test desc")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_8)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_8)
with self.subTest('With 9 records present'):
# Create record.
- self.connector.records.insert('category', '(9, "test name", "test descr")')
+ self.connector.records.insert('category', '(9, "test name", "test descr")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_9)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_9)
with self.subTest('With 10 records present'):
# Create record.
- self.connector.records.insert('category', '(10, "test name", "test descri")')
+ self.connector.records.insert('category', '(10, "test name", "test descri")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_10)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_10)
with self.subTest('With 11 records present'):
# Create record.
- self.connector.records.insert('category', '(101, "test name", "test descrip")')
+ self.connector.records.insert('category', '(101, "test name", "test descrip")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_11)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_11)
with self.subTest('With 12 records present'):
# Create record.
@@ -478,34 +481,34 @@ class CoreDisplayRecordsMixin:
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_12)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_12)
with self.subTest('With 13 records present'):
# Create record.
- self.connector.records.insert('category', '(10101, "test name", "test descripti")')
+ self.connector.records.insert('category', '(10101, "test name", "test descripti")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_13)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_13)
with self.subTest('With 14 records present'):
# Create record.
- self.connector.records.insert('category', '(101010, "test name", "test descriptio")')
+ self.connector.records.insert('category', '(101010, "test name", "test descriptio")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_14)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_14)
with self.subTest('With 15 records present'):
# Create record.
- self.connector.records.insert('category', '(1010101, "test name", "test description")')
+ self.connector.records.insert('category', '(1010101, "test name", "test description")', display_query=False)
# Capture logging output.
with self.assertLogs(None, 'INFO') as ilog:
self.connector.records.select('category')
self.assertText(self.get_logging_output(ilog, 0), 'SELECT * FROM category;')
- self.assertText(self.get_logging_output(ilog, 8), self.expected_output.records.SELECT__PT_15)
+ self.assertText(self.get_logging_output(ilog, 7), self.expected_output.records.SELECT__PT_15)
--
GitLab