def test_grant_type_password(self):
        # Build ACL
        InitCommand.build_acl()

        # Create client
        client = InitCommand.save_client('Test', 'Py Test Client', 'test')

        # Create user
        user = UserCommand.save_user('admin', 'Admin', 'adminpwd', '[email protected]', 'admin')
        username = user.username
        email = user.email

        # Request token
        data = dict(grant_type='password', password='adminpwd', username=user.username, scope='test', client_id=client.client_id, client_secret=client.client_secret)
        response = self.app.get('/api/v1/auth?%s' % url_encode(data))
        token_data = self.check_api_response(response)
        assert token_data['token_type'] == 'Bearer'
        assert token_data['scope'] == 'test'

        # Get user info
        data = dict(access_token=token_data['access_token'])
        response = self.app.get('/api/v1/account?%s' % url_encode(data))
        user_data = self.check_api_response(response)
        assert user_data['account']['email'] == email
        assert user_data['account']['username'] == username

    def test_get(self):
        user_access_token = self.get_access_token_for_role('user')
        admin_access_token = self.get_access_token_for_role('admin')

        device_1 = model.zwave.device.Device(zway_id=5, name='test_device', description='Test device 1')
        db.session.add(device_1)
        device_1 = model.zwave.device.Device.query.filter_by(zway_id=5).first()
        device_2 = model.zwave.device.Device(zway_id=3, name='test_device_2', description='Test device 2')
        db.session.add(device_2)
        device_2 = model.zwave.device.Device.query.filter_by(zway_id=3).first()

        user_access_token.user.devices = [device_1]

        db.session.commit()

        # Get device info
        data = dict(access_token=user_access_token.access_token)
        response = self.app.get('/api/v1/zwave/devices/%d?%s' % (device_1.id, url_encode(data)))
        devices_data = self.check_api_response(response)
        assert devices_data['device']['zway_id'] == 5
        assert devices_data['device']['name'] == 'test_device'
        assert devices_data['device']['description'] == 'Test device 1'

        response = self.app.get('/api/v1/zwave/devices/%d?%s' % (device_2.id, url_encode(data)))
        self.check_api_response(response, 404)

        data = dict(access_token=admin_access_token.access_token)
        response = self.app.get('/api/v1/zwave/devices/%d?%s' % (device_2.id, url_encode(data)))
        devices_data = self.check_api_response(response)
        assert devices_data['device']['zway_id'] == 3
        assert devices_data['device']['name'] == 'test_device_2'
        assert devices_data['device']['description'] == 'Test device 2'

    def test_delete(self):
        admin_access_token = self.get_access_token_for_role('admin')
        other_access_token = self.get_access_token_for_role('user')

        # Create a second token to delete
        expire_time = datetime.now().replace(microsecond=0)
        token = model.auth.bearer_token.BearerToken(client=self.client, user=admin_access_token.user, token_type='bearer', access_token='Xqd8px7X6OO2gqc0vhlmJd1oUYaj2X', refresh_token='iRwV7aDH5VCwsvIZvJGWAui9O1wiP1', expires=expire_time, remote_address=None, user_agent='', _scopes='zway')
        db.session.add(token)
        token = model.auth.bearer_token.BearerToken.query.filter_by(user=admin_access_token.user, access_token='Xqd8px7X6OO2gqc0vhlmJd1oUYaj2X').first()

        db.session.commit()

        # Check that there are two sessions
        data = dict(access_token=admin_access_token.access_token)
        response = self.app.get('/api/v1/account/sessions?%s' % url_encode(data))
        token_data = self.check_api_response(response)
        assert len(token_data['sessions']) == 2

        # Delete session
        data = dict(access_token=admin_access_token.access_token)
        response = self.app.delete('/api/v1/account/sessions/%s?%s' % (token.id, url_encode(data)))
        token_data = self.check_api_response(response)
        assert token_data['session']['user']['username'] == admin_access_token.user.username

        # Get sessions info
        data = dict(access_token=admin_access_token.access_token)
        response = self.app.get('/api/v1/account/sessions?%s' % url_encode(data))
        token_data = self.check_api_response(response)
        assert len(token_data['sessions']) == 1

        # Try to delete session of another user
        data = dict(access_token=admin_access_token.access_token)
        response = self.app.delete('/api/v1/account/sessions/%s?%s' % (other_access_token.id, url_encode(data)))
        token_data = self.check_api_response(response, 404)

    def test_put_password(self):
        admin_access_token = self.get_access_token_for_role('admin')

        # Update password
        data = dict(access_token=admin_access_token.access_token)
        post_data = dict(old_password='pwd', new_password='newpwd', password_repeat='newpwd', options='password')
        response = self.app.put('/api/v1/account?%s' % url_encode(data), data=json.dumps(post_data), content_type='application/json')
        user_data = self.check_api_response(response)
        assert user_data['account']['email'] == admin_access_token.user.email
        assert user_data['account']['username'] == admin_access_token.user.username
        assert user_data['account']['fullname'] == admin_access_token.user.fullname

        user = model.auth.user.User.query.first()
        assert user.check_password('newpwd')

        # Try to update with wrong old password
        data = dict(access_token=admin_access_token.access_token)
        post_data = dict(old_password='wrongpwd', new_password='newpwd', password_repeat='newpwd', options='password')
        response = self.app.put('/api/v1/account?%s' % url_encode(data), data=json.dumps(post_data), content_type='application/json')
        user_data = self.check_api_response(response, 409)

        user = model.auth.user.User.query.first()
        assert user.check_password('newpwd')

        # Try to update with wrong repeated password
        data = dict(access_token=admin_access_token.access_token)
        post_data = dict(old_password='newpwd', new_password='anothernewpwd', password_repeat='wrongnewpwd', options='password')
        response = self.app.put('/api/v1/account?%s' % url_encode(data), data=json.dumps(post_data), content_type='application/json')
        user_data = self.check_api_response(response, 409)

        user = model.auth.user.User.query.first()
        assert user.check_password('newpwd')

