NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

test_http.py (5197B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\cherrypy\test\test_http.py
      2 import mimetypes
      3 import cherrypy
      4 from cherrypy._cpcompat import HTTPConnection, HTTPSConnection, ntob
      5 
      6 def encode_multipart_formdata(files):
      7     BOUNDARY = '________ThIs_Is_tHe_bouNdaRY_$'
      8     L = []
      9     for key, filename, value in files:
     10         L.append('--' + BOUNDARY)
     11         L.append('Content-Disposition: form-data; name="%s"; filename="%s"' % (key, filename))
     12         ct = mimetypes.guess_type(filename)[0] or 'application/octet-stream'
     13         L.append('Content-Type: %s' % ct)
     14         L.append('')
     15         L.append(value)
     16 
     17     L.append('--' + BOUNDARY + '--')
     18     L.append('')
     19     body = '\r\n'.join(L)
     20     content_type = 'multipart/form-data; boundary=%s' % BOUNDARY
     21     return (content_type, body)
     22 
     23 
     24 from cherrypy.test import helper
     25 
     26 class HTTPTests(helper.CPWebCase):
     27 
     28     def setup_server():
     29 
     30         class Root:
     31 
     32             def index(self, *args, **kwargs):
     33                 return 'Hello world!'
     34 
     35             index.exposed = True
     36 
     37             def no_body(self, *args, **kwargs):
     38                 return 'Hello world!'
     39 
     40             no_body.exposed = True
     41             no_body._cp_config = {'request.process_request_body': False}
     42 
     43             def post_multipart(self, file):
     44                 contents = file.file.read()
     45                 summary = []
     46                 curchar = ''
     47                 count = 0
     48                 for c in contents:
     49                     if c == curchar:
     50                         count += 1
     51                     else:
     52                         if count:
     53                             summary.append('%s * %d' % (curchar, count))
     54                         count = 1
     55                         curchar = c
     56 
     57                 if count:
     58                     summary.append('%s * %d' % (curchar, count))
     59                 return ', '.join(summary)
     60 
     61             post_multipart.exposed = True
     62 
     63         cherrypy.tree.mount(Root())
     64         cherrypy.config.update({'server.max_request_body_size': 30000000})
     65 
     66     setup_server = staticmethod(setup_server)
     67 
     68     def test_no_content_length(self):
     69         if self.scheme == 'https':
     70             c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT))
     71         else:
     72             c = HTTPConnection('%s:%s' % (self.interface(), self.PORT))
     73         c.request('POST', '/no_body')
     74         response = c.getresponse()
     75         self.body = response.fp.read()
     76         self.status = str(response.status)
     77         self.assertStatus(200)
     78         self.assertBody(ntob('Hello world!'))
     79         if self.scheme == 'https':
     80             c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT))
     81         else:
     82             c = HTTPConnection('%s:%s' % (self.interface(), self.PORT))
     83         c.request('POST', '/')
     84         response = c.getresponse()
     85         self.body = response.fp.read()
     86         self.status = str(response.status)
     87         self.assertStatus(411)
     88 
     89     def test_post_multipart(self):
     90         alphabet = 'abcdefghijklmnopqrstuvwxyz'
     91         contents = ''.join([ c * 65536 for c in alphabet ])
     92         files = [('file', 'file.txt', contents)]
     93         content_type, body = encode_multipart_formdata(files)
     94         body = body.encode('Latin-1')
     95         if self.scheme == 'https':
     96             c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT))
     97         else:
     98             c = HTTPConnection('%s:%s' % (self.interface(), self.PORT))
     99         c.putrequest('POST', '/post_multipart')
    100         c.putheader('Content-Type', content_type)
    101         c.putheader('Content-Length', str(len(body)))
    102         c.endheaders()
    103         c.send(body)
    104         response = c.getresponse()
    105         self.body = response.fp.read()
    106         self.status = str(response.status)
    107         self.assertStatus(200)
    108         self.assertBody(', '.join([ '%s * 65536' % c for c in alphabet ]))
    109 
    110     def test_malformed_request_line(self):
    111         if getattr(cherrypy.server, 'using_apache', False):
    112             return self.skip('skipped due to known Apache differences...')
    113         if self.scheme == 'https':
    114             c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT))
    115         else:
    116             c = HTTPConnection('%s:%s' % (self.interface(), self.PORT))
    117         c._output(ntob('GET /'))
    118         c._send_output()
    119         if hasattr(c, 'strict'):
    120             response = c.response_class(c.sock, strict=c.strict, method='GET')
    121         else:
    122             response = c.response_class(c.sock, method='GET')
    123         response.begin()
    124         self.assertEqual(response.status, 400)
    125         self.assertEqual(response.fp.read(22), ntob('Malformed Request-Line'))
    126         c.close()
    127 
    128     def test_malformed_header(self):
    129         if self.scheme == 'https':
    130             c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT))
    131         else:
    132             c = HTTPConnection('%s:%s' % (self.interface(), self.PORT))
    133         c.putrequest('GET', '/')
    134         c.putheader('Content-Type', 'text/plain')
    135         c._output(ntob('Re, 1.2.3.4#015#012'))
    136         c.endheaders()
    137         response = c.getresponse()
    138         self.status = str(response.status)
    139         self.assertStatus(400)
    140         self.body = response.fp.read(20)
    141         self.assertBody('Illegal header line.')