Azure Functionのテスト作成（TimerTrigger、HTTPTrigger）

Azure Functionでコード組んだ時に、テストについてpytest周りとかどうすればよいのかわからなかったので備忘録として。

ディレクトリ構成

本来はhost.jsonとかいろいろあるけど今回はテストコードだけ書きたいのでそのあたりは割愛

azfunc_pytest
├─functions
│  ├─http_trigger
|  |  └─function_app.py
│  └─timer_trigger
|     └─function_app.py
└─tests
    ├─http_trigger
    | └─test_http_function_app.py
    └─timer_trigger
      └─test_timer_function_app.py

HttpTrigger

プロダクトコード

functions/http_trigger/function_app.py

import logging

import azure.functions as func

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)


@app.route(route="http_trigger")
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("Python HTTP trigger function processed a request.")

    name = req.params.get("name")
    if not name:
        try:
            req_body = req.get_json()
        except Exception:
            return func.HttpResponse(
                "Please set a name in the query string or in the request body",
                status_code=400,
            )
        else:
            name = req_body.get("name")

    if name:
        return func.HttpResponse(
            f"Hello, {name}. This HTTP triggered function executed successfully."
        )
    else:
        return func.HttpResponse(
            "This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
            status_code=200,
        )

VSCode のAzure拡張機能から自動で作成したHttpTriggerのコードをそのまま使用

テストコード

tests/http_trigger/test_http_function_app.py

import os
import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))

import azure.functions as func
import pytest

from functions.http_trigger.function_app import http_trigger


@pytest.fixture
def client():
    return http_trigger.build().get_user_function()


def test_get_http_trigger_success(client):
    """GETリクエストのテストケース"""
    req = func.HttpRequest(
        method="GET",
        body=None,
        url="/api/http_trigger",
        params={"name": "GET"},
    )
    resp = client(req)
    assert (
        resp.get_body()
        == b"Hello, GET. This HTTP triggered function executed successfully."
    )
    assert resp.status_code == 200


def test_post_http_trigger_success(client):
    """POSTリクエストのテストケース"""
    req = func.HttpRequest(
        method="POST",
        body=b'{"name": "POST"}',
        url="/api/http_trigger",
    )
    resp = client(req)
    assert (
        resp.get_body()
        == b"Hello, POST. This HTTP triggered function executed successfully."
    )
    assert resp.status_code == 200


def test_http_trigger_no_name(client):
    """nameパラメータが指定されていない場合のテストケース"""
    req = func.HttpRequest(
        method="GET",
        body=None,
        url="/api/http_trigger",
    )
    resp = client(req)
    assert (
        resp.get_body()
        == b"Please set a name in the query string or in the request body"
    )
    assert resp.status_code == 400

ポイントはclientのfixture定義部分.http_trigger.build().get_user_function()でAzure Functionのハンドラーを取得しているらしく、これでAzure Function特有の面倒な制約を排除して純粋にコードのテストができるらしい.
これについてはHTTPTriggerよりも後述のTimerTriggerの方が恩恵が分かりやすいかもしれない.

TimerTrigger

プロダクトコード

functions/timer_trigger/function_app.py

import logging

import azure.functions as func

app = func.FunctionApp()


@app.schedule(
    schedule="* * * * * *", arg_name="myTimer", run_on_startup=False, use_monitor=False
)
def timer_trigger(myTimer: func.TimerRequest) -> None:
    if myTimer.past_due:
        logging.info("The timer is past due!")
        return

    logging.info("Python timer trigger function executed.")

HttpTriggerの時と同様にVSCode のAzure拡張機能から自動で作成したTimerTriggerのコードを少し修正.
具体的にはif myTimer.past_dueの時にpast_dueのログ出したらそこで終わるように修正.

テストコード

tests/timer_trigger/test_timer_function_app.py

import os
import sys

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))

import logging

import pytest
from azure.functions import TimerRequest

from functions.timer_trigger.function_app import timer_trigger


class CustomTimerRequest(TimerRequest):
    def past_due(self):
        return False


@pytest.fixture
def timer_request():
    return CustomTimerRequest()


@pytest.fixture
def client():
    return timer_trigger.build().get_user_function()


def test_timer_trigger(client, timer_request, caplog):
    caplog.set_level(logging.INFO)

    timer_request.past_due = False
    client(timer_request)

    assert len(caplog.records) == 1
    assert "Python timer trigger function executed." in caplog.records[0].message


def test_timer_trigger_past_due(client, timer_request, caplog):
    caplog.set_level(logging.INFO)

    timer_request.past_due = True
    client(timer_request)

    assert len(caplog.records) == 1
    assert "The timer is past due!" in caplog.records[0].message

HttpTriggerと同じくclient fixtureを定義. これがないとapp.scheduleデコレータが邪魔をして処理が走らずまともにテストできない. 逆に言えばこれがあるとスケジュール設定など考慮せずテストが走る.

テストの際はプロジェクトルート（今回の例ではazfunc_pytest）からpytestを実行するだけ.
通常TimerTriggerを実装した時はBlobにデータを作成したりDBにレコード登録したりがあると思うが割愛. とりあえずログ出力でもってテストを行う.

とりあえずよく使いそうな（私はよく使ってた）トリガーのテストについてまとめてみた.
次はEventHubTrigger（IoTHubTrigger）とBlobTriggerについて書いていければよいなと.