test_util.py

# coding: utf-8
import platform
from datetime import date, datetime, timedelta, tzinfo
from functools import partial
from types import ModuleType

import pytest
import pytz
import six
import sys

from apscheduler.job import Job
from apscheduler.util import (
    asint, asbool, astimezone, convert_to_datetime, datetime_to_utc_timestamp,
    utc_timestamp_to_datetime, timedelta_seconds, datetime_ceil, get_callable_name, obj_to_ref,
    ref_to_obj, maybe_ref, check_callable_args, datetime_repr, repr_escape,
    iscoroutinefunction_partial
)
from tests.conftest import minpython, maxpython

try:
    from unittest.mock import Mock
except ImportError:
    from mock import Mock


class DummyClass(object):
    def meth(self):
        pass

    @staticmethod
    def staticmeth():
        pass

    @classmethod
    def classmeth(cls):
        pass

    def __call__(self):
        pass

    class InnerDummyClass(object):
        @classmethod
        def innerclassmeth(cls):
            pass


class InheritedDummyClass(Job):
    pass


class TestAsint(object):
    @pytest.mark.parametrize('value', ['5s', 'shplse'], ids=['digit first', 'text'])
    def test_invalid_value(self, value):
        pytest.raises(ValueError, asint, value)

    def test_number(self):
        assert asint('539') == 539

    def test_none(self):
        assert asint(None) is None


class TestAsbool(object):
    @pytest.mark.parametrize(
        'value',
        [' True', 'true ', 'Yes', ' yes ', '1  ', True],
        ids=['capital true', 'lowercase true', 'capital yes', 'lowercase yes', 'one', 'True'])
    def test_true(self, value):
        assert asbool(value) is True

    @pytest.mark.parametrize(
        'value',
        [' False', 'false ', 'No', ' no ', '0  ', False],
        ids=['capital', 'lowercase false', 'capital no', 'lowercase no', 'zero', 'False'])
    def test_false(self, value):
        assert asbool(value) is False

    def test_bad_value(self):
        pytest.raises(ValueError, asbool, 'yep')


class TestAstimezone(object):
    def test_str(self):
        value = astimezone('Europe/Helsinki')
        assert isinstance(value, tzinfo)

    def test_tz(self):
        tz = pytz.timezone('Europe/Helsinki')
        value = astimezone(tz)
        assert tz is value

    def test_none(self):
        assert astimezone(None) is None

    def test_bad_timezone_type(self):
        exc = pytest.raises(TypeError, astimezone, tzinfo())
        assert 'Only timezones from the pytz library are supported' in str(exc.value)

    def test_bad_local_timezone(self):
        zone = Mock(tzinfo, localize=None, normalize=None, zone='local')
        exc = pytest.raises(ValueError, astimezone, zone)
        assert 'Unable to determine the name of the local timezone' in str(exc.value)

    def test_bad_value(self):
        exc = pytest.raises(TypeError, astimezone, 4)
        assert 'Expected tzinfo, got int instead' in str(exc.value)


class TestConvertToDatetime(object):
    @pytest.mark.parametrize('input,expected', [
        (None, None),
        (date(2009, 8, 1), datetime(2009, 8, 1)),
        (datetime(2009, 8, 1, 5, 6, 12), datetime(2009, 8, 1, 5, 6, 12)),
        ('2009-8-1', datetime(2009, 8, 1)),
        ('2009-8-1 5:16:12', datetime(2009, 8, 1, 5, 16, 12)),
        ('2009-8-1T5:16:12Z', datetime(2009, 8, 1, 5, 16, 12, tzinfo=pytz.utc)),
        ('2009-8-1T5:16:12+02:30',
         pytz.FixedOffset(150).localize(datetime(2009, 8, 1, 5, 16, 12))),
        ('2009-8-1T5:16:12-05:30',
         pytz.FixedOffset(-330).localize(datetime(2009, 8, 1, 5, 16, 12))),
        (pytz.FixedOffset(-60).localize(datetime(2009, 8, 1)),
         pytz.FixedOffset(-60).localize(datetime(2009, 8, 1)))
    ], ids=['None', 'date', 'datetime', 'date as text', 'datetime as text', 'utc', 'tzoffset',
            'negtzoffset', 'existing tzinfo'])
    def test_date(self, timezone, input, expected):
        returned = convert_to_datetime(input, timezone, None)
        if expected is not None:
            assert isinstance(returned, datetime)
            expected = timezone.localize(expected) if not expected.tzinfo else expected

        assert returned == expected

    def test_invalid_input_type(self, timezone):
        exc = pytest.raises(TypeError, convert_to_datetime, 92123, timezone, 'foo')
        assert str(exc.value) == 'Unsupported type for foo: int'

    def test_invalid_input_value(self, timezone):
        exc = pytest.raises(ValueError, convert_to_datetime, '19700-12-1', timezone, None)
        assert str(exc.value) == 'Invalid date string'

    def test_missing_timezone(self):
        exc = pytest.raises(ValueError, convert_to_datetime, '2009-8-1', None, 'argname')
        assert str(exc.value) == ('The "tz" argument must be specified if argname has no timezone '
                                  'information')

    def test_text_timezone(self):
        returned = convert_to_datetime('2009-8-1', 'UTC', None)
        assert returned == datetime(2009, 8, 1, tzinfo=pytz.utc)

    def test_bad_timezone(self):
        exc = pytest.raises(TypeError, convert_to_datetime, '2009-8-1', tzinfo(), None)
        assert str(exc.value) == ('Only pytz timezones are supported (need the localize() and '
                                  'normalize() methods)')


