path: root/tests/test_klass.py

# Copyright: 2006-2007 Brian Harring <ferringb@gmail.com>
# License: BSD/GPL2

from functools import partial
import math
import re
from time import time

from snakeoil import klass
from snakeoil.test import TestCase, mk_cpy_loadable_testcase, not_a_test


class Test_native_GetAttrProxy(TestCase):
    kls = staticmethod(klass.native_GetAttrProxy)

    def test_it(self):
        class foo1(object):
            def __init__(self, obj):
                self.obj = obj
            __getattr__ = self.kls('obj')

        class foo2(object):
            pass

        o2 = foo2()
        o = foo1(o2)
        self.assertRaises(AttributeError, getattr, o, "blah")
        self.assertEqual(o.obj, o2)
        o2.foon = "dar"
        self.assertEqual(o.foon, "dar")
        o.foon = "foo"
        self.assertEqual(o.foon, 'foo')

    def test_attrlist(self):
        def make_class(attr_list=None):
            class foo(object, metaclass=self.kls):
                if attr_list is not None:
                    locals()['__attr_comparison__'] = attr_list

        self.assertRaises(TypeError, make_class)
        self.assertRaises(TypeError, make_class, ['foon'])
        self.assertRaises(TypeError, make_class, [None])

    def test_instancemethod(self):
        class foo(object):
            bar = "baz"

        class Test(object):
            method = self.kls('test')
            test = foo()

        test = Test()
        self.assertEqual(test.method('bar'), foo.bar)


class Test_CPY_GetAttrProxy(Test_native_GetAttrProxy):

    kls = staticmethod(klass.GetAttrProxy)
    if klass.GetAttrProxy is klass.native_GetAttrProxy:
        skip = "cpython extension isn't available"

    def test_sane_recursion_bail(self):
        # people are stupid; if protection isn't in place, we wind up blowing
        # the c stack, which doesn't result in a friendly Exception being
        # thrown.
        # results in a segfault.. so if it's horked, this will bail the test
        # runner.

        class c(object):
            __getattr__ = self.kls("obj")

        o = c()
        o.obj = o
        # now it's cyclical.
        self.assertRaises((AttributeError, RuntimeError), getattr, o, "hooey")


class TestDirProxy(TestCase):

    @staticmethod
    def noninternal_attrs(obj):
        return sorted(x for x in dir(obj) if not re.match(r'__\w+__', x))

    def test_combined(self):
        class foo1(object):
            def __init__(self, obj):
                self.obj = obj
            __dir__ = klass.DirProxy('obj')

        class foo2(object):
            def __init__(self):
                self.attr = 'foo'

        o2 = foo2()
        o = foo1(o2)
        self.assertEqual(self.noninternal_attrs(o), ['attr', 'obj'])

    def test_empty(self):
        class foo1(object):
            def __init__(self, obj):
                self.obj = obj
            __dir__ = klass.DirProxy('obj')

        class foo2(object):
            pass

        o2 = foo2()
        o = foo1(o2)
        self.assertEqual(self.noninternal_attrs(o2), [])
        self.assertEqual(self.noninternal_attrs(o), ['obj'])


class Test_native_contains(TestCase):
    func = staticmethod(klass.native_contains)

    def test_it(self):
        class c(dict):
            __contains__ = self.func
        d = c({"1": 2})
        self.assertIn("1", d)
        self.assertNotIn(1, d)


class Test_CPY_contains(Test_native_contains):
    func = staticmethod(klass.contains)

    if klass.contains is klass.native_contains:
        skip = "cpython extension isn't available"


class Test_native_get(TestCase):
    func = staticmethod(klass.native_get)

    def test_it(self):
        class c(dict):
            get = self.func
        d = c({"1": 2})
        self.assertEqual(d.get("1"), 2)
        self.assertEqual(d.get("1", 3), 2)
        self.assertEqual(d.get(1), None)
        self.assertEqual(d.get(1, 3), 3)

class Test_CPY_get(Test_native_get):
    func = staticmethod(klass.get)

    if klass.get is klass.native_get:
        skip = "cpython extension isn't available"


