import math
import os
import time
from datetime import datetime, timedelta
from itertools import count
from time import monotonic
from unittest.mock import patch

try:
    from zoneinfo import ZoneInfo  # Python 3.9+
except ImportError:
    from backports.zoneinfo import ZoneInfo  # Python 3.8

import pytest
from celery.schedules import crontab, schedule, solar
from django.contrib.admin.sites import AdminSite
from django.contrib.messages.storage.fallback import FallbackStorage
from django.test import RequestFactory, override_settings
from django.utils import timezone

from django_celery_beat import schedulers
from django_celery_beat.admin import PeriodicTaskAdmin
from django_celery_beat.clockedschedule import clocked
from django_celery_beat.models import (DAYS, ClockedSchedule, CrontabSchedule,
                                       IntervalSchedule, PeriodicTask,
                                       PeriodicTasks, SolarSchedule)
from django_celery_beat.utils import NEVER_CHECK_TIMEOUT, make_aware

_ids = count(0)


@pytest.fixture(autouse=True)
def no_multiprocessing_finalizers(patching):
    patching('multiprocessing.util.Finalize')
    patching('django_celery_beat.schedulers.Finalize')


class EntryTrackSave(schedulers.ModelEntry):

    def __init__(self, *args, **kwargs):
        self.saved = 0
        super().__init__(*args, **kwargs)

    def save(self):
        self.saved += 1
        super().save()


class EntrySaveRaises(schedulers.ModelEntry):

    def save(self):
        raise RuntimeError('this is expected')


class TrackingScheduler(schedulers.DatabaseScheduler):
    Entry = EntryTrackSave

    def __init__(self, *args, **kwargs):
        self.flushed = 0
        schedulers.DatabaseScheduler.__init__(self, *args, **kwargs)

    def sync(self):
        self.flushed += 1
        schedulers.DatabaseScheduler.sync(self)


@pytest.mark.django_db
class SchedulerCase:

    def create_model_interval(self, schedule, **kwargs):
        interval = IntervalSchedule.from_schedule(schedule)
        interval.save()
        return self.create_model(interval=interval, **kwargs)

    def create_model_crontab(self, schedule, **kwargs):
        crontab = CrontabSchedule.from_schedule(schedule)
        crontab.save()
        return self.create_model(crontab=crontab, **kwargs)

    def create_model_solar(self, schedule, **kwargs):
        solar = SolarSchedule.from_schedule(schedule)
        solar.save()
        return self.create_model(solar=solar, **kwargs)

    def create_model_clocked(self, schedule, **kwargs):
        clocked = ClockedSchedule.from_schedule(schedule)
        clocked.save()
        return self.create_model(clocked=clocked, one_off=True, **kwargs)

    def create_conf_entry(self):
        name = f'thefoo{next(_ids)}'
        return name, dict(
            task=f'djcelery.unittest.add{next(_ids)}',
            schedule=timedelta(0, 600),
            args=(),
            relative=False,
            kwargs={},
            options={'queue': 'extra_queue'}
        )

    def create_model(self, Model=PeriodicTask, **kwargs):
        entry = dict(
            name=f'thefoo{next(_ids)}',
            task=f'djcelery.unittest.add{next(_ids)}',
            args='[2, 2]',
            kwargs='{"callback": "foo"}',
            queue='xaz',
            routing_key='cpu',
            priority=1,
            headers='{"_schema_name": "foobar"}',
            exchange='foo',
        )
        return Model(**dict(entry, **kwargs))

    def create_interval_schedule(self):
        return IntervalSchedule.objects.create(every=10, period=DAYS)

    def create_crontab_schedule(self):
        return CrontabSchedule.objects.create()


