mostovik-backend/tests/apps/core/test_management_commands.py

"""Тесты для базового класса management commands."""

from io import StringIO

from apps.core.management.commands.base import BaseAppCommand
from django.core.management.base import CommandError, OutputWrapper
from django.test import TestCase


class TestCommand(BaseAppCommand):
    """Тестовая команда для проверки BaseAppCommand."""

    help = "Test command"

    def execute_command(self, *args, **options):
        if options.get("fail"):
            raise ValueError("Test error")
        return "Success"


class TransactionCommand(BaseAppCommand):
    """Команда для тестов транзакций и dry-run."""

    use_transaction = True

    def execute_command(self, *args, **options):
        from apps.core.models import BackgroundJob

        BackgroundJob.objects.create(
            task_id=options.get("task_id", "tx-task"),
            task_name="test.command",
        )
        if options.get("fail"):
            raise ValueError("tx error")
        return "OK"


class BaseAppCommandTest(TestCase):
    """Тесты для BaseAppCommand."""

    def test_base_command_has_dry_run_argument(self):
        """Проверка наличия аргумента --dry-run."""
        cmd = BaseAppCommand()
        parser = cmd.create_parser("manage.py", "test")
        # Парсер должен принимать --dry-run
        args = parser.parse_args(["--dry-run"])
        self.assertTrue(args.dry_run)

    def test_base_command_has_silent_argument(self):
        """Проверка наличия аргумента --silent."""
        cmd = BaseAppCommand()
        parser = cmd.create_parser("manage.py", "test")
        args = parser.parse_args(["--silent"])
        self.assertTrue(args.silent)

    def test_log_methods_exist(self):
        """Проверка наличия методов логирования."""
        cmd = BaseAppCommand()
        cmd.stdout = StringIO()
        cmd.stderr = StringIO()
        cmd.silent = False
        cmd.verbosity = 2

        # Методы должны существовать и не падать
        cmd.log_info("Test info")
        cmd.log_success("Test success")
        cmd.log_warning("Test warning")
        cmd.log_error("Test error")
        cmd.log_debug("Test debug")

    def test_progress_iter(self):
        """Тест итератора с прогрессом."""
        cmd = BaseAppCommand()
        cmd.stdout = StringIO()
        cmd.silent = True  # Без вывода

        items = list(range(10))
        result = list(cmd.progress_iter(items, "Processing"))

        self.assertEqual(result, items)

    def test_confirm_in_dry_run(self):
        """Тест подтверждения в dry-run режиме."""
        cmd = BaseAppCommand()
        cmd.stdout = StringIO()
        cmd.dry_run = True
        cmd.silent = False

        # В dry-run confirm всегда возвращает True
        result = cmd.confirm("Continue?")
        self.assertTrue(result)

    def test_confirm_with_input_func(self):
        cmd = BaseAppCommand()
        cmd.stdout = OutputWrapper(StringIO())
        cmd.dry_run = False
        cmd.silent = True
        cmd.input_func = lambda: "y"
        self.assertTrue(cmd.confirm("Proceed?"))

    def test_abort_raises_command_error(self):
        """Тест прерывания команды."""
        cmd = BaseAppCommand()

        with self.assertRaises(CommandError):
            cmd.abort("Test abort")

    def test_timed_operation(self):
        """Тест контекстного менеджера для измерения времени."""
        cmd = BaseAppCommand()
        cmd.stdout = StringIO()
        cmd.verbosity = 2

        with cmd.timed_operation("Test operation"):
            pass  # Операция

        # Не должно падать

    def test_handle_success(self):
        cmd = TestCommand()
        cmd.stdout = StringIO()
        cmd.stderr = StringIO()
        result = cmd.handle()
        self.assertEqual(result, "Success")

    def test_handle_raises_command_error(self):
        cmd = TestCommand()
        cmd.stdout = StringIO()
        cmd.stderr = StringIO()
        with self.assertRaises(CommandError):
            cmd.handle(fail=True)

    def test_handle_dry_run_rolls_back(self):
        cmd = TransactionCommand()
        cmd.stdout = StringIO()
        cmd.stderr = StringIO()
        cmd.handle(dry_run=True, silent=True)

        from apps.core.models import BackgroundJob

        self.assertEqual(BackgroundJob.objects.count(), 0)

    def test_handle_transaction_persists(self):
        cmd = TransactionCommand()
        cmd.stdout = StringIO()
        cmd.stderr = StringIO()
        cmd.handle(dry_run=False, silent=True, task_id="persist-task")

        from apps.core.models import BackgroundJob

        self.assertTrue(BackgroundJob.objects.filter(task_id="persist-task").exists())

    def test_progress_iter_without_total(self):
        cmd = BaseAppCommand()
        cmd.stdout = StringIO()
        cmd.silent = True

        def generator():
            for idx in range(3):
                yield idx

        result = list(cmd.progress_iter(generator(), "Iter"))
        self.assertEqual(result, [0, 1, 2])