def test_datetime_to_utc_timestamp(timezone):
    dt = timezone.localize(datetime(2014, 3, 12, 5, 40, 13, 254012))
    timestamp = datetime_to_utc_timestamp(dt)
    dt2 = utc_timestamp_to_datetime(timestamp)
    assert dt2 == dt


def test_timedelta_seconds():
    delta = timedelta(minutes=2, seconds=30)
    seconds = timedelta_seconds(delta)
    assert seconds == 150


@pytest.mark.parametrize('input,expected', [
    (datetime(2009, 4, 7, 2, 10, 16, 4000), datetime(2009, 4, 7, 2, 10, 17)),
    (datetime(2009, 4, 7, 2, 10, 16), datetime(2009, 4, 7, 2, 10, 16))
], ids=['milliseconds', 'exact'])
def test_datetime_ceil(input, expected):
    assert datetime_ceil(input) == expected


@pytest.mark.parametrize('input,expected', [
    (None, 'None'),
    (pytz.timezone('Europe/Helsinki').localize(datetime(2014, 5, 30, 7, 12, 20)),
     '2014-05-30 07:12:20 EEST')
], ids=['None', 'datetime+tzinfo'])
def test_datetime_repr(input, expected):
    assert datetime_repr(input) == expected


class TestGetCallableName(object):
    @pytest.mark.parametrize('input,expected', [
        (asint, 'asint'),
        (DummyClass.staticmeth, 'DummyClass.staticmeth' if
         hasattr(DummyClass, '__qualname__') else 'staticmeth'),
        (DummyClass.classmeth, 'DummyClass.classmeth'),
        (DummyClass.meth, 'meth' if sys.version_info[:2] == (3, 2) else 'DummyClass.meth'),
        (DummyClass().meth, 'DummyClass.meth'),
        (DummyClass, 'DummyClass'),
        (DummyClass(), 'DummyClass')
    ], ids=['function', 'static method', 'class method', 'unbounded method', 'bounded method',
            'class', 'instance'])
    def test_inputs(self, input, expected):
        assert get_callable_name(input) == expected

    def test_bad_input(self):
        pytest.raises(TypeError, get_callable_name, object())


class TestObjToRef(object):
    @pytest.mark.parametrize('obj, error', [
        (partial(DummyClass.meth), 'Cannot create a reference to a partial()'),
        (lambda: None, 'Cannot create a reference to a lambda')
    ], ids=['partial', 'lambda'])
    def test_errors(self, obj, error):
        exc = pytest.raises(ValueError, obj_to_ref, obj)
        assert str(exc.value) == error

    @pytest.mark.skipif(sys.version_info[:2] < (3, 3),
                        reason='Requires __qualname__ (Python 3.3+)')
    def test_nested_function_error(self):
        def nested():
            pass

        exc = pytest.raises(ValueError, obj_to_ref, nested)
        assert str(exc.value) == 'Cannot create a reference to a nested function'

    @pytest.mark.parametrize('input,expected', [
        pytest.mark.skipif(sys.version_info[:2] == (3, 2),
                           reason="Unbound methods can't be resolved on Python 3.2")(
            (DummyClass.meth, 'tests.test_util:DummyClass.meth')
        ),
        (DummyClass.classmeth, 'tests.test_util:DummyClass.classmeth'),
        pytest.mark.skipif(sys.version_info < (3, 3),
                           reason="Requires __qualname__ (Python 3.3+)")(
            (DummyClass.InnerDummyClass.innerclassmeth,
             'tests.test_util:DummyClass.InnerDummyClass.innerclassmeth')
        ),
        pytest.mark.skipif(sys.version_info < (3, 3),
                           reason="Requires __qualname__ (Python 3.3+)")(
            (DummyClass.staticmeth, 'tests.test_util:DummyClass.staticmeth')
        ),
        pytest.mark.skipif(sys.version_info >= (3, 2),
                           reason="Unbound methods (Python 3.2) and __qualname__ (Python 3.3+)")(
            (InheritedDummyClass.pause, 'tests.test_util:InheritedDummyClass.pause')
        ),
        (timedelta, 'datetime:timedelta'),
    ], ids=['unbound method', 'class method', 'inner class method', 'static method',
            'inherited class method', 'timedelta'])
    def test_valid_refs(self, input, expected):
        assert obj_to_ref(input) == expected