@pytest.mark.django_db
class test_ModelEntry(SchedulerCase):
    Entry = EntryTrackSave

    def test_entry(self):
        m = self.create_model_interval(schedule(timedelta(seconds=10)))
        e = self.Entry(m, app=self.app)

        assert e.args == [2, 2]
        assert e.kwargs == {'callback': 'foo'}
        assert e.schedule
        assert e.total_run_count == 0
        assert isinstance(e.last_run_at, datetime)
        assert e.options['queue'] == 'xaz'
        assert e.options['exchange'] == 'foo'
        assert e.options['routing_key'] == 'cpu'
        assert e.options['priority'] == 1
        assert e.options['headers']['_schema_name'] == 'foobar'
        assert e.options['headers']['periodic_task_name'] == m.name

        right_now = self.app.now()
        m2 = self.create_model_interval(
            schedule(timedelta(seconds=10)),
            last_run_at=right_now,
        )

        assert m2.last_run_at
        e2 = self.Entry(m2, app=self.app)
        assert e2.last_run_at is right_now

        e3 = e2.next()
        assert e3.last_run_at > e2.last_run_at
        assert e3.total_run_count == 1

    @override_settings(
        USE_TZ=False,
        DJANGO_CELERY_BEAT_TZ_AWARE=False
    )
    @pytest.mark.usefixtures('depends_on_current_app')
    @timezone.override('Europe/Berlin')
    @pytest.mark.celery(timezone='Europe/Berlin')
    def test_entry_is_due__no_use_tz(self):
        old_tz = os.environ.get("TZ")
        os.environ["TZ"] = "Europe/Berlin"
        if hasattr(time, "tzset"):
            time.tzset()
        assert self.app.timezone.key == 'Europe/Berlin'

        # simulate last_run_at from DB - not TZ aware but localtime
        right_now = datetime.utcnow()

        m = self.create_model_crontab(
            crontab(minute='*/10'),
            last_run_at=right_now,
        )
        e = self.Entry(m, app=self.app)

        assert e.is_due().is_due is False
        assert e.is_due().next <= 600  # 10 minutes; see above
        if old_tz is not None:
            os.environ["TZ"] = old_tz
        else:
            del os.environ["TZ"]
        if hasattr(time, "tzset"):
            time.tzset()

    @override_settings(
        USE_TZ=False,
        DJANGO_CELERY_BEAT_TZ_AWARE=False
    )
    @pytest.mark.usefixtures('depends_on_current_app')
    @timezone.override('Europe/Berlin')
    @pytest.mark.celery(timezone='Europe/Berlin')
    def test_entry_and_model_last_run_at_with_utc_no_use_tz(self, monkeypatch):
        old_tz = os.environ.get("TZ")
        os.environ["TZ"] = "Europe/Berlin"
        if hasattr(time, "tzset"):
            time.tzset()
        assert self.app.timezone.key == 'Europe/Berlin'
        # simulate last_run_at from DB - not TZ aware but localtime
        right_now = datetime.utcnow()
        # make sure to use fixed date time
        monkeypatch.setattr(self.Entry, '_default_now', lambda o: right_now)
        m = self.create_model_crontab(
            crontab(minute='*/10')
        )
        m.save()
        e = self.Entry(m, app=self.app)
        e.save()
        m.refresh_from_db()

        assert m.last_run_at == e.last_run_at

        if old_tz is not None:
            os.environ["TZ"] = old_tz
        else:
            del os.environ["TZ"]
        if hasattr(time, "tzset"):
            time.tzset()

    @override_settings(
        USE_TZ=False,
        DJANGO_CELERY_BEAT_TZ_AWARE=False
    )
    @pytest.mark.usefixtures('depends_on_current_app')
    @timezone.override('Europe/Berlin')
    @pytest.mark.celery(timezone='Europe/Berlin')
    def test_entry_and_model_last_run_at_when_model_changed(self, monkeypatch):
        old_tz = os.environ.get("TZ")
        os.environ["TZ"] = "Europe/Berlin"
        if hasattr(time, "tzset"):
            time.tzset()
        assert self.app.timezone.key == 'Europe/Berlin'
        # simulate last_run_at from DB - not TZ aware but localtime
        right_now = datetime.utcnow()
        # make sure to use fixed date time
        monkeypatch.setattr(self.Entry, '_default_now', lambda o: right_now)
        m = self.create_model_crontab(
            crontab(minute='*/10')
        )
        m.save()
        e = self.Entry(m, app=self.app)
        e.save()
        m.refresh_from_db()

        # The just created model has no value for date_changed
        # so last_run_at should be set to the Entry._default_now()
        assert m.last_run_at == e.last_run_at

        # If the model has been updated and the entry.last_run_at is None,
        # entry.last_run_at should be set to the value of model.date_changed.
        # see #717
        m2 = self.create_model_crontab(
            crontab(minute='*/10')
        )
        m2.save()
        m2.refresh_from_db()
        assert m2.date_changed is not None
        e2 = self.Entry(m2, app=self.app)
        e2.save()
        assert m2.last_run_at == m2.date_changed

        if old_tz is not None:
            os.environ["TZ"] = old_tz
        else:
            del os.environ["TZ"]
        if hasattr(time, "tzset"):
            time.tzset()

    @override_settings(
        USE_TZ=False,
        DJANGO_CELERY_BEAT_TZ_AWARE=False,
        TIME_ZONE="Europe/Berlin",
        CELERY_TIMEZONE="America/New_York"
    )
    @pytest.mark.usefixtures('depends_on_current_app')
    @timezone.override('Europe/Berlin')
    @pytest.mark.celery(timezone='America/New_York')
    def test_entry_is_due__celery_timezone_doesnt_match_time_zone(self):
        old_tz = os.environ.get("TZ")
        os.environ["TZ"] = "Europe/Berlin"
        if hasattr(time, "tzset"):
            time.tzset()
        assert self.app.timezone.key == 'America/New_York'

        # simulate last_run_at all none, doing the same thing that
        # _default_now() would do
        right_now = datetime.utcnow()

        m = self.create_model_crontab(
            crontab(minute='*/10'),
            last_run_at=right_now,
        )
        e = self.Entry(m, app=self.app)

        assert e.is_due().is_due is False
        assert e.is_due().next <= 600  # 10 minutes; see above
        if old_tz is not None:
            os.environ["TZ"] = old_tz
        else:
            del os.environ["TZ"]
        if hasattr(time, "tzset"):
            time.tzset()

    def test_task_with_start_time(self):
        interval = 10
        right_now = self.app.now()
        one_interval_ago = right_now - timedelta(seconds=interval)
        m = self.create_model_interval(
            schedule(timedelta(seconds=interval)),
            start_time=right_now,
            last_run_at=one_interval_ago
        )
        e = self.Entry(m, app=self.app)
        isdue, delay = e.is_due()
        assert isdue
        assert delay == interval

        tomorrow = right_now + timedelta(days=1)
        m2 = self.create_model_interval(
            schedule(timedelta(seconds=interval)),
            start_time=tomorrow,
            last_run_at=one_interval_ago
        )
        e2 = self.Entry(m2, app=self.app)
        isdue, delay = e2.is_due()
        assert not isdue
        assert delay == math.ceil(
            (tomorrow - right_now).total_seconds()
        )

    def test_one_off_task(self):
        interval = 10
        right_now = self.app.now()
        one_interval_ago = right_now - timedelta(seconds=interval)
        m = self.create_model_interval(
            schedule(timedelta(seconds=interval)),
            one_off=True,
            last_run_at=one_interval_ago,
            total_run_count=0
        )
        e = self.Entry(m, app=self.app)
        isdue, delay = e.is_due()
        assert isdue
        assert delay == interval

        m2 = self.create_model_interval(
            schedule(timedelta(seconds=interval)),
            one_off=True,
            last_run_at=one_interval_ago,
            total_run_count=1
        )
        e2 = self.Entry(m2, app=self.app)
        isdue, delay = e2.is_due()
        assert not isdue
        assert delay == NEVER_CHECK_TIMEOUT

    def test_task_with_expires(self):
        interval = 10
        right_now = self.app.now()
        one_second_later = right_now + timedelta(seconds=1)
        m = self.create_model_interval(
            schedule(timedelta(seconds=interval)),
            start_time=right_now,
            expires=one_second_later
        )
        e = self.Entry(m, app=self.app)
        isdue, delay = e.is_due()
        assert isdue
        assert delay == interval

        m2 = self.create_model_interval(
            schedule(timedelta(seconds=interval)),
            start_time=right_now,
            expires=right_now
        )
        e2 = self.Entry(m2, app=self.app)
        isdue, delay = e2.is_due()
        assert not isdue
        assert delay == NEVER_CHECK_TIMEOUT

        one_second_ago = right_now - timedelta(seconds=1)
        m2 = self.create_model_interval(
            schedule(timedelta(seconds=interval)),
            start_time=right_now,
            expires=one_second_ago
        )
        e2 = self.Entry(m2, app=self.app)
        isdue, delay = e2.is_due()
        assert not isdue
        assert delay == NEVER_CHECK_TIMEOUT


