Web   ·   Wiki   ·   Activities   ·   Blog   ·   Lists   ·   Chat   ·   Meeting   ·   Bugs   ·   Git   ·   Translate   ·   Archive   ·   People   ·   Donate
summaryrefslogtreecommitdiffstats
path: root/websdk/werkzeug/testsuite/local.py
blob: 236e7539b945bfe975463720d0f1f6638cf6da09 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
# -*- coding: utf-8 -*-
"""
    werkzeug.testsuite.local
    ~~~~~~~~~~~~~~~~~~~~~~~~

    Local and local proxy tests.

    :copyright: (c) 2011 by Armin Ronacher.
    :license: BSD, see LICENSE for more details.
"""
import time
import unittest
from threading import Thread

from werkzeug.testsuite import WerkzeugTestCase

from werkzeug import local


class LocalTestCase(WerkzeugTestCase):

    def test_basic_local(self):
        l = local.Local()
        l.foo = 0
        values = []
        def value_setter(idx):
            time.sleep(0.01 * idx)
            l.foo = idx
            time.sleep(0.02)
            values.append(l.foo)
        threads = [Thread(target=value_setter, args=(x,))
                   for x in [1, 2, 3]]
        for thread in threads:
            thread.start()
        time.sleep(0.2)
        assert sorted(values) == [1, 2, 3]

        def delfoo():
            del l.foo
        delfoo()
        self.assert_raises(AttributeError, lambda: l.foo)
        self.assert_raises(AttributeError, delfoo)

        local.release_local(l)

    def test_local_release(self):
        loc = local.Local()
        loc.foo = 42
        local.release_local(loc)
        assert not hasattr(loc, 'foo')

        ls = local.LocalStack()
        ls.push(42)
        local.release_local(ls)
        assert ls.top is None

    def test_local_proxy(self):
        foo = []
        ls = local.LocalProxy(lambda: foo)
        ls.append(42)
        ls.append(23)
        ls[1:] = [1, 2, 3]
        assert foo == [42, 1, 2, 3]
        assert repr(foo) == repr(ls)
        assert foo[0] == 42
        foo += [1]
        assert list(foo) == [42, 1, 2, 3, 1]

    def test_local_stack(self):
        ident = local.get_ident()

        ls = local.LocalStack()
        assert ident not in ls._local.__storage__
        assert ls.top is None
        ls.push(42)
        assert ident in ls._local.__storage__
        assert ls.top == 42
        ls.push(23)
        assert ls.top == 23
        ls.pop()
        assert ls.top == 42
        ls.pop()
        assert ls.top is None
        assert ls.pop() is None
        assert ls.pop() is None

        proxy = ls()
        ls.push([1, 2])
        assert proxy == [1, 2]
        ls.push((1, 2))
        assert proxy == (1, 2)
        ls.pop()
        ls.pop()
        assert repr(proxy) == '<LocalProxy unbound>'

        assert ident not in ls._local.__storage__

    def test_local_proxies_with_callables(self):
        foo = 42
        ls = local.LocalProxy(lambda: foo)
        assert ls == 42
        foo = [23]
        ls.append(42)
        assert ls == [23, 42]
        assert foo == [23, 42]

    def test_custom_idents(self):
        ident = 0
        loc = local.Local()
        stack = local.LocalStack()
        mgr = local.LocalManager([loc, stack], ident_func=lambda: ident)

        loc.foo = 42
        stack.push({'foo': 42})
        ident = 1
        loc.foo = 23
        stack.push({'foo': 23})
        ident = 0
        assert loc.foo == 42
        assert stack.top['foo'] == 42
        stack.pop()
        assert stack.top is None
        ident = 1
        assert loc.foo == 23
        assert stack.top['foo'] == 23
        stack.pop()
        assert stack.top is None


def suite():
    suite = unittest.TestSuite()
    suite.addTest(unittest.makeSuite(LocalTestCase))
    return suite