<commit_before># -*- coding: utf-8 -*-

import unittest

from flask import Flask, Blueprint, jsonify
from flask_restful import Api, Resource
from flask import g

from drift.tests import DriftTestCase
from drift.core.extensions.jwt import jwtsetup, verify_token, jwt_not_required, current_user, check_jwt_authorization

app = Flask(__name__)
app.testing = True


bp = Blueprint("jwt_api", __name__)
api = Api(bp)


# Endpoints closed from public access
class APIClosed(Resource):

    def get(self):
        ret = {
            "message2": "hi there",
            "current_user": dict(current_user) if current_user else None,
        }
        return ret


api.add_resource(APIClosed, '/apiclosed', endpoint="apiclosed")


@bp.route("/closed")
def closedfunc():
    return jsonify({"message": "this endpoint is closed"})


# Endpoints open for public access
class APIOpen(Resource):

    no_jwt_check = ["GET"]  # Only GET is public, DELETE is closed

    def get(self):
        ret = {
            "message2": "hi there, this is open",
            "current_user": dict(current_user) if current_user else None,
        }
        return ret

    def delete(self):
        ret = {
            "message2": "aawwww, deleting me?",
            "current_user": dict(current_user) if current_user else None,
        }

        return ret


api.add_resource(APIOpen, '/apiopen', endpoint="apiopen")


@jwt_not_required
@bp.route("/open", methods=["GET", "POST"])
def openfunc():
    return jsonify(
        {
            "message": "this is wide open",
            "current_user": dict(current_user) if current_user else None,
        }
    )


@jwt_not_required
@app.route('/api', methods=['GET'])
def this_func():
    """This is a function. It does nothing."""
    return jsonify({'result': ''})


@jwt_not_required
@app.route('/api/help', methods=['GET'])
def help():
    """Print available functions."""
    func_list = {}
    for rule in app.url_map.iter_rules():
        if rule.endpoint != 'static':
            func_list[rule.rule] = app.view_functions[rule.endpoint].__doc__
    return jsonify(func_list)


private_key = '''
-----BEGIN RSA PRIVATE KEY-----
MIIBygIBAAJhAOOEkKLzpVY5zNbn2zZlz/JlRe383fdnsuy2mOThXpJc9Tq+GuI+
PJJXsNa5wuPBy32r46/N8voe/zUG4qYrrRCRyjmV0yu4kZeNPSdO4uM4K98P1obr
UaYrik9cpwnu8QIDAQABAmA+BSAMW5CBfcYZ+yAlpwFVmUfDxT+YtpruriPlmI3Y
JiDvP21CqSaH2gGptv+qaGQVq8E1xcxv9jT1qK3b7wm7+xoxTYyU0XqZC3K+lGeW
5L+77H59RwQznG21FvjtRgECMQDzihOiirv8LI2S7yg11/DjC4c4lIzupjnhX2ZH
weaLJcjGogS/labJO3b2Q8RUimECMQDvKKKl1KiAPNvuylcrDw6+yXOBDw+qcwiP
rKysATJ2iCsOgnLC//Rk3+SN3R2+TpECMGjAglOOsu7zxu1levk16cHu6nm2w6u+
yfSbkSXaTCyb0vFFLR+u4e96aV/hpCfs4QIwd/I0aOFYRUDAuWmoAEOEDLHyiSbp
n34kLBLZY0cSbRpsJdHNBvniM/mKoo/ki/7RAjEAtpt6ixFoEP3w/2VLh5cut61x
E74vGa3+G/KdGO94ZnI9uxySb/czhnhvOGkpd9/p
-----END RSA PRIVATE KEY-----
'''

public_test_key = "ssh-rsa A<KEY>" \
    "zW59s2Zc/yZUXt/N33Z7Lstpjk4V6SXPU6vhriPjySV7DWucLjwct9q+Ovz" \
    "fL6Hv81BuKmK60Qkco5ldMruJGXjT0nTuLjOCvfD9aG61GmK4pPXKcJ7vE=" \
    " <EMAIL>"


