test_session.py (13675B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\cherrypy\test\test_session.py 2 import os 3 localDir = os.path.dirname(__file__) 4 import sys 5 import threading 6 import time 7 import cherrypy 8 from cherrypy._cpcompat import copykeys, HTTPConnection, HTTPSConnection 9 from cherrypy.lib import sessions 10 from cherrypy.lib.httputil import response_codes 11 12 def http_methods_allowed(methods = ['GET', 'HEAD']): 13 method = cherrypy.request.method.upper() 14 if method not in methods: 15 cherrypy.response.headers['Allow'] = ', '.join(methods) 16 raise cherrypy.HTTPError(405) 17 18 19 cherrypy.tools.allow = cherrypy.Tool('on_start_resource', http_methods_allowed) 20 21 def setup_server(): 22 23 class Root: 24 _cp_config = {'tools.sessions.on': True, 25 'tools.sessions.storage_type': 'ram', 26 'tools.sessions.storage_path': localDir, 27 'tools.sessions.timeout': 1.0 / 60, 28 'tools.sessions.clean_freq': 1.0 / 60} 29 30 def clear(self): 31 cherrypy.session.cache.clear() 32 33 clear.exposed = True 34 35 def data(self): 36 cherrypy.session['aha'] = 'foo' 37 return repr(cherrypy.session._data) 38 39 data.exposed = True 40 41 def testGen(self): 42 counter = cherrypy.session.get('counter', 0) + 1 43 cherrypy.session['counter'] = counter 44 yield str(counter) 45 46 testGen.exposed = True 47 48 def testStr(self): 49 counter = cherrypy.session.get('counter', 0) + 1 50 cherrypy.session['counter'] = counter 51 return str(counter) 52 53 testStr.exposed = True 54 55 def setsessiontype(self, newtype): 56 self.__class__._cp_config.update({'tools.sessions.storage_type': newtype}) 57 if hasattr(cherrypy, 'session'): 58 del cherrypy.session 59 cls = getattr(sessions, newtype.title() + 'Session') 60 if cls.clean_thread: 61 cls.clean_thread.stop() 62 cls.clean_thread.unsubscribe() 63 del cls.clean_thread 64 65 setsessiontype.exposed = True 66 setsessiontype._cp_config = {'tools.sessions.on': False} 67 68 def index(self): 69 sess = cherrypy.session 70 c = sess.get('counter', 0) + 1 71 time.sleep(0.01) 72 sess['counter'] = c 73 return str(c) 74 75 index.exposed = True 76 77 def keyin(self, key): 78 return str(key in cherrypy.session) 79 80 keyin.exposed = True 81 82 def delete(self): 83 cherrypy.session.delete() 84 sessions.expire() 85 return 'done' 86 87 delete.exposed = True 88 89 def delkey(self, key): 90 del cherrypy.session[key] 91 return 'OK' 92 93 delkey.exposed = True 94 95 def blah(self): 96 return self._cp_config['tools.sessions.storage_type'] 97 98 blah.exposed = True 99 100 def iredir(self): 101 raise cherrypy.InternalRedirect('/blah') 102 103 iredir.exposed = True 104 105 def restricted(self): 106 return cherrypy.request.method 107 108 restricted.exposed = True 109 restricted._cp_config = {'tools.allow.on': True, 110 'tools.allow.methods': ['GET']} 111 112 def regen(self): 113 cherrypy.tools.sessions.regenerate() 114 return 'logged in' 115 116 regen.exposed = True 117 118 def length(self): 119 return str(len(cherrypy.session)) 120 121 length.exposed = True 122 123 def session_cookie(self): 124 cherrypy.session.load() 125 return cherrypy.session.id 126 127 session_cookie.exposed = True 128 session_cookie._cp_config = {'tools.sessions.path': '/session_cookie', 129 'tools.sessions.name': 'temp', 130 'tools.sessions.persistent': False} 131 132 cherrypy.tree.mount(Root()) 133 134 135 from cherrypy.test import helper 136 137 class SessionTest(helper.CPWebCase): 138 setup_server = staticmethod(setup_server) 139 140 def tearDown(self): 141 for fname in os.listdir(localDir): 142 if fname.startswith(sessions.FileSession.SESSION_PREFIX): 143 os.unlink(os.path.join(localDir, fname)) 144 145 def test_0_Session(self): 146 self.getPage('/setsessiontype/ram') 147 self.getPage('/clear') 148 self.getPage('/data') 149 self.assertBody("{'aha': 'foo'}") 150 c = self.cookies[0] 151 self.getPage('/data', self.cookies) 152 self.assertEqual(self.cookies[0], c) 153 self.getPage('/testStr') 154 self.assertBody('1') 155 cookie_parts = dict([ p.strip().split('=') for p in self.cookies[0][1].split(';') ]) 156 self.assertEqual(set(cookie_parts.keys()), set(['session_id', 'expires', 'Path'])) 157 self.getPage('/testGen', self.cookies) 158 self.assertBody('2') 159 self.getPage('/testStr', self.cookies) 160 self.assertBody('3') 161 self.getPage('/data', self.cookies) 162 self.assertBody("{'aha': 'foo', 'counter': 3}") 163 self.getPage('/length', self.cookies) 164 self.assertBody('2') 165 self.getPage('/delkey?key=counter', self.cookies) 166 self.assertStatus(200) 167 self.getPage('/setsessiontype/file') 168 self.getPage('/testStr') 169 self.assertBody('1') 170 self.getPage('/testGen', self.cookies) 171 self.assertBody('2') 172 self.getPage('/testStr', self.cookies) 173 self.assertBody('3') 174 self.getPage('/delkey?key=counter', self.cookies) 175 self.assertStatus(200) 176 time.sleep(2) 177 self.getPage('/') 178 self.assertBody('1') 179 self.getPage('/length', self.cookies) 180 self.assertBody('1') 181 self.getPage('/keyin?key=counter', self.cookies) 182 self.assertBody('True') 183 cookieset1 = self.cookies 184 self.getPage('/') 185 self.getPage('/length', self.cookies) 186 self.assertBody('2') 187 self.getPage('/delete', self.cookies) 188 self.assertBody('done') 189 self.getPage('/delete', cookieset1) 190 self.assertBody('done') 191 f = lambda : [ x for x in os.listdir(localDir) if x.startswith('session-') ] 192 self.assertEqual(f(), []) 193 self.getPage('/') 194 f = lambda : [ x for x in os.listdir(localDir) if x.startswith('session-') ] 195 self.assertNotEqual(f(), []) 196 time.sleep(2) 197 self.assertEqual(f(), []) 198 199 def test_1_Ram_Concurrency(self): 200 self.getPage('/setsessiontype/ram') 201 self._test_Concurrency() 202 203 def test_2_File_Concurrency(self): 204 self.getPage('/setsessiontype/file') 205 self._test_Concurrency() 206 207 def _test_Concurrency(self): 208 client_thread_count = 5 209 request_count = 30 210 self.getPage('/') 211 self.assertBody('1') 212 cookies = self.cookies 213 data_dict = {} 214 errors = [] 215 216 def request(index): 217 if self.scheme == 'https': 218 c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT)) 219 else: 220 c = HTTPConnection('%s:%s' % (self.interface(), self.PORT)) 221 for i in range(request_count): 222 c.putrequest('GET', '/') 223 for k, v in cookies: 224 c.putheader(k, v) 225 226 c.endheaders() 227 response = c.getresponse() 228 body = response.read() 229 if response.status != 200 or not body.isdigit(): 230 errors.append((response.status, body)) 231 else: 232 data_dict[index] = max(data_dict[index], int(body)) 233 234 ts = [] 235 for c in range(client_thread_count): 236 data_dict[c] = 0 237 t = threading.Thread(target=request, args=(c,)) 238 ts.append(t) 239 t.start() 240 241 for t in ts: 242 t.join() 243 244 hitcount = max(data_dict.values()) 245 expected = 1 + client_thread_count * request_count 246 for e in errors: 247 print e 248 249 self.assertEqual(hitcount, expected) 250 251 def test_3_Redirect(self): 252 self.getPage('/testStr') 253 self.getPage('/iredir', self.cookies) 254 self.assertBody('file') 255 256 def test_4_File_deletion(self): 257 self.getPage('/testStr') 258 id = self.cookies[0][1].split(';', 1)[0].split('=', 1)[1] 259 path = os.path.join(localDir, 'session-' + id) 260 os.unlink(path) 261 self.getPage('/testStr', self.cookies) 262 263 def test_5_Error_paths(self): 264 self.getPage('/unknown/page') 265 self.assertErrorPage(404, "The path '/unknown/page' was not found.") 266 self.getPage('/restricted', self.cookies, method='POST') 267 self.assertErrorPage(405, response_codes[405]) 268 269 def test_6_regenerate(self): 270 self.getPage('/testStr') 271 id1 = self.cookies[0][1].split(';', 1)[0].split('=', 1)[1] 272 self.getPage('/regen') 273 self.assertBody('logged in') 274 id2 = self.cookies[0][1].split(';', 1)[0].split('=', 1)[1] 275 self.assertNotEqual(id1, id2) 276 self.getPage('/testStr') 277 id1 = self.cookies[0][1].split(';', 1)[0].split('=', 1)[1] 278 self.getPage('/testStr', headers=[('Cookie', 'session_id=maliciousid; expires=Sat, 27 Oct 2017 04:18:28 GMT; Path=/;')]) 279 id2 = self.cookies[0][1].split(';', 1)[0].split('=', 1)[1] 280 self.assertNotEqual(id1, id2) 281 self.assertNotEqual(id2, 'maliciousid') 282 283 def test_7_session_cookies(self): 284 self.getPage('/setsessiontype/ram') 285 self.getPage('/clear') 286 self.getPage('/session_cookie') 287 cookie_parts = dict([ p.strip().split('=') for p in self.cookies[0][1].split(';') ]) 288 self.assertEqual(set(cookie_parts.keys()), set(['temp', 'Path'])) 289 id1 = cookie_parts['temp'] 290 self.assertEqual(copykeys(sessions.RamSession.cache), [id1]) 291 self.getPage('/session_cookie', self.cookies) 292 cookie_parts = dict([ p.strip().split('=') for p in self.cookies[0][1].split(';') ]) 293 self.assertEqual(set(cookie_parts.keys()), set(['temp', 'Path'])) 294 self.assertBody(id1) 295 self.assertEqual(copykeys(sessions.RamSession.cache), [id1]) 296 self.getPage('/session_cookie') 297 cookie_parts = dict([ p.strip().split('=') for p in self.cookies[0][1].split(';') ]) 298 self.assertEqual(set(cookie_parts.keys()), set(['temp', 'Path'])) 299 id2 = cookie_parts['temp'] 300 self.assertNotEqual(id1, id2) 301 self.assertEqual(set(sessions.RamSession.cache.keys()), set([id1, id2])) 302 time.sleep(2.5) 303 cache = copykeys(sessions.RamSession.cache) 304 if cache: 305 if cache == [id2]: 306 self.fail('The second session did not time out.') 307 else: 308 self.fail('Unknown session id in cache: %r', cache) 309 310 311 import socket 312 try: 313 import memcache 314 host, port = ('127.0.0.1', 11211) 315 for res in socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM): 316 af, socktype, proto, canonname, sa = res 317 s = None 318 try: 319 s = socket.socket(af, socktype, proto) 320 s.settimeout(1.0) 321 s.connect((host, port)) 322 s.close() 323 except socket.error: 324 if s: 325 s.close() 326 raise 327 328 break 329 330 except (ImportError, socket.error): 331 332 class MemcachedSessionTest(helper.CPWebCase): 333 setup_server = staticmethod(setup_server) 334 335 def test(self): 336 return self.skip('memcached not reachable ') 337 338 339 else: 340 341 class MemcachedSessionTest(helper.CPWebCase): 342 setup_server = staticmethod(setup_server) 343 344 def test_0_Session(self): 345 self.getPage('/setsessiontype/memcached') 346 self.getPage('/testStr') 347 self.assertBody('1') 348 self.getPage('/testGen', self.cookies) 349 self.assertBody('2') 350 self.getPage('/testStr', self.cookies) 351 self.assertBody('3') 352 self.getPage('/length', self.cookies) 353 self.assertErrorPage(500) 354 self.assertInBody('NotImplementedError') 355 self.getPage('/delkey?key=counter', self.cookies) 356 self.assertStatus(200) 357 time.sleep(1.25) 358 self.getPage('/') 359 self.assertBody('1') 360 self.getPage('/keyin?key=counter', self.cookies) 361 self.assertBody('True') 362 self.getPage('/delete', self.cookies) 363 self.assertBody('done') 364 365 def test_1_Concurrency(self): 366 client_thread_count = 5 367 request_count = 30 368 self.getPage('/') 369 self.assertBody('1') 370 cookies = self.cookies 371 data_dict = {} 372 373 def request(index): 374 for i in range(request_count): 375 self.getPage('/', cookies) 376 377 if not self.body.isdigit(): 378 self.fail(self.body) 379 data_dict[index] = v = int(self.body) 380 381 ts = [] 382 for c in range(client_thread_count): 383 data_dict[c] = 0 384 t = threading.Thread(target=request, args=(c,)) 385 ts.append(t) 386 t.start() 387 388 for t in ts: 389 t.join() 390 391 hitcount = max(data_dict.values()) 392 expected = 1 + client_thread_count * request_count 393 self.assertEqual(hitcount, expected) 394 395 def test_3_Redirect(self): 396 self.getPage('/testStr') 397 self.getPage('/iredir', self.cookies) 398 self.assertBody('memcached') 399 400 def test_5_Error_paths(self): 401 self.getPage('/unknown/page') 402 self.assertErrorPage(404, "The path '/unknown/page' was not found.") 403 self.getPage('/restricted', self.cookies, method='POST') 404 self.assertErrorPage(405, response_codes[405])