@pytest.mark.django_db
class test_DatabaseSchedulerFromAppConf(SchedulerCase):
    Scheduler = TrackingScheduler

    @pytest.mark.django_db
    @pytest.fixture(autouse=True)
    def setup_scheduler(self, app):
        self.app = app

        self.entry_name, entry = self.create_conf_entry()
        self.app.conf.beat_schedule = {self.entry_name: entry}
        self.m1 = PeriodicTask(name=self.entry_name,
                               interval=self.create_interval_schedule())

    def test_constructor(self):
        s = self.Scheduler(app=self.app)

        assert isinstance(s._dirty, set)
        assert s._last_sync is None
        assert s.sync_every

    def test_periodic_task_model_enabled_schedule(self):
        s = self.Scheduler(app=self.app)
        sched = s.schedule
        assert len(sched) == 2
        assert 'celery.backend_cleanup' in sched
        assert self.entry_name in sched
        for n, e in sched.items():
            assert isinstance(e, s.Entry)
            if n == 'celery.backend_cleanup':
                assert e.options['expires'] == 12 * 3600
                assert e.model.expires is None
                assert e.model.expire_seconds == 12 * 3600

    def test_periodic_task_model_disabled_schedule(self):
        self.m1.enabled = False
        self.m1.save()

        s = self.Scheduler(app=self.app)
        sched = s.schedule
        assert sched
        assert len(sched) == 1
        assert 'celery.backend_cleanup' in sched
        assert self.entry_name not in sched

    def test_periodic_task_model_schedule_type_change(self):
        self.m1.interval = None
        self.m1.crontab = self.create_crontab_schedule()
        self.m1.save()

        self.Scheduler(app=self.app)
        self.m1.refresh_from_db()
        assert self.m1.interval
        assert self.m1.crontab is None


