'Test HTTP Digest auth'

def test_digest_auth(self):

    from calibre.srv.http_request import normalize_header_name
    from calibre.srv.utils import parse_http_dict
    r = router()
    with TestServer(r.dispatch) as server:
        r.auth_controller.log = server.log
        def test(conn, path, headers={}, status=httplib.OK, body='', request_body=''):
            conn.request(u'GET', path, request_body, headers)
            r = conn.getresponse()
            self.ae(r.status, status)
            self.ae(r.read(), body)
            return {normalize_header_name(k): v for (k, v) in r.getheaders()}
        conn = server.connect()
        test(conn, u'/open', body='open')
        auth = parse_http_dict(test(conn, u'/closed', status=httplib.UNAUTHORIZED)[u'WWW-Authenticate'].partition('    ')[2])
        nonce = auth[u'nonce']
        auth = parse_http_dict(test(conn, u'/closed', status=httplib.UNAUTHORIZED)[u'WWW-Authenticate'].partition('    ')[2])
        self.assertNotEqual(nonce, auth[u'nonce'], u'nonce    was    re-used')
        (self.ae(auth['realm'], bytes(REALM)), self.ae(auth['algorithm'], 'MD5'), self.ae(auth['qop'], 'auth'))
        self.assertNotIn(u'stale', auth)
        args = auth.copy()
        (args[u'un'], args[u'pw'], args[u'uri']) = (u'testuser', u'testpw', u'/closed')
        def ok_test(conn, dh, **args):
            args[u'body'] = args.get(u'body', 'closed')
            return test(conn, u'/closed', headers={u'Authorization': dh}, **args)
        ok_test(conn, digest(**args))
        ok_test(conn, digest(**args))
        warnings = []
        server.loop.log.warn = (lambda *args, **kwargs: warnings.append(u'    '.join(args)))
        (orig, r.auth_controller.max_age_seconds) = (r.auth_controller.max_age_seconds, (-1))
        auth = parse_http_dict(test(conn, u'/closed', headers={u'Authorization': digest(**args)}, status=httplib.UNAUTHORIZED)[u'WWW-Authenticate'].partition('    ')[2])
        self.assertIn(u'stale', auth)
        r.auth_controller.max_age_seconds = orig
        ok_test(conn, digest(**args))
        def fail_test(conn, modify, **kw):
            kw[u'body'] = kw.get(u'body', '')
            kw[u'status'] = kw.get(u'status', httplib.UNAUTHORIZED)
            args[u'modify'] = modify
            return test(conn, u'/closed', headers={u'Authorization': digest(**args)}, **kw)
        fail_test(conn, (lambda da: setattr(da, u'nonce', u'xyz')))
        fail_test(conn, (lambda da: setattr(da, u'nonce', (u'x' + da.nonce))))
        fail_test(conn, (lambda da: setattr(da, u'uri', u'/')))
        fail_test(conn, (lambda da: setattr(da, u'uri', u'/closed2')))
        fail_test(conn, (lambda da: setattr(da, u'uri', u'/closed/2')))
        fail_test(conn, (lambda da: setattr(da, u'pw', u'/')))
        fail_test(conn, (lambda da: setattr(da, u'username', u'/')))
        fail_test(conn, (lambda da: setattr(da, u'username', u'')))
        fail_test(conn, (lambda da: setattr(da, u'pw', u'')))
        fail_test(conn, (lambda da: (setattr(da, u'pw', u''), setattr(da, u'username', u''))))
        self.ae(urlopen(server).read(), 'closed')
        curl = find_executable(u'curl')
        if curl:
            def docurl(data, *args):
                cmd = (([curl] + list(args)) + [(u'http://localhost:%d/closed' % server.address[1])])
                p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=open(os.devnull, u'wb'))
                x = p.stdout.read()
                p.wait()
                self.ae(x, data)
            docurl('')
            docurl('', u'--digest', u'--user', u'xxxx:testpw')
            docurl('', u'--digest', u'--user', u'testuser:xtestpw')
            docurl('closed', u'--digest', u'--user', u'testuser:testpw')
