More than 1 year has passed since last update.

Flaskとdynamodb_localでDB認証とセッション管理を試してみる(テストくどめ、Selenium入り)

Last updated at 2022-01-30Posted at 2022-01-30

はじめに

PythonのWebアプリのフレームワークのFlaskとなります。
dynamodb_localを利用し、DB認証とセッション管理のサンプル実装となります。

各種バージョンは以下のとおりです。
Python 3.9.6
Flask 2.0.2
Flask-Login 0.5.0
Flask-Sessionstore 0.4.5
pynamodb 5.2.0
pytest 6.2.5
pytest-flask 1.2.0
selenium 4.1.0

ソースはgithubに登録しております。

サンプル概要

ファイル構成(github)

ファイル構成(github)は以下のとおりです。

プロジェクトルート
├─login_app
│  │─models
│  │  │─sessions.py
│  │  │─users.py
│  │
│  │─templates
│  │  │─login.html
│  │  │─top.html
│  │
│  │─views
│  │  │─__init__.py
│  │  │─common.py
│  │  │─user_auth.py
│  │
│  │─__init__.py
│  │─config.py
│  │─init_db.py
│  │─server.py
│  │─user_loader.py
│ 
├─tests
│  │─__init__.py
│  │─login_app_selenium_test.py
│  │─login_app_test.py
│  │─login_app_unit_test.py
│  │─test_common.py

ファイル構成(github未登録)

プロジェクトルート
├─login_app
│ │─dynamodb_local_latest
├─ chromedriver.exe

dynamodb_local_latestフォルダには
dynamodb_local_latestからダウンロードし解凍したファイルを格納してください。

dynamodb_local_latestの中身は以下のようなイメージとなります。

プロジェクトルート直下に
chromedriver
からダウンロードしたchromedriver.exeを配置してください。

chromedriver.exeのバージョンは、ご利用されているchromeのバージョンと同じ物を選択してください。

実際の動作イメージ

ルートにアクセスすると「Log In」リンクがあり、クリックするとログイン画面に遷移し、ログインに成功するとルートにリダイレクトされ、「Log In」リンクは表示されず、「Log Out」リンクが表示され、「Log Out」リンクをクリックするとログアウトし、ログイン画面にリダイレクトます。

あとは、ログインしているときにuser_idを指定し、ユーザ詳細(全然詳細じゃないですが)が見える画面を実装しております。
http://localhost:5000/users/detail/user_id_1のようにアクセスします。
ログインしていない場合はログイン画面にリダイレクトます。

dynamodb_localの起動方法

Java ランタイム環境 (JRE) 6.x 以降のバージョンが必要です。

dynamodb_local_latestに移動後し、カレントディレクトリがdynamodb_local_latestを前提としてたコマンド例とまります。

アプリ動作用のDB起動

java "-Djava.library.path=./DynamoDBLocal_lib" -jar DynamoDBLocal.jar -sharedDb
を実行します。

以下のように出力されてDBが起動します。

Initializing DynamoDB Local with the following configuration:
Port:   8000
InMemory:       false
DbPath: null
SharedDb:       true
shouldDelayTransientStatuses:   false
CorsParams:     *

テスト用のDB起動

テスト実行時に利用するDBをポート:9000で起動します。2つDBを起動しなくても問題ないのですが、config.pyで定義した設定が切り替わる事を確認できるように、との意図となります。

java "-Djava.library.path=./DynamoDBLocal_lib" -jar DynamoDBLocal.jar -sharedDb -port 9000
を実行します。

サンプルの説明

config.py

各種設定を含んだモジュールファイル
dummyやDUMMYを含む設定は、dynamodb_localでは利用しないので適当な値をセットしてください。
必要な設定はDYNAMODB_ENDPOINT_URL 、SESSION_DYNAMODB_ENDPOINT_URL 、SESSION_TYPE 、SESSION_DYNAMODB_TABLEとなります。

config.py

import os


class Config(object):
    DEBUG = True

    # AWS_ACCESS_KEY_ID ローカルなのでなんでもOK
    AWS_ACCESS_KEY_ID = "AWS_ACCESS_KEY_ID_DUMMY"
    # AWS_SECRET_ACCESS_KEY ローカルなのでなんでもOK
    AWS_SECRET_ACCESS_KEY = "AWS_SECRET_ACCESS_KEY_DUMMY"

    SECRET_KEY = "secret key dummy"
    USERNAME = "dummy_user"
    PASSWORD = "dummy_pass"

    # session情報をdynamodbに保存
    SESSION_TYPE = "dynamodb"
    SESSION_DYNAMODB_TABLE = "sessions"