@pytest.mark.django_db
class test_DatabaseScheduler(SchedulerCase):
    Scheduler = TrackingScheduler

    @pytest.mark.django_db
    @pytest.fixture(autouse=True)
    def setup_scheduler(self, app):
        self.app = app
        self.app.conf.beat_schedule = {}

        self.m1 = self.create_model_interval(
            schedule(timedelta(seconds=10)))
        self.m1.save()
        self.m1.refresh_from_db()

        self.m2 = self.create_model_interval(
            schedule(timedelta(minutes=20)))
        self.m2.save()
        self.m2.refresh_from_db()

        self.m3 = self.create_model_crontab(
            crontab(minute='2,4,5'))
        self.m3.save()
        self.m3.refresh_from_db()

        self.m4 = self.create_model_solar(
            solar('solar_noon', 48.06, 12.86))
        self.m4.save()
        self.m4.refresh_from_db()

        # disabled, should not be in schedule
        self.m5 = self.create_model_interval(
            schedule(timedelta(seconds=1)))
        self.m5.enabled = False
        self.m5.save()

        # near future time (should be in schedule)
        now = datetime.now()
        two_minutes_later = now + timedelta(minutes=2)
        dt_aware = make_aware(
            datetime(
                day=two_minutes_later.day,
                month=two_minutes_later.month,
                year=two_minutes_later.year,
                hour=two_minutes_later.hour,
                minute=two_minutes_later.minute
            )
        )
        self.m6 = self.create_model_clocked(
            clocked(dt_aware)
        )
        self.m6.save()
        self.m6.refresh_from_db()

        # distant future time (should not be in schedule)
        ten_minutes_later = now + timedelta(minutes=10)
        distant_dt_aware = make_aware(
            datetime(
                day=ten_minutes_later.day,
                month=ten_minutes_later.month,
                year=ten_minutes_later.year,
                hour=ten_minutes_later.hour,
                minute=ten_minutes_later.minute
            )
        )
        self.m7 = self.create_model_clocked(
            clocked(distant_dt_aware)
        )
        self.m7.save()
        self.m7.refresh_from_db()

        now_hour = timezone.localtime(timezone.now()).hour
        # near future time (should be in schedule)
        self.m8 = self.create_model_crontab(
            crontab(hour=str(now_hour)))
        self.m8.save()
        self.m8.refresh_from_db()
        self.m9 = self.create_model_crontab(
            crontab(hour=str((now_hour + 1) % 24)))
        self.m9.save()
        self.m9.refresh_from_db()
        self.m10 = self.create_model_crontab(
            crontab(hour=str((now_hour - 1) % 24)))
        self.m10.save()
        self.m10.refresh_from_db()

        # distant future time (should not be in schedule)
        self.m11 = self.create_model_crontab(
            crontab(hour=str((now_hour + 3) % 24)))
        self.m11.save()
        self.m11.refresh_from_db()

        self.s = self.Scheduler(app=self.app)

    def test_constructor(self):
        assert isinstance(self.s._dirty, set)
        assert self.s._last_sync is None
        assert self.s.sync_every

    def test_all_as_schedule(self):
        sched = self.s.schedule
        assert sched

        # Check for presence of standard tasks
        expected_task_names = {
            self.m1.name,  # interval task
            self.m2.name,  # interval task
            self.m3.name,  # crontab task
            self.m4.name,  # solar task
            self.m6.name,  # clocked task (near future)
            self.m8.name,  # crontab task (current hour)
            self.m9.name,  # crontab task (current hour + 1)
            self.m10.name,  # crontab task (current hour - 1)
            'celery.backend_cleanup'  # auto-added by system
        }

        # The distant future crontab task (hour + 3) should be excluded
        distant_task_name = self.m11.name

        # But if it's hour is 4 (or converts to 4 after timezone adjustment),
        # it would be included because of the special handling for hour=4
        current_hour = timezone.localtime(timezone.now()).hour
        is_hour_four_task = False

        # Check if the task would have hour 4.
        if self.m11.crontab.hour == '4' or (current_hour + 3) % 24 == 4:
            is_hour_four_task = True

        # Add to expected tasks if it's an hour=4 task
        if is_hour_four_task:
            expected_task_names.add(distant_task_name)

        # Verify all expected tasks are present in the schedule
        schedule_task_names = set(sched.keys())
        assert schedule_task_names == expected_task_names, (
            f"Task mismatch. Expected: {expected_task_names}, "
            f"Got: {schedule_task_names}"
        )

        # Verify all entries are the right type
        for n, e in sched.items():
            assert isinstance(e, self.s.Entry)

    def test_schedule_changed(self):
        self.m2.args = '[16, 16]'
        self.m2.save()
        e2 = self.s.schedule[self.m2.name]
        assert e2.args == [16, 16]

        self.m1.args = '[32, 32]'
        self.m1.save()
        e1 = self.s.schedule[self.m1.name]
        assert e1.args == [32, 32]
        e1 = self.s.schedule[self.m1.name]
        assert e1.args == [32, 32]

        self.m3.delete()
        with pytest.raises(KeyError):
            self.s.schedule.__getitem__(self.m3.name)

    def test_should_sync(self):
        assert self.s.should_sync()
        self.s._last_sync = monotonic()
        assert not self.s.should_sync()
        self.s._last_sync -= self.s.sync_every
        assert self.s.should_sync()

    def test_reserve(self):
        e1 = self.s.schedule[self.m1.name]
        self.s.schedule[self.m1.name] = self.s.reserve(e1)
        assert self.s.flushed == 1

        e2 = self.s.schedule[self.m2.name]
        self.s.schedule[self.m2.name] = self.s.reserve(e2)
        assert self.s.flushed == 1
        assert self.m2.name in self.s._dirty

    def test_sync_not_saves_last_run_at_while_schedule_changed(self):
        # Update e1 last_run_at and add to dirty
        e1 = self.s.schedule[self.m2.name]
        time.sleep(3)
        e1.model.last_run_at = e1._default_now()
        self.s._dirty.add(e1.model.name)

        # Record e1 pre sync last_run_at
        e1_pre_sync_last_run_at = e1.model.last_run_at

        # Set schedule_changed() == True
        self.s._last_timestamp = monotonic()
        # Do sync
        self.s.sync()

        # Record e1 post sync last_run_at
        e1_m = PeriodicTask.objects.get(pk=e1.model.pk)
        e1_post_sync_last_run_at = e1_m.last_run_at

        # Check
        assert e1_pre_sync_last_run_at == e1_post_sync_last_run_at

    def test_sync_saves_last_run_at(self):
        e1 = self.s.schedule[self.m2.name]
        last_run = e1.last_run_at
        last_run2 = last_run - timedelta(days=1)
        e1.model.last_run_at = last_run2
        self.s._dirty.add(self.m2.name)
        self.s.sync()

        e2 = self.s.schedule[self.m2.name]
        assert e2.last_run_at == last_run2

    def test_sync_syncs_before_save(self):
        # Get the entry for m2
        e1 = self.s.schedule[self.m2.name]

        # Increment the entry (but make sure it doesn't sync)
        self.s._last_sync = monotonic()
        e2 = self.s.schedule[e1.name] = self.s.reserve(e1)
        assert self.s.flushed == 1

        # Fetch the raw object from db, change the args
        # and save the changes.
        m2 = PeriodicTask.objects.get(pk=self.m2.pk)
        m2.args = '[16, 16]'
        m2.save()

        # get_schedule should now see the schedule has changed.
        # and also sync the dirty objects.
        e3 = self.s.schedule[self.m2.name]
        assert self.s.flushed == 2
        assert e3.last_run_at == e2.last_run_at
        assert e3.args == [16, 16]

    def test_periodic_task_disabled_and_enabled(self):
        # Get the entry for m2
        e1 = self.s.schedule[self.m2.name]

        # Increment the entry (but make sure it doesn't sync)
        self.s._last_sync = monotonic()
        self.s.schedule[e1.name] = self.s.reserve(e1)
        assert self.s.flushed == 1

        # Fetch the raw object from db, change the args
        # and save the changes.
        m2 = PeriodicTask.objects.get(pk=self.m2.pk)
        m2.enabled = False
        m2.save()

        # get_schedule should now see the schedule has changed.
        # and remove entry for m2
        assert self.m2.name not in self.s.schedule
        assert self.s.flushed == 2

        m2.enabled = True
        m2.save()

        # get_schedule should now see the schedule has changed.
        # and add entry for m2
        assert self.m2.name in self.s.schedule
        assert self.s.flushed == 3

    def test_periodic_task_disabled_while_reserved(self):
        # Get the entry for m2
        e1 = self.s.schedule[self.m2.name]

        # Increment the entry (but make sure it doesn't sync)
        self.s._last_sync = monotonic()
        e2 = self.s.schedule[e1.name] = self.s.reserve(e1)
        assert self.s.flushed == 1

        # Fetch the raw object from db, change the args
        # and save the changes.
        m2 = PeriodicTask.objects.get(pk=self.m2.pk)
        m2.enabled = False
        m2.save()

        # reserve is called because the task gets called from
        # tick after the database change is made
        self.s.reserve(e2)

        # get_schedule should now see the schedule has changed.
        # and remove entry for m2
        assert self.m2.name not in self.s.schedule
        assert self.s.flushed == 2

    def test_sync_not_dirty(self):
        self.s._dirty.clear()
        self.s.sync()

    def test_sync_object_gone(self):
        self.s._dirty.add('does-not-exist')
        self.s.sync()

    def test_sync_rollback_on_save_error(self):
        self.s.schedule[self.m1.name] = EntrySaveRaises(self.m1, app=self.app)
        self.s._dirty.add(self.m1.name)
        with pytest.raises(RuntimeError):
            self.s.sync()

    def test_update_scheduler_heap_invalidation(self, monkeypatch):
        # mock "schedule_changed" to always trigger update for
        # all calls to schedule, as a change may occur at any moment
        monkeypatch.setattr(self.s, 'schedule_changed', lambda: True)
        self.s.tick()

    def test_heap_size_is_constant(self, monkeypatch):
        # heap size is constant unless the schedule changes
        monkeypatch.setattr(self.s, 'schedule_changed', lambda: True)
        expected_heap_size = len(self.s.schedule.values())
        self.s.tick()
        assert len(self.s._heap) == expected_heap_size
        self.s.tick()
        assert len(self.s._heap) == expected_heap_size

    def test_scheduler_schedules_equality_on_change(self, monkeypatch):
        monkeypatch.setattr(self.s, 'schedule_changed', lambda: False)
        assert self.s.schedules_equal(self.s.schedule, self.s.schedule)

        monkeypatch.setattr(self.s, 'schedule_changed', lambda: True)
        assert not self.s.schedules_equal(
            self.s.schedule, self.s.schedule
        )

    def test_heap_always_return_the_first_item(self):
        interval = 10

        s1 = schedule(timedelta(seconds=interval))
        m1 = self.create_model_interval(s1, enabled=False)
        m1.last_run_at = self.app.now() - timedelta(seconds=interval + 2)
        m1.save()
        m1.refresh_from_db()

        s2 = schedule(timedelta(seconds=interval))
        m2 = self.create_model_interval(s2, enabled=True)
        m2.last_run_at = self.app.now() - timedelta(seconds=interval + 1)
        m2.save()
        m2.refresh_from_db()

        e1 = EntryTrackSave(m1, self.app)
        # because the disabled task e1 runs first, e2 will never be executed
        e2 = EntryTrackSave(m2, self.app)

        s = self.Scheduler(app=self.app)
        s.schedule.clear()
        s.schedule[e1.name] = e1
        s.schedule[e2.name] = e2

        tried = set()
        for _ in range(len(s.schedule) * 8):
            tick_interval = s.tick()
            if tick_interval and tick_interval > 0.0:
                tried.add(s._heap[0].entry.name)
                time.sleep(tick_interval)
                if s.should_sync():
                    s.sync()
        assert len(tried) == 1 and tried == {e1.name}

    def test_starttime_trigger(self, monkeypatch):
        # Ensure there is no heap block in case of new task with start_time
        PeriodicTask.objects.all().delete()
        s = self.Scheduler(app=self.app)
        assert not s._heap
        m1 = self.create_model_interval(schedule(timedelta(seconds=3)))
        m1.save()
        s.tick()
        assert len(s._heap) == 2
        m2 = self.create_model_interval(
            schedule(timedelta(days=1)),
            start_time=make_aware(
                datetime.now() + timedelta(seconds=2)))
        m2.save()
        s.tick()
        assert s._heap[0][2].name == m2.name
        assert len(s._heap) == 3
        assert s._heap[0]
        time.sleep(2)
        s.tick()
        assert s._heap[0]
        assert s._heap[0][2].name == m1.name

    def test_crontab_with_start_time_between_now_and_crontab(self, app):
        now = app.now()
        delay_minutes = 2

        test_start_time = now + timedelta(minutes=delay_minutes)

        crontab_time = test_start_time + timedelta(minutes=delay_minutes)

        task = self.create_model_crontab(
            crontab(minute=f'{crontab_time.minute}'),
            start_time=test_start_time)

        entry = EntryTrackSave(task, app=app)

        is_due, next_check = entry.is_due()

        expected_delay = 2 * delay_minutes * 60

        assert not is_due
        assert next_check == pytest.approx(expected_delay, abs=60)

    def test_crontab_with_start_time_after_crontab(self, app):
        now = app.now()

        delay_minutes = 2

        crontab_time = now + timedelta(minutes=delay_minutes)

        test_start_time = crontab_time + timedelta(minutes=delay_minutes)

        task = self.create_model_crontab(
            crontab(minute=f'{crontab_time.minute}'),
            start_time=test_start_time)

        entry = EntryTrackSave(task, app=app)

        is_due, next_check = entry.is_due()

        expected_delay = delay_minutes * 60 + 3600

        assert not is_due
        assert next_check == pytest.approx(expected_delay, abs=60)

    def test_crontab_with_start_time_different_time_zone(self, app):
        now = app.now()

        delay_minutes = 2

        test_start_time = now + timedelta(minutes=delay_minutes)

        crontab_time = test_start_time + timedelta(minutes=delay_minutes)

        tz = ZoneInfo('Asia/Shanghai')
        test_start_time = test_start_time.astimezone(tz)

        task = self.create_model_crontab(
            crontab(minute=f'{crontab_time.minute}'),
            start_time=test_start_time)

        entry = EntryTrackSave(task, app=app)

        is_due, next_check = entry.is_due()

        expected_delay = 2 * delay_minutes * 60

        assert not is_due
        assert next_check == pytest.approx(expected_delay, abs=60)

        now = app.now()

        crontab_time = now + timedelta(minutes=delay_minutes)

        test_start_time = crontab_time + timedelta(minutes=delay_minutes)

        tz = ZoneInfo('Asia/Shanghai')
        test_start_time = test_start_time.astimezone(tz)

        task = self.create_model_crontab(
            crontab(minute=f'{crontab_time.minute}'),
            start_time=test_start_time)

        entry = EntryTrackSave(task, app=app)

        is_due, next_check = entry.is_due()

        expected_delay = delay_minutes * 60 + 3600

        assert not is_due
        assert next_check == pytest.approx(expected_delay, abs=60)

    def test_crontab_with_start_time_tick(self, app):
        PeriodicTask.objects.all().delete()
        s = self.Scheduler(app=self.app)
        assert not s._heap

        m1 = self.create_model_interval(schedule(timedelta(seconds=3)))
        m1.save()

        now = timezone.now()
        start_time = now + timedelta(minutes=1)
        crontab_trigger_time = now + timedelta(minutes=2)

        m2 = self.create_model_crontab(
            crontab(minute=f'{crontab_trigger_time.minute}'),
            start_time=start_time)
        m2.save()

        e2 = EntryTrackSave(m2, app=self.app)
        is_due, _ = e2.is_due()

        max_iterations = 1000
        iterations = 0
        while (not is_due and iterations < max_iterations):
            s.tick()
            assert s._heap[0][2].name != m2.name
            is_due, _ = e2.is_due()

    @pytest.mark.django_db
    def test_crontab_exclusion_logic_basic(self):
        current_hour = datetime.now().hour
        cron_current_hour = CrontabSchedule.objects.create(
            hour=str(current_hour)
        )
        cron_plus_one = CrontabSchedule.objects.create(
            hour=str((current_hour + 1) % 24)
        )
        cron_plus_two = CrontabSchedule.objects.create(
            hour=str((current_hour + 2) % 24)
        )
        cron_minus_one = CrontabSchedule.objects.create(
            hour=str((current_hour - 1) % 24)
        )
        cron_minus_two = CrontabSchedule.objects.create(
            hour=str((current_hour - 2) % 24)
        )
        cron_outside = CrontabSchedule.objects.create(
            hour=str((current_hour + 5) % 24)
        )

        # Create a special hour 4 schedule that should always be included
        cron_hour_four = CrontabSchedule.objects.create(hour='4')

        # Create periodic tasks using these schedules
        task_current = self.create_model(
            name='task-current',
            crontab=cron_current_hour
        )
        task_current.save()

        task_plus_one = self.create_model(
            name='task-plus-one',
            crontab=cron_plus_one
        )
        task_plus_one.save()

        task_plus_two = self.create_model(
            name='task-plus-two',
            crontab=cron_plus_two
        )
        task_plus_two.save()

        task_minus_one = self.create_model(
            name='task-minus-one',
            crontab=cron_minus_one
        )
        task_minus_one.save()

        task_minus_two = self.create_model(
            name='task-minus-two',
            crontab=cron_minus_two
        )
        task_minus_two.save()

        task_outside = self.create_model(
            name='task-outside',
            crontab=cron_outside
        )
        task_outside.save()

        task_hour_four = self.create_model(
            name='task-hour-four',
            crontab=cron_hour_four
        )
        task_hour_four.save()

        # Run the scheduler's exclusion logic
        exclude_query = self.s._get_crontab_exclude_query()

        # Get excluded task IDs
        excluded_tasks = set(
            PeriodicTask.objects.filter(exclude_query).values_list(
                'id', flat=True
            )
        )

        # The test that matters is that hour 4 is always included
        assert task_hour_four.id not in excluded_tasks

        # Assert that tasks within the time window are not excluded
        for task_id in [task_current.id, task_plus_one.id, task_plus_two.id,
                        task_minus_one.id, task_minus_two.id]:
            assert task_id not in excluded_tasks

        if task_outside.crontab.hour != '4':
            assert task_outside.id in excluded_tasks

    @pytest.mark.django_db
    def test_crontab_special_hour_four(self):
        """
        Test that schedules with hour=4 are always included, regardless of
        the current time.
        """
        # Create a task with hour=4
        cron_hour_four = CrontabSchedule.objects.create(hour='4')
        task_hour_four = self.create_model(
            name='special-cleanup-task',
            crontab=cron_hour_four
        )

        # Run the scheduler's exclusion logic
        exclude_query = self.s._get_crontab_exclude_query()

        # Get excluded task IDs
        excluded_tasks = set(
            PeriodicTask.objects.filter(exclude_query).values_list(
                'id', flat=True
            )
        )

        # The hour=4 task should never be excluded
        assert task_hour_four.id not in excluded_tasks

    @pytest.mark.django_db
    @patch('django_celery_beat.schedulers.aware_now')
    @patch('django.utils.timezone.get_current_timezone')
    def test_crontab_timezone_conversion(self, mock_get_tz, mock_aware_now):
        # Set up mocks for server timezone and current time
        from datetime import datetime
        server_tz = ZoneInfo("Asia/Tokyo")

        mock_get_tz.return_value = server_tz

        # Server time is 17:00 Tokyo time
        mock_now_dt = datetime(2023, 1, 1, 17, 0, 0, tzinfo=server_tz)
        mock_aware_now.return_value = mock_now_dt

        # Create tasks with the following crontab schedules:
        # 1. UTC task at hour 8 - equivalent to 17:00 Tokyo time (current hour)
        #    - should be included
        # 2. New York task at hour 3 - equivalent to 17:00 Tokyo time
        #    (current hour) - should be included
        # 3. UTC task at hour 3 - equivalent to 12:00 Tokyo time
        #    - should be excluded (outside window)

        # Create crontab schedules in different timezones
        utc_current_hour = CrontabSchedule.objects.create(
            hour='8', timezone='UTC'
        )
        ny_current_hour = CrontabSchedule.objects.create(
            hour='3', timezone='America/New_York'
        )
        utc_outside_window = CrontabSchedule.objects.create(
            hour='3', timezone='UTC'
        )

        # Create periodic tasks using these schedules
        task_utc_current = self.create_model(
            name='utc-current-hour',
            crontab=utc_current_hour
        )
        task_utc_current.save()

        task_ny_current = self.create_model(
            name='ny-current-hour',
            crontab=ny_current_hour
        )
        task_ny_current.save()

        task_utc_outside = self.create_model(
            name='utc-outside-window',
            crontab=utc_outside_window
        )
        task_utc_outside.save()

        # Run the scheduler's exclusion logic
        exclude_query = self.s._get_crontab_exclude_query()

        # Get excluded task IDs
        excluded_tasks = set(
            PeriodicTask.objects.filter(exclude_query).values_list(
                'id', flat=True
            )
        )

        # Current hour tasks in different timezones should not be excluded
        assert task_utc_current.id not in excluded_tasks, (
            "UTC current hour task should be included"
        )
        assert task_ny_current.id not in excluded_tasks, (
            "New York current hour task should be included"
        )

        # Task outside window should be excluded
        assert task_utc_outside.id in excluded_tasks, (
            "UTC outside window task should be excluded"
        )

    @pytest.mark.django_db
    @patch('django.utils.timezone.get_current_timezone')
    @patch('django_celery_beat.schedulers.aware_now')
    def test_crontab_timezone_conversion_with_negative_offset_and_dst(
        self, mock_aware_now, mock_get_tz
    ):
        # Set up mocks for server timezone and current time
        from datetime import datetime

        server_tz = ZoneInfo("UTC")

        mock_get_tz.return_value = server_tz

        # Server time is 17:00 UTC in June
        mock_now_dt = datetime(2023, 6, 1, 17, 0, 0, tzinfo=server_tz)
        mock_aware_now.return_value = mock_now_dt

        # Create tasks with the following crontab schedules:
        # 1. Asia/Tokyo task at hour 2 - equivalent to 17:00 UTC (current hour)
        #    - should be included
        # 2. Europe/Paris task at hour 19 - equivalent to 17:00 UTC
        #    (current hour) - should be included
        # 3. Europe/Paris task at hour 15 - equivalent to 13:00 UTC
        #    - should be excluded (outside window)

        # Create crontab schedules in different timezones
        tokyo_current_hour = CrontabSchedule.objects.create(
            hour='2', timezone='Asia/Tokyo'
        )
        paris_current_hour = CrontabSchedule.objects.create(
            hour='19', timezone='Europe/Paris'
        )
        paris_outside_window = CrontabSchedule.objects.create(
            hour='14', timezone='Europe/Paris'
        )

        # Create periodic tasks using these schedules
        task_utc_current = self.create_model(
            name='tokyo-current-hour',
            crontab=tokyo_current_hour
        )
        task_utc_current.save()

        task_ny_current = self.create_model(
            name='paros-current-hour',
            crontab=paris_current_hour
        )
        task_ny_current.save()

        task_utc_outside = self.create_model(
            name='paris-outside-window',
            crontab=paris_outside_window
        )
        task_utc_outside.save()

        # Run the scheduler's exclusion logic
        exclude_query = self.s._get_crontab_exclude_query()

        # Get excluded task IDs
        excluded_tasks = set(
            PeriodicTask.objects.filter(exclude_query).values_list(
                'id', flat=True
            )
        )

        # Current hour tasks in different timezones should not be excluded
        assert task_utc_current.id not in excluded_tasks, (
            "Tokyo current hour task should be included"
        )
        assert task_ny_current.id not in excluded_tasks, (
            "Paris current hour task should be included"
        )

        # Task outside window should be excluded
        assert task_utc_outside.id in excluded_tasks, (
            "Paris outside window task should be excluded"
        )