class Test_chained_getter(TestCase):

    kls = klass.chained_getter

    def test_hash(self):
        self.assertEqual(hash(self.kls("foon")), hash("foon"))
        self.assertEqual(hash(self.kls("foon.dar")), hash("foon.dar"))

    def test_caching(self):
        l = [id(self.kls("fa2341f%s" % x)) for x in "abcdefghij"]
        self.assertEqual(id(self.kls("fa2341fa")), l[0])

    def test_eq(self):
        self.assertEqual(self.kls("asdf", disable_inst_caching=True),
                         self.kls("asdf", disable_inst_caching=True))

        self.assertNotEqual(self.kls("asdf2", disable_inst_caching=True),
                            self.kls("asdf", disable_inst_caching=True))

    def test_it(self):
        class maze(object):
            def __init__(self, kwargs):
                self.__data__ = kwargs

            def __getattr__(self, attr):
                return self.__data__.get(attr, self)

        d = {}
        m = maze(d)
        f = self.kls
        self.assertEqual(f('foon')(m), m)
        d["foon"] = 1
        self.assertEqual(f('foon')(m), 1)
        self.assertEqual(f('dar.foon')(m), 1)
        self.assertEqual(f('.'.join(['blah']*10))(m), m)
        self.assertRaises(AttributeError, f('foon.dar'), m)


