-
Notifications
You must be signed in to change notification settings - Fork 33
/
Copy pathtest_middleware.py
218 lines (183 loc) · 7.43 KB
/
test_middleware.py
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
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
from engineauth.middleware import AuthMiddleware
from engineauth.middleware import EngineAuthRequest
from engineauth import models
import test_base
import webapp2
__author__ = '[email protected] (Kyle Finley)'
app = AuthMiddleware(webapp2.WSGIApplication())
class TestAuthMiddleware(test_base.BaseTestCase):
def setUp(self):
super(TestAuthMiddleware, self).setUp()
# def test_load_config(self):
# req = EngineAuthRequest.blank('/auth/google')
# resp = req.get_response(app)
# self.assertEqual(resp, '/auth')
def test_load_strategy(self):
from engineauth.strategies.google import GoogleStrategy
strategy_class = app._load_strategy('google')
self.assertEqual(strategy_class, GoogleStrategy)
self.assertRaises(Exception, app._load_strategy, 'enron')
from engineauth.strategies.appengine_openid import\
AppEngineOpenIDStrategy
strategy_class = app._load_strategy('appengine_openid')
self.assertEqual(strategy_class, AppEngineOpenIDStrategy)
def test_load_session_no_session(self):
req = EngineAuthRequest.blank('/auth/google')
# No Session
s_count = models.Session.query().count()
self.assertTrue(s_count == 0)
sess = req._load_session()
s_count = models.Session.query().count()
self.assertTrue(s_count == 1)
def test_laod_session_session_id_no_user_id(self):
# Cookie session_id but no user_id
s = models.Session.create()
s_count = models.Session.query().count()
self.assertTrue(s_count == 1)
req = EngineAuthRequest.blank('/auth/google')
req.cookies['_eauth'] = s.serialize()
req._load_session()
self.assertTrue(req.session.session_id == s.session_id)
# Assert No new session was created
s_count2 = models.Session.query().count()
self.assertTrue(s_count2 == 1)
def test_laod_session_session_id_and_user_id(self):
# Cookie session_id and user_id
s = models.Session.create()
s_count = models.Session.query().count()
self.assertTrue(s_count == 1)
req = EngineAuthRequest.blank('/auth/google')
req.cookies['_eauth'] = s.serialize()
req._load_session()
self.assertTrue(req.session.session_id == s.session_id)
# Assert No new session was created
s_count2 = models.Session.query().count()
self.assertTrue(s_count2 == 1)
def test_laod_session_cookie_and_no_session(self):
# Cookie and not session
s = models.Session.create()
old_sid = s.session_id
s_serialized = s.serialize()
s.key.delete()
s_count = models.Session.query().count()
self.assertTrue(s_count == 0)
req = EngineAuthRequest.blank('/auth/google')
req.cookies['_eauth'] = s_serialized
req._load_session()
# Assert that a new session was created
self.assertTrue(req.session.session_id != old_sid)
# Assert No new session was created
s_count2 = models.Session.query().count()
self.assertTrue(s_count2 == 1)
def test_save_session(self):
# Cookie session_id but no user_id
s = models.Session.create()
s_count = models.Session.query().count()
self.assertTrue(s_count == 1)
req = EngineAuthRequest.blank('/auth/google')
req.cookies['_eauth'] = s.serialize()
resp = req.get_response(app)
resp.request = req
resp._save_session()
self.assertTrue(resp.request.session.session_id == s.session_id)
# Assert No new session was created
s_count2 = models.Session.query().count()
self.assertTrue(s_count2 == 1)
# Add a user_id to session
resp.request.session.user_id = '1'
resp._save_session()
# a new session should be created with the user_id as it's id
# self.assertEqual(resp.request.session.key.id(), '1')
s_count = models.Session.query().count()
self.assertTrue(s_count == 1)
s1 = models.Session.query().get()
self.assertEqual(s1.key.id(), '1')
def test__load_user(self):
user = models.User.create_user('test:12345')
req = EngineAuthRequest.blank('/auth/google')
req._load_session()
req.session.user_id = user.get_id()
req._load_user()
self.assertEqual(user, req.user)
def test__load_user_by_profile(self):
# No existing User no logged in User
auth_id = 'test:12345'
user_info = {
'auth_id': auth_id,
'info': {},
}
# create profile
p = models.UserProfile.get_or_create(auth_id, user_info)
req = EngineAuthRequest.blank('/auth/google')
req._load_session()
req._load_user()
# User Count before
user_count = models.User.query().count()
self.assertEqual(user_count, 0)
req.load_user_by_profile(p)
# User Count after
user_count = models.User.query().count()
self.assertEqual(user_count, 1)
user = models.User.query().get()
self.assertTrue(p.key.id() in user.auth_ids)
# Yes existing User no logged in User
req = EngineAuthRequest.blank('/auth/google')
req._load_session()
req._load_user()
req.load_user_by_profile(p)
# Test to no new User was created
user_count = models.User.query().count()
self.assertEqual(user_count, 1)
# Yes existing User yes logged in User new Profile
auth_id = 'test:abc'
user_info = {
'auth_id': auth_id,
'info': {},
}
# create profile
p1 = models.UserProfile.get_or_create(auth_id, user_info)
req.load_user_by_profile(p1)
# Test to no new User was created
user_count = models.User.query().count()
self.assertEqual(user_count, 1)
def test_add_message(self):
req = EngineAuthRequest.blank('/auth/google')
req._load_session()
msgs = req.get_messages()
self.assertEquals(msgs, None)
req.add_message('TEST MESSAGE')
msgs = req.get_messages()
self.assertEquals(msgs, [{'level': None, 'message':'TEST MESSAGE' }])
# Get again should be none.
msgs = req.get_messages()
self.assertEquals(msgs, None)
# add message with level error
req.add_message('TEST1', 'error')
# add another message with level error
req.add_message('TEST2', 'success')
msgs = req.get_messages()
self.assertEquals(msgs, [
{'level': 'error', 'message':'TEST1' },
{'level': 'success', 'message':'TEST2' },
])
# Get again should be none.
msgs = req.get_messages()
self.assertEquals(msgs, None)
# Test with different key.
# add message with level error
req.add_message('TEST1', 'error')
# add another message with level error
req.add_message('TEST2', 'success', '_mykey')
msgs = req.get_messages()
self.assertEquals(msgs, [
{'level': 'error', 'message':'TEST1' },
])
msgs_key = req.get_messages('_mykey')
self.assertEquals(msgs_key, [
{'level': 'success', 'message':'TEST2' },
])
# Get again should be none.
msgs = req.get_messages()
self.assertEquals(msgs, None)
msgs_key = req.get_messages()
self.assertEquals(msgs_key, None)