~sirn/fanboi2

ref: 6ac25b00afeb1848d62d228ccfd0d67e61be9076 fanboi2/fanboi2/tests/__init__.py -rw-r--r-- 4.3 KiB
6ac25b00Kridsada Thanabulpong Initial development Vagrantfile. 7 years ago
                                                                                
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
import os
import transaction
from fanboi2.models import DBSession, Base, redis_conn
from pyramid import testing
from sqlalchemy import create_engine


DATABASE_URI = os.environ.get(
    'POSTGRESQL_TEST_DATABASE',
    'postgresql://fanboi2:fanboi2@localhost:5432/fanboi2_test')


class DummyRedis(object):

    @classmethod
    def from_url(cls, *args, **kwargs):
        return cls()

    def __init__(self):
        self._store = {}
        self._expire = {}

    def get(self, key):
        return self._store.get(key, None)

    def set(self, key, value):
        try:
            value = bytes(value.encode('utf-8'))
        except AttributeError:
            pass
        self._store[key] = value

    def setnx(self, key, value):
        if not self.get(key):
            self.set(key, value)

    def exists(self, key):
        return key in self._store

    def expire(self, key, time):
        self._expire[key] = time

    def ttl(self, key):
        return self._expire.get(key, 0)

    def ping(self):
        return True


class _ModelInstanceSetup(object):

    def _makeBoard(self, **kwargs):
        from fanboi2.models import Board
        board = Board(**kwargs)
        DBSession.add(board)
        DBSession.flush()
        return board

    def _makeTopic(self, **kwargs):
        from fanboi2.models import Topic
        topic = Topic(**kwargs)
        DBSession.add(topic)
        DBSession.flush()
        return topic

    def _makePost(self, **kwargs):
        from fanboi2.models import Post
        if not kwargs.get('ip_address', None):
            kwargs['ip_address'] = '0.0.0.0'
        post = Post(**kwargs)
        DBSession.add(post)
        DBSession.flush()
        return post


class ModelMixin(_ModelInstanceSetup):

    @classmethod
    def tearDownClass(cls):
        Base.metadata.bind = None
        DBSession.remove()

    @classmethod
    def setUpClass(cls):
        engine = create_engine(DATABASE_URI)
        DBSession.configure(bind=engine)
        Base.metadata.bind = engine

    def setUp(self):
        redis_conn._redis = DummyRedis()
        Base.metadata.drop_all()
        Base.metadata.create_all()
        transaction.begin()
        self.request = self._makeRequest()
        self.registry = self._makeRegistry()
        self.config = testing.setUp(
            request=self.request,
            registry=self.registry)

    def tearDown(self):
        redis_conn._redis = None
        testing.tearDown()
        transaction.abort()

    def _makeRequest(self):
        """:rtype: pyramid.request.Request"""
        request = testing.DummyRequest()
        request.user_agent = 'Mock/1.0'
        request.referrer = None
        return request

    def _makeRegistry(self):
        """:rtype: pyramid.registry.Registry"""
        from pyramid.registry import Registry
        registry = Registry()
        registry.settings = {
            'app.timezone': 'Asia/Bangkok',
            'app.secret': 'Silently test in secret',
        }
        return registry


class TaskMixin(object):

    @classmethod
    def setUpClass(cls):
        from fanboi2.tasks import celery
        celery.config_from_object({'CELERY_ALWAYS_EAGER': True})
        super(TaskMixin, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        from fanboi2.tasks import celery
        celery.config_from_object({'CELERY_ALWAYS_EAGER': False})
        super(TaskMixin, cls).tearDownClass()


class ViewMixin(object):

    def _make_csrf(self, request):
        import hmac
        import os
        from hashlib import sha1
        request.session['csrf'] = sha1(os.urandom(64)).hexdigest()
        request.params['csrf_token'] = hmac.new(
            bytes(request.registry.settings['app.secret'].encode('utf8')),
            bytes(request.session['csrf'].encode('utf8')),
            digestmod=sha1,
        ).hexdigest()
        return request

    def _POST(self, data=None):
        from webob.multidict import MultiDict
        request = self.request
        request.method = 'POST'
        request.remote_addr = "127.0.0.1"
        request.params = MultiDict(data)
        return request

    def _GET(self, data=None):
        from webob.multidict import MultiDict
        request = self.request
        if data is None:
            data = {}
        request.remote_addr = "127.0.0.1"
        request.params = MultiDict(data)
        return request