class DevConfig(Config):
    # dynamodb_localのURL
    DYNAMODB_ENDPOINT_URL = "http://localhost:8000"

    # この設定ないとAWSに接続しにいく
    SESSION_DYNAMODB_ENDPOINT_URL = DYNAMODB_ENDPOINT_URL


class TestConfig(Config):
    # dynamodb_localのURL
    DYNAMODB_ENDPOINT_URL = "http://localhost:9000"

    # この設定ないとAWSに接続しにいく
    SESSION_DYNAMODB_ENDPOINT_URL = DYNAMODB_ENDPOINT_URL


class DevDbInitConfig(DevConfig):
    # sessionを利用しない
    SESSION_TYPE = ""


class TestDbInitConfig(TestConfig):
    # sessionを利用しない
    SESSION_TYPE = ""

必要な設定に指定する情報は以下のとおりです。

設定対象	説明
DYNAMODB_ENDPOINT_URL	dynamodb_localにアクセスするためのURL
SESSION_DYNAMODB_ENDPOINT_URL	dynamodb_local(session情報格納DB)にアクセスするためのURL
SESSION_TYPE	session情報格納の保存タイプで、null:利用しない、redis、 memcached、filesystem、mongodb、sqlalchemyが選択可能
SESSION_DYNAMODB_TABLE	session情報を格納するテーブル名

login_appのinit.py

__init__.py


import os
from flask import session
from flask import Flask
from flask_login import LoginManager
from flask_sessionstore import Session
from logging import getLogger, StreamHandler, DEBUG

logger = getLogger(__name__)
handler = StreamHandler()
handler.setLevel(DEBUG)
logger.setLevel(DEBUG)
logger.addHandler(handler)
logger.propagate = False


app = Flask(__name__, instance_relative_config=True)

# 各種設定の読み込み
try:
    run_flag = os.environ["RUN_FLAG"]
except KeyError:
    run_flag = "1"

config_name = "login_app.config.DevConfig"

"""
RuntimeError: The table sessions does not exist in DynamoDB for the requested region・・・
と怒られるのでmanage_db.pyの呼び出しの時はDevDbInitConfig or TestDbInitConfigにして
SESSION_TYPE = ""としセッション機能を無効にする。
"""
if run_flag == "1":
    config_name = "login_app.config.DevConfig"
elif run_flag == "2":
    config_name = "login_app.config.TestConfig"
elif run_flag == "3":
    config_name = "login_app.config.DevDbInitConfig"
elif run_flag == "4":
    config_name = "login_app.config.TestDbInitConfig"

logger.debug(f"run_flag={run_flag} config_name={config_name}")

app.config.from_object(config_name)

# user_auth, commonをimportしないとルーティングされない
from login_app.views import user_auth, common

# セッション管理を有効化
Session(app)

# flask_loginの初期化
login_manager = LoginManager()
login_manager.init_app(app)

from login_app.user_loader import setup_auth

# userをロードするためのcallbackを定義
setup_auth(login_manager)

# @login_requiredが付与されたurlにアクセスし、未ログインの場合はここにリダイレクトされる。
login_manager.login_view = "login"
login_manager.login_message = "ログインしてください"

環境変数のRUN_FLAGとconfig.py

コメントに記載しているとおりなのですが、SESSION_DYNAMODB_TABLEで指定しているsessionsテーブルが存在しない場合、appに対するsession管理が有効化できません。既にsessionsテーブルが作成されているDBであれば問題ないのですが、DBの初期化時もappを利用するので、DBの初期化時はsession管理を無効にする必要があります。

環境変数のRUN_FLAGでconfig.pyで定義した設定の読み込みを分岐しています。

RUN_FLAGの値で利用DBを分岐

RUN_FLAG=1 or RUN_FLAG = 3の時
8000で動作しているdynamodb_localを利用します。
RUN_FLAG=2 or RUN_FLAG = 4の時
9000で動作しているdynamodb_localを利用します。

RUN_FLAGの値でsession管理の有効・無効を分岐

RUN_FLAG=1 or RUN_FLAG = 2の時
SESSION_TYPE = "dynamodb"となるのでsession管理は有効
RUN_FLAG=3 or RUN_FLAG = 4の時
SESSION_TYPE = ""となるのでsession管理は無効

login_app.viewsのcommon.py

