Dialogflow CX Agentからの単体テストケース作成を自動化してみる

本記事は 夏休みクラウド自由研究 8/2付の記事です

こんにちは。SCSKの江木です。

夏休みクラウド自由研究2024の2日目の記事になります。

みなさん、単体テストケース作成に時間がかかりすぎていませんか?

先日Dialogflow CXの単体テストケースを作成していて、抜け漏れチェックに追われて、かなり時間がかかってしまいました。

今回はテストケース作成の時間を削減すべく、グラフ理論を使って、単体テストケース作成を自動化する方法を実装してみました。

このグラフ理論の考え方は他にも使えると思うので、ぜひご覧ください。

Dialogflow CXにおける単体テストケースについて

単体テストの方法

Dialogflow CX(以下、Dialogflow)は他のサービスと連携して使用されることが多いのが現状です。したがって、本記事におけるDialogflowの単体テストはDialogflow単体で動作するかという意味で使用していますので、ご了承ください。

これを踏まえて、Dialogflowの単体テストは以下の方法で行うことを想定しております。

  1. 基本的にはページレベルでテスト
  2. ループのテストはフローレベルで
  3. テストは正常系、異常系、ループに分ける

それぞれの観点について詳しく説明していきます。

基本的にはページレベルでテスト

一般的にプログラムの単体テストは関数/メソッドといった単位で行いますが、このプログラムにおける関数/メソッドをDialogflowで置き換えるとページと捉えることができます。

従って、ページレベル単位でテストを行います。

※一般的な単体テストと捉えていただけるとわかりやすいと思います。

ループのテストはフローレベルで

聞き返し機能をチャット/ボイスボットに備えるため、DialogflowのAgentにはループが存在することがあります。

ループのテストはページレベルで行うとテストケースが複雑になるため、フローレベルで行います。

テストは正常系、異常系、ループに分ける

テストは正常系、異常系、ループの3つに分けて行います。

正常系ではパラメータ入力およびイベント発生における動作確認を行います。異常系では無効な入力に対する動作確認を行います。

また、正常系・異常系は以下の観点でテストを行います。

  • 正しく遷移するか
  • 画面が仕様通りに表示されているか
  • 取得情報に誤りがないか

一方で、ループは以下の観点でテストを行います。

  • ループがフローに影響がないか

ループは正常系に入れてもよいのですが、テスト観点が異なるためテストジャンルを分けました。

作成するテストケースについて

上記のテスト観点を踏まえて、テストケースを作成していきます。作成したテストケースは以下のようなエクセルシートで出力したいと思います。また、出力されるエクセルシートは全部埋められているわけではなく、抜け漏れチェックのためにエクセルシートの一部をテスターが埋めるようにしています。

※横に長いため、分割しています。

実装

アーキテクチャ

アーキテクチャは以下の図を想定しています。

Cloud StorageのバケットにAgentのファイルをアップロードすると、テストケースが同じバケットに出力されるように設計しています。

実装方針

テストケース作成にはページの遷移先、ループの検出が不可欠です。そこで、グラフ理論の考え方でDialogflowのフローは有向グラフであることに着目し、Pythonのグラフ系ライブラリであるNetworkXを使って、実装しました。NetworkXを使うことで、ループ検出、ルート検索が容易になります。

有向グラフとは?

有向グラフとは、下図のように頂点と、それらを結ぶ辺が向きを持つグラフです。辺の向きは矢印で表され、矢印の先端が頂点の終点、矢印の根元が始点となります。

いざ実装!

それでは実際に実装していきます。Dialogflowのフローを有向グラフに置き換える箇所を中心にプログラムを抜粋して紹介していきます。

トリガーの設定

Cloud Storageのトリガーを設定していきます。Cloud StorageのバケットにAgentのファイルが格納された際にCloud Functionsが起動するようにしたいので、以下のように「google.cloud.storage.object.v1.finalized」で設定します。

 

Agentファイルの読み込み

CloudEvent関数を使用して、Cloud Storageにアップロードされたファイル名の取得を行います。その後、ファイル名からAgentをjson形式でCloud Storageから読み込みます。以下は、アップロードされたAgentのファイルから辞書形式でAgentの詳細を取得するプログラムです。

@functions_framework.cloud_event
def main(cloud_event):
    #GCSバケットの名前
    bucket_name = cloud_event.data['bucket']
    #zipファイルの名前
    blob_name = cloud_event.data['name']
    #Agentファイル以外がアップロードされた場合は終了
    if blob_name.split(".")[1] != "zip":
        return

  #flowごとのjsonが格納された辞書を取得
  Agent_dict = get_dict(bucket_name,blob_name)

また、辞書形式でAgentを取得するget_json関数の詳細は以下の通りです。