class Test_native_jit_attr(TestCase):

    kls = staticmethod(klass._native_internal_jit_attr)

    @property
    def jit_attr(self):
        return partial(klass.jit_attr, kls=self.kls)

    @property
    def jit_attr_named(self):
        return partial(klass.jit_attr_named, kls=self.kls)

    @property
    def jit_attr_ext_method(self):
        return partial(klass.jit_attr_ext_method, kls=self.kls)

    def mk_inst(self, attrname='_attr', method_lookup=False,
                use_cls_setattr=False, func=None,
                singleton=klass._uncached_singleton):

        f = func
        if not func:
            def f(self):
                self._invokes.append(self)
                return 54321

        class cls(object):

            def __init__(self):
                sf = partial(object.__setattr__, self)
                sf('_sets', [])
                sf('_reflects', [])
                sf('_invokes', [])

            attr = self.kls(f, attrname, singleton, use_cls_setattr)

            def __setattr__(self, attr, value):
                self._sets.append(self)
                object.__setattr__(self, attr, value)

            def reflect(self):
                self._reflects.append(self)
                return 12345

        return cls()

    def assertState(self, instance, sets=0, reflects=0, invokes=0, value=54321):
        self.assertEqual(instance.attr, value)
        sets = [instance] * sets
        reflects = [instance] * reflects
        invokes = [instance] * invokes
        msg = ("checking %s: got(%r), expected(%r); state was sets=%r, "
               "reflects=%r, invokes=%r" % (
                   "%s", "%s", "%s", instance._sets, instance._reflects,
                   instance._invokes))
        self.assertEqual(instance._sets, sets,
                         msg=(msg % ("sets", instance._sets, sets,)))
        self.assertEqual(instance._reflects, reflects,
                         msg=(msg % ("reflects", instance._reflects,
                                     reflects,)))
        self.assertEqual(instance._invokes, invokes,
                         msg=(msg % ("invokes", instance._invokes, invokes,)))

    def test_implementation(self):
        obj = self.mk_inst()

        # default state is use_cls_setattr = False
        self.assertState(obj, invokes=1)
        self.assertState(obj, invokes=1)
        del obj._attr
        self.assertState(obj, invokes=2)
        self.assertState(obj, invokes=2)

        # basic caching is now verified.
        obj = self.mk_inst(use_cls_setattr=True)
        self.assertState(obj, sets=1, invokes=1)
        self.assertState(obj, sets=1, invokes=1)
        del obj._attr
        self.assertState(obj, sets=2, invokes=2)
        self.assertState(obj, sets=2, invokes=2)

    def test_jit_attr(self):
        now = time()

        class cls(object):
            @self.jit_attr
            def my_attr(self):
                return now

        o = cls()
        self.assertEqual(o.my_attr, now)
        self.assertEqual(o._my_attr, now)

        class cls(object):
            @self.jit_attr
            def attr2(self):
                return now

            def __setattr__(self, attr, value):
                raise AssertionError("setattr was invoked")

        o = cls()
        self.assertEqual(o.attr2, now)
        self.assertEqual(o._attr2, now)
        del o._attr2
        self.assertEqual(o.attr2, now)
        self.assertEqual(o._attr2, now)

    def test_jit_attr_named(self):
        now = time()

        # check attrname control and default object.__setattr__ avoidance
        class cls(object):
            @self.jit_attr_named("_blah")
            def my_attr(self):
                return now

            def __setattr__(self, attr, value):
                raise AssertionError("setattr was invoked")

        o = cls()
        self.assertEqual(o.my_attr, now)
        self.assertEqual(o._blah, now)

        class cls(object):
            @self.jit_attr_named("_blah2", use_cls_setattr=True)
            def my_attr(self):
                return now

            def __setattr__(self, attr, value):
                object.__setattr__(self, "invoked", True)
                object.__setattr__(self, attr, value)

        o = cls()
        self.assertFalse(hasattr(o, 'invoked'))
        self.assertEqual(o.my_attr, now)
        self.assertEqual(o._blah2, now)
        self.assertTrue(o.invoked)

    def test_jit_attr_ext_method(self):
        now = time()
        now2 = now + 100

        class base(object):
            def f1(self):
                return now

            def f2(self):
                return now2

            def __setattr__(self, attr, value):
                if not getattr(self, '_setattr_allowed', False):
                    raise TypeError("setattr isn't allowed for %s" % attr)
                object.__setattr__(self, attr, value)

        base.attr = self.jit_attr_ext_method('f1', '_attr')
        o = base()
        self.assertEqual(o.attr, now)
        self.assertEqual(o._attr, now)
        self.assertEqual(o.attr, now)

        base.attr = self.jit_attr_ext_method('f1', '_attr', use_cls_setattr=True)
        o = base()
        self.assertRaises(TypeError, getattr, o, 'attr')
        base._setattr_allowed = True
        self.assertEqual(o.attr, now)

        base.attr = self.jit_attr_ext_method('f2', '_attr2')
        o = base()
        self.assertEqual(o.attr, now2)
        self.assertEqual(o._attr2, now2)

        # finally, check that it's doing lookups rather then storing the func.
        base.attr = self.jit_attr_ext_method('func', '_attr2')
        o = base()
        # no func...
        self.assertRaises(AttributeError, getattr, o, 'attr')
        base.func = base.f1
        self.assertEqual(o.attr, now)
        self.assertEqual(o._attr2, now)
        # check caching...
        base.func = base.f2
        self.assertEqual(o.attr, now)
        del o._attr2
        self.assertEqual(o.attr, now2)

    def test_check_singleton_is_compare(self):
        def throw_assert(*args, **kwds):
            raise AssertionError("I shouldn't be invoked: %s, %s" % (args, kwds,))

        class puker(object):
            __eq__ = throw_assert

        puker_singleton = puker()

        obj = self.mk_inst(singleton=puker_singleton)
        obj._attr = puker_singleton
        # force attr access. if it's done wrong, it'll puke.
        # pylint: disable=pointless-statement
        obj.attr

    def test_cached_property(self):
        l = []
        class foo(object):
            @klass.cached_property
            def blah(self, l=l, i=iter(range(5))):
                l.append(None)
                return next(i)
        f = foo()
        self.assertEqual(f.blah, 0)
        self.assertEqual(len(l), 1)
        self.assertEqual(f.blah, 0)
        self.assertEqual(len(l), 1)
        del f.blah
        self.assertEqual(f.blah, 1)
        self.assertEqual(len(l), 2)

    def test_cached_property(self):
        l = []

        def named(self, l=l, i=iter(range(5))):
            l.append(None)
            return next(i)

        class foo(object):
            blah = klass.cached_property_named("blah")(named)

        f = foo()
        self.assertEqual(f.blah, 0)
        self.assertEqual(len(l), 1)
        self.assertEqual(f.blah, 0)
        self.assertEqual(len(l), 1)
        del f.blah
        self.assertEqual(f.blah, 1)
        self.assertEqual(len(l), 2)


class Test_cpy_jit_attr(Test_native_jit_attr):

    kls = staticmethod(klass._internal_jit_attr)
    if klass._internal_jit_attr is klass._native_internal_jit_attr:
        skip = "extension is missing"