/に対する処理
404になるパスの処理
/favicon.icoの処理
を実装しています。

common.py

import os
from flask import redirect, url_for, render_template, send_from_directory
from login_app import app

from logging import getLogger, StreamHandler, DEBUG

logger = getLogger(__name__)
handler = StreamHandler()
handler.setLevel(DEBUG)
logger.setLevel(DEBUG)
logger.addHandler(handler)
logger.propagate = False

# parameter defaults to ``["GET"]``. ``HEAD`` and
#        ``OPTIONS`` are added automatically.
# だそうです。
@app.route("/")
def top():
    logger.debug("start top")
    return render_template("top.html")


@app.errorhandler(404)
def non_existant_route(error):
    logger.debug("start non_existant_route")
    # 404のときはloginにリダイレクト
    return redirect(url_for("login"))


@app.route("/favicon.ico")
def favicon():
    return send_from_directory(
        os.path.join(app.root_path, "static/img"),
        "favicon.ico",
    )

/favicon.icoは必要な処理ではないのですが、ブラウザからアクセスすると
127.0.0.1 - - [29/Jan/2022 11:55:46] "GET /favicon.ico HTTP/1.1" 302 -]
のログが出力され気持ち悪いので追加しております。

top.html

current_userはログインしているユーザの情報となります。
詳細はlogin_appのuser_loader.py
で説明させていたきます。

top.html

<ul>
    {% if current_user.is_authenticated %}
    <li><a href="{{ url_for('logout') }}">Log Out</a></li>
    {% else %}
    <li>You are not logined. <a href="{{ url_for('login') }}">Log In</a></li>
    {% endif %}
</ul>

login.html

シンプルなログイン画面です。

<form action="{{ url_for('login') }}" method=post>
    <div>
        <label for="InputUserId">ユーザーID</label>
        <input type="text" id="InputUserId" name=user_id>
    </div>

    <div>
        <label for="InputPassword">パスワード</label>
        <input type="password" id="InputPassword" name=password>
    </div>

    <button type="submit" id="LoginButton">Login Button</button>

    {% for message in get_flashed_messages() %}
        <div role="alert">
            {{ message }}
        </div>
    {% endfor %}
</form>

login_app.viewsのuser_auth.py

/loginに対するGET, POSTの処理を
/logoutに対する処理
/users/detail/<user_id>"に対する処理
を実装しています。

user_auth.py

from flask import request, redirect, url_for, render_template, flash, session
from flask_login import login_user, logout_user, login_required

from logging import getLogger, StreamHandler, DEBUG
from pynamodb.exceptions import DoesNotExist
from login_app import app


logger = getLogger(__name__)
handler = StreamHandler()
handler.setLevel(DEBUG)
logger.setLevel(DEBUG)
logger.addHandler(handler)
logger.propagate = False


@app.route("/login", methods=["GET", "POST"])
def login():
    logger.debug("login start")

    # 以前にflashした値をクリア
    session.pop("_flashes", None)

    if request.method == "POST":
        user_id = request.form["user_id"]
        password = request.form["password"]

        if user_id:
            try:
                from login_app.models.users import User

                user = User.get(user_id)

                if user.password != password:
                    logger.error(f"パスワードが一致しません。")

                    flash("ユーザーIDもしくはパスワードが不正です。")
                else:
                    logger.debug(f"login成功 id={user.id}")

                    login_user(user)

                    logger.debug(f"call login_user end")

                    flash("ログインしました。")

                    return redirect(url_for("top"))
            except DoesNotExist:
                flash("ユーザーIDもしくはパスワードが不正です。")
                logger.error(f"ユーザが存在しません。user_id={user_id}")
        else:
            flash("ユーザーIDが指定されていません。")

    return render_template("login.html")


@app.route("/logout")
def logout():
    logger.debug("logout start")

    logout_user()
    flash("ログアウトしました。")
    return redirect(url_for("login"))


@app.route("/users/detail/<user_id>", methods=["GET"])
@login_required
def user_detail(user_id):
    logger.debug(f"user_detail user_id={user_id} start")

    try:
        from login_app.models.users import User

        user = User.get(user_id)
        return render_template("user_detail.html", user=user)

    except DoesNotExist:
        flash("ユーザが存在しません。")
        logger.error(f"user_detail ユーザが存在しません。user_id={user_id}")

        return redirect(url_for("top"))

loginメソッド

httpのメソッドがPOSTの場合

user_id = request.form["user_id"]
password = request.form["password"]
でリクエストからパラメタを取り出して