def test_url_encoding():
    strict_eq(urls.url_encode({"foo": "bar 45"}), "foo=bar+45")
    d = {"foo": 1, "bar": 23, "blah": u"Hänsel"}
    strict_eq(urls.url_encode(d, sort=True), "bar=23&blah=H%C3%A4nsel&foo=1")
    strict_eq(
        urls.url_encode(d, sort=True, separator=u";"), "bar=23;blah=H%C3%A4nsel;foo=1"
    )

    def test_get_authorize(self):
        rv = self.client.get('/oauth/authorize')
        assert 'Missing+client_id+parameter' in rv.location

        client = self.client
        rv = client.get('/oauth/authorize?client_id=ios&response_type=code')
        assert rv.status_code == 200

        user = self.login()

        rv = client.get('/oauth/authorize?client_id=ios&response_type=code')
        assert rv.status_code == 200
        assert to_bytes(user.username) in rv.data

        oauth_client = OAuthClient.query.first()

        rv = client.get('/oauth/authorize?%s' % url_encode({
            'client_id': oauth_client.client_id,
            'response_type': 'code',
            'scope': 'user',
        }))
        assert b'user:email' in rv.data
        assert b'user:write' in rv.data

        rv = client.get('/oauth/authorize?%s' % url_encode({
            'client_id': oauth_client.client_id,
            'response_type': 'code',
            'scope': 'user:email',
        }))
        assert b'user:email' in rv.data
        assert b'user:write' not in rv.data

    def test_get_redirect_url_with_query_string(self, url, qs):
        instance = core.RedirectView(url=url, query_string=True)

        result = url_parse(url).replace(query=url_encode(qs)).to_url()

        with patch.object(core, 'request') as m:
            m.environ = {'QUERY_STRING': url_encode(qs)}

            assert instance.get_redirect_url() == result

 def test_quoting(self):
     self.assert_strict_equal(urls.url_quote(u'\xf6\xe4\xfc'), '%C3%B6%C3%A4%C3%BC')
     self.assert_strict_equal(urls.url_unquote(urls.url_quote(u'#%="\xf6')), u'#%="\xf6')
     self.assert_strict_equal(urls.url_quote_plus('foo bar'), 'foo+bar')
     self.assert_strict_equal(urls.url_unquote_plus('foo+bar'), u'foo bar')
     self.assert_strict_equal(urls.url_encode({b'a': None, b'b': b'foo bar'}), 'b=foo+bar')
     self.assert_strict_equal(urls.url_encode({u'a': None, u'b': u'foo bar'}), 'b=foo+bar')
     self.assert_strict_equal(urls.url_fix(u'http://de.wikipedia.org/wiki/Elf (Begriffsklärung)'),
            'http://de.wikipedia.org/wiki/Elf%20(Begriffskl%C3%A4rung)')

