diff --git a/evennia/contrib/utils/database_backup/database_backup.py b/evennia/contrib/utils/database_backup/database_backup.py index d1c6bdef1d..ff92837351 100644 --- a/evennia/contrib/utils/database_backup/database_backup.py +++ b/evennia/contrib/utils/database_backup/database_backup.py @@ -30,6 +30,17 @@ class DatabaseBackupScript(DefaultScript): self.desc = "Database backups" self.persistent = True + def log(self, message): + global _MUDINFO_CHANNEL + if not _MUDINFO_CHANNEL and settings.CHANNEL_MUDINFO: + channels = search.search_channel(settings.CHANNEL_MUDINFO["key"]) + if channels: + _MUDINFO_CHANNEL = channels[0] + + if _MUDINFO_CHANNEL: + _MUDINFO_CHANNEL.msg(message) + logger.log_sec(message) + def backup_postgres(self, db_name, db_user, output_file_path): """ Run `pg_dump` on the postgreSQL database and save the output. @@ -44,7 +55,7 @@ class DatabaseBackupScript(DefaultScript): stdout=open(output_file_path, "w"), check=True, ) - return f"|wPostgreSQL db backed up in: {BACKUP_FOLDER}|n" + self.log(f"|wpostgresql db backed up in: {BACKUP_FOLDER}|n") def backup_sqlite3(self, db_name, output_file_path): """ @@ -57,14 +68,9 @@ class DatabaseBackupScript(DefaultScript): output_file_path += ".db3" os.makedirs(os.path.dirname(output_file_path), exist_ok=True) shutil.copy(db_name, output_file_path) - return f"|wsqlite3 db backed up in {BACKUP_FOLDER}|n" + self.log(f"|wsqlite3 db backed up in: {BACKUP_FOLDER}|n") def at_repeat(self): - global _MUDINFO_CHANNEL - if not _MUDINFO_CHANNEL: - if settings.CHANNEL_MUDINFO: - _MUDINFO_CHANNEL = ChannelDB.objects.get(db_key=settings.CHANNEL_MUDINFO["key"]) - databases = settings.DATABASES db = databases["default"] engine = db.get("ENGINE") @@ -78,13 +84,10 @@ class DatabaseBackupScript(DefaultScript): output_file_path = os.path.join(BACKUP_FOLDER, output_file) if "postgres" in engine: - message = self.backup_postgres(db_name, db_user, output_file_path) + self.backup_postgres(db_name, db_user, output_file_path) elif "sqlite3" in engine: - message = self.backup_sqlite3(db_name, output_file_path) + self.backup_sqlite3(db_name, output_file_path) - logger.log_sec(message) - if _MUDINFO_CHANNEL: - _MUDINFO_CHANNEL.msg(message) except Exception as e: logger.log_err("Backup failed: {}".format(e)) diff --git a/evennia/contrib/utils/database_backup/tests.py b/evennia/contrib/utils/database_backup/tests.py new file mode 100644 index 0000000000..0c617de42f --- /dev/null +++ b/evennia/contrib/utils/database_backup/tests.py @@ -0,0 +1,88 @@ +""" +Tests for database backups. +""" + +import evennia.contrib.utils.database_backup.database_backup as backup +from evennia.commands.default.tests import BaseEvenniaCommandTest +from unittest.mock import patch + +EXCEPTION_STR = "failed" + + +class TestDatabaseBackupScript(BaseEvenniaCommandTest): + mocked_db_setting_postgres = patch( + "django.conf.settings.DATABASES", + { + "default": { + "ENGINE": "django.db.backends.postgresql_psycopg2", + "NAME": "fake_name", + "USER": "fake_user", + } + }, + ) + + def setUp(self): + super().setUp() + + @patch("shutil.copy") + @patch("evennia.utils.logger.log_sec") + def test_sqlite_success(self, mock_logger, mock_copy): + mock_copy.return_value.returncode = 0 + self.call( + backup.CmdBackup(), + "300", + "You have scheduled backups to run every 300 seconds.", + caller=self.char1, + ) + + mock_logger.assert_called_with(f"|wsqlite3 db backed up in: {backup.BACKUP_FOLDER}|n") + + self.call( + backup.CmdBackup(), + "/stop", + "DB backup script deleted.", + caller=self.char1, + ) + + @patch("shutil.copy") + @patch("evennia.utils.logger.log_err") + def test_sqlite_failure(self, mock_logger, mock_copy): + mock_copy.return_value.returncode = 1 + mock_copy.side_effect = Exception(EXCEPTION_STR) + + self.call( + backup.CmdBackup(), + "", + "You have scheduled backups to run every 86400 seconds.", + caller=self.char1, + ) + mock_logger.assert_called_with(f"Backup failed: {EXCEPTION_STR}") + + @mocked_db_setting_postgres + @patch("subprocess.run") + @patch("evennia.utils.logger.log_sec") + def test_postgres_success(self, mock_logger, mock_run): + mock_run.return_value.returncode = 0 + + self.call( + backup.CmdBackup(), + "", + "You have scheduled backups to run every 86400 seconds.", + caller=self.char1, + ) + mock_logger.assert_called_with(f"|wpostgresql db backed up in: {backup.BACKUP_FOLDER}|n") + + @mocked_db_setting_postgres + @patch("subprocess.run") + @patch("evennia.utils.logger.log_err") + def test_postgres_failure(self, mock_logger, mock_run): + mock_run.return_value.returncode = 0 + mock_run.side_effect = Exception(EXCEPTION_STR) + + self.call( + backup.CmdBackup(), + "", + "You have scheduled backups to run every 86400 seconds.", + caller=self.char1, + ) + mock_logger.assert_called_with(f"Backup failed: {EXCEPTION_STR}")