import gc
import random
import sys
import threading
import weakref
from functools import partial
from itertools import product
from time import sleep

import pytest

from tls_property import tls_property


if sys.version_info >= (3,):  # Python 3.x and above
    create_daemon_thread = partial(threading.Thread, daemon=True)

    get_thread_ident = threading.get_ident

else:  # Python 2.7 and below
    def create_daemon_thread(**kwargs):
        thread = threading.Thread(**kwargs)
        thread.daemon = True
        return thread

    import thread
    get_thread_ident = thread.get_ident


@pytest.fixture
def cls():
    # tuples doesn't support weak references "even when subclassed"
    #  (according to the documentation of weakref module)
    class Result(object):
        def __init__(self, thread_ident, i):
            self.thread_ident = thread_ident
            self.i = i

    class Test(object):
        def __init__(self):
            self.tls = threading.local()

        def _prop(self):
            i = getattr(self.tls, 'i', 0)
            setattr(self.tls, 'i', i + 1)
            return Result(get_thread_ident(), i)

        prop = tls_property(_prop)

    return Test


def test_tls_property_fixture(cls):
    obj = cls()
    for i in range(20):
        p = obj._prop()
        assert p.thread_ident == get_thread_ident()
        assert p.i == i, 'i not increasing'


def test_tls_property_once(cls):
    objs = [cls() for _ in range(100)]
    rounds = list(product(objs, range(20)))
    random.shuffle(rounds)
    for obj, _ in rounds:
        p = obj.prop
        assert p
        assert p.i == 0, 'Function under @tls_property ' \
                         'was called more than once'


def test_tls_property_threading(cls):
    exceptions = []
    obj = cls()

    def target():
        try:
            for i in range(20):
                assert obj.prop.thread_ident == get_thread_ident(), \
                    'Value created in another thread'
                sleep(random.random() / 10)
        except Exception as exc:
            exceptions.append(exc)

    threads = []
    for _ in range(10):
        thread = create_daemon_thread(target=target)
        thread.start()
        threads.append(thread)
    for thread in threads:
        thread.join()
    for exc in exceptions:
        raise exc


def test_tls_property_gc(cls):
    obj = cls()
    obj_ref = weakref.ref(obj)

    p = obj.prop
    assert p
    p_ref = weakref.ref(p)

    del p, obj
    gc.collect()
    assert obj_ref() is None, 'Object is still alive?!'
    assert p_ref() is None, 'Value is still alive'


def test_tls_property_get(cls):
    assert isinstance(cls.prop, tls_property)


def test_tls_property_reset(cls):
    obj = cls()

    del obj.prop  # should not fail

    value1 = obj.prop
    value2 = obj.prop
    assert value1 is value2, 'Something wrong with basics before reset?!'

    del obj.prop

    value3 = obj.prop
    value4 = obj.prop
    assert value3 is value4, 'Something wrong with basics after reset'

    assert value1 is not value3, 'Reset not working'


def test_tls_property_gc_reset(cls):
    obj = cls()

    p = obj.prop
    assert p
    p_ref = weakref.ref(p)

    del p, obj.prop
    gc.collect()
    assert p_ref() is None, 'Value is still alive after reset'

import gc
import random
import sys
import threading
import weakref
from functools import partial
from itertools import product
from time import sleep

import pytest

from tls_property import tls_property


if sys.version_info >= (3,):  # Python 3.x and above
    create_daemon_thread = partial(threading.Thread, daemon=True)

    get_thread_ident = threading.get_ident

else:  # Python 2.7 and below
    def create_daemon_thread(**kwargs):
        thread = threading.Thread(**kwargs)
        thread.daemon = True
        return thread

    import thread
    get_thread_ident = thread.get_ident


@pytest.fixture
def cls():
    # tuples doesn't support weak references "even when subclassed"
    #  (according to the documentation of weakref module)
    class Result(object):
        def __init__(self, thread_ident, i):
            self.thread_ident = thread_ident
            self.i = i

    class Test(object):
        def __init__(self):
            self.tls = threading.local()

        def _prop(self):
            i = getattr(self.tls, 'i', 0)
            setattr(self.tls, 'i', i + 1)
            return Result(get_thread_ident(), i)

        prop = tls_property(_prop)

    return Test


def test_tls_property_fixture(cls):
    obj = cls()
    for i in range(20):
        p = obj._prop()
        assert p.thread_ident == get_thread_ident()
        assert p.i == i, 'i not increasing'


def test_tls_property_once(cls):
    objs = [cls() for _ in range(100)]
    rounds = list(product(objs, range(20)))
    random.shuffle(rounds)
    for obj, _ in rounds:
        p = obj.prop
        assert p
        assert p.i == 0, 'Function under @tls_property ' \
                         'was called more than once'


def test_tls_property_threading(cls):
    exceptions = []
    obj = cls()

    def target():
        try:
            for i in range(20):
                assert obj.prop.thread_ident == get_thread_ident(), \
                    'Value created in another thread'
                sleep(random.random() / 10)
        except Exception as exc:
            exceptions.append(exc)

    threads = []
    for _ in range(10):
        thread = create_daemon_thread(target=target)
        thread.start()
        threads.append(thread)
    for thread in threads:
        thread.join()
    for exc in exceptions:
        raise exc


def test_tls_property_gc(cls):
    obj = cls()
    obj_ref = weakref.ref(obj)

    p = obj.prop
    assert p
    p_ref = weakref.ref(p)

    del p, obj
    gc.collect()
    assert obj_ref() is None, 'Object is still alive?!'
    assert p_ref() is None, 'Value is still alive'


def test_tls_property_get(cls):
    assert isinstance(cls.prop, tls_property)


def test_tls_property_reset(cls):
    obj = cls()

    del obj.prop  # should not fail

    value1 = obj.prop
    value2 = obj.prop
    assert value1 is value2, 'Something wrong with basics before reset?!'

    del obj.prop

    value3 = obj.prop
    value4 = obj.prop
    assert value3 is value4, 'Something wrong with basics after reset'

    assert value1 is not value3, 'Reset not working'


def test_tls_property_gc_reset(cls):
    obj = cls()

    p = obj.prop
    assert p
    p_ref = weakref.ref(p)

    del p, obj.prop
    gc.collect()
    assert p_ref() is None, 'Value is still alive after reset'

# Python 3.x and above # Python 2.7 and below # tuples doesn't support weak references "even when subclassed" #  (according to the documentation of weakref module) # should not fail