manyi/venv/Lib/site-packages/celery/tests/utils/test_timeutils.py

from __future__ import absolute_import

import pytz

from datetime import datetime, timedelta, tzinfo
from mock import Mock, patch
from pytz import AmbiguousTimeError

from celery.utils.timeutils import (
    delta_resolution,
    humanize_seconds,
    maybe_iso8601,
    maybe_timedelta,
    timedelta_seconds,
    timezone,
    rate,
    remaining,
    make_aware,
    maybe_make_aware,
    localize,
    LocalTimezone,
    ffwd,
    utcoffset,
)
from celery.utils.iso8601 import parse_iso8601
from celery.tests.case import Case


class test_LocalTimezone(Case):

    def test_daylight(self):
        with patch('celery.utils.timeutils._time') as time:
            time.timezone = 3600
            time.daylight = False
            x = LocalTimezone()
            self.assertEqual(x.STDOFFSET, timedelta(seconds=-3600))
            self.assertEqual(x.DSTOFFSET, x.STDOFFSET)
            time.daylight = True
            time.altzone = 3600
            y = LocalTimezone()
            self.assertEqual(y.STDOFFSET, timedelta(seconds=-3600))
            self.assertEqual(y.DSTOFFSET, timedelta(seconds=-3600))

            self.assertTrue(repr(y))

            y._isdst = Mock()
            y._isdst.return_value = True
            self.assertTrue(y.utcoffset(datetime.now()))
            self.assertFalse(y.dst(datetime.now()))
            y._isdst.return_value = False
            self.assertTrue(y.utcoffset(datetime.now()))
            self.assertFalse(y.dst(datetime.now()))

            self.assertTrue(y.tzname(datetime.now()))


class test_iso8601(Case):

    def test_parse_with_timezone(self):
        d = datetime.utcnow().replace(tzinfo=pytz.utc)
        self.assertEqual(parse_iso8601(d.isoformat()), d)
        # 2013-06-07T20:12:51.775877+00:00
        iso = d.isoformat()
        iso1 = iso.replace('+00:00', '-01:00')
        d1 = parse_iso8601(iso1)
        self.assertEqual(d1.tzinfo._minutes, -60)
        iso2 = iso.replace('+00:00', '+01:00')
        d2 = parse_iso8601(iso2)
        self.assertEqual(d2.tzinfo._minutes, +60)


class test_timeutils(Case):

    def test_delta_resolution(self):
        D = delta_resolution
        dt = datetime(2010, 3, 30, 11, 50, 58, 41065)
        deltamap = ((timedelta(days=2), datetime(2010, 3, 30, 0, 0)),
                    (timedelta(hours=2), datetime(2010, 3, 30, 11, 0)),
                    (timedelta(minutes=2), datetime(2010, 3, 30, 11, 50)),
                    (timedelta(seconds=2), dt))
        for delta, shoulda in deltamap:
            self.assertEqual(D(dt, delta), shoulda)

    def test_timedelta_seconds(self):
        deltamap = ((timedelta(seconds=1), 1),
                    (timedelta(seconds=27), 27),
                    (timedelta(minutes=3), 3 * 60),
                    (timedelta(hours=4), 4 * 60 * 60),
                    (timedelta(days=3), 3 * 86400))
        for delta, seconds in deltamap:
            self.assertEqual(timedelta_seconds(delta), seconds)

    def test_timedelta_seconds_returns_0_on_negative_time(self):
        delta = timedelta(days=-2)
        self.assertEqual(timedelta_seconds(delta), 0)

    def test_humanize_seconds(self):
        t = ((4 * 60 * 60 * 24, '4.00 days'),
             (1 * 60 * 60 * 24, '1.00 day'),
             (4 * 60 * 60, '4.00 hours'),
             (1 * 60 * 60, '1.00 hour'),
             (4 * 60, '4.00 minutes'),
             (1 * 60, '1.00 minute'),
             (4, '4.00 seconds'),
             (1, '1.00 second'),
             (4.3567631221, '4.36 seconds'),
             (0, 'now'))

        for seconds, human in t:
            self.assertEqual(humanize_seconds(seconds), human)

        self.assertEqual(humanize_seconds(4, prefix='about '),
                         'about 4.00 seconds')

    def test_maybe_iso8601_datetime(self):
        now = datetime.now()
        self.assertIs(maybe_iso8601(now), now)

    def test_maybe_timedelta(self):
        D = maybe_timedelta

        for i in (30, 30.6):
            self.assertEqual(D(i), timedelta(seconds=i))

        self.assertEqual(D(timedelta(days=2)), timedelta(days=2))

    def test_remaining_relative(self):
        remaining(datetime.utcnow(), timedelta(hours=1), relative=True)


