datestex/webflask/auth.py

# https://flask.palletsprojects.com/en/2.0.x/tutorial/views/
# --------------------------------------------------------------
import functools
import traceback
import os
from flask import (
    Blueprint, flash, g, redirect, render_template, request, session, url_for
)
from werkzeug.security import check_password_hash, generate_password_hash
import basic.program
import basic.constants as B
import basic.toolHandling
import model.user
from webflask.db import get_db

bp = Blueprint('auth', __name__, url_prefix='/auth')

@bp.route('/login', methods=('GET', 'POST'))
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        job = basic.program.Job("webflask", {})
        session['datest_job'] = job
        dbtype = job.conf[B.TOPIC_NODE_DB][B.ATTR_TYPE]
        userentity = model.user.User()
        dbi = basic.toolHandling.getDbTool(job, userentity, dbtype)
        error = None
        row = dbi.selectRows("user", job, "WHERE username = \'" + username + "\'")

        if row is None or len(row[B.DATA_NODE_DATA]) == 0:
            error = 'Incorrect username.'
        elif not check_password_hash(row[B.DATA_NODE_DATA][0]['password'], password):
            error = 'Incorrect password.'

        if error is None:
            session.clear()
            session['user_id'] = row[B.DATA_NODE_DATA][0]['id']
            return redirect(url_for('testcase.overview'))


        flash(error)

    return render_template('auth/login.html')

@bp.route('/register', methods=('GET', 'POST'))
def register():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        job = basic.program.Job("webflask", {})
        dbtype = job.conf[B.TOPIC_NODE_DB][B.ATTR_TYPE]
        dbi = basic.toolHandling.getDbTool(job, None, "mysql")
        # db = get_db()
        error = None

        if not username:
            error = 'Username is required.'
        elif not password:
            error = 'Password is required.'

        if error is None:
            try:
                sql = "INSERT INTO datest.user (username, password) "
                sql += "VALUES ( '"+username+"', '"+generate_password_hash(password)+"');"
                dbi.execStatement(job.conf[B.TOPIC_NODE_DB], sql )
                #db.commit()
            except Exception as e:
                error = str(e)
                # error = f"User {username} is already registered."
            else:
                return redirect(url_for("auth.login"))

        flash(error)

    return render_template('auth/register.html')

@bp.before_app_request
def load_logged_in_user():
    user_id = session.get('user_id')

    if user_id is None:
        g.user = None
    else:
        job = basic.program.Job("webflask", {})
        dbtype = job.conf[B.TOPIC_NODE_DB][B.ATTR_TYPE]
        userentity = basic.user.User(job)
        dbi = basic.toolHandling.getDbTool(job, userentity, dbtype)
        error = None
        sql = "WHERE id = " + str(user_id)
        row = dbi.selectRows("user", job, sql)
        if len(row[B.DATA_NODE_DATA]) > 0:
            g.user = row[B.DATA_NODE_DATA][0]
        else:
            g.user = None
        #    get_db().execute(
        #    'SELECT * FROM user WHERE id = ?', (user_id,)
        #).fetchone()

@bp.route('/logout')
def logout():
    session.clear()
    return redirect(url_for('index'))

def login_required(view):
    @functools.wraps(view)
    def wrapped_view(**kwargs):
        if g.user is None:
            return redirect(url_for('auth.login'))

        return view(**kwargs)

    return wrapped_view