import datetime
import os

try:
    from zoneinfo import ZoneInfo, available_timezones
except ImportError:
    from backports.zoneinfo import available_timezones, ZoneInfo

import pytest
from celery import schedules
from django.apps import apps
from django.conf import settings
from django.db.migrations.autodetector import MigrationAutodetector
from django.db.migrations.loader import MigrationLoader
from django.db.migrations.questioner import NonInteractiveMigrationQuestioner
from django.db.migrations.state import ProjectState
from django.test import TestCase, override_settings
from django.utils import timezone

from django_celery_beat import migrations as beat_migrations
from django_celery_beat.models import (DAYS, ClockedSchedule, CrontabSchedule,
                                       IntervalSchedule, PeriodicTasks,
                                       SolarSchedule,
                                       crontab_schedule_celery_timezone)
from t.proj.models import O2OToPeriodicTasks


class MigrationTests(TestCase):
    def test_no_future_duplicate_migration_numbers(self):
        """Verify no duplicate migration numbers.

        Migration files with the same number can cause issues with
        backward migrations, so avoid them.
        """
        path = os.path.dirname(beat_migrations.__file__)
        files = [f[:4] for f in os.listdir(path) if f.endswith('.py')]
        expected_duplicates = [
            (3, '0006'),
        ]
        duplicates_extra = sum(count - 1 for count, _ in expected_duplicates)
        duplicates_numbers = [number for _, number in expected_duplicates]
        self.assertEqual(
            len(files), len(set(files)) + duplicates_extra,
            msg=('Detected migration files with the same migration number'
                 ' (besides {})'.format(' and '.join(duplicates_numbers))))

    def test_models_match_migrations(self):
        """Make sure that no model changes exist.

        This logic is taken from django's makemigrations.py file.
        Here just detect if model changes exist that require
        a migration, and if so we fail.
        """
        app_labels = ['django_celery_beat']
        loader = MigrationLoader(None, ignore_no_migrations=True)
        questioner = NonInteractiveMigrationQuestioner(
            specified_apps=app_labels, dry_run=False)
        autodetector = MigrationAutodetector(
            loader.project_state(),
            ProjectState.from_apps(apps),
            questioner,
        )
        changes = autodetector.changes(
            graph=loader.graph,
            trim_to_apps=app_labels,
            convert_apps=app_labels,
            migration_name='fake_name',
        )
        self.assertTrue(
            not changes,
            msg='Model changes exist that do not have a migration')


class TestDuplicatesMixin:
    def _test_duplicate_schedules(self, cls, kwargs, schedule=None):
        sched1 = cls.objects.create(**kwargs)
        cls.objects.create(**kwargs)
        self.assertEqual(cls.objects.filter(**kwargs).count(), 2)
        # try to create a duplicate schedule from a celery schedule
        if schedule is None:
            schedule = sched1.schedule
        sched3 = cls.from_schedule(schedule)
        # the schedule should be the first of the 2 previous duplicates
        self.assertEqual(sched3, sched1)
        # and the duplicates should not be deleted !
        self.assertEqual(cls.objects.filter(**kwargs).count(), 2)


class CrontabScheduleTestCase(TestCase, TestDuplicatesMixin):
    FIRST_VALID_TIMEZONE = available_timezones().pop()

    def test_default_timezone_without_settings_config(self):
        assert crontab_schedule_celery_timezone() == "UTC"

    @override_settings(CELERY_TIMEZONE=FIRST_VALID_TIMEZONE)
    def test_default_timezone_with_settings_config(self):
        assert crontab_schedule_celery_timezone() == self.FIRST_VALID_TIMEZONE

    def test_duplicate_schedules(self):
        # See: https://github.com/celery/django-celery-beat/issues/322
        kwargs = {
            "minute": "*",
            "hour": "4",
            "day_of_week": "*",
            "day_of_month": "*",
            "month_of_year": "*",
        }
        schedule = schedules.crontab(hour="4")
        self._test_duplicate_schedules(CrontabSchedule, kwargs, schedule)


class SolarScheduleTestCase(TestCase):
    EVENT_CHOICES = SolarSchedule._meta.get_field("event").choices

    def test_celery_solar_schedules_sorted(self):
        assert all(
            self.EVENT_CHOICES[i] <= self.EVENT_CHOICES[i + 1]
            for i in range(len(self.EVENT_CHOICES) - 1)
        ), "SolarSchedule event choices are unsorted"

    def test_celery_solar_schedules_included_as_event_choices(self):
        """Make sure that all Celery solar schedules are included
        in SolarSchedule `event` field choices, keeping synchronized
        Celery solar events with `django-celery-beat` supported solar
        events.

        This test is necessary because Celery solar schedules are
        hardcoded at models so that Django can discover their translations.
        """
        event_choices_values = [value for value, tr in self.EVENT_CHOICES]
        for solar_event in schedules.solar._all_events:
            assert solar_event in event_choices_values

        for event_choice in event_choices_values:
            assert event_choice in schedules.solar._all_events