def get_dict(bucket_name: str,blob_name: str) -> dict:
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blobs = storage_client.list_blobs(bucket_name)
    blob = bucket.blob(blob_name)
    content = blob.download_as_string()

    # zipファイルを読み込む
    zip_f = zipfile.ZipFile(BytesIO(content))

    # zipファイルの中身のファイル名を取得
    lst = zip_f.namelist()

    selected_items = [item for item in lst if 'flows' in item]
    
    Agent_dict = {}
    for item in selected_items:
        flow_name = item.split('/')[1]
        tag = item.split('/')[-1].split('.')[0]
        if flow_name not in Agent_dict:
            Agent_dict[flow_name] = {}
        with zip_f.open(item) as myfile: # openを利用してファイルにアクセス
            json_detail = json.load(myfile)
        Agent_dict[flow_name][tag] = json_detail

    return Agent_dict
有向グラフの作成

それでは、辞書型のAgentからグラフを作成していきます。以下は辞書型のAgentから、page名をkey、そのpageからの遷移先の配列をvalueとした辞書を作成し、その辞書からグラフを作成していくプログラムになります。また、page_dictはflowごとのpage一覧が格納された辞書を表しています。

def make_graph(page_dict: dict,flow_name: str,Agent_dict: dict) -> Graph:
        #page名が入った配列を作成。End Sessionページが入っていないので追加する
        page_list = page_dict[flow_name]
        page_list.append("End Session")

        #page名をkey,遷移先の配列をvalueとした辞書を作成
        dict_flow = {}
        #page名からflowを探索して、遷移先の配列を作成
        for page in page_list
            if page == "Start Page":
                 dict_flow[page] = search(Agent_dict[flow_name][flow_name])
            elif page == "End Session":
                 dict_flow[page] = []
            else:
                dict_flow[page] = search(Agent_dict[flow_name][page])

     #グラフ作成
        g = nx.DiGraph(dict_flow)
        return g

 

テストケースの作成

正常系、異常系、ループの3つのジャンルで作成していきます。

正常系テストケースの作成

正常系のテストケースを作成するプログラムは以下の通りです。イベントを確認するテストケースとパラメータを確認するテストケースに分けて作成してします。rowは操作するエクセルシートの行、wsはエクセルシート、flow_dictはflowの辞書を表しています。

def make_normal(row: int,page: str,ws: sheet,flow_dict: dict) -> Tuple[int,sheet]:
    test_category = "正常系"

    #イベントテストケース作成
    #イベント一覧の取得
    event_list,event_json = get_event(page,flow_dict)
    #異常系のイベントを除去
    event_list_remove_abnormal = [event for event in event_list if 'sys.no-match' not in event and 'sys.no-input' not in event and 'long-utterance' not in event]
    if len(event_list_remove_abnormal) != 0:
        for event in event_list_remove_abnormal:
            category_sentence = "イベント(" + event + ")"
            transition = "遷移しない"
            for event_detail in event_json["page_event"]:
                if event_detail["event"] == event and "targetPage" in event_detail.keys():
                    transition = event_detail["targetPage"]
            #テストケースをエクセルへ書き込み
            row,ws = add_page_testcase(row,page,test_category,category_sentence,transition,ws)

    #パラメータテストケース作成
    #パラメータ一覧の取得
    parameter_list = get_parameter(page,flow_dict)
    #遷移先一覧の取得
    transitionRoute_list,transitionRoute_json = get_transitionRoutes(page,flow_dict)
    if len(parameter_list) != 0:
        param_all = ','.join(parameter_list)
        category_sentence = "パラメータ(" + param_all + ")"
        for route in transitionRoute_json:
            for param in parameter_list:
                transition = "遷移しない"
                if "$page.params.status" in route["condition"] or param in route["condition"]:#page.paramsかパラメータ名が条件に入っていれば、そのルートをテストケースとする
                    if "targetPage" in route.keys():
                        transition = route["targetPage"]
           #テストケースをエクセルへ書き込み
                    row,ws = add_page_testcase(row,page,test_category,category_sentence,transition,ws)

    return row,ws
異常系テストケースの作成

異常系のテストケースを作成するプログラムは以下の通りです。no-match、no-inputイベントとlong-utteranceイベントに分けてテストケース作成を行います。no-match、no-inputイベントは設定された回数分テストケースを作成します。long-utteranceイベントはループ回数の指定があれば、指定の回数分テストケースを作成しますが、指定がない際はテストケースを1つのみ作成します。

