# Path: invenio_oauth2server/models.py
# class Token(db.Model):
#     """A bearer token is the final token that can be used by the client."""
# 
#     __tablename__ = 'oauth2server_token'
#     __table_args__ = (
#         Index('ix_oauth2server_token_access_token',
#               'access_token',
#               unique=True,
#               mysql_length=255),
#         Index('ix_oauth2server_token_refresh_token',
#               'refresh_token',
#               unique=True,
#               mysql_length=255),
#     )
# 
#     id = db.Column(db.Integer, primary_key=True, autoincrement=True)
#     """Object ID."""
# 
#     client_id = db.Column(
#         db.String(255),
#         db.ForeignKey(Client.client_id, ondelete='CASCADE'),
#         nullable=False, index=True
#     )
#     """Foreign key to client application."""
# 
#     client = db.relationship(
#         'Client',
#         backref=db.backref(
#             'oauth2tokens',
#             cascade="all, delete-orphan"
#         ))
#     """SQLAlchemy relationship to client application."""
# 
#     user_id = db.Column(
#         db.Integer, db.ForeignKey(User.id, ondelete='CASCADE'), nullable=True,
#         index=True
#     )
#     """Foreign key to user."""
# 
#     user = db.relationship(
#         User,
#         backref=db.backref(
#             "oauth2tokens",
#             cascade="all, delete-orphan",
#         )
#     )
#     """SQLAlchemy relationship to user."""
# 
#     token_type = db.Column(db.String(255), default='bearer')
#     """Token type - only bearer is supported at the moment."""
# 
#     access_token = db.Column(
#         EncryptedType(
#             type_in=db.String(255),
#             key=secret_key,
#         ),
#     )
# 
#     refresh_token = db.Column(
#         EncryptedType(
#             type_in=db.String(255),
#             key=secret_key,
#             engine=NoneAesEngine,
#         ),
#         nullable=True,
#     )
# 
#     expires = db.Column(db.DateTime, nullable=True)
# 
#     _scopes = db.Column(db.Text)
# 
#     is_personal = db.Column(db.Boolean(name='is_personal'), default=False)
#     """Personal accesss token."""
# 
#     is_internal = db.Column(db.Boolean(name='is_internal'), default=False)
#     """Determines if token is an internally generated token."""
# 
#     @property
#     def scopes(self):
#         """Return all scopes.
# 
#         :returns: A list of scopes.
#         """
#         if self._scopes:
#             return self._scopes.split()
#         return []
# 
#     @scopes.setter
#     def scopes(self, scopes):
#         """Set scopes.
# 
#         :param scopes: The list of scopes.
#         """
#         validate_scopes(scopes)
#         self._scopes = " ".join(set(scopes)) if scopes else ""
# 
#     def get_visible_scopes(self):
#         """Get list of non-internal scopes for token.
# 
#         :returns: A list of scopes.
#         """
#         return [k for k, s in current_oauth2server.scope_choices()
#                 if k in self.scopes]
# 
#     @classmethod
#     def create_personal(cls, name, user_id, scopes=None, is_internal=False):
#         """Create a personal access token.
# 
#         A token that is bound to a specific user and which doesn't expire, i.e.
#         similar to the concept of an API key.
# 
#         :param name: Client name.
#         :param user_id: User ID.
#         :param scopes: The list of permitted scopes. (Default: ``None``)
#         :param is_internal: If ``True`` it's a internal access token.
#              (Default: ``False``)
#         :returns: A new access token.
#         """
#         with db.session.begin_nested():
#             scopes = " ".join(scopes) if scopes else ""
# 
#             c = Client(
#                 name=name,
#                 user_id=user_id,
#                 is_internal=True,
#                 is_confidential=False,
#                 _default_scopes=scopes
#             )
#             c.gen_salt()
# 
#             t = Token(
#                 client_id=c.client_id,
#                 user_id=user_id,
#                 access_token=gen_salt(
#                     current_app.config.get(
#                         'OAUTH2SERVER_TOKEN_PERSONAL_SALT_LEN')
#                 ),
#                 expires=None,
#                 _scopes=scopes,
#                 is_personal=True,
#                 is_internal=is_internal,
#             )
# 
#             db.session.add(c)
#             db.session.add(t)
# 
#         return t
# 
# Path: invenio_oauth2server/utils.py
# def rebuild_access_tokens(old_key):
#     """Rebuild the access_token field when the SECRET_KEY is changed.
# 
#     Needed to fix the access tokens used in the REST API calls.
# 
#     :param old_key: the old SECRET_KEY.
#     """
#     current_app.logger.info('rebuilding Token.access_token...')
#     rebuild_encrypted_properties(old_key, Token,
#                                  ['access_token', 'refresh_token'])


import sys
import pytest
from invenio_db import db
from invenio_oauth2server.models import Token
from invenio_oauth2server.utils import rebuild_access_tokens
# -*- coding: utf-8 -*-
#
# This file is part of Invenio.
# Copyright (C) 2017-2018 CERN.
#
# Invenio is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.

"""Test case for rebuilding access tokens."""





def test_rebuilding_access_tokens(models_fixture):
    """Test rebuilding access tokens with random new SECRET_KEY."""
    app = models_fixture
    with app.app_context():
        old_secret_key = app.secret_key
        tokens_before = Token.query.order_by(Token.id).all()

        # Changing application SECRET_KEY
        app.secret_key = "NEW_SECRET_KEY"
        db.session.expunge_all()

        # Asserting the decoding error occurs with the stale SECRET_KEY
        if sys.version_info[0] < 3:  # python 2