@pytest.mark.django_db
class test_models(SchedulerCase):

    def test_IntervalSchedule_unicode(self):
        assert (str(IntervalSchedule(every=1, period='seconds'))
                == 'every second')
        assert (str(IntervalSchedule(every=10, period='seconds'))
                == 'every 10 seconds')

    def test_CrontabSchedule_unicode(self):
        assert str(CrontabSchedule(
            minute=3,
            hour=3,
            day_of_week=None,
        )) == '3 3 * * * (m/h/dM/MY/d) UTC'
        assert str(CrontabSchedule(
            minute=3,
            hour=3,
            day_of_week='tue',
            day_of_month='*/2',
            month_of_year='4,6',
        )) == '3 3 */2 4,6 tue (m/h/dM/MY/d) UTC'

    def test_PeriodicTask_unicode_interval(self):
        p = self.create_model_interval(schedule(timedelta(seconds=10)))
        assert str(p) == f'{p.name}: every 10.0 seconds'

    def test_PeriodicTask_unicode_crontab(self):
        p = self.create_model_crontab(crontab(
            hour='4, 5',
            day_of_week='4, 5',
        ))
        assert str(p) == """{}: * 4,5 * * 4,5 (m/h/dM/MY/d) UTC""".format(
            p.name
        )

    def test_PeriodicTask_unicode_solar(self):
        p = self.create_model_solar(
            solar('solar_noon', 48.06, 12.86), name='solar_event'
        )
        assert str(p) == 'solar_event: {} ({}, {})'.format(
            'Solar noon', '48.06', '12.86'
        )

    def test_PeriodicTask_unicode_clocked(self):
        time = make_aware(datetime.now())
        p = self.create_model_clocked(
            clocked(time), name='clocked_event'
        )
        assert str(p) == '{}: {}'.format(
            'clocked_event', str(time)
        )

    def test_PeriodicTask_schedule_property(self):
        p1 = self.create_model_interval(schedule(timedelta(seconds=10)))
        s1 = p1.schedule
        assert s1.run_every.total_seconds() == 10

        p2 = self.create_model_crontab(crontab(
            hour='4, 5',
            minute='10,20,30',
            day_of_month='1-7',
            month_of_year='*/3',
        ))
        s2 = p2.schedule
        assert s2.hour == {4, 5}
        assert s2.minute == {10, 20, 30}
        assert s2.day_of_week == {0, 1, 2, 3, 4, 5, 6}
        assert s2.day_of_month == {1, 2, 3, 4, 5, 6, 7}
        assert s2.month_of_year == {1, 4, 7, 10}

    def test_PeriodicTask_unicode_no_schedule(self):
        p = self.create_model()
        assert str(p) == f'{p.name}: {{no schedule}}'

    def test_CrontabSchedule_schedule(self):
        s = CrontabSchedule(
            minute='3, 7',
            hour='3, 4',
            day_of_week='*',
            day_of_month='1, 16',
            month_of_year='1, 7',
        )
        assert s.schedule.minute == {3, 7}
        assert s.schedule.hour == {3, 4}
        assert s.schedule.day_of_week == {0, 1, 2, 3, 4, 5, 6}
        assert s.schedule.day_of_month == {1, 16}
        assert s.schedule.month_of_year == {1, 7}

    def test_CrontabSchedule_long_schedule(self):
        s = CrontabSchedule(
            minute=str(list(range(60)))[1:-1],
            hour=str(list(range(24)))[1:-1],
            day_of_week=str(list(range(7)))[1:-1],
            day_of_month=str(list(range(1, 32)))[1:-1],
            month_of_year=str(list(range(1, 13)))[1:-1]
        )
        assert s.schedule.minute == set(range(60))
        assert s.schedule.hour == set(range(24))
        assert s.schedule.day_of_week == set(range(7))
        assert s.schedule.day_of_month == set(range(1, 32))
        assert s.schedule.month_of_year == set(range(1, 13))
        fields = [
            'minute', 'hour', 'day_of_week', 'day_of_month', 'month_of_year'
        ]
        for field in fields:
            str_length = len(str(getattr(s.schedule, field)))
            field_length = s._meta.get_field(field).max_length
            assert str_length <= field_length

    def test_SolarSchedule_schedule(self):
        s = SolarSchedule(event='solar_noon', latitude=48.06, longitude=12.86)
        dt = datetime(day=26, month=7, year=2050, hour=1, minute=0)
        dt_lastrun = make_aware(dt)

        assert s.schedule is not None
        isdue, nextcheck = s.schedule.is_due(dt_lastrun)
        assert isdue is False  # False means task isn't due, but keep checking.
        assert (nextcheck > 0) and (isdue is False) or \
               (nextcheck == s.max_interval) and (isdue is True)

        s2 = SolarSchedule(event='solar_noon', latitude=48.06, longitude=12.86)
        dt2 = datetime(day=26, month=7, year=2000, hour=1, minute=0)
        dt2_lastrun = make_aware(dt2)

        assert s2.schedule is not None
        isdue2, nextcheck2 = s2.schedule.is_due(dt2_lastrun)
        assert isdue2 is True  # True means task is due and should run.
        assert (nextcheck2 > 0) and (isdue2 is True) or \
               (nextcheck2 == s2.max_interval) and (isdue2 is False)

    def test_ClockedSchedule_schedule(self):
        due_datetime = make_aware(datetime(
            day=26,
            month=7,
            year=3000,
            hour=1,
            minute=0
        ))  # future time
        s = ClockedSchedule(clocked_time=due_datetime)
        dt = datetime(day=25, month=7, year=2050, hour=1, minute=0)
        dt_lastrun = make_aware(dt)

        assert s.schedule is not None
        isdue, nextcheck = s.schedule.is_due(dt_lastrun)
        # False means task isn't due, but keep checking.
        assert isdue is False
        assert (nextcheck > 0) and (isdue is False) or \
               (nextcheck == s.max_interval) and (isdue is True)

        due_datetime = make_aware(datetime.now())
        s = ClockedSchedule(clocked_time=due_datetime)
        dt2_lastrun = make_aware(datetime.now())

        assert s.schedule is not None
        isdue2, nextcheck2 = s.schedule.is_due(dt2_lastrun)
        assert isdue2 is True  # True means task is due and should run.
        assert (nextcheck2 == NEVER_CHECK_TIMEOUT) and (isdue2 is True)