user = User.get(user_id)
でDBからuser_idに対応するユーザを検索し

                if user.password != password:
                    logger.error(f"パスワードが一致しません。")

                    flash("ユーザーIDもしくはパスワードが不正です。")
                else:
                    logger.debug(f"login成功 id={user.id}")

                    login_user(user)
                    flash("ログインしました。")

                    return redirect(url_for("top"))

でパスワードが一致しない場合はtop(アプリケーションルート)にリダイレクトします。

user_idに対応するユーザがDBに存在しない場合は、DoesNotExistがスローされ、
return render_template("login.html")に処理が流れるので、login画面が「ユーザーIDもしくはパスワードが不正です。」のメッセージ付きで表示されます。

httpのメソッドがGETの場合

return render_template("login.html")
でlogin画面が表示されます

user_detail

GETのパスで指定されたuser_idでUserを検索し、
user_detail.htmlを表示するときのパラメタでuserを指定しているので、user_detail.htmlから{{ user.id }}のように参照可能となります。
また、@login_requiredが付与されているので、ログインしていない場合は、login_manager.login_view = "login"で定義したloginにリダイレクトされます。

@app.route("/users/detail/<user_id>", methods=["GET"])
@login_required
def user_detail(user_id):
    logger.debug(f"user_detail user_id={user_id} start")

    try:
        from login_app.models.users import User

        user = User.get(user_id)
        return render_template("user_detail.html", user=user)

    except DoesNotExist:
        flash("ユーザが存在しません。")
        logger.error(f"user_detail ユーザが存在しません。user_id={user_id}")

        return redirect(url_for("top"))

user_detail.html

<h2>id : {{ user.id }} mame : {{ user.name }}</h2>

login_appのuser_loader.py

認証されいるユーザーの情報をロードする処理を実施します。

テンプレートのhtmlでcurrent_userを利用すると@login_manager.user_loaderが付与されている関数が呼び出され、ログインしているユーザ情報返却されるのでテンプレートから参照可能となります。

from logging import getLogger, StreamHandler, DEBUG

logger = getLogger(__name__)
handler = StreamHandler()
handler.setLevel(DEBUG)
logger.setLevel(DEBUG)
logger.addHandler(handler)
logger.propagate = False


def setup_auth(login_manager):
    @login_manager.user_loader
    def load_user(user_id):
        logger.debug(f"load_user user_id={user_id}")
        from login_app.models.users import User

        return User(user_id)

modelsのsessions.py

テーブル名やリージョンなどをconfigから読み込んでセットします。
テーブルの属性としてSessionIdとSessionをそれぞれ文字列として定義しています。
SessionIdがプライマリキーとなります。

sessions.py

from datetime import datetime
from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute, NumberAttribute, UTCDateTimeAttribute
from login_app import app

class Session(Model):
    class Meta:
        table_name = app.config.get("SESSION_DYNAMODB_TABLE")
        region = app.config.get("DYNAMODB_REGION")
        aws_access_key_id = app.config.get("AWS_ACCESS_KEY_ID")
        aws_secret_access_key = app.config.get("AWS_SECRET_ACCESS_KEY")
        host = app.config.get("DYNAMODB_ENDPOINT_URL")

    SessionId = UnicodeAttribute(hash_key=True, null=False)
    Session = UnicodeAttribute(null=True)

modelsのusers.py

class Meta:ですが、table_nameは直で指定している以外はsessions.pyと同様となります。

テーブルの属性としてid、password、hostをそれぞれ文字列として定義しています。
idがプライマリキーとなります。

users.py

from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute
from login_app import app


class User(Model):
    class Meta:
        table_name = "users"
        region = app.config.get("DYNAMODB_REGION")
        aws_access_key_id = app.config.get("AWS_ACCESS_KEY_ID")
        aws_secret_access_key = app.config.get("AWS_SECRET_ACCESS_KEY")
        host = app.config.get("DYNAMODB_ENDPOINT_URL")

    id = UnicodeAttribute(hash_key=True, null=False)
    password = UnicodeAttribute(null=False)
    name = UnicodeAttribute(null=False)

    def get_id(self):
        return self.id

    def is_authenticated(self):
        return True

    def is_active(self):
        return True

    def is_anonymous(self):
        return False