class TestRefToObj(object):
    def test_valid_ref(self):
        from logging.handlers import RotatingFileHandler
        assert ref_to_obj('logging.handlers:RotatingFileHandler') is RotatingFileHandler

    def test_complex_path(self):
        pkg1 = ModuleType('pkg1')
        pkg1.pkg2 = 'blah'
        pkg2 = ModuleType('pkg1.pkg2')
        pkg2.varname = 'test'
        sys.modules['pkg1'] = pkg1
        sys.modules['pkg1.pkg2'] = pkg2
        assert ref_to_obj('pkg1.pkg2:varname') == 'test'

    @pytest.mark.parametrize('input,error', [
        (object(), TypeError),
        ('module', ValueError),
        ('module:blah', LookupError)
    ], ids=['raw object', 'module', 'module attribute'])
    def test_lookup_error(self, input, error):
        pytest.raises(error, ref_to_obj, input)


@pytest.mark.parametrize('input,expected', [
    ('datetime:timedelta', timedelta),
    (timedelta, timedelta)
], ids=['textref', 'direct'])
def test_maybe_ref(input, expected):
    assert maybe_ref(input) == expected


@pytest.mark.parametrize('input,expected', [
    (b'T\xc3\xa9st'.decode('utf-8'), 'T\\xe9st' if six.PY2 else 'Tést'),
    (1, 1)
], ids=['string', 'int'])
@maxpython(3)
def test_repr_escape_py2(input, expected):
    assert repr_escape(input) == expected


class TestCheckCallableArgs(object):
    def test_invalid_callable_args(self):
        """
        Tests that attempting to create a job with an invalid number of arguments raises an
        exception.

        """
        exc = pytest.raises(ValueError, check_callable_args, lambda x: None, [1, 2], {})
        assert str(exc.value) == (
            'The list of positional arguments is longer than the target callable can handle '
            '(allowed: 1, given in args: 2)')

    def test_invalid_callable_kwargs(self):
        """
        Tests that attempting to schedule a job with unmatched keyword arguments raises an
        exception.

        """
        exc = pytest.raises(ValueError, check_callable_args, lambda x: None, [], {'x': 0, 'y': 1})
        assert str(exc.value) == ('The target callable does not accept the following keyword '
                                  'arguments: y')

    def test_missing_callable_args(self):
        """Tests that attempting to schedule a job with missing arguments raises an exception."""
        exc = pytest.raises(ValueError, check_callable_args, lambda x, y, z: None, [1], {'y': 0})
        assert str(exc.value) == 'The following arguments have not been supplied: z'

    def test_default_args(self):
        """Tests that default values for arguments are properly taken into account."""
        exc = pytest.raises(ValueError, check_callable_args, lambda x, y, z=1: None, [1], {})
        assert str(exc.value) == 'The following arguments have not been supplied: y'

    def test_conflicting_callable_args(self):
        """
        Tests that attempting to schedule a job where the combination of args and kwargs are in
        conflict raises an exception.

        """
        exc = pytest.raises(ValueError, check_callable_args, lambda x, y: None, [1, 2], {'y': 1})
        assert str(exc.value) == 'The following arguments are supplied in both args and kwargs: y'

    def test_signature_positional_only(self):
        """Tests that a function where signature() fails is accepted."""
        check_callable_args(object().__setattr__, ('blah', 1), {})

    @minpython(3, 4)
    @pytest.mark.skipif(platform.python_implementation() == 'PyPy',
                        reason='PyPy does not expose signatures of builtins')
    def test_positional_only_args(self):
        """
        Tests that an attempt to use keyword arguments for positional-only arguments raises an
        exception.

        """
        exc = pytest.raises(ValueError, check_callable_args, object.__setattr__, ['blah'],
                            {'value': 1})
        assert str(exc.value) == ('The following arguments cannot be given as keyword arguments: '
                                  'value')

    @minpython(3)
    def test_unfulfilled_kwargs(self):
        """
        Tests that attempting to schedule a job where not all keyword-only arguments are fulfilled
        raises an exception.

        """
        func = eval("lambda x, *, y, z=1: None")
        exc = pytest.raises(ValueError, check_callable_args, func, [1], {})
        assert str(exc.value) == ('The following keyword-only arguments have not been supplied in '
                                  'kwargs: y')


class TestIsCoroutineFunctionPartial(object):
    @staticmethod
    def not_a_coro(x):
        pass

    @staticmethod
    async def a_coro(x):
        pass

    def test_non_coro(self):
        assert not iscoroutinefunction_partial(self.not_a_coro)

    def test_coro(self):
        assert iscoroutinefunction_partial(self.a_coro)

    def test_coro_partial(self):
        assert iscoroutinefunction_partial(partial(self.a_coro, 1))