import os
from io import StringIO
from unittest import mock

from admin_scripts.tests import AdminScriptTestCase

from django.apps import apps
from django.core import management
from django.core.management import BaseCommand, CommandError, find_commands
from django.core.management.utils import find_command, popen_wrapper
from django.db import connection
from django.test import SimpleTestCase, override_settings
from django.test.utils import captured_stderr, extend_sys_path
from django.utils import translation

from .management.commands import dance


# A minimal set of apps to avoid system checks running on all apps.
@override_settings(
    INSTALLED_APPS=[
        'django.contrib.auth',
        'django.contrib.contenttypes',
        'user_commands',
    ],
)
class CommandTests(SimpleTestCase):
    def test_command(self):
        out = StringIO()
        management.call_command('dance', stdout=out)
        self.assertIn("I don't feel like dancing Rock'n'Roll.\n", out.getvalue())

    def test_command_style(self):
        out = StringIO()
        management.call_command('dance', style='Jive', stdout=out)
        self.assertIn("I don't feel like dancing Jive.\n", out.getvalue())
        # Passing options as arguments also works (thanks argparse)
        management.call_command('dance', '--style', 'Jive', stdout=out)
        self.assertIn("I don't feel like dancing Jive.\n", out.getvalue())

    def test_language_preserved(self):
        out = StringIO()
        with translation.override('fr'):
            management.call_command('dance', stdout=out)
            self.assertEqual(translation.get_language(), 'fr')

    def test_explode(self):
        """ An unknown command raises CommandError """
        with self.assertRaisesMessage(CommandError, "Unknown command: 'explode'"):
            management.call_command(('explode',))

    def test_system_exit(self):
        """ Exception raised in a command should raise CommandError with
            call_command, but SystemExit when run from command line
        """
        with self.assertRaises(CommandError):
            management.call_command('dance', example="raise")
        dance.Command.requires_system_checks = False
        try:
            with captured_stderr() as stderr, self.assertRaises(SystemExit):
                management.ManagementUtility(['manage.py', 'dance', '--example=raise']).execute()
        finally:
            dance.Command.requires_system_checks = True
        self.assertIn("CommandError", stderr.getvalue())

    def test_no_translations_deactivate_translations(self):
        """
        When the Command handle method is decorated with @no_translations,
        translations are deactivated inside the command.
        """
        current_locale = translation.get_language()
        with translation.override('pl'):
            result = management.call_command('no_translations', stdout=StringIO())
            self.assertIsNone(result)
        self.assertEqual(translation.get_language(), current_locale)

    def test_find_command_without_PATH(self):
        """
        find_command should still work when the PATH environment variable
        doesn't exist (#22256).
        """
        current_path = os.environ.pop('PATH', None)

        try:
            self.assertIsNone(find_command('_missing_'))
        finally:
            if current_path is not None:
                os.environ['PATH'] = current_path

    def test_discover_commands_in_eggs(self):
        """
        Management commands can also be loaded from Python eggs.
        """
        egg_dir = '%s/eggs' % os.path.dirname(__file__)
        egg_name = '%s/basic.egg' % egg_dir
        with extend_sys_path(egg_name):
            with self.settings(INSTALLED_APPS=['commandegg']):
                cmds = find_commands(os.path.join(apps.get_app_config('commandegg').path, 'management'))
        self.assertEqual(cmds, ['eggcommand'])

    def test_call_command_option_parsing(self):
        """
        When passing the long option name to call_command, the available option
        key is the option dest name (#22985).
        """
        out = StringIO()
        management.call_command('dance', stdout=out, opt_3=True)
        self.assertIn("option3", out.getvalue())
        self.assertNotIn("opt_3", out.getvalue())
        self.assertNotIn("opt-3", out.getvalue())

    def test_call_command_option_parsing_non_string_arg(self):
        """
        It should be possible to pass non-string arguments to call_command.
        """
        out = StringIO()
        management.call_command('dance', 1, verbosity=0, stdout=out)
        self.assertIn("You passed 1 as a positional argument.", out.getvalue())

    def test_calling_a_command_with_only_empty_parameter_should_ends_gracefully(self):
        out = StringIO()
        management.call_command('hal', "--empty", stdout=out)
        self.assertIn("Dave, I can't do that.\n", out.getvalue())

    def test_calling_command_with_app_labels_and_parameters_should_be_ok(self):
        out = StringIO()
        management.call_command('hal', 'myapp', "--verbosity", "3", stdout=out)
        self.assertIn("Dave, my mind is going. I can feel it. I can feel it.\n", out.getvalue())

    def test_calling_command_with_parameters_and_app_labels_at_the_end_should_be_ok(self):
        out = StringIO()
        management.call_command('hal', "--verbosity", "3", "myapp", stdout=out)
        self.assertIn("Dave, my mind is going. I can feel it. I can feel it.\n", out.getvalue())

    def test_calling_a_command_with_no_app_labels_and_parameters_should_raise_a_command_error(self):
        with self.assertRaises(CommandError):
            management.call_command('hal', stdout=StringIO())

    def test_output_transaction(self):
        output = management.call_command('transaction', stdout=StringIO(), no_color=True)
        self.assertTrue(output.strip().startswith(connection.ops.start_transaction_sql()))
        self.assertTrue(output.strip().endswith(connection.ops.end_transaction_sql()))

    def test_call_command_no_checks(self):
        """
        By default, call_command should not trigger the check framework, unless
        specifically asked.
        """
        self.counter = 0

        def patched_check(self_, **kwargs):
            self.counter += 1

        saved_check = BaseCommand.check
        BaseCommand.check = patched_check
        try:
            management.call_command("dance", verbosity=0)
            self.assertEqual(self.counter, 0)
            management.call_command("dance", verbosity=0, skip_checks=False)
            self.assertEqual(self.counter, 1)
        finally:
            BaseCommand.check = saved_check

    def test_check_migrations(self):
        requires_migrations_checks = dance.Command.requires_migrations_checks
        self.assertIs(requires_migrations_checks, False)
        try:
            with mock.patch.object(BaseCommand, 'check_migrations') as check_migrations:
                management.call_command('dance', verbosity=0)
                self.assertFalse(check_migrations.called)
                dance.Command.requires_migrations_checks = True
                management.call_command('dance', verbosity=0)
                self.assertTrue(check_migrations.called)
        finally:
            dance.Command.requires_migrations_checks = requires_migrations_checks

    def test_call_command_unrecognized_option(self):
        msg = (
            'Unknown option(s) for dance command: unrecognized. Valid options '
            'are: example, help, integer, no_color, opt_3, option3, '
            'pythonpath, settings, skip_checks, stderr, stdout, style, '
            'traceback, verbosity, version.'
        )
        with self.assertRaisesMessage(TypeError, msg):
            management.call_command('dance', unrecognized=1)

        msg = (
            'Unknown option(s) for dance command: unrecognized, unrecognized2. '
            'Valid options are: example, help, integer, no_color, opt_3, '
            'option3, pythonpath, settings, skip_checks, stderr, stdout, '
            'style, traceback, verbosity, version.'
        )
        with self.assertRaisesMessage(TypeError, msg):
            management.call_command('dance', unrecognized=1, unrecognized2=1)

    def test_call_command_with_required_parameters_in_options(self):
        out = StringIO()
        management.call_command('required_option', need_me='foo', needme2='bar', stdout=out)
        self.assertIn('need_me', out.getvalue())
        self.assertIn('needme2', out.getvalue())

    def test_call_command_with_required_parameters_in_mixed_options(self):
        out = StringIO()
        management.call_command('required_option', '--need-me=foo', needme2='bar', stdout=out)
        self.assertIn('need_me', out.getvalue())
        self.assertIn('needme2', out.getvalue())

    def test_subparser(self):
        out = StringIO()
        management.call_command('subparser', 'foo', 12, stdout=out)
        self.assertIn('bar', out.getvalue())

    def test_subparser_invalid_option(self):
        msg = "Error: invalid choice: 'test' (choose from 'foo')"
        with self.assertRaisesMessage(CommandError, msg):
            management.call_command('subparser', 'test', 12)