設定対象	説明
get_id()	ユーザを一意に識別できる文字列を返却するように実装します。idをintで管理している場合では文字列に変換して返却します。user_loaderに渡す値はこのメソッドから取得されます。
is_authenticated()	ユーザが認証済み(ログインしているか)の場合はTrueを返却します。@login_requiredが付与されたURLにアクセスする時に呼び出されます。
is_active()	このサンプルでは意味ないですが、どこかのサイトにアカウントを新規登録して、メール認証やSMS認証が完了していないようなユーザはFalseを返却するように実装します。
is_anonymous()	anonymousユーザの場合はTrueを返却します。

init_db.py

DBの初期化を行うモジュールとなります。
sessionsテーブルとusersテーブルの作成とusersの初期データの登録を行います。

プロジェクトルートに移動後に、Windowsの場合は
8000ポートのDB利用時はset RUN_FLAG=3
9000ポートのDB利用時はset RUN_FLAG=4
を実行後にinit_db.pyを実行してください。
python login_app/init_db.py

RUN_FLAGの値は
1:8000ポートのDB利用でアプリ起動モード
2:9000ポートのDB利用でアプリ起動モード
3:8000ポートのDB利用でDBセットアップモード
4:9000ポートのDB利用でDBセットアップモード
を意味しています。

init_db.py

from logging import getLogger, StreamHandler, DEBUG
from login_app.models.sessions import Session
from login_app.models.users import User


logger = getLogger(__name__)
handler = StreamHandler()
handler.setLevel(DEBUG)
logger.setLevel(DEBUG)
logger.addHandler(handler)
logger.propagate = False


def init():
    logger.debug("init start")

    if not Session.exists():
        Session.create_table(read_capacity_units=5, write_capacity_units=2)
    if not User.exists():
        User.create_table(read_capacity_units=5, write_capacity_units=2)

    user_1 = User(id="user_id_1", password="user_id_1_pass", name="user_id_1_name")
    user_1.save()
    logger.debug("user_id_1.save() end")

    user_2 = User(id="user_id_2", password="user_id_2_pass", name="user_id_2_name")
    user_2.save()
    logger.debug("user_id_2.save() end")


init()

server.py

アプリの起動を行うモジュールとなります。

serer.py

from login_app import app

if __name__ == "__main__":
    app.run()

プロジェクトルートに移動後に
init.pyでRUN_FLAGの環境変数で読み込むconfigを分岐していますので
Windowsの場合は
8000ポートのDB利用時はset RUN_FLAG=1
9000ポートのDB利用時はset RUN_FLAG=2
を実行後にpython ./login_app/server.pyを実行するとアプリが起動します。

