""" Tests for MySQL backup functionality in backup_manager.py """ import os import sys import zipfile import tempfile import unittest from unittest.mock import patch, MagicMock, call sys.path.insert(0, os.path.dirname(__file__)) import backup_manager class TestPerformMysqlBackupValidation(unittest.TestCase): """Tests for input validation in perform_mysql_backup.""" def test_invalid_mode_returns_error(self): success, msg, path = backup_manager.perform_mysql_backup( {}, db_mode='invalid' ) self.assertFalse(success) self.assertIn('invalid', msg.lower()) self.assertIsNone(path) def test_single_mode_without_db_name_returns_error(self): success, msg, path = backup_manager.perform_mysql_backup( {}, db_mode='single', db_name='' ) self.assertFalse(success) self.assertIn('single', msg) self.assertIsNone(path) def test_single_mode_with_none_db_name_returns_error(self): success, msg, path = backup_manager.perform_mysql_backup( {}, db_mode='single', db_name=None ) self.assertFalse(success) self.assertIn('single', msg) self.assertIsNone(path) def test_list_mode_without_db_list_returns_error(self): success, msg, path = backup_manager.perform_mysql_backup( {}, db_mode='list', db_list='' ) self.assertFalse(success) self.assertIn('list', msg) self.assertIsNone(path) def test_list_mode_with_only_commas_returns_error(self): success, msg, path = backup_manager.perform_mysql_backup( {}, db_mode='list', db_list=',,,,' ) self.assertFalse(success) self.assertIn('list', msg) self.assertIsNone(path) def test_list_mode_with_none_returns_error(self): success, msg, path = backup_manager.perform_mysql_backup( {}, db_mode='list', db_list=None ) self.assertFalse(success) self.assertIn('list', msg) self.assertIsNone(path) class TestPerformMysqlBackupSingleMode(unittest.TestCase): """Tests for single-database backup mode.""" def setUp(self): self.tmpdir = tempfile.mkdtemp() def tearDown(self): import shutil shutil.rmtree(self.tmpdir, ignore_errors=True) @patch('subprocess.run') def test_single_mode_success(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='-- SQL DUMP --', stderr='') cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'secret'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='single', db_name='mydb', backup_dir=self.tmpdir, prefix='test_backup' ) self.assertTrue(success) self.assertIn('mydb', msg) self.assertIsNotNone(path) self.assertTrue(os.path.exists(path)) # Verify zip contains the dump with zipfile.ZipFile(path) as zf: self.assertEqual(len(zf.namelist()), 1) self.assertTrue(zf.namelist()[0].startswith('mydb_')) @patch('subprocess.run') def test_single_mode_mysqldump_failure(self, mock_run): mock_run.return_value = MagicMock( returncode=1, stdout='', stderr='Access denied' ) cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'wrong'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='single', db_name='mydb', backup_dir=self.tmpdir, prefix='test_backup' ) self.assertFalse(success) self.assertIn('failed', msg.lower()) # Zip should be removed when all databases fail self.assertIsNone(path) @patch('subprocess.run') def test_password_not_in_subprocess_args(self, mock_run): """Password must not appear in the subprocess command arguments.""" mock_run.return_value = MagicMock(returncode=0, stdout='-- SQL --', stderr='') cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'supersecret'} backup_manager.perform_mysql_backup( cfg, db_mode='single', db_name='mydb', backup_dir=self.tmpdir, prefix='test_backup' ) args = mock_run.call_args[0][0] self.assertNotIn('supersecret', args) # Password is passed via environment variable env = mock_run.call_args[1].get('env', {}) self.assertEqual(env.get('MYSQL_PWD'), 'supersecret') class TestPerformMysqlBackupListMode(unittest.TestCase): """Tests for list-of-databases backup mode.""" def setUp(self): self.tmpdir = tempfile.mkdtemp() def tearDown(self): import shutil shutil.rmtree(self.tmpdir, ignore_errors=True) @patch('subprocess.run') def test_list_mode_all_success(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='-- SQL --', stderr='') cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'pass'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='list', db_list='db1, db2, db3', backup_dir=self.tmpdir, prefix='test_backup' ) self.assertTrue(success) self.assertEqual(mock_run.call_count, 3) with zipfile.ZipFile(path) as zf: self.assertEqual(len(zf.namelist()), 3) @patch('subprocess.run') def test_list_mode_partial_failure(self, mock_run): def side_effect(cmd, **kwargs): db = cmd[-1] if db == 'db2': return MagicMock(returncode=1, stdout='', stderr='DB not found') return MagicMock(returncode=0, stdout='-- SQL --', stderr='') mock_run.side_effect = side_effect cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'pass'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='list', db_list='db1,db2,db3', backup_dir=self.tmpdir, prefix='test_backup' ) # Partial success: returns True with a warning message self.assertTrue(success) self.assertIn('error', msg.lower()) self.assertIsNotNone(path) with zipfile.ZipFile(path) as zf: self.assertEqual(len(zf.namelist()), 2) # only db1 and db3 @patch('subprocess.run') def test_list_mode_all_fail_removes_zip(self, mock_run): mock_run.return_value = MagicMock(returncode=1, stdout='', stderr='Error') cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'pass'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='list', db_list='db1,db2', backup_dir=self.tmpdir, prefix='test_backup' ) self.assertFalse(success) self.assertIsNone(path) class TestPerformMysqlBackupAllMode(unittest.TestCase): """Tests for all-databases backup mode.""" def setUp(self): self.tmpdir = tempfile.mkdtemp() def tearDown(self): import shutil shutil.rmtree(self.tmpdir, ignore_errors=True) @patch('subprocess.run') @patch('pymysql.connect') def test_all_mode_excludes_system_databases(self, mock_connect, mock_run): mock_cursor = MagicMock() mock_cursor.__enter__ = lambda s: s mock_cursor.__exit__ = MagicMock(return_value=False) mock_cursor.fetchall.return_value = [ {'Database': 'myapp'}, {'Database': 'information_schema'}, {'Database': 'performance_schema'}, {'Database': 'mysql'}, {'Database': 'sys'}, ] mock_conn = MagicMock() mock_conn.cursor.return_value = mock_cursor mock_connect.return_value = mock_conn mock_run.return_value = MagicMock(returncode=0, stdout='-- SQL --', stderr='') cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'pass'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='all', backup_dir=self.tmpdir, prefix='test_backup' ) self.assertTrue(success) # Only 1 non-system DB should be backed up self.assertEqual(mock_run.call_count, 1) backed_up_db = mock_run.call_args[0][0][-1] self.assertEqual(backed_up_db, 'myapp') @patch('pymysql.connect') def test_all_mode_connection_failure(self, mock_connect): mock_connect.side_effect = Exception("Connection refused") cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'pass'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='all', backup_dir=self.tmpdir, prefix='test_backup' ) self.assertFalse(success) self.assertIn('list databases', msg.lower()) self.assertIsNone(path) @patch('subprocess.run') @patch('pymysql.connect') def test_all_mode_no_user_databases(self, mock_connect, mock_run): mock_cursor = MagicMock() mock_cursor.__enter__ = lambda s: s mock_cursor.__exit__ = MagicMock(return_value=False) mock_cursor.fetchall.return_value = [ {'Database': 'information_schema'}, {'Database': 'mysql'}, ] mock_conn = MagicMock() mock_conn.cursor.return_value = mock_cursor mock_connect.return_value = mock_conn cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'pass'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='all', backup_dir=self.tmpdir, prefix='test_backup' ) self.assertFalse(success) self.assertIn('no databases', msg.lower()) self.assertIsNone(path) class TestPerformMysqlBackupOutputFile(unittest.TestCase): """Tests for output file naming and directory creation.""" def setUp(self): self.tmpdir = tempfile.mkdtemp() def tearDown(self): import shutil shutil.rmtree(self.tmpdir, ignore_errors=True) @patch('subprocess.run') def test_output_file_has_timestamp_and_prefix(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='-- SQL --', stderr='') cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'pass'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='single', db_name='testdb', backup_dir=self.tmpdir, prefix='myjob' ) self.assertTrue(success) filename = os.path.basename(path) self.assertTrue(filename.startswith('myjob_')) self.assertTrue(filename.endswith('.zip')) @patch('subprocess.run') def test_creates_backup_dir_if_missing(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='-- SQL --', stderr='') new_dir = os.path.join(self.tmpdir, 'newsubdir') self.assertFalse(os.path.exists(new_dir)) cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'pass'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='single', db_name='testdb', backup_dir=new_dir, prefix='myjob' ) self.assertTrue(success) self.assertTrue(os.path.exists(new_dir)) @patch('subprocess.run') def test_zip_uses_compression(self, mock_run): mock_run.return_value = MagicMock(returncode=0, stdout='-- SQL DUMP CONTENT --', stderr='') cfg = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'pass'} success, msg, path = backup_manager.perform_mysql_backup( cfg, db_mode='single', db_name='testdb', backup_dir=self.tmpdir, prefix='test' ) self.assertTrue(success) with zipfile.ZipFile(path) as zf: info = zf.infolist()[0] self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED) if __name__ == '__main__': unittest.main()