def make_abnormal(row: int,page: str,ws: sheet,flow_dict: dict) -> Tuple[int,sheet]:
    test_category = "異常系"

    #イベント一覧の取得
    event_list,event_json = get_event(page,flow_dict)
    event_list_abnormal = [event for event in event_list if "sys.no-match" in event or "sys.no-input" in event or "long-utterance" in event]
    event_list_abnormal_not_longutterance = [event for event in event_list if "sys.no-match" in event or "sys.no-input" in event]

    if len(event_list_abnormal_not_longutterance) != 0:
        for event in event_list_abnormal_not_longutterance:
            category_sentence = event + "イベント"
            #pageのほうにある場合もある
            transition = "遷移しない"
            if "param_event" in event_json.keys():
                for event_detail in event_json["param_event"]:
                    if event_detail["event"] == event and "targetPage" in event_detail.keys():
                       transition = event_detail["targetPage"]
            elif "page_event" in event_json.keys():
                for event_detail in event_json["page_event"]:
                    if event_detail["event"] == event and "targetPage" in event_detail.keys():
                        transition = event_detail["targetPage"]
            #テストケースをエクセルへ書き込み
            row,ws = add_page_testcase(row,page,test_category,category_sentence,transition,ws)
   
    #long-utteranceの時の処理
    parameter_list = get_parameter(page,flow_dict)
    transitionRoute_list,transitionRoute_json = get_transitionRoutes(page,flow_dict)
    for event in event_list_abnormal:
        if "long-utterance" in event:
            #loop-countのパラメータがなければ(not-added)、テストケースは1つだけ作成する
            flag = "not-added"
            for route in transitionRoute_json:
                #long-utteranceのループ回数をカウントするパラメータを探索して、存在すればそのループカウント分だけテストケースを作成。
                if "loop" in route["condition"] and "count" in route["condition"] and "$session.params" in route["condition"]:
                    split_condition = route["condition"].split("=")
                    loop_count_max = int(split_condition[1].strip())
                    for i in range(loop_count_max):
                        category_sentence = event + "-" + str(i+1) +"イベント"
                        if i+1 != loop_count_max:
                            transition = "遷移しない"
                        else:
                            transition = route["targetPage"]
                        #テストケースをエクセルへ書き込み
                        row,ws = add_page_testcase(row,page,test_category,category_sentence,transition,ws)
                    flag = "added"
            if flag == "not-added":
                category_sentence = event + "イベント"
                transition = "遷移しない"
                #テストケースをエクセルへ書き込み
                row,ws = add_page_testcase(row,page,test_category,category_sentence,transition,ws)

    return row,ws
ループテストケースの作成

ループのテストケースを作成するプログラムは以下の通りです。NetworkXの閉路を検知する関数を使って、ループ検知を行います。

def make_loop_test(row: int,pages: str,ws: sheet,g: Graph) -> Tuple[int,sheet]:

    #有向グラフからサイクル検知でloopを検知
    loop_list = list(nx.simple_cycles(g))

    if len(loop_list) != 0:
        for loop in loop_list:
            loop.append(loop[0])
            page_sentence = "⇒".join(loop)
            test_category = "-"
            category_sentence = "、".join(loop)
            category_sentence = category_sentence + "におけるloopをカウントするパラメータの確認"
            #テストケースをエクセルへ書き込み
            row,ws = add_loop_testcase(row,page_sentence,test_category,category_sentence,ws)    

    return row,ws
ファイルの出力・保存

エクセルファイルをCloud Storageのバケットへ出力するプログラムは以下の通りです。エクセルファイルを一時ファイル的に保存するためにtmpフォルダを使用しています。(地味に苦労したポイントです!!)tmpフォルダへ格納した後、Cloud Storageのバケットへアップロードします。

def repository(wb: book,bucket_name: str,file_name: str) -> None:
    #エクセルの一時ファイルを保存
    wb.save("/tmp/"+excel_filename)
  
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)

    # エクセルの一時ファイルをCloud Storageのバケットへアップロード
    blob.upload_from_filename("/tmp/" + file_name)
    os.remove("/tmp/" + file_name)

出力結果

以下のAgentのテストケースを出力してみます。

以下のようにうまく出力することができました!(テストケースが多いので、抜粋しています)

また、ループの検知がうまくできていることも以下のように見て取れます。

おわりに

今回はグラフ理論を使って、Dialogflow CXのテストケース作成の自動化を実装してみました。

本記事で実装しているのはループの検出のみですが、ルートの検出ももちろんできます。

ループ、ルートの検出が出来ると何がうれしいかといいますと、テストケースの抜け漏れを防げるということです!

Dialogflowの他にも、世の中にはグラフに置き換えることができるサービスで溢れていますので、それらのサービスをグラフに置き換えることができれば、今回のように何らかの自動化ができるかもしれませんね…

最後まで読んでいただき、ありがとうございました。

タイトルとURLをコピーしました