python ./login_app/server.py
run_flag=2 config_name=login_app.config.TestConfig
 * Serving Flask app 'login_app' (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Restarting with stat
run_flag=2 config_name=login_app.config.TestConfig
 * Debugger is active!
 * Debugger PIN: 219-915-120
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

サンプルのテスト

test_common.py

テストで利用する共通処理を実装しているモジュールです。

test_common.py


from logging import getLogger, StreamHandler, DEBUG

logger = getLogger(__name__)
handler = StreamHandler()
handler.setLevel(DEBUG)
logger.setLevel(DEBUG)
logger.addHandler(handler)
logger.propagate = False

BASE_URL = "http:\\localhost:5000"
LOGIN_FORM_URL = f"{BASE_URL}/login"


def request_login(client, user_id, password):
    logger.debug(f"request_login user_id={user_id} password={password} ")

    return client.post(
        "/login", data=dict(user_id=user_id, password=password), follow_redirects=True
    )


def request_user_detail(client, user_id):
    logger.debug(f"request_user_detail user_id={user_id}")

    return client.get(
        f"/users/detail/{user_id}",
        follow_redirects=True,
    )

login_app_test.py

flaskのapp.test_client()を利用し、DBにアクセスするテストは以下のとおりです。

login_app_test.py

import pytest

from login_app import app
from logging import getLogger, StreamHandler, DEBUG

from tests.test_common import request_login, request_user_detail


logger = getLogger(__name__)
handler = StreamHandler()
handler.setLevel(DEBUG)
logger.setLevel(DEBUG)
logger.addHandler(handler)
logger.propagate = False


@pytest.fixture
def client():
    logger.debug("client start")

    with app.test_client() as client:
        yield client


def test_top_not_logined(client):
    rv = client.get("/")
    assert b"Log In" in rv.data


@pytest.fixture(
    params=[
        ("user_id_1", "wrong_password"),
        ("wrong_user_id", "user_id_1_pass"),
        ("", ""),
    ]
)
def login_fail_fixture(request):
    return (request.getfixturevalue("client"), request.param[0], request.param[1])


def test_login_fail(login_fail_fixture):
    client, user_id, password = login_fail_fixture

    # ログインが失敗すると"Login Button"を含む
    rv = request_login(client, user_id, password)
    assert b"Login Button" in rv.data


def test_top_logined(client):
    # loginが成功するとログイン状態でtopにリダイレクトされるので"Log Out"を含む
    rv = request_login(client, "user_id_1", "user_id_1_pass")
    assert b"Log Out" in rv.data

    # ログイン状態でtopにアクセスすると"Log Out"を含む
    rv = client.get("/")
    assert b"Log Out" in rv.data


def test_detail_logined(client):
    # まずはログイン
    rv = request_login(client, "user_id_1", "user_id_1_pass")
    assert b"Log Out" in rv.data

    # ログイン状態でuser_detailにアクセス
    rv = request_user_detail(client, "user_id_1")
    assert b"id : user_id_1 mame : user_id_1_name" in rv.data


def test_detail_not_logined(client):
    # 未ログイン状態でuser_detailにアクセスするとloginにリダイレクトされる
    rv = request_user_detail(client, "user_id_1")
    assert b"Login Button" in rv.data


def test_detail_logined_not_exist_user(client):
    # まずはログイン
    rv = request_login(client, "user_id_1", "user_id_1_pass")
    assert b"Log Out" in rv.data

    # userが存在しない場合はtopにリダイレクトされる
    rv = request_user_detail(client, "dummy_user_id")
    # のでtop画面でログインしているのでLog Outリンクが存在
    assert b"Log Out" in rv.data

test_top_not_logined

未ログインでトップ(/)にアクセスする時のテストとなります。

@pytest.fixture
def client():
    logger.debug("client start")

    with app.test_client() as client:
        yield client


def test_top_not_logined(client):
    rv = client.get("/")
    assert b"Log In" in rv.data

app.test_client() でテスト用のオブジェクトが取得できますので、client.getやclient.postなどを呼び出し、得られた結果を検証します。

''sert b"Log In" in rv.data''のように結果に"Log In"が含まれる事を検証しています。rv.dataはバイナリですので、b"Log In"と指定する必要があります。

test_login_fail

ログインが失敗するテストとなります。

@pytest.fixture(
    params=[
        ("user_id_1", "wrong_password"),
        ("wrong_user_id", "user_id_1_pass"),
        ("", ""),
    ]
)
def login_fail_fixture(request):
    return (request.getfixturevalue("client"), request.param[0], request.param[1])


def test_login_fail(login_fail_fixture):
    client, user_id, password = login_fail_fixture

    # ログインが失敗すると"Login Button"を含む
    rv = request_login(client, user_id, password)
    assert b"Login Button" in rv.data

ログインが失敗するuser_idとpasswordの繰り合わせでloignをpostで呼び出すので、結果に"Login Button"が含まれます。

test_top_logined

ログインに成功したらtopにリダイレクトされ、"Log Out"が含まれる。
ログイン状態でtop(/)にアクセスすると"Log Out"が含まれる。
とのテスト内容となります。

def test_top_logined(client):
    # loginが成功するとログイン状態でtopにリダイレクトされるので"Log Out"を含む
    rv = request_login(client, "user_id_1", "user_id_1_pass")
    assert b"Log Out" in rv.data

    # ログイン状態でtopにアクセスすると"Log Out"を含む
    rv = client.get("/")
    assert b"Log Out" in rv.data

test_detail_logined

ログイン状態でuser_detail("/users/detail/")にアクセスするテストとなります。

def test_detail_logined(client):
    # まずはログイン
    rv = request_login(client, "user_id_1", "user_id_1_pass")
    assert b"Log Out" in rv.data

    # ログイン状態でuser_detailにアクセス
    rv = request_user_detail(client, "user_id_1")
    assert b"id : user_id_1 mame : user_id_1_name" in rv.data

test_detail_logined

未ログイン状態でuser_detail("/users/detail/")にアクセスするテストとなります。

def test_detail_not_logined(client):
    # 未ログイン状態でuser_detailにアクセスするとloginにリダイレクトされる
    rv = request_user_detail(client, "user_id_1")
    assert b"Login Button" in rv.data

未ログイン状態なのでlogin(性格にはlogin_manager.login_viewの値)にリダイレクトされるので、結果に"Login Button"が含まれる事を検証しています。

test_detail_logined_not_exist_user

ログイン状態で、存在しないuser_idを指定してuser_detail("/users/detail/")にアクセスするテストとなります。

def test_detail_logined_not_exist_user(client):
    # まずはログイン
    rv = request_login(client, "user_id_1", "user_id_1_pass")
    assert b"Log Out" in rv.data

    # userが存在しない場合はtopにリダイレクトされる
    rv = request_user_detail(client, "dummy_user_id")
    # のでtop画面でログインしているのでLog Outリンクが存在
    assert b"Log Out" in rv.data

login_app_unit_test.py

flaskのapp.test_client()を利用し、DBアクセス処理をモック化したテストは以下のとおりです。
DBにアクセス処理をモック化の部分にフォーカスするために、login_app_test.pyに対応する全てのテストは実装しておりません。

mocker.patch.object(User, "get", return_value=user)でUser.getが呼ばれた時の結果を固定化している。
mocker.patch.object(User, "get", side_effect=DoesNotExist)でUser.getが呼ばれた時にDoesNotExistがスローされる。
ぐらいの違いしかありまんが・・・

login_app_unit_test.py

import pytest
from pynamodb.exceptions import DoesNotExist

from login_app import app
from login_app.models.users import User
from tests.test_common import request_login


@pytest.fixture
def client():
    with app.test_client() as client:
        yield client


def test_login_fail(mocker, client):
    user = User(id="id", password="password")
    mocker.patch.object(User, "get", return_value=user)

    # ログインが失敗すると"Login Button"が存在する
    rv = request_login(client, "id", "wrong_password")
    assert b"Login Button" in rv.data


def test_login_does_not_exist(mocker, client):
    mocker.patch.object(User, "get", side_effect=DoesNotExist)

    # ログインが失敗すると"Login Button"が存在する
    rv = request_login(client, "id", "wrong_password")
    assert b"Login Button" in rv.data


def test_top_logined(mocker, client):
    user = User(id="id", password="password")
    mocker.patch.object(User, "get", return_value=user)

    # loginが成功するとログイン状態でtopにリダイレクトされるので"Log Out"を含む
    rv = request_login(client, user.id, user.password)
    assert b"Log Out" in rv.data

    # ログイン状態でtopにアクセスすると"Log Out"を含む
    rv = client.get("/")
    assert b"Log Out" in rv.data

login_app_selenium_test.py

seleniumでChromeを動かしてテストを実施するモジュールとなります。

login_app_selenium_test.py

import pytest
from selenium import webdriver
from selenium.webdriver.common.by import By
from logging import getLogger, StreamHandler, DEBUG
from tests.test_common import BASE_URL, LOGIN_FORM_URL


logger = getLogger(__name__)
handler = StreamHandler()
handler.setLevel(DEBUG)
logger.setLevel(DEBUG)
logger.addHandler(handler)
logger.propagate = False


@pytest.fixture
def webdriver_chrome() -> webdriver:
    webdriver_chrome = webdriver.Chrome()
    yield webdriver_chrome
    webdriver_chrome.quit()


def test_top_not_logined(webdriver_chrome: webdriver):
    webdriver_chrome.get(BASE_URL)

    login_text = webdriver_chrome.find_element(
        by=By.XPATH, value='//*[contains(., "Log In")]'
    )

    assert not login_text is None


@pytest.fixture(
    params=[
        ("user_id_1", "wrong_password"),
        ("wrong_user_id", "user_id_1_pass"),
        ("", ""),
    ]
)
def login_fail_fixture(request):
    return (
        request.getfixturevalue("webdriver_chrome"),
        request.param[0],
        request.param[1],
    )


def test_login_fail(login_fail_fixture):
    webdriver_chrome, user_id, password = login_fail_fixture

    login(webdriver_chrome, user_id, password)

    # ログインが失敗すると"Login Button"が存在する
    login_button = webdriver_chrome.find_element(by=By.ID, value="LoginButton")
    assert not login_button == None


def login(webdriver_chrome: webdriver, user_id: str, password: str):
    logger.debug(f"login user_id={user_id} password={password} ")

    webdriver_chrome.get(LOGIN_FORM_URL)

    input_user_id = webdriver_chrome.find_element(by=By.ID, value="InputUserId")
    input_user_id.send_keys(user_id)

    input_password = webdriver_chrome.find_element(by=By.ID, value="InputPassword")
    input_password.send_keys(password)

    login_button = webdriver_chrome.find_element(by=By.ID, value="LoginButton")
    login_button.click()


def test_top_logined(webdriver_chrome: webdriver):

    # loginが成功するとログイン状態でtopにリダイレクトされるので"Log Out"を含む
    login(webdriver_chrome, "user_id_1", "user_id_1_pass")

    logout_text = webdriver_chrome.find_element(
        by=By.XPATH, value='//*[contains(., "Log Out")]'
    )
    assert not logout_text == None

    # ログイン状態でtopにアクセスすると"Log Out"を含む
    webdriver_chrome.get(BASE_URL)

    logout_text = webdriver_chrome.find_element(
        by=By.XPATH, value='//*[contains(., "Log Out")]'
    )
    assert not logout_text == None

サンプルを動作させた時のログやDBデータ

@login_manager.user_loaderや''User.get_id()''などは動作イメージが掴みにいと思いますので、手でブラウザを操作した時のログを記載いたします。

ログイン時のログ

login start
127.0.0.1 - - [29/Jan/2022 17:23:27] "GET /login HTTP/1.1" 200 -
127.0.0.1 - - [29/Jan/2022 17:23:27] "GET /favicon.ico HTTP/1.1" 304 -
login start
login成功 id=user_id_1
get_id start
call login_user end
127.0.0.1 - - [29/Jan/2022 17:23:32] "POST /login HTTP/1.1" 302 -
start top
load_user user_id=user_id_1
127.0.0.1 - - [29/Jan/2022 17:23:32] "GET / HTTP/1.1" 200 -

user_auth.py

                    logger.debug(f"login成功 id={user.id}")

                    login_user(user)

                    logger.debug(f"call login_user end")

                    flash("ログインしました。")

                    return redirect(url_for("top"))

ですので、login_user(user)からUser.get_id()が呼び出されている事が分かります。

127.0.0.1 - - [29/Jan/2022 17:23:32] "POST /login HTTP/1.1" 302 -
でtop(/)にリダイレクトされ

top.html

{% if current_user.is_authenticated %}

のようにcurrent_userのis_authenticatedが参照されており、load_userがuser_id="user_id_1"で呼びされている事が分かります。

user_loader.py

    @login_manager.user_loader
    def load_user(user_id):
        logger.debug(f"load_user user_id={user_id}")
        from login_app.models.users import User

        return User(user_id)

load_user user_id=user_id_1

sessionsテーブルのデータ

ログイン後に
aws dynamodb scan --table-name sessions --endpoint-url http://localhost:9000
でsessionsのデータを取得してみます。

{
    "Items": [
        {
            "SessionId": {
                "S": "session:06d96df5-f0b9-446d-a773-c6d6268afd2f"
            },
            "Session": {
                "S": "{\"_flashes\":[{\" t\":[\"message\",\"\\u30ed\\u30b0\\u30a4\\u30f3\\u3057\\u307e\\u3057\\u305f\\u3002\"]}],\"_fresh\":true,\"_id\":\"afbe5f0825480344b9fb04f9eaf777bd554fc35119445b76f64c2ec1b5018c14bd5e16d176b95c2ba8133608772c600ce3474490e9e1c245088e736bb3ee2e1a\",\"_permanent\":true,\"_user_id\":\"user_id_1\"}"
            }
        }
    ],
    "Count": 1,
    "ScannedCount": 1,
    "ConsumedCapacity": null
}

それらしいデータが登録されている事が分かります。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

Flaskとdynamodb_localでDB認証とセッション管理を試してみる(テストくどめ、Selenium入り)

はじめに

サンプル概要

ファイル構成(github)

ファイル構成(github未登録)

実際の動作イメージ

dynamodb_localの起動方法

アプリ動作用のDB起動

テスト用のDB起動

サンプルの説明

config.py

login_appの__init__.py

環境変数のRUN_FLAGとconfig.py

RUN_FLAGの値で利用DBを分岐

RUN_FLAGの値でsession管理の有効・無効を分岐

login_app.viewsのcommon.py

common.py

top.html

login.html

login_app.viewsのuser_auth.py

loginメソッド

httpのメソッドがPOSTの場合

httpのメソッドがGETの場合

user_detail

login_appのuser_loader.py

modelsのsessions.py

modelsのusers.py

init_db.py

server.py

サンプルのテスト

test_common.py

login_app_test.py

test_top_not_logined

test_login_fail

test_top_logined

test_detail_logined

test_detail_logined

test_detail_logined_not_exist_user

login_app_unit_test.py

login_app_selenium_test.py

サンプルを動作させた時のログやDBデータ

ログイン時のログ

sessionsテーブルのデータ

login_appのinit.py