class CommandRunTests(AdminScriptTestCase):
    """
    Tests that need to run by simulating the command line, not by call_command.
    """
    def tearDown(self):
        self.remove_settings('settings.py')

    def test_script_prefix_set_in_commands(self):
        self.write_settings('settings.py', apps=['user_commands'], sdict={
            'ROOT_URLCONF': '"user_commands.urls"',
            'FORCE_SCRIPT_NAME': '"/PREFIX/"',
        })
        out, err = self.run_manage(['reverse_url'])
        self.assertNoOutput(err)
        self.assertEqual(out.strip(), '/PREFIX/some/url/')

    def test_disallowed_abbreviated_options(self):
        """
        To avoid conflicts with custom options, commands don't allow
        abbreviated forms of the --setting and --pythonpath options.
        """
        self.write_settings('settings.py', apps=['user_commands'])
        out, err = self.run_manage(['set_option', '--set', 'foo'])
        self.assertNoOutput(err)
        self.assertEqual(out.strip(), 'Set foo')


class UtilsTests(SimpleTestCase):

    def test_no_existent_external_program(self):
        msg = 'Error executing a_42_command_that_doesnt_exist_42'
        with self.assertRaisesMessage(CommandError, msg):
            popen_wrapper(['a_42_command_that_doesnt_exist_42'])