# Path: invenio_oauth2server/models.py
# class Token(db.Model):
#     """A bearer token is the final token that can be used by the client."""
# 
#     __tablename__ = 'oauth2server_token'
#     __table_args__ = (
#         Index('ix_oauth2server_token_access_token',
#               'access_token',
#               unique=True,
#               mysql_length=255),
#         Index('ix_oauth2server_token_refresh_token',
#               'refresh_token',
#               unique=True,
#               mysql_length=255),
#     )
# 
#     id = db.Column(db.Integer, primary_key=True, autoincrement=True)
#     """Object ID."""
# 
#     client_id = db.Column(
#         db.String(255),
#         db.ForeignKey(Client.client_id, ondelete='CASCADE'),
#         nullable=False, index=True
#     )
#     """Foreign key to client application."""
# 
#     client = db.relationship(
#         'Client',
#         backref=db.backref(
#             'oauth2tokens',
#             cascade="all, delete-orphan"
#         ))
#     """SQLAlchemy relationship to client application."""
# 
#     user_id = db.Column(
#         db.Integer, db.ForeignKey(User.id, ondelete='CASCADE'), nullable=True,
#         index=True
#     )
#     """Foreign key to user."""
# 
#     user = db.relationship(
#         User,
#         backref=db.backref(
#             "oauth2tokens",
#             cascade="all, delete-orphan",
#         )
#     )
#     """SQLAlchemy relationship to user."""
# 
#     token_type = db.Column(db.String(255), default='bearer')
#     """Token type - only bearer is supported at the moment."""
# 
#     access_token = db.Column(
#         EncryptedType(
#             type_in=db.String(255),
#             key=secret_key,
#         ),
#     )
# 
#     refresh_token = db.Column(
#         EncryptedType(
#             type_in=db.String(255),
#             key=secret_key,
#             engine=NoneAesEngine,
#         ),
#         nullable=True,
#     )
# 
#     expires = db.Column(db.DateTime, nullable=True)
# 
#     _scopes = db.Column(db.Text)
# 
#     is_personal = db.Column(db.Boolean(name='is_personal'), default=False)
#     """Personal accesss token."""
# 
#     is_internal = db.Column(db.Boolean(name='is_internal'), default=False)
#     """Determines if token is an internally generated token."""
# 
#     @property
#     def scopes(self):
#         """Return all scopes.
# 
#         :returns: A list of scopes.
#         """
#         if self._scopes:
#             return self._scopes.split()
#         return []
# 
#     @scopes.setter
#     def scopes(self, scopes):
#         """Set scopes.
# 
#         :param scopes: The list of scopes.
#         """
#         validate_scopes(scopes)
#         self._scopes = " ".join(set(scopes)) if scopes else ""
# 
#     def get_visible_scopes(self):
#         """Get list of non-internal scopes for token.
# 
#         :returns: A list of scopes.
#         """
#         return [k for k, s in current_oauth2server.scope_choices()
#                 if k in self.scopes]
# 
#     @classmethod
#     def create_personal(cls, name, user_id, scopes=None, is_internal=False):
#         """Create a personal access token.
# 
#         A token that is bound to a specific user and which doesn't expire, i.e.
#         similar to the concept of an API key.
# 
#         :param name: Client name.
#         :param user_id: User ID.
#         :param scopes: The list of permitted scopes. (Default: ``None``)
#         :param is_internal: If ``True`` it's a internal access token.
#              (Default: ``False``)
#         :returns: A new access token.
#         """
#         with db.session.begin_nested():
#             scopes = " ".join(scopes) if scopes else ""
# 
#             c = Client(
#                 name=name,
#                 user_id=user_id,
#                 is_internal=True,
#                 is_confidential=False,
#                 _default_scopes=scopes
#             )
#             c.gen_salt()
# 
#             t = Token(
#                 client_id=c.client_id,
#                 user_id=user_id,
#                 access_token=gen_salt(
#                     current_app.config.get(
#                         'OAUTH2SERVER_TOKEN_PERSONAL_SALT_LEN')
#                 ),
#                 expires=None,
#                 _scopes=scopes,
#                 is_personal=True,
#                 is_internal=is_internal,
#             )
# 
#             db.session.add(c)
#             db.session.add(t)
# 
#         return t
# 
# Path: invenio_oauth2server/utils.py
# def rebuild_access_tokens(old_key):
#     """Rebuild the access_token field when the SECRET_KEY is changed.
# 
#     Needed to fix the access tokens used in the REST API calls.
# 
#     :param old_key: the old SECRET_KEY.
#     """
#     current_app.logger.info('rebuilding Token.access_token...')
#     rebuild_encrypted_properties(old_key, Token,
#                                  ['access_token', 'refresh_token'])


# Path: tests/test_utils.py
import sys
import pytest
from invenio_db import db
from invenio_oauth2server.models import Token
from invenio_oauth2server.utils import rebuild_access_tokens

# -*- coding: utf-8 -*-
#
# This file is part of Invenio.
# Copyright (C) 2017-2018 CERN.
#
# Invenio is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.

"""Test case for rebuilding access tokens."""





def test_rebuilding_access_tokens(models_fixture):
    """Test rebuilding access tokens with random new SECRET_KEY."""
    app = models_fixture
    with app.app_context():
        old_secret_key = app.secret_key
        tokens_before = Token.query.order_by(Token.id).all()

        # Changing application SECRET_KEY
        app.secret_key = "NEW_SECRET_KEY"
        db.session.expunge_all()

        # Asserting the decoding error occurs with the stale SECRET_KEY
        if sys.version_info[0] < 3:  # python 2

            token = Token.query.filter(Token.id == tokens_before[0].id).one()