class JWTCase(DriftTestCase):

    def create_app(self):
        app = super(JWTCase, self).create_app()
        app.register_blueprint(bp)
        jwtsetup(app)

        # check for a valid JWT/JTI access token in the request header
        # and populate current_user
        @app.before_request
        def jwt_check_hook(*args, **kw):
            check_jwt_authorization()

        # Deployables implement the authenticate() callback function
        # as well as providing a private key for signing tokens. Here
        # we do this as Drift is not a deployable by itself.
        import drift.core.extensions.jwt as jwtsetupmodule
        jwtsetupmodule.authenticate = self.authenticate
        jwtsetupmodule.private_key = self.private_key

        return app

    custom_payload = {
        "user_id": 123,
        "identity_id": 555,
        "player_id": 10050,
        "player_name": "A player name",
        "roles": [],
    }

    def authenticate(self, username, password):
        self.custom_payload['user_name'] = username
        return self.custom_payload

    def private_key(self):
        return private_key

    def test_access_control(self):
        g.driftenv = {}
        g.driftenv["tier_name"] = 'unittest_tier'
        g.driftenv["name"] = 'unittest_tenant'

        # Make myself a trusted issuer
        issuer = {
            'iss': self.app.config["name"],
            'pub_rsa': public_test_key,
        }
        self.app.config["jwt_trusted_issuers"] = [issuer]
<commit_msg>Fixing unit test for jwt.<commit_after># -*- coding: utf-8 -*-

import unittest

from flask import Flask, Blueprint, jsonify
from flask_restful import Api, Resource
from flask import g

from drift.tests import DriftTestCase
from drift.core.extensions.jwt import jwtsetup, verify_token, jwt_not_required, current_user, check_jwt_authorization

app = Flask(__name__)
app.testing = True


bp = Blueprint("jwt_api", __name__)
api = Api(bp)


# Endpoints closed from public access
class APIClosed(Resource):

    def get(self):
        ret = {
            "message2": "hi there",
            "current_user": dict(current_user) if current_user else None,
        }
        return ret


api.add_resource(APIClosed, '/apiclosed', endpoint="apiclosed")


@bp.route("/closed")
def closedfunc():
    return jsonify({"message": "this endpoint is closed"})


# Endpoints open for public access
class APIOpen(Resource):

    no_jwt_check = ["GET"]  # Only GET is public, DELETE is closed

    def get(self):
        ret = {
            "message2": "hi there, this is open",
            "current_user": dict(current_user) if current_user else None,
        }
        return ret

    def delete(self):
        ret = {
            "message2": "aawwww, deleting me?",
            "current_user": dict(current_user) if current_user else None,
        }

        return ret


api.add_resource(APIOpen, '/apiopen', endpoint="apiopen")


@jwt_not_required
@bp.route("/open", methods=["GET", "POST"])
def openfunc():
    return jsonify(
        {
            "message": "this is wide open",
            "current_user": dict(current_user) if current_user else None,
        }
    )


@jwt_not_required
@app.route('/api', methods=['GET'])
def this_func():
    """This is a function. It does nothing."""
    return jsonify({'result': ''})


@jwt_not_required
@app.route('/api/help', methods=['GET'])
def help():
    """Print available functions."""
    func_list = {}
    for rule in app.url_map.iter_rules():
        if rule.endpoint != 'static':
            func_list[rule.rule] = app.view_functions[rule.endpoint].__doc__
    return jsonify(func_list)


private_key = '''
-----<KEY>
'''

public_test_key = "ssh-rsa <KEY> <KEY> \
    "<KEY> \
    " <EMAIL>"


class JWTCase(DriftTestCase):

    def create_app(self):
        app = super(JWTCase, self).create_app()
        app.register_blueprint(bp)
        jwtsetup(app)

        # check for a valid JWT/JTI access token in the request header
        # and populate current_user
        @app.before_request
        def jwt_check_hook(*args, **kw):
            check_jwt_authorization()

        # Deployables implement the authenticate() callback function
        # as well as providing a private key for signing tokens. Here
        # we do this as Drift is not a deployable by itself.
        app.config['private_key'] = private_key
        import drift.core.extensions.jwt as jwtsetupmodule
        jwtsetupmodule.authenticate = self.authenticate

        return app

    custom_payload = {
        "user_id": 123,
        "identity_id": 555,
        "player_id": 10050,
        "player_name": "A player name",
        "roles": [],
    }

    def authenticate(self, username, password):
        self.custom_payload['user_name'] = username
        return self.custom_payload

    def private_key(self):
        return private_key

    def test_access_control(self):
        g.driftenv = {}
        g.driftenv["tier_name"] = 'unittest_tier'
        g.driftenv["name"] = 'unittest_tenant'

        # Make myself a trusted issuer
        issuer = {
            'iss': self.app.config["name"],
            'pub_rsa': public_test_key,
        }
        self.app.config["jwt_trusted_issuers"] = [issuer]

{'fraction_non_alphanumeric': 0.11329451722464823, 'fraction_numerical': 0.017467248908296942, 'mean_word_length': 2.9544364508393284, 'pattern_counts': {'":': 28, '<': 10, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3801510', 'n_tokens_mistral': 3008, 'n_tokens_neox': 2868, 'n_words': 611}