'Test HTTP protocol responses'

def test_http_response(self):

    from calibre.srv.http_response import parse_multipart_byterange
    def handler(conn):
        return conn.generate_static_output(u'test', (lambda : u''.join(conn.path)))
    with NamedTemporaryFile(suffix=u'test.epub') as f:
        with open(P(u'localization/locales.zip'), u'rb') as lf:
            with TestServer(handler, timeout=1, compress_min_size=0) as server:
                fdata = (string.ascii_letters * 100)
                (f.write(fdata), f.seek(0))
                conn = server.connect()
                conn.request(u'GET', u'/an_etagged_path')
                r = conn.getresponse()
                (self.ae(r.status, httplib.OK), self.ae(r.read(), 'an_etagged_path'))
                etag = r.getheader(u'ETag')
                self.ae(etag, (u'"%s"' % hashlib.sha1(u'an_etagged_path').hexdigest()))
                conn.request(u'GET', u'/an_etagged_path', headers={u'If-None-Match': etag})
                r = conn.getresponse()
                self.ae(r.status, httplib.NOT_MODIFIED)
                self.ae(r.read(), '')
                raw = ('a' * 20000)
                server.change_handler((lambda conn: raw))
                conn = server.connect()
                conn.request(u'GET', u'/an_etagged_path', headers={u'Accept-Encoding': u'gzip'})
                r = conn.getresponse()
                self.ae(str(len(raw)), r.getheader(u'Calibre-Uncompressed-Length'))
                (self.ae(r.status, httplib.OK), self.ae(zlib.decompress(r.read(), (16 + zlib.MAX_WBITS)), raw))
                num_calls = [0]
                def edfunc():
                    num_calls[0] += 1
                    return 'data'
                server.change_handler((lambda conn: conn.etagged_dynamic_response(u'xxx', edfunc)))
                conn = server.connect()
                conn.request(u'GET', u'/an_etagged_path')
                r = conn.getresponse()
                (self.ae(r.status, httplib.OK), self.ae(r.read(), 'data'))
                etag = r.getheader(u'ETag')
                self.ae(etag, '"xxx"')
                self.ae(r.getheader(u'Content-Length'), u'4')
                conn.request(u'GET', u'/an_etagged_path', headers={u'If-None-Match': etag})
                r = conn.getresponse()
                self.ae(r.status, httplib.NOT_MODIFIED)
                self.ae(r.read(), '')
                self.ae(num_calls[0], 1)
                for use_sendfile in (True, False):
                    server.change_handler((lambda conn: f))
                    server.loop.opts.use_sendfile = use_sendfile
                    conn = server.connect()
                    conn.request(u'GET', u'/test')
                    r = conn.getresponse()
                    etag = type(u'')(r.getheader(u'ETag'))
                    self.assertTrue(etag)
                    self.ae(r.getheader(u'Content-Type'), guess_type(f.name)[0])
                    self.ae(type(u'')(r.getheader(u'Accept-Ranges')), u'bytes')
                    self.ae(int(r.getheader(u'Content-Length')), len(fdata))
                    (self.ae(r.status, httplib.OK), self.ae(r.read(), fdata))
                    conn.request(u'GET', u'/test', headers={u'Range': u'bytes=2-25'})
                    r = conn.getresponse()
                    self.ae(r.status, httplib.PARTIAL_CONTENT)
                    self.ae(type(u'')(r.getheader(u'Accept-Ranges')), u'bytes')
                    self.ae(type(u'')(r.getheader(u'Content-Range')), (u'bytes    2-25/%d' % len(fdata)))
                    self.ae(int(r.getheader(u'Content-Length')), 24)
                    self.ae(r.read(), fdata[2:26])
                    conn.request(u'GET', u'/test', headers={u'Range': u'bytes=100000-'})
                    r = conn.getresponse()
                    self.ae(r.status, httplib.REQUESTED_RANGE_NOT_SATISFIABLE)
                    self.ae(type(u'')(r.getheader(u'Content-Range')), (u'bytes    */%d' % len(fdata)))
                    conn.request(u'GET', u'/test', headers={u'Range': u'bytes=25-50', u'If-Range': etag})
                    r = conn.getresponse()
                    (self.ae(r.status, httplib.PARTIAL_CONTENT), self.ae(r.read(), fdata[25:51]))
                    self.ae(int(r.getheader(u'Content-Length')), 26)
                    conn.request(u'GET', u'/test', headers={u'Range': u'bytes=0-1000000'})
                    r = conn.getresponse()
                    (self.ae(r.status, httplib.PARTIAL_CONTENT), self.ae(r.read(), fdata))
                    conn.request(u'GET', u'/test', headers={u'Range': u'bytes=25-50', u'If-Range': u'"nomatch"'})
                    r = conn.getresponse()
                    (self.ae(r.status, httplib.OK), self.ae(r.read(), fdata))
                    self.assertFalse(r.getheader(u'Content-Range'))
                    self.ae(int(r.getheader(u'Content-Length')), len(fdata))
                    conn.request(u'GET', u'/test', headers={u'Range': u'bytes=0-25,26-50'})
                    r = conn.getresponse()
                    self.ae(r.status, httplib.PARTIAL_CONTENT)
                    clen = int(r.getheader(u'Content-Length'))
                    data = r.read()
                    self.ae(clen, len(data))
                    buf = BytesIO(data)
                    self.ae(parse_multipart_byterange(buf, r.getheader(u'Content-Type')), [(0, fdata[:26]), (26, fdata[26:51])])
                    start_time = monotonic()
                    lf.seek(0)
                    data = lf.read()
                    server.change_handler((lambda conn: lf))
                    conn = server.connect(timeout=1)
                    conn.request(u'GET', u'/test')
                    r = conn.getresponse()
                    self.ae(r.status, httplib.OK)
                    rdata = r.read()
                    self.ae(len(data), len(rdata))
                    self.ae(hashlib.sha1(data).hexdigest(), hashlib.sha1(rdata).hexdigest())
                    self.ae(data, rdata)
                    time_taken = (monotonic() - start_time)
                    self.assertLess(time_taken, 1, u'Large    file    transfer    took    too    long')