class IntervalScheduleTestCase(TestCase, TestDuplicatesMixin):

    def test_duplicate_schedules(self):
        kwargs = {'every': 1, 'period': IntervalSchedule.SECONDS}
        schedule = schedules.schedule(run_every=1.0)
        self._test_duplicate_schedules(IntervalSchedule, kwargs, schedule)


class ClockedScheduleTestCase(TestCase, TestDuplicatesMixin):

    def test_duplicate_schedules(self):
        kwargs = {'clocked_time': timezone.now()}
        self._test_duplicate_schedules(ClockedSchedule, kwargs)

    # IMPORTANT: we must have a valid timezone (not UTC) for accurate testing
    @override_settings(TIME_ZONE='Africa/Cairo')
    def test_timezone_format(self):
        """Ensure scheduled time is not shown in UTC when timezone is used"""
        tz_info = datetime.datetime.now(ZoneInfo(settings.TIME_ZONE))
        schedule, created = ClockedSchedule.objects.get_or_create(
            clocked_time=tz_info)
        # testnig str(schedule) calls make_aware() internally
        assert str(schedule.clocked_time) == str(schedule)


@pytest.mark.django_db
class OneToOneRelTestCase(TestCase):
    """
    Make sure that when OneToOne relation Model changed,
    the `PeriodicTasks.last_update` will be update.
    """

    @classmethod
    def setUpTestData(cls):
        super().setUpTestData()
        cls.interval_schedule = IntervalSchedule.objects.create(
            every=10, period=DAYS
        )

    def test_trigger_update_when_saved(self):
        o2o_to_periodic_tasks = O2OToPeriodicTasks.objects.create(
            name='name1',
            task='task1',
            enabled=True,
            interval=self.interval_schedule
        )
        not_changed_dt = PeriodicTasks.last_change()
        o2o_to_periodic_tasks.enabled = True  # Change something on instance.
        o2o_to_periodic_tasks.save()
        has_changed_dt = PeriodicTasks.last_change()
        self.assertTrue(
            not_changed_dt != has_changed_dt,
            'The `PeriodicTasks.last_update` has not be update.'
        )
        # Check the `PeriodicTasks` does be updated.

    def test_trigger_update_when_deleted(self):
        o2o_to_periodic_tasks = O2OToPeriodicTasks.objects.create(
            name='name1',
            task='task1',
            enabled=True,
            interval=self.interval_schedule
        )
        not_changed_dt = PeriodicTasks.last_change()
        o2o_to_periodic_tasks.delete()
        has_changed_dt = PeriodicTasks.last_change()
        self.assertTrue(
            not_changed_dt != has_changed_dt,
            'The `PeriodicTasks.last_update` has not be update.'
        )
        # Check the `PeriodicTasks` does be updated.


class HumanReadableTestCase(TestCase):
    def test_good(self):
        """Valid crontab display."""
        cron = CrontabSchedule.objects.create(
            hour="2",
            minute="0",
            day_of_week="mon",
        )
        self.assertNotEqual(
            cron.human_readable, "0 2 * * mon UTC"
        )

    def test_invalid(self):
        """Invalid crontab display."""
        cron = CrontabSchedule.objects.create(
            hour="2",
            minute="0",
            day_of_week="xxx",
        )
        self.assertEqual(
            cron.human_readable, "0 2 * * xxx UTC"
        )

    def test_long_name(self):
        """Long day name display."""
        for day_day_of_week, expected in (
            ("1", ", only on Monday"),
            ("mon", ", only on Monday"),
            ("Monday,tue", ", only on Monday and Tuesday"),
            ("sat-sun/2", ", only on Saturday"),
            ("mon-wed", ", only on Monday, Tuesday, and Wednesday"),
            ("*", ""),
            ("0-6", ""),
            ("2-1", ""),
            ("mon-sun", ""),
            ("tue-mon", ""),
        ):
            cron = CrontabSchedule.objects.create(
                hour="2",
                minute="0",
                day_of_week=day_day_of_week,
            )

            self.assertEqual(
                cron.human_readable,
                f"At 02:00 AM{expected} UTC",
                day_day_of_week,
            )