neo4jbackup-tests.py

import unittest
from unittest.mock import patch, MagicMock
import os
import subprocess
from neo4j_backup import perform_backup

class TestPerformBackup(unittest.TestCase):

    @patch('subprocess.check_output')
    @patch('os.path.getsize')
    def test_successful_backup(self, mock_getsize, mock_check_output):
        # Mock environment variables
        os.environ["COMPRESS"] = "true"
        os.environ["KEEP_FAILED"] = "true"
        os.environ["TO_PATH"] = "/path/to/backup"
        os.environ["BACKUP_TYPE"] = "full"
        os.environ["NEO4J_HOST"] = "localhost"
        os.environ["NEO4J_PORT"] = "7687"
        os.environ["DATABASE"] = ""
        os.environ["MIN_SIZE_BYTES"] = "1073741824"

        # Mock subprocess.check_output
        mock_check_output.return_value = b"Backup completed successfully"

        # Mock os.path.getsize
        mock_getsize.return_value = 2000000000  # 2GB

        # Call the function
        perform_backup()

        # Assert that subprocess.check_output was called with the correct command
        mock_check_output.assert_called_once_with('neo4j-admin database backup --compress=true --keep-failed=true --to-path=/path/to/backup --type=full --from=localhost:7687 ', stderr=subprocess.STDOUT, shell=True)

        # Assert that os.path.getsize was called with the correct path
        mock_getsize.assert_called_once_with('/path/to/backup')

    @patch('subprocess.check_output')
    @patch('os.path.getsize')
    def test_failed_backup(self, mock_getsize, mock_check_output):
        # Mock environment variables
        os.environ["COMPRESS"] = "true"
        os.environ["KEEP_FAILED"] = "true"
        os.environ["TO_PATH"] = "/path/to/backup"
        os.environ["BACKUP_TYPE"] = "full"
        os.environ["NEO4J_HOST"] = "localhost"
        os.environ["NEO4J_PORT"] = "7687"
        os.environ["DATABASE"] = ""
        os.environ["MIN_SIZE_BYTES"] = "1073741824"

        # Mock subprocess.check_output to raise an exception
        mock_check_output.side_effect = subprocess.CalledProcessError(1, "command", output=b"Backup failed")

        # Call the function
        perform_backup()

        # Assert that subprocess.check_output was called with the correct command
        mock_check_output.assert_called_once_with('neo4j-admin database backup --compress=true --keep-failed=true --to-path=/path/to/backup --type=full --from=localhost:7687 ', stderr=subprocess.STDOUT, shell=True)

        # Assert that os.path.getsize was not called
        mock_getsize.assert_not_called()

    @patch('subprocess.check_output')
    @patch('os.path.getsize')
    @patch('boto3.client')
    def test_backup_size_less_than_min_size(self, mock_boto3_client, mock_getsize, mock_check_output):
        # Mock environment variables
        os.environ["COMPRESS"] = "true"
        os.environ["KEEP_FAILED"] = "true"
        os.environ["TO_PATH"] = "/path/to/backup"
        os.environ["BACKUP_TYPE"] = "full"
        os.environ["NEO4J_HOST"] = "localhost"
        os.environ["NEO4J_PORT"] = "7687"
        os.environ["DATABASE"] = ""
        os.environ["MIN_SIZE_BYTES"] = "1073741824"

        # Mock subprocess.check_output
        mock_check_output.return_value = b"Backup completed successfully"

        # Mock os.path.getsize to return a smaller size than the minimum size
        mock_getsize.return_value = 500000000  # 500MB

        # Call the function
        perform_backup()

        # Assert that subprocess.check_output was called with the correct command
        mock_check_output.assert_called_once_with('neo4j-admin database backup --compress=true --keep-failed=true --to-path=/path/to/backup --type=full --from=localhost:7687 ', stderr=subprocess.STDOUT, shell=True)

        # Assert that os.path.getsize was called with the correct path
        mock_getsize.assert_called_once_with('/path/to/backup')

        # Assert that boto3.client.publish was called with the error message
        mock_boto3_client.return_value.publish.assert_called_once_with(TopicArn=None, Message='Backup size is less than 1073741824 bytes')

if __name__ == '__main__':
    unittest.main()