@pytest.mark.django_db
class test_model_PeriodicTasks(SchedulerCase):

    def test_track_changes(self):
        assert PeriodicTasks.last_change() is None
        m1 = self.create_model_interval(schedule(timedelta(seconds=10)))
        m1.save()
        x = PeriodicTasks.last_change()
        assert x
        m1.args = '(23, 24)'
        m1.save()
        y = PeriodicTasks.last_change()
        assert y
        assert y > x


@pytest.mark.django_db
class test_modeladmin_PeriodicTaskAdmin(SchedulerCase):
    @pytest.mark.django_db
    @pytest.fixture(autouse=True)
    def setup_scheduler(self, app):
        self.app = app
        self.site = AdminSite()
        self.request_factory = RequestFactory()

        interval_schedule = self.create_interval_schedule()

        entry_name, entry = self.create_conf_entry()
        self.app.conf.beat_schedule = {entry_name: entry}
        self.m1 = PeriodicTask(name=entry_name, interval=interval_schedule)
        self.m1.task = 'celery.backend_cleanup'
        self.m1.save()

        entry_name, entry = self.create_conf_entry()
        self.app.conf.beat_schedule = {entry_name: entry}
        self.m2 = PeriodicTask(name=entry_name, interval=interval_schedule)
        self.m2.task = 'celery.backend_cleanup'
        self.m2.save()

    def patch_request(self, request):
        """patch request to allow for django messages storage"""
        setattr(request, 'session', 'session')
        messages = FallbackStorage(request)
        setattr(request, '_messages', messages)
        return request

    # don't hang if broker is down
    # https://github.com/celery/celery/issues/4627
    @pytest.mark.timeout(5)
    def test_run_task(self):
        ma = PeriodicTaskAdmin(PeriodicTask, self.site)
        self.request = self.patch_request(self.request_factory.get('/'))
        ma.run_tasks(self.request, PeriodicTask.objects.filter(id=self.m1.id))
        assert len(self.request._messages._queued_messages) == 1
        queued_message = self.request._messages._queued_messages[0].message
        assert queued_message == '1 task was successfully run'

    # don't hang if broker is down
    # https://github.com/celery/celery/issues/4627
    @pytest.mark.timeout(5)
    def test_run_tasks(self):
        ma = PeriodicTaskAdmin(PeriodicTask, self.site)
        self.request = self.patch_request(self.request_factory.get('/'))
        ma.run_tasks(self.request, PeriodicTask.objects.all())
        assert len(self.request._messages._queued_messages) == 1
        queued_message = self.request._messages._queued_messages[0].message
        assert queued_message == '2 tasks were successfully run'

    @pytest.mark.timeout(5)
    def test_run_task_headers(self, monkeypatch):
        def mock_apply_async(*args, **kwargs):
            self.captured_headers = kwargs.get('headers', {})

        monkeypatch.setattr('celery.app.task.Task.apply_async',
                            mock_apply_async)
        ma = PeriodicTaskAdmin(PeriodicTask, self.site)
        self.request = self.patch_request(self.request_factory.get('/'))
        ma.run_tasks(self.request, PeriodicTask.objects.filter(id=self.m1.id))
        assert 'periodic_task_name' in self.captured_headers
        assert self.captured_headers['periodic_task_name'] == self.m1.name