def test_quoting():
    strict_eq(urls.url_quote(u'\xf6\xe4\xfc'), '%C3%B6%C3%A4%C3%BC')
    strict_eq(urls.url_unquote(urls.url_quote(u'#%="\xf6')), u'#%="\xf6')
    strict_eq(urls.url_quote_plus('foo bar'), 'foo+bar')
    strict_eq(urls.url_unquote_plus('foo+bar'), u'foo bar')
    strict_eq(urls.url_quote_plus('foo+bar'), 'foo%2Bbar')
    strict_eq(urls.url_unquote_plus('foo%2Bbar'), u'foo+bar')
    strict_eq(urls.url_encode({b'a': None, b'b': b'foo bar'}), 'b=foo+bar')
    strict_eq(urls.url_encode({u'a': None, u'b': u'foo bar'}), 'b=foo+bar')
    strict_eq(urls.url_fix(u'http://de.wikipedia.org/wiki/Elf (Begriffsklärung)'),
           'http://de.wikipedia.org/wiki/Elf%20(Begriffskl%C3%A4rung)')
    strict_eq(urls.url_quote_plus(42), '42')
    strict_eq(urls.url_quote(b'\xff'), '%FF')

def with_qs(url, **args):
    """Updates query string part from the ``url``.  Parameters to update
    are given by keywords.

    """
    try:
        pos = url.index('?')
    except ValueError:
        return url + '?' + url_encode(args)
    pos += 1
    query = url_decode(url[pos:], cls=dict)
    query.update(args)
    return url[:pos] + url_encode(query)

 def url_for_page(self, pagenum):
     current_args = flask.request.args
     args_dict = current_args.to_dict()
     if pagenum != 1:
         args_dict['page'] = pagenum
     elif 'page' in args_dict:
         del args_dict['page']
         
     encoded = url_encode(args_dict)
     if encoded:
         url = "".join([flask.request.path, "?",url_encode(args_dict)])
         return url
     else:
         return flask.request.path

def test_sorted_url_encode():
    strict_eq(
        urls.url_encode(
            {u"a": 42, u"b": 23, 1: 1, 2: 2}, sort=True, key=lambda i: text_type(i[0])
        ),
        "1=1&2=2&a=42&b=23",
    )
    strict_eq(
        urls.url_encode(
            {u"A": 1, u"a": 2, u"B": 3, "b": 4},
            sort=True,
            key=lambda x: x[0].lower() + x[0],
        ),
        "A=1&a=2&B=3&b=4",
    )

def test_quoting():
    strict_eq(urls.url_quote(u"\xf6\xe4\xfc"), "%C3%B6%C3%A4%C3%BC")
    strict_eq(urls.url_unquote(urls.url_quote(u'#%="\xf6')), u'#%="\xf6')
    strict_eq(urls.url_quote_plus("foo bar"), "foo+bar")
    strict_eq(urls.url_unquote_plus("foo+bar"), u"foo bar")
    strict_eq(urls.url_quote_plus("foo+bar"), "foo%2Bbar")
    strict_eq(urls.url_unquote_plus("foo%2Bbar"), u"foo+bar")
    strict_eq(urls.url_encode({b"a": None, b"b": b"foo bar"}), "b=foo+bar")
    strict_eq(urls.url_encode({u"a": None, u"b": u"foo bar"}), "b=foo+bar")
    strict_eq(
        urls.url_fix(u"http://de.wikipedia.org/wiki/Elf (Begriffsklärung)"),
        "http://de.wikipedia.org/wiki/Elf%20(Begriffskl%C3%A4rung)",
    )
    strict_eq(urls.url_quote_plus(42), "42")
    strict_eq(urls.url_quote(b"\xff"), "%FF")

    def _redirect_to_record(cls, model, res_id, access_token=None, **kwargs):
        """ If the current user doesn't have access to the document, but provided
        a valid access token, redirect him to the front-end view.
        If the partner_id and hash parameters are given, add those parameters to the redirect url
        to authentify the recipient in the chatter, if any.

        :param model: the model name of the record that will be visualized
        :param res_id: the id of the record
        :param access_token: token that gives access to the record
            bypassing the rights and rules restriction of the user.
        :param kwargs: Typically, it can receive a partner_id and a hash (sign_token).
            If so, those two parameters are used to authentify the recipient in the chatter, if any.
        :return:
        """
        if issubclass(type(request.env[model]), request.env.registry['portal.mixin']):
            uid = request.session.uid or request.env.ref('base.public_user').id
            record_sudo = request.env[model].sudo().browse(res_id).exists()
            try:
                record_sudo.sudo(uid).check_access_rights('read')
                record_sudo.sudo(uid).check_access_rule('read')
            except AccessError:
                if record_sudo.access_token and access_token and consteq(record_sudo.access_token, access_token):
                    record_action = record_sudo.with_context(force_website=True).get_access_action()
                    if record_action['type'] == 'ir.actions.act_url':
                        pid = kwargs.get('pid')
                        hash = kwargs.get('hash')
                        url = record_action['url']
                        if pid and hash:
                            url = urls.url_parse(url)
                            url_params = url.decode_query()
                            url_params.update([("pid", pid), ("hash", hash)])
                            url = url.replace(query=urls.url_encode(url_params)).to_url()
                        return werkzeug.utils.redirect(url)
        return super(MailController, cls)._redirect_to_record(model, res_id, access_token=access_token)