class test_timezone(Case):

    def test_get_timezone_with_pytz(self):
        self.assertTrue(timezone.get_timezone('UTC'))

    def test_tz_or_local(self):
        self.assertEqual(timezone.tz_or_local(), timezone.local)
        self.assertTrue(timezone.tz_or_local(timezone.utc))

    def test_to_local(self):
        self.assertTrue(
            timezone.to_local(make_aware(datetime.utcnow(), timezone.utc)),
        )
        self.assertTrue(
            timezone.to_local(datetime.utcnow())
        )

    def test_to_local_fallback(self):
        self.assertTrue(
            timezone.to_local_fallback(
                make_aware(datetime.utcnow(), timezone.utc)),
        )
        self.assertTrue(
            timezone.to_local_fallback(datetime.utcnow())
        )


class test_make_aware(Case):

    def test_tz_without_localize(self):
        tz = tzinfo()
        self.assertFalse(hasattr(tz, 'localize'))
        wtz = make_aware(datetime.utcnow(), tz)
        self.assertEqual(wtz.tzinfo, tz)

    def test_when_has_localize(self):

        class tzz(tzinfo):
            raises = False

            def localize(self, dt, is_dst=None):
                self.localized = True
                if self.raises and is_dst is None:
                    self.raised = True
                    raise AmbiguousTimeError()
                return 1  # needed by min() in Python 3 (None not hashable)

        tz = tzz()
        make_aware(datetime.utcnow(), tz)
        self.assertTrue(tz.localized)

        tz2 = tzz()
        tz2.raises = True
        make_aware(datetime.utcnow(), tz2)
        self.assertTrue(tz2.localized)
        self.assertTrue(tz2.raised)

    def test_maybe_make_aware(self):
        aware = datetime.utcnow().replace(tzinfo=timezone.utc)
        self.assertTrue(maybe_make_aware(aware), timezone.utc)
        naive = datetime.utcnow()
        self.assertTrue(maybe_make_aware(naive))


class test_localize(Case):

    def test_tz_without_normalize(self):
        tz = tzinfo()
        self.assertFalse(hasattr(tz, 'normalize'))
        self.assertTrue(localize(make_aware(datetime.utcnow(), tz), tz))

    def test_when_has_normalize(self):

        class tzz(tzinfo):
            raises = None

            def normalize(self, dt, **kwargs):
                self.normalized = True
                if self.raises and kwargs and kwargs.get('is_dst') is None:
                    self.raised = True
                    raise self.raises
                return 1  # needed by min() in Python 3 (None not hashable)

        tz = tzz()
        localize(make_aware(datetime.utcnow(), tz), tz)
        self.assertTrue(tz.normalized)

        tz2 = tzz()
        tz2.raises = AmbiguousTimeError()
        localize(make_aware(datetime.utcnow(), tz2), tz2)
        self.assertTrue(tz2.normalized)
        self.assertTrue(tz2.raised)

        tz3 = tzz()
        tz3.raises = TypeError()
        localize(make_aware(datetime.utcnow(), tz3), tz3)
        self.assertTrue(tz3.normalized)
        self.assertTrue(tz3.raised)


class test_rate_limit_string(Case):

    def test_conversion(self):
        self.assertEqual(rate(999), 999)
        self.assertEqual(rate(7.5), 7.5)
        self.assertEqual(rate('2.5/s'), 2.5)
        self.assertEqual(rate('1456/s'), 1456)
        self.assertEqual(rate('100/m'),
                         100 / 60.0)
        self.assertEqual(rate('10/h'),
                         10 / 60.0 / 60.0)

        for zero in (0, None, '0', '0/m', '0/h', '0/s', '0.0/s'):
            self.assertEqual(rate(zero), 0)


class test_ffwd(Case):

    def test_repr(self):
        x = ffwd(year=2012)
        self.assertTrue(repr(x))

    def test_radd_with_unknown_gives_NotImplemented(self):
        x = ffwd(year=2012)
        self.assertEqual(x.__radd__(object()), NotImplemented)


class test_utcoffset(Case):

    def test_utcoffset(self):
        with patch('celery.utils.timeutils._time') as _time:
            _time.daylight = True
            self.assertIsNotNone(utcoffset())
            _time.daylight = False
            self.assertIsNotNone(utcoffset())