class test_aliased_attr(TestCase):

    func = staticmethod(klass.alias_attr)

    def test_it(self):
        class cls(object):
            attr = self.func("dar.blah")

        o = cls()
        self.assertRaises(AttributeError, getattr, o, 'attr')
        o.dar = "foon"

        self.assertRaises(AttributeError, getattr, o, 'attr')
        o.dar = o
        o.blah = "monkey"

        self.assertEqual(o.attr, 'monkey')

        # verify it'll cross properties...
        class blah(object):
            target = object()

        class cls(object):
            @property
            def foon(self):
                return blah()

            alias = self.func("foon.target")
        o = cls()
        self.assertIdentical(o.alias, blah.target)


class test_cached_hash(TestCase):
    func = staticmethod(klass.cached_hash)

    def test_it(self):
        now = int(time())
        class cls(object):
            invoked = []
            @self.func
            def __hash__(self):
                self.invoked.append(self)
                return now
        o = cls()
        self.assertEqual(hash(o), now)
        self.assertEqual(o.invoked, [o])
        # ensure it cached...
        self.assertEqual(hash(o), now)
        self.assertEqual(o.invoked, [o])
        self.assertEqual(o._hash, now)


class test_native_reflective_hash(TestCase):
    func = staticmethod(klass.native_reflective_hash)

    def test_it(self):
        class cls(object):
            __hash__ = self.func('_hash')

        obj = cls()
        self.assertRaises(AttributeError, hash, obj)
        obj._hash = 1
        self.assertEqual(hash(obj), 1)
        obj._hash = 123123123
        self.assertEqual(hash(obj), 123123123)
        # verify it's not caching in any form
        del obj._hash
        self.assertRaises(AttributeError, hash, obj)

        class cls2(object):
            __hash__ = self.func('_dar')
        obj = cls2()
        self.assertRaises(AttributeError, hash, obj)
        obj._dar = 4
        self.assertEqual(hash(obj), 4)


class test_cpy_reflective_hash(test_native_reflective_hash):

    kls = staticmethod(klass.reflective_hash)
    if klass.reflective_hash is klass.native_reflective_hash:
        skip = "cpython extension isn't available"


cpy_loaded_Test = mk_cpy_loadable_testcase(
    "snakeoil._klass", "snakeoil.klass", "reflective_hash", "reflective_hash")


class TestImmutableInstance(TestCase):

    def test_metaclass(self):
        def f(scope):
            scope["__metaclass__"] = klass.immutable_instance

        self.common_test(f)

    def test_injection(self):

        def f(scope):
            klass.inject_immutable_instance(scope)

        self.common_test(f)

    @not_a_test
    def common_test(self, modify_kls):
        class kls(object):
            modify_kls(locals())

        o = kls()
        self.assertRaises(AttributeError, setattr, o, "dar", "foon")
        self.assertRaises(AttributeError, delattr, o, "dar")

        object.__setattr__(o, 'dar', 'foon')
        self.assertRaises(AttributeError, delattr, o, "dar")

        # ensure it only sets it if nothing is in place already.

        class kls(object):
            def __setattr__(self, attr, value):
                raise TypeError(self)

            modify_kls(locals())

        o = kls()
        self.assertRaises(TypeError, setattr, o, "dar", "foon")
        self.assertRaises(AttributeError, delattr, o, "dar")


class TestAliasMethod(TestCase):

    func = staticmethod(klass.alias_method)

    def test_alias_method(self):
        class kls(object):
            __len__ = lambda s: 3
            lfunc = self.func("__len__")

        c = kls()
        self.assertEqual(c.__len__(), c.lfunc())
        c.__len__ = lambda: 4
        self.assertEqual(c.__len__(), c.lfunc())


class TestPatch(TestCase):

    def setUp(self):
        # cache original methods
        self._math_ceil = math.ceil
        self._math_floor = math.floor

    def tearDown(self):
        # restore original methods
        math.ceil = self._math_ceil
        math.floor = self._math_floor

    def test_patch(self):
        n = 0.1
        self.assertEqual(math.ceil(n), 1)

        @klass.patch('math.ceil')
        def ceil(orig_ceil, n):
            return math.floor(n)

        self.assertEqual(math.ceil(n), 0)

    def test_multiple_patches(self):
        n = 1.1
        self.assertEqual(math.ceil(n), 2)
        self.assertEqual(math.floor(n), 1)

        @klass.patch('math.ceil')
        @klass.patch('math.floor')
        def zero(orig_func, n):
            return 0

        self.assertEqual(math.ceil(n), 0)
        self.assertEqual(math.floor(n), 0)