def login_url(login_view, next_url=None, next_field='next'):
    '''
    Creates a URL for redirecting to a login page. If only `login_view` is
    provided, this will just return the URL for it. If `next_url` is provided,
    however, this will append a ``next=URL`` parameter to the query string
    so that the login view can redirect back to that URL.

    :param login_view: The name of the login view. (Alternately, the actual
                       URL to the login view.)
    :type login_view: str
    :param next_url: The URL to give the login view for redirection.
    :type next_url: str
    :param next_field: What field to store the next URL in. (It defaults to
                       ``next``.)
    :type next_field: str
    '''
    if login_view.startswith(('https://', 'http://', '/')):
        base = login_view
    else:
        base = url_for(login_view)

    if next_url is None:
        return base

    parts = list(urlparse(base))
    md = url_decode(parts[4])
    md[next_field] = make_next_param(base, next_url)
    parts[4] = url_encode(md, sort=True)
    return urlunparse(parts)

    def get_environ(self):
        """返回内置环境。"""
        input_stream = self.input_stream
        content_length = self.content_length
        content_type = self.content_type

        if input_stream is not None:
            start_pos = input_stream.tell()
            input_stream.seek(0, 2)
            end_pos = input_stream.tell()
            input_stream.seek(start_pos)
            content_length = end_pos - start_pos
        elif content_type == "multipart/form-data":
            values = CombinedMultiDict([self.form, self.files])
            input_stream, content_length, boundary = stream_encode_multipart(values, charset=self.charset)
            content_type += '; boundary="%s"' % boundary
        elif content_type == "application/x-www-form-urlencoded":
            # py2v3 review
            values = url_encode(self.form, charset=self.charset)
            values = values.encode("ascii")
            content_length = len(values)
            input_stream = BytesIO(values)
        else:
            input_stream = _empty_stream

        result = {}
        if self.environ_base:
            result.update(self.environ_base)

        def _path_encode(x):
            return wsgi_encoding_dance(url_unquote(x, self.charset), self.charset)

        qs = wsgi_encoding_dance(self.query_string)

        result.update(
            {
                "REQUEST_METHOD": self.method,
                "SCRIPT_NAME": _path_encode(self.script_root),
                "PATH_INFO": _path_encode(self.path),
                "QUERY_STRING": qs,
                "SERVER_NAME": self.server_name,
                "SERVER_PORT": str(self.server_port),
                "HTTP_HOST": self.host,
                "SERVER_PROTOCOL": self.server_protocol,
                "CONTENT_TYPE": content_type or "",
                "CONTENT_LENGTH": str(content_length or "0"),
                "wsgi.version": self.wsgi_version,
                "wsgi.url_scheme": self.url_scheme,
                "wsgi.input": input_stream,
                "wsgi.errors": self.errors_stream,
                "wsgi.multithread": self.multithread,
                "wsgi.multiprocess": self.multiprocess,
                "wsgi.run_once": self.run_once,
            }
        )
        for key, value in self.headers.to_wsgi_list():
            result["HTTP_%s" % key.upper().replace("-", "_")] = value
        if self.environ_overrides:
            result.update(self.environ_overrides)
        return result

 def ccavenue_form_generate_values(self, values):
     self.ensure_one()
     base_url = self.env['ir.config_parameter'].get_param('web.base.url')
     ccavenue_values = dict(access_code=self.ccavenue_access_code,
                            merchant_id=self.ccavenue_merchant_id,
                            order_id=values.get('reference'),
                            currency=values.get('currency').name,
                            amount=values.get('amount'),
                            redirect_url=urls.url_join(base_url, CCAvenueController._return_url),
                            cancel_url=urls.url_join(base_url, CCAvenueController._cancel_url),
                            language='EN',
                            customer_identifier=values.get('partner_email'),
                            delivery_name=values.get('partner_name'),
                            delivery_address=values.get('partner_address'),
                            delivery_city=values.get('partner_city'),
                            delivery_state=values.get('partner_state').name,
                            delivery_zip=values.get('partner_zip'),
                            delivery_country=values.get('partner_country').name,
                            delivery_tel=values.get('partner_phone'),
                            billing_name=values.get('billing_partner_name'),
                            billing_address=values.get('billing_partner_address'),
                            billing_city=values.get('billing_partner_city'),
                            billing_state=values.get('billing_partner_state').name,
                            billing_zip=values.get('billing_partner_zip'),
                            billing_country=values.get('billing_partner_country').name,
                            billing_tel=values.get('billing_partner_phone'),
                            billing_email=values.get('billing_partner_email'),
                            )
     ccavenue_values['encRequest'] = self._ccavenue_encrypt_text(urls.url_encode(ccavenue_values))
     return ccavenue_values

