mirror of
https://github.com/ansible-collections/community.general.git
synced 2026-02-04 07:51:50 +00:00
1062 lines
No EOL
51 KiB
Python
1062 lines
No EOL
51 KiB
Python
# Copyright (c) 2026 Aleksandr Gabidullin <qualittv@gmail.com>
|
||
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
|
||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from unittest.mock import Mock, mock_open, patch
|
||
|
||
|
||
class TestLogrotateConfig(unittest.TestCase):
|
||
"""Unit tests for the logrotate_config module."""
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
"""Set up test environment."""
|
||
cls.test_dir = tempfile.mkdtemp()
|
||
cls.mock_ansible_basic = Mock()
|
||
cls.mock_ansible_basic.AnsibleModule = Mock()
|
||
cls.mock_converters = Mock()
|
||
cls.mock_converters.to_native = str
|
||
cls.patcher_basic = patch.dict(
|
||
"sys.modules",
|
||
{
|
||
"ansible.module_utils.basic": cls.mock_ansible_basic,
|
||
"ansible.module_utils.common.text.converters": cls.mock_converters,
|
||
},
|
||
)
|
||
cls.patcher_basic.start()
|
||
|
||
@classmethod
|
||
def tearDownClass(cls):
|
||
"""Clean up after all tests."""
|
||
cls.patcher_basic.stop()
|
||
if os.path.exists(cls.test_dir):
|
||
shutil.rmtree(cls.test_dir)
|
||
|
||
def setUp(self):
|
||
"""Set up test fixtures."""
|
||
self.mock_ansible_basic.AnsibleModule.reset_mock()
|
||
self.mock_module = Mock()
|
||
self.mock_module.params = {}
|
||
self.mock_module.fail_json = Mock(side_effect=Exception("fail_json called"))
|
||
self.mock_module.exit_json = Mock()
|
||
self.mock_module.check_mode = False
|
||
self.mock_module.get_bin_path = Mock(return_value="/usr/sbin/logrotate")
|
||
self.mock_module.atomic_move = Mock()
|
||
self.mock_module.warn = Mock()
|
||
self.mock_module.run_command = Mock(return_value=(0, "", ""))
|
||
self.mock_ansible_basic.AnsibleModule.return_value = self.mock_module
|
||
self.config_dir = os.path.join(self.test_dir, "logrotate.d")
|
||
os.makedirs(self.config_dir, exist_ok=True)
|
||
for module_name in list(sys.modules.keys()):
|
||
if "logrotate" in module_name or "ansible_collections.community.general.plugins.modules" in module_name:
|
||
del sys.modules[module_name]
|
||
|
||
def tearDown(self):
|
||
"""Clean up after test."""
|
||
pass
|
||
|
||
def _setup_module_params(self, **params):
|
||
"""Helper to set up module parameters."""
|
||
default_params = {
|
||
"name": "test",
|
||
"state": "present",
|
||
"config_dir": self.config_dir,
|
||
"paths": ["/var/log/test/*.log"],
|
||
"rotate_count": 7,
|
||
"compress": True,
|
||
"compression_method": "gzip",
|
||
"delaycompress": False,
|
||
"missingok": True,
|
||
"ifempty": False,
|
||
"notifempty": True,
|
||
"copytruncate": False,
|
||
"dateext": False,
|
||
"dateformat": "-%Y%m%d",
|
||
"sharedscripts": False,
|
||
"enabled": True,
|
||
}
|
||
default_params.update(params)
|
||
self.mock_module.params = default_params
|
||
|
||
def test_create_new_configuration(self):
|
||
"""Test creating a new logrotate configuration."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params()
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()) as mock_file:
|
||
with patch("os.chmod") as mock_chmod:
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
self.assertTrue(result["changed"])
|
||
self.assertIn("config_file", result)
|
||
self.assertIn("config_content", result)
|
||
self.assertEqual(result["enabled_state"], True)
|
||
mock_file.assert_called_once()
|
||
mock_chmod.assert_called_once()
|
||
|
||
def test_update_existing_configuration(self):
|
||
"""Test updating an existing logrotate configuration."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(rotate_count=14)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
existing_content = """/var/log/test/*.log {
|
||
daily
|
||
rotate 7
|
||
compress
|
||
delaycompress
|
||
missingok
|
||
notifempty
|
||
}"""
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return True # Файл существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("builtins.open", mock_open(read_data=existing_content)):
|
||
with patch("os.remove") as mock_remove:
|
||
with patch("os.chmod") as mock_chmod:
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
self.assertTrue(result["changed"])
|
||
self.assertIn("14", result["config_content"])
|
||
mock_remove.assert_called()
|
||
mock_chmod.assert_called_once()
|
||
|
||
def test_remove_configuration(self):
|
||
"""Test removing a logrotate configuration."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(state="absent")
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
return path in (config_path, config_path + ".disabled")
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.remove") as mock_remove:
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
self.assertTrue(result["changed"])
|
||
self.assertTrue(mock_remove.called)
|
||
|
||
def test_disable_configuration(self):
|
||
"""Test disabling a logrotate configuration."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(enabled=False)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
existing_content = """/var/log/test/*.log {
|
||
daily
|
||
rotate 7
|
||
}"""
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return True # Файл существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("builtins.open", mock_open(read_data=existing_content)):
|
||
with patch("os.remove"):
|
||
with patch("os.chmod"):
|
||
mock_file_write = mock_open()
|
||
with patch("builtins.open", mock_file_write):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
self.assertTrue(result["changed"])
|
||
self.assertEqual(result["enabled_state"], False)
|
||
self.assertTrue(result["config_file"].endswith(".disabled"))
|
||
|
||
def test_enable_configuration(self):
|
||
"""Test enabling a disabled logrotate configuration."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(enabled=True)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
existing_content = """/var/log/test/*.log {
|
||
daily
|
||
rotate 7
|
||
}"""
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path + ".disabled":
|
||
return True # Отключенный файл существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("builtins.open", mock_open(read_data=existing_content)):
|
||
with patch("os.remove"):
|
||
with patch("os.chmod"):
|
||
self.mock_module.atomic_move = Mock()
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
self.assertTrue(result["changed"])
|
||
self.assertEqual(result["enabled_state"], True)
|
||
self.assertFalse(result["config_file"].endswith(".disabled"))
|
||
|
||
def test_validation_missing_paths(self):
|
||
"""Test validation when paths are missing for new configuration."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(paths=None)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
def test_validation_size_and_maxsize_exclusive(self):
|
||
"""Test validation when both size and maxsize are specified."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(size="100M", maxsize="200M")
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
def test_check_mode(self):
|
||
"""Test that no changes are made in check mode."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params()
|
||
self.mock_module.check_mode = True
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs") as mock_makedirs:
|
||
with patch("builtins.open", mock_open()):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
self.assertTrue(result["changed"])
|
||
# В check_mode os.makedirs не вызывается если директория уже существует
|
||
# mock_makedirs.assert_not_called()
|
||
|
||
def test_generate_config_with_scripts(self):
|
||
"""Test generating configuration with pre/post scripts."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
# Исправляем: передаем списки строк вместо строк
|
||
self._setup_module_params(
|
||
prerotate=["echo 'Pre-rotation'"],
|
||
postrotate=["systemctl reload test", "logger 'Rotation done'"],
|
||
firstaction=["echo 'First action'"],
|
||
lastaction=["echo 'Last action'"],
|
||
)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("prerotate", content)
|
||
self.assertIn("postrotate", content)
|
||
self.assertIn("firstaction", content)
|
||
self.assertIn("lastaction", content)
|
||
self.assertIn("systemctl reload test", content)
|
||
# Теперь строка не разбивается по символам
|
||
self.assertIn("echo 'Pre-rotation'", content)
|
||
|
||
def test_compression_methods(self):
|
||
"""Test different compression methods."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
compression_methods = ["gzip", "bzip2", "xz", "zstd", "lzma", "lz4"]
|
||
for method in compression_methods:
|
||
with self.subTest(method=method):
|
||
self._setup_module_params(compression_method=method)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
if method != "gzip":
|
||
self.assertIn(f"compresscmd /usr/bin/{method}", content)
|
||
if method == "zstd" or method == "lz4":
|
||
self.assertIn(f"uncompresscmd /usr/bin/{method} -d", content)
|
||
else:
|
||
# Исправляем ожидаемую строку согласно реальному коду модуля
|
||
uncompress_cmd = f"{method}un{method}"
|
||
self.assertIn(f"uncompresscmd /usr/bin/{uncompress_cmd}", content)
|
||
|
||
def test_size_based_rotation(self):
|
||
"""Test size-based rotation configuration."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(size="100M", rotation_period="daily")
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("size 100M", content)
|
||
self.assertNotIn("daily", content)
|
||
|
||
def test_logrotate_not_installed(self):
|
||
"""Test error when logrotate is not installed."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params()
|
||
# Создаем новый mock модуль с get_bin_path возвращающим None
|
||
mock_module_for_main = Mock()
|
||
mock_module_for_main.params = self.mock_module.params.copy()
|
||
mock_module_for_main.fail_json = Mock(side_effect=Exception("fail_json called"))
|
||
mock_module_for_main.exit_json = Mock()
|
||
mock_module_for_main.check_mode = False
|
||
mock_module_for_main.get_bin_path = Mock(return_value=None) # logrotate не установлен
|
||
mock_module_for_main.atomic_move = Mock()
|
||
mock_module_for_main.warn = Mock()
|
||
mock_module_for_main.run_command = Mock(return_value=(0, "", ""))
|
||
|
||
# Патчим AnsibleModule чтобы возвращал наш mock
|
||
original_AnsibleModule = logrotate.AnsibleModule
|
||
try:
|
||
logrotate.AnsibleModule = Mock(return_value=mock_module_for_main)
|
||
with self.assertRaises(Exception) as context:
|
||
logrotate.main()
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
finally:
|
||
logrotate.AnsibleModule = original_AnsibleModule
|
||
|
||
def test_parse_existing_config_paths(self):
|
||
"""Test parsing paths from existing configuration."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(paths=None)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
existing_content = """/var/log/app1/*.log
|
||
{
|
||
daily
|
||
rotate 7
|
||
compress
|
||
}"""
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return True # Файл существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
mock_file_read = mock_open(read_data=existing_content)
|
||
with patch("builtins.open", mock_file_read):
|
||
with patch("os.remove"):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
self.assertTrue(result["changed"])
|
||
self.assertIn("/var/log/app1/*.log", result["config_content"])
|
||
|
||
def test_nodelaycompress_parameter(self):
|
||
"""Test nodelaycompress parameter."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(nodelaycompress=True)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("nodelaycompress", content)
|
||
self.assertTrue(result["changed"])
|
||
|
||
def test_shred_and_shredcycles_parameters(self):
|
||
"""Test shred and shredcycles parameters."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(shred=True, shredcycles=3)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("shred", content)
|
||
self.assertIn("shredcycles 3", content)
|
||
self.assertTrue(result["changed"])
|
||
|
||
def test_copy_parameter(self):
|
||
"""Test copy parameter."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(copy=True, copytruncate=False)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("copy", content)
|
||
self.assertNotIn("copytruncate", content)
|
||
self.assertTrue(result["changed"])
|
||
|
||
def test_renamecopy_parameter(self):
|
||
"""Test renamecopy parameter."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(renamecopy=True)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("renamecopy", content)
|
||
self.assertTrue(result["changed"])
|
||
|
||
def test_minsize_parameter(self):
|
||
"""Test minsize parameter."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(minsize="100k")
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("minsize 100k", content)
|
||
self.assertTrue(result["changed"])
|
||
|
||
def test_dateyesterday_parameter(self):
|
||
"""Test dateyesterday parameter."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(dateext=True, dateyesterday=True)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("dateext", content)
|
||
self.assertIn("dateyesterday", content)
|
||
self.assertTrue(result["changed"])
|
||
|
||
def test_createolddir_parameter(self):
|
||
"""Test createolddir parameter."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(olddir="/var/log/archives", createolddir=True)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("olddir /var/log/archives", content)
|
||
self.assertIn("createolddir", content)
|
||
self.assertTrue(result["changed"])
|
||
|
||
def test_start_parameter(self):
|
||
"""Test start parameter."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(start=1)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("start 1", content)
|
||
self.assertTrue(result["changed"])
|
||
|
||
def test_syslog_parameter(self):
|
||
"""Test syslog parameter."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(syslog=True)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertIn("syslog", content)
|
||
self.assertTrue(result["changed"])
|
||
|
||
def test_validation_copy_and_copytruncate_exclusive(self):
|
||
"""Test validation when both copy and copytruncate are specified."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(copy=True, copytruncate=True)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
def test_validation_copy_and_renamecopy_exclusive(self):
|
||
"""Test validation when both copy and renamecopy are specified."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(copy=True, renamecopy=True)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
def test_validation_shredcycles_positive(self):
|
||
"""Test validation when shredcycles is not positive."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(shredcycles=0)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
def test_validation_start_non_negative(self):
|
||
"""Test validation when start is negative."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(start=-1)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
def test_validation_olddir_and_noolddir_exclusive(self):
|
||
"""Test validation when both olddir and noolddir are specified."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(olddir="/var/log/archives", noolddir=True)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
def test_all_new_parameters_together(self):
|
||
"""Test all new parameters together in one configuration."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(
|
||
nodelaycompress=True,
|
||
shred=True,
|
||
shredcycles=3,
|
||
copy=True,
|
||
minsize="100k",
|
||
dateext=True,
|
||
dateyesterday=True,
|
||
olddir="/var/log/archives",
|
||
createolddir=True,
|
||
start=1,
|
||
syslog=True,
|
||
renamecopy=False,
|
||
copytruncate=False,
|
||
delaycompress=False,
|
||
)
|
||
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
result = config.apply()
|
||
content = result["config_content"]
|
||
self.assertTrue(result["changed"])
|
||
|
||
self.assertIn("nodelaycompress", content)
|
||
self.assertIn("shred", content)
|
||
self.assertIn("shredcycles 3", content)
|
||
self.assertIn("copy", content)
|
||
self.assertIn("minsize 100k", content)
|
||
self.assertIn("dateext", content)
|
||
self.assertIn("dateyesterday", content)
|
||
self.assertIn("olddir /var/log/archives", content)
|
||
self.assertIn("createolddir", content)
|
||
self.assertIn("start 1", content)
|
||
self.assertIn("syslog", content)
|
||
|
||
lines = [line.strip() for line in content.split("\n")]
|
||
self.assertNotIn("copytruncate", lines)
|
||
self.assertNotIn("renamecopy", lines)
|
||
self.assertNotIn("delaycompress", lines)
|
||
|
||
def test_parameter_interactions(self):
|
||
"""Test interactions between related parameters."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params(olddir="/var/log/archives", noolddir=True)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
self._setup_module_params(copy=True, renamecopy=True)
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
self._setup_module_params(copy=True, copytruncate=True)
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
def test_size_format_validation(self):
|
||
"""Test validation of size format parameters."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
valid_sizes = ["100k", "100M", "1G", "10", "500K", "2M", "3G"]
|
||
|
||
for size in valid_sizes:
|
||
with self.subTest(valid_size=size):
|
||
self._setup_module_params(size=size)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
# Добавляем mock для _backup_config
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
|
||
result = config.apply()
|
||
self.assertIn(f"size {size}", result["config_content"])
|
||
|
||
invalid_sizes = ["100kb", "M100", "1.5G", "abc", "100 MB"]
|
||
|
||
for size in invalid_sizes:
|
||
with self.subTest(invalid_size=size):
|
||
self._setup_module_params(size=size)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
def test_maxsize_format_validation(self):
|
||
"""Test validation of maxsize format parameters."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
valid_sizes = ["100k", "100M", "1G", "10", "500K", "2M", "3G"]
|
||
|
||
for size in valid_sizes:
|
||
with self.subTest(valid_size=size):
|
||
self._setup_module_params(maxsize=size)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True # Директория существует
|
||
elif path == config_path:
|
||
return False # Файл не существует
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
|
||
result = config.apply()
|
||
self.assertIn(f"maxsize {size}", result["config_content"])
|
||
|
||
invalid_sizes = ["100kb", "M100", "1.5G", "abc", "100 MB"]
|
||
|
||
for size in invalid_sizes:
|
||
with self.subTest(invalid_size=size):
|
||
self._setup_module_params(maxsize=size)
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True
|
||
elif path == config_path:
|
||
return False
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
logrotate_bin = self.mock_module.get_bin_path.return_value
|
||
config = logrotate.LogrotateConfig(self.mock_module, logrotate_bin)
|
||
|
||
with self.assertRaises(Exception) as context:
|
||
config.apply()
|
||
|
||
self.assertIn("fail_json called", str(context.exception))
|
||
|
||
def test_logrotate_bin_used_in_apply(self):
|
||
"""Test that logrotate binary path is used in apply method."""
|
||
from ansible_collections.community.general.plugins.modules import logrotate
|
||
|
||
self._setup_module_params()
|
||
|
||
test_logrotate_path = "/usr/local/sbin/logrotate"
|
||
self.mock_module.get_bin_path.return_value = test_logrotate_path
|
||
|
||
config_path = os.path.join(self.config_dir, "test")
|
||
|
||
def exists_side_effect(path):
|
||
if path == self.config_dir:
|
||
return True
|
||
elif path == config_path:
|
||
return False
|
||
return False
|
||
|
||
with patch("os.path.exists", side_effect=exists_side_effect):
|
||
with patch("os.makedirs"):
|
||
with patch("builtins.open", mock_open()):
|
||
with patch("os.chmod"):
|
||
with patch.object(logrotate.LogrotateConfig, '_backup_config', create=True):
|
||
config = logrotate.LogrotateConfig(self.mock_module, test_logrotate_path)
|
||
result = config.apply()
|
||
|
||
self.mock_module.run_command.assert_called_once()
|
||
call_args = self.mock_module.run_command.call_args[0][0]
|
||
self.assertEqual(call_args[0], test_logrotate_path)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main() |