@pytest.mark.django_db
class test_timezone_offset_handling:
    def setup_method(self):
        self.app = patch("django_celery_beat.schedulers.current_app").start()

    def teardown_method(self):
        patch.stopall()

    @patch("django_celery_beat.schedulers.aware_now")
    def test_server_timezone_handling_with_zoneinfo(self, mock_aware_now):
        """Test handling when server timezone is already a ZoneInfo instance."""

        # Create a mock scheduler with only the methods we need to test
        class MockScheduler:
            _get_timezone_offset = schedulers.DatabaseScheduler._get_timezone_offset

        s = MockScheduler()

        tokyo_tz = ZoneInfo("Asia/Tokyo")
        mock_now = datetime(2023, 1, 1, 12, 0, 0, tzinfo=tokyo_tz)
        mock_aware_now.return_value = mock_now

        # Test with a different timezone
        new_york_tz = "America/New_York"
        offset = s._get_timezone_offset(new_york_tz)  # Pass self explicitly

        # Tokyo is UTC+9, New York is UTC-5, so difference should be 14 hours
        assert offset == 14
        assert mock_aware_now.called

    @patch("django_celery_beat.schedulers.aware_now")
    def test_timezone_offset_with_zoneinfo_object_param(self, mock_aware_now):
        """Test handling when timezone_name parameter is a ZoneInfo object."""

        class MockScheduler:
            _get_timezone_offset = schedulers.DatabaseScheduler._get_timezone_offset

        s = MockScheduler()

        tokyo_tz = ZoneInfo("Asia/Tokyo")
        mock_now = datetime(2023, 1, 1, 12, 0, 0, tzinfo=tokyo_tz)
        mock_aware_now.return_value = mock_now

        # Test with a ZoneInfo object as parameter
        new_york_tz = ZoneInfo("America/New_York")
        offset = s._get_timezone_offset(new_york_tz)  # Pass self explicitly

        # Tokyo is UTC+9, New York is UTC-5, so difference should be 14 hours
        assert offset == 14