278 lines
11 KiB
Python
278 lines
11 KiB
Python
|
import os
|
||
|
|
import re
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
from pygadmin.database_dumper import DatabaseDumper
|
||
|
|
from pygadmin.configurator import global_app_configurator
|
||
|
|
|
||
|
|
|
||
|
|
class TestDatabaseDumperMethods(unittest.TestCase):
|
||
|
|
"""
|
||
|
|
Test the class for dumping a database and its behavior.
|
||
|
|
"""
|
||
|
|
|
||
|
|
def test_valid_database_dump(self):
|
||
|
|
"""
|
||
|
|
Use valid data for dumping a database and get the result.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Use a valid connection and valid dump data.
|
||
|
|
database_dumper = DatabaseDumper("testuser", "testdb", "localhost", 5432, "Table", "test")
|
||
|
|
# Get the result of a database dump.
|
||
|
|
result = database_dumper.dump_database()
|
||
|
|
|
||
|
|
assert result is not None
|
||
|
|
|
||
|
|
def test_invalid_database_dump(self):
|
||
|
|
"""
|
||
|
|
Use invalid data for dumping a database and check the invalid result.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Use an invalid connection and invalid dump data.
|
||
|
|
database_dumper = DatabaseDumper("testuser", "testdb", "localhost", 1337, "Table", "test")
|
||
|
|
# Get the result of a database dump.
|
||
|
|
result = database_dumper.dump_database()
|
||
|
|
|
||
|
|
assert result is None
|
||
|
|
|
||
|
|
def test_valid_create_pass_file(self):
|
||
|
|
"""
|
||
|
|
Test the method for creating a pass file with valid data.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Define the relevant data for a dump.
|
||
|
|
user = "testuser"
|
||
|
|
database = "testdb"
|
||
|
|
host = "localhost"
|
||
|
|
port = 5432
|
||
|
|
dump_information = "Table"
|
||
|
|
table_name = "test"
|
||
|
|
|
||
|
|
# Define a password identifier, which is relevant for the creating of a pass file.
|
||
|
|
password_identifier = "{}@{}:{}".format(user, host, port)
|
||
|
|
|
||
|
|
# Create a dumper.
|
||
|
|
database_dumper = DatabaseDumper(user, database, host, port, dump_information, table_name)
|
||
|
|
|
||
|
|
# Get the file path and handler.
|
||
|
|
file_path, file_handler = database_dumper.create_pass_file(password_identifier)
|
||
|
|
|
||
|
|
# Check the file path and the file handler for its existence.
|
||
|
|
assert os.path.exists(file_path) is True
|
||
|
|
assert os.path.exists(file_handler) is True
|
||
|
|
|
||
|
|
def test_invalid_create_pass_file(self):
|
||
|
|
"""
|
||
|
|
Test the creation and usage of a pass file with invalid data, so the host, user and port cannot be found in the
|
||
|
|
password manager. The file creation and usage should function normally.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Define the relevant data for a dump.
|
||
|
|
user = "testasdf"
|
||
|
|
database = "testdb"
|
||
|
|
host = "unknown"
|
||
|
|
port = 1337
|
||
|
|
dump_information = "Table"
|
||
|
|
table_name = "test"
|
||
|
|
|
||
|
|
# Define a password identifier, which is relevant for the creating of a pass file.
|
||
|
|
password_identifier = "{}@{}:{}".format(user, host, port)
|
||
|
|
|
||
|
|
# Create a dumper.
|
||
|
|
database_dumper = DatabaseDumper(user, database, host, port, dump_information, table_name)
|
||
|
|
|
||
|
|
# Get the file path and handler.
|
||
|
|
file_path, file_handler = database_dumper.create_pass_file(password_identifier)
|
||
|
|
|
||
|
|
# Check the file path and the file handler for its existence, because they should also exist for invalid data.
|
||
|
|
assert os.path.exists(file_path) is True
|
||
|
|
assert os.path.exists(file_handler) is True
|
||
|
|
|
||
|
|
def test_pg_dump_path(self):
|
||
|
|
"""
|
||
|
|
Test the functionality of the method for getting the pg dump path with also checking for the path in the global
|
||
|
|
app configurator.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Get the pg dump path out of the global app configurator.
|
||
|
|
configurator_dump_path = global_app_configurator.get_single_configuration("pg_dump_path")
|
||
|
|
|
||
|
|
# Create a dumper.
|
||
|
|
database_dumper = DatabaseDumper("testuser", "testdb", "localhost", 5432, "Table", "test")
|
||
|
|
# Get the dump path as accessible attribute of the class.
|
||
|
|
database_dumper.get_pg_dump_path()
|
||
|
|
|
||
|
|
# If a path in the configurator does exist, check for the correct set.
|
||
|
|
if configurator_dump_path is not None and os.path.exists(configurator_dump_path):
|
||
|
|
# The path in the configurator should be the pg dump path of the dumper.
|
||
|
|
assert database_dumper.pg_dump_path == configurator_dump_path
|
||
|
|
|
||
|
|
# The path should not be a None, so it has been set.
|
||
|
|
assert database_dumper.pg_dump_path is not None
|
||
|
|
|
||
|
|
def test_database_pg_dump_statement(self):
|
||
|
|
"""
|
||
|
|
Test the creation of a pg dump statement for a database.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Define the relevant parameters as local variables.
|
||
|
|
user = "testuser"
|
||
|
|
database = "testdb"
|
||
|
|
host = "localhost"
|
||
|
|
port = 5432
|
||
|
|
dump_information = "Database"
|
||
|
|
information_name = "testdb"
|
||
|
|
|
||
|
|
# Create a dumper.
|
||
|
|
database_dumper = DatabaseDumper(user, database, host, port, dump_information, information_name)
|
||
|
|
# Get the dump path as attribute of the dumper, which is necessary for the dump statement.
|
||
|
|
database_dumper.get_pg_dump_path()
|
||
|
|
# Get the dump statement as attribute of the dumper.
|
||
|
|
database_dumper.get_pg_dump_statement()
|
||
|
|
|
||
|
|
# The statement is a list, so get the list.
|
||
|
|
statement_list = database_dumper.pg_dump_statement
|
||
|
|
|
||
|
|
# Check for the correct instance of the list.
|
||
|
|
assert isinstance(statement_list, list)
|
||
|
|
# Check for the relevant strings and statements as part of the statement list.
|
||
|
|
assert database_dumper.pg_dump_path in statement_list
|
||
|
|
assert "-T*" in statement_list
|
||
|
|
assert "--create" in statement_list
|
||
|
|
assert "--dbname=postgresql://{}@{}:{}/{}".format(user, host, port, database) in statement_list
|
||
|
|
|
||
|
|
def test_table_pg_dump_statement(self):
|
||
|
|
"""
|
||
|
|
Test the creation of a pg dump statement for a table.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Define the relevant parameters as local variables.
|
||
|
|
user = "testuser"
|
||
|
|
database = "testdb"
|
||
|
|
host = "localhost"
|
||
|
|
port = 5432
|
||
|
|
dump_information = "Table"
|
||
|
|
information_name = "test"
|
||
|
|
|
||
|
|
# Create a dumper.
|
||
|
|
database_dumper = DatabaseDumper(user, database, host, port, dump_information, information_name)
|
||
|
|
# Get the dump path as attribute of the dumper, which is necessary for the dump statement.
|
||
|
|
database_dumper.get_pg_dump_path()
|
||
|
|
# Get the dump statement as attribute of the dumper.
|
||
|
|
database_dumper.get_pg_dump_statement()
|
||
|
|
|
||
|
|
# The statement is a list, so get the list.
|
||
|
|
statement_list = database_dumper.pg_dump_statement
|
||
|
|
|
||
|
|
# Check for the correct instance of the list.
|
||
|
|
assert isinstance(statement_list, list)
|
||
|
|
# Check for the relevant strings and statements as part of the statement list.
|
||
|
|
assert database_dumper.pg_dump_path in statement_list
|
||
|
|
assert "--table={}".format(information_name) in statement_list
|
||
|
|
assert "--schema-only" in statement_list
|
||
|
|
assert "--dbname=postgresql://{}@{}:{}/{}".format(user, host, port, database) in statement_list
|
||
|
|
|
||
|
|
def test_valid_database_dump_clean_result(self):
|
||
|
|
"""
|
||
|
|
Test the dump of a database with valid data and test the clean of the result.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Use a valid connection and valid dump data.
|
||
|
|
database_dumper = DatabaseDumper("testuser", "testdb", "localhost", 5432, "Table", "test")
|
||
|
|
# Get the result of a database dump.
|
||
|
|
result = database_dumper.dump_database_and_clean_result()
|
||
|
|
|
||
|
|
# The result should be a list.
|
||
|
|
assert isinstance(result, list)
|
||
|
|
|
||
|
|
def test_invalid_database_dump_clean_result(self):
|
||
|
|
"""
|
||
|
|
Test the dump of a database with invalid data and test the clean of the result.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Use an invalid connection and invalid dump data.
|
||
|
|
database_dumper = DatabaseDumper("testuser", "testdb", "localhost", 1337, "Table", "test")
|
||
|
|
# Get the result of a database dump.
|
||
|
|
result = database_dumper.dump_database()
|
||
|
|
|
||
|
|
assert result is None
|
||
|
|
|
||
|
|
def test_clean_database_result(self):
|
||
|
|
"""
|
||
|
|
Test the clean of the database result.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Use a valid connection and valid dump data.
|
||
|
|
database_dumper = DatabaseDumper("testuser", "testdb", "localhost", 5432, "Database", "testdb")
|
||
|
|
# Get the result of a database dump.
|
||
|
|
result = database_dumper.dump_database()
|
||
|
|
# Split the result into lines.
|
||
|
|
result_lines = result.stdout.split(os.linesep)
|
||
|
|
# Clean the result.
|
||
|
|
result_list = database_dumper.clean_database_result(result_lines)
|
||
|
|
|
||
|
|
# Check for the correct clean, so only lines with CREATE or ALTER are valid.
|
||
|
|
for line in result_list:
|
||
|
|
assert re.search("CREATE|ALTER", line) is not None
|
||
|
|
|
||
|
|
def test_clean_table_result(self):
|
||
|
|
"""
|
||
|
|
Test the clean of the table result.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Use a valid connection and valid dump data.
|
||
|
|
database_dumper = DatabaseDumper("testuser", "testdb", "localhost", 5432, "Table", "test")
|
||
|
|
# Get the result of a database dump.
|
||
|
|
result = database_dumper.dump_database()
|
||
|
|
# Split the result into lines.
|
||
|
|
result_lines = result.stdout.split(os.linesep)
|
||
|
|
# Clean the table result.
|
||
|
|
result_list = database_dumper.clean_table_result(result_lines)
|
||
|
|
# Define a bracket count, which counts opened and closed brackets.
|
||
|
|
bracket_count = 0
|
||
|
|
|
||
|
|
# Check every line for a CREATE or open/closed brackets.
|
||
|
|
for line in result_list:
|
||
|
|
assert re.search("CREATE", line) or bracket_count != 0
|
||
|
|
for character in line:
|
||
|
|
if character == "(":
|
||
|
|
bracket_count += 1
|
||
|
|
|
||
|
|
if character == ")":
|
||
|
|
bracket_count -= 1
|
||
|
|
|
||
|
|
def test_clean_view_result(self):
|
||
|
|
"""
|
||
|
|
Test the clean of a view result.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Use a valid connection and valid dump data.
|
||
|
|
database_dumper = DatabaseDumper("testuser", "testdb", "localhost", 5432, "View", "testview")
|
||
|
|
# Get the result of a database dump.
|
||
|
|
result = database_dumper.dump_database()
|
||
|
|
# Split the result into lines.
|
||
|
|
result_lines = result.stdout.split(os.linesep)
|
||
|
|
# Clean the table result
|
||
|
|
result_list = database_dumper.clean_view_result(result_lines)
|
||
|
|
# This line is set to True after a CREATE VIEW and is set to False after a semicolon, which ends the command.
|
||
|
|
create_view = False
|
||
|
|
|
||
|
|
# Check the result lines.
|
||
|
|
for line in result_list:
|
||
|
|
# The line should contain a CREATE VIEW or the create view should be in a previous line.
|
||
|
|
assert re.search("CREATE VIEW", line) or create_view is True
|
||
|
|
|
||
|
|
create_view = True
|
||
|
|
|
||
|
|
# Check for a semicolon to end the command.
|
||
|
|
for character in line:
|
||
|
|
if character == ";":
|
||
|
|
create_view = False
|
||
|
|
|
||
|
|
def test_invalid_clean_result(self):
|
||
|
|
# Use a valid connection and valid dump data.
|
||
|
|
database_dumper = DatabaseDumper("testuser", "testdb", "localhost", 5432, "Database", "testdb")
|
||
|
|
|
||
|
|
database_dumper.clean_database_result(None)
|
||
|
|
|