def yahoo_handle_oauth2_response(args):
  access_args = {
    'code': args.get('code'),
    'client_id': yahoo.consumer_key,
    'client_secret': yahoo.consumer_secret,
    'redirect_uri': flask.session.get('%s_oauthredir' % yahoo.name),
    'state': args.get('state'),
  }
  access_args.update(yahoo.access_token_params)
  auth_header = 'Basic %s' % base64.b64encode(
    ('%s:%s' % (yahoo.consumer_key, yahoo.consumer_secret)).encode('latin1')
  ).strip().decode('latin1')
  response, content = yahoo.http_request(
    yahoo.expand_url(yahoo.access_token_url),
    method=yahoo.access_token_method,
    data=urls.url_encode(access_args),
    headers={
      'Authorization': auth_header,
      'User-Agent': config.USER_AGENT,
      'Content-Type': 'application/x-www-form-urlencoded',
    },
  )
  data = client.parse_response(response, content)
  if response.code not in (200, 201):
    raise client.OAuthException(
      'Invalid response from %s' % yahoo.name,
      type='invalid_response', data=data,
    )
  return data

def login(request):
  from kay.auth import login

  if settings.AUTH_POST_LOGIN_SUCCESS_DEFAULT_URL:
  	next = unquote_plus(request.values.get("next", settings.AUTH_POST_LOGIN_SUCCESS_DEFAULT_URL))
  else:
  	next = unquote_plus(request.values.get("next", "/"))
  owned_domain_hack = request.values.get("owned_domain_hack")
  message = ""
  form = LoginForm()
  if request.method == "POST":
    if form.validate(request.form):
      result = login(request, user_name=form.data['user_name'],
                              password=form.data['password'])
      if result:
        if owned_domain_hack == 'True':
          original_host_url = unquote_plus(
            request.values.get("original_host_url"))
          url = original_host_url[:-1] + url_for("auth/post_session")
          url += '?' + url_encode({'session_id': result.key().name(),
                                   'next': next})
          return redirect(url)
        else:
          return redirect(next)
      else:
        message = _("Failed to login successfully with those credentials, try another or click the 'Forgot Password' link below.")
  return render_to_response("auth/loginform.html",
                            {"form": form.as_widget(),
                             "message": message})

    def _get_share_url(self, redirect=False, signup_partner=False, pid=None):
        """
        Build the url of the record  that will be sent by mail and adds additional parameters such as
        access_token to bypass the recipient's rights,
        signup_partner to allows the user to create easily an account,
        hash token to allow the user to be authenticated in the chatter of the record portal view, if applicable
        :param redirect : Send the redirect url instead of the direct portal share url
        :param signup_partner: allows the user to create an account with pre-filled fields.
        :param pid: = partner_id - when given, a hash is generated to allow the user to be authenticated
            in the portal chatter, if any in the target page,
            if the user is redirected to the portal instead of the backend.
        :return: the url of the record with access parameters, if any.
        """
        self.ensure_one()
        params = {
            'model': self._name,
            'res_id': self.id,
        }
        if hasattr(self, 'access_token'):
            params['access_token'] = self._portal_ensure_token()
        if pid:
            params['pid'] = pid
            params['hash'] = self._sign_token(pid)
        if signup_partner and hasattr(self, 'partner_id') and self.partner_id:
            params.update(self.partner_id.signup_get_auth_param()[self.partner_id.id])

        return '%s?%s' % ('/mail/view' if redirect else self.access_url, url_encode(params))