APRESIA Technical Blog

behaveにて繰り返し処理を考慮したテスト自動化の テストケースとスクリプトを書いてみた

はじめに

behaveにて条件分岐を考慮したテスト自動化のテストケースとスクリプトを書いてみた」 にて、behaveによる条件分岐のサンプルについてご説明しました。今回のテックブログは、その続編となります。
テストシナリオの繰り返し処理についてサンプルを作成してご説明します。この繰り返し処理は、現行のbehave には具備されておりません。Featureファイル 上で既に実行済みの行に処理を戻す必要がありますが、behave のシナリオ実行処理箇所のソースコードを変更して実現しました。

ファイル配置

  • 配置は前回と同様の構成になります。ただし、model.py は 独自修正を加えたライブラリのファイルです。
+--- jump.feature     // Feature ファイル
+--- environment.py   // environment.py
+--- steps
     +--- jump.py     // Stepファイル
  • ここで model.py とは、Feature/Scenario/ScenarioOutline/Stepといった要素クラスを定義し、Featureファイルに含まれる各要素を管理・実行するモジュールです。model.py は筆者の環境では pip install behave にて下記ディレクトリにインストールされていました。
/usr/local/lib/python3.10/site-packages/behave/model.py

繰り返し処理の実装方針について

behave は Feature クラスの配列 (scenarios) に実行するシナリオ群を格納しています。Feature クラスでは scenarios を for ループで先頭から順に実行しています。また、Scenario クラスでは step 群を管理しています。今回は、シナリオについて繰り返し処理を実装します。

  • 実装の概要は下記となります。
  • 繰り返し回数を指定するStepを新たに定義します。さらに environment.py にあるフック関数からシナリオのインデックスを指定できるようにします。behave の Feature クラスの run 関数をそのインデックスに沿って動作するように拡張します。

  • 上記より、実装対象ファイルは下記となります。
  • ◊Feature クラスの run 関数が実装されているmodel.py
    ◊サンプルテストの environment.py
    ◊サンプルテストの Feature ファイル
    ◊サンプルテストの Step ファイル

Feature クラスの run 関数の変更箇所

  • シナリオインデックスとして、nxt を利用します。
  • Feature.run()を外部からシナリオを指定できるように変更します。このとき、ジャンプ先のシナリオインデックスはcontext.condition[next] から取得します。
  • ジャンプ先が指定されていない場合、nxt は +1 します。
  • scenariosの最後まで到達したら、 break で while 文を抜けます。
  • diff は下記となります。
diff -u ~/python/model.py /usr/local/lib/python3.10/site-packages/behave/model.py
--- /root/python/model.py       2022-07-09 07:22:21.494878832 +0000
+++ /usr/local/lib/python3.10/site-packages/behave/model.py     2022-07-12 02:05:57.188915295 +0000
@@ -311,19 +311,33 @@
                     formatter.background(self.background)

         if not skip_feature_untested:
-            for scenario in self.scenarios:
-                # -- OPTIONAL: Select scenario by name (regular expressions).
+            nxt = 0
+            while True:
                 if (runner.config.name and
-                        not scenario.should_run_with_name_select(runner.config)):
-                    scenario.mark_skipped()
+                        not self.scenarios[nxt].should_run_with_name_select(runner.config)):
+                    self.scenarios[nxt].mark_skipped()
+                    nxt += 1
+                    if nxt >= len(self.scenarios):
+                        break
                     continue

-                failed = scenario.run(runner)
+                failed = self.scenarios[nxt].run(runner)
                 if failed:
                     failed_count += 1
                     if runner.config.stop or runner.aborted:
                         # -- FAIL-EARLY: Stop after first failure.
                         break
+                cond = getattr(runner.context, "condition", None)
+                if cond['enable'] and cond['kind'] == 'scenario':
+                    n = int(cond['next'])
+                    if n >= 0:
+                        nxt = int(n)
+                    else:
+                        nxt += 1
+                else:
+                    nxt += 1
+                if nxt >= len(self.scenarios):
+                    break

         self.clear_status()  # -- ENFORCE: compute_status() after run.
         if not self.scenarios and not run_feature:

Featureファイル

  • 前回の条件分岐のデグレ確認のため、Scenario B の先頭行の Given でテストを失敗させて、次の Given ステップまでジャンプします。
  • テスト用のステップ “retry_test count <リトライ回数>を追加して、ステップを任意の回数失敗できるようにしています。
  • 本Featureでは3回失敗させています。
  • Scenario B の 2 個目のGivenで retry_count が 3 になるまで Scenario A にジャンプします。retry_count が 3 になった場合は、Scenario B の次行(when) 以降が逐次実行されます。

※今回、内部カウンタの初期値は 0 としているため、Scenario A は初回+3 回リトライの計4回実行されます。

Stepファイル

  • @step(‘event sleep {sec}‘)は、前回と同様です。
  • @step(‘retry_test count {count}’)にて定義されるステップ関数にて、下記処理を実施します
    ◊ジャンプ条件やジャンプ先などをcontext.conditionに格納
    ◊g_retry_count をインクリメントしつつ、count よりも小さい場合は Exception を発行してジャンプ処理を実施
  • context.conditionのプロパティ一覧

# -*- coding: utf-8 -*-

from __future__ import unicode_literals
from __future__ import print_function
from behave import *
import time

g_retry_count = 0

@step('event sleep {sec}')
def step_impl(context, sec):
    sec = _set_condition(context, sec)
    time.sleep(int(sec))

@step('retry_test count {count}')
def step_impl(context, count):
    global g_retry_count
    count = _set_condition(context, count)
    count = int(count)

    if count > g_retry_count:
        g_retry_count += 1
        print(g_retry_count)
        raise Exception
    else:
        g_retry_count = 0

def _set_condition(context, str):
    split = str.split('@')

    if len(split) > 1:
        condition = split[1].split(' ')
        context.condition['cond'] = condition[1]
        context.condition['kind'] = condition[3]
        context.condition['to'] = condition[4]
        context.condition['enable'] = True

    return split[0]

environment.py

フック関数:before_all()

  • 各種初期化を実施します。context.condition[‘next’] を追加しています。

フック関数:before_scenario()

  • シナリオに入る前にcontextに保存したconditionの内容をクリアするかを判定しています。

フック関数:before_step()

  • ステップを実行する前にcontextに保存したスキップ条件をチェックし、条件を満たすまで、hook_failed = Trueに設定することで当該ステップは未実行となり ます。また、status は skipped に設定しておきます。
  • 指定したスキップ先に到達した後は、context.conditionに保存した条件を削除します。

フック関数:after_scenario()

  • シナリオから抜ける際にジャンプ条件をチェックし、必要に応じてシナリオ名からインデックスに変更した値をcontext.condition[‘next‘] に設定します。

フック関数:after_step()

  • ステップから抜ける際に、条件を確認し必要に応じてクリア処理を実施します。
# -*- coding: utf-8 -*-

from __future__ import unicode_literals

from behave.model_core import Status
from behave.model import Scenario

import sys

def before_all(context):
    sys.tracebacklimit = -1
    userdata = context.config.userdata
    continue_after_failed = userdata.getbool('runner.continue_after_failed_step', False)
    Scenario.continue_after_failed_step = continue_after_failed

    context.condition = {
        'enable': False,
        'cond': None,
        'kind': None,
        'to': None,
        'next': None
    }

def before_scenario(context, scenario):
    if context.condition['enable']:
        if context.condition['kind'] == 'scenario':
            _clear_condition(context)

def before_step(context, step):
    if context.condition['enable']:
        if context.condition['kind'] == 'scenario':
            step.hook_failed = True
            step.status = Status.skipped
        elif context.condition['kind'] == 'step':
            if step.keyword != context.condition['to']:
                step.hook_failed = True
                step.status = Status.skipped
            else:
                _clear_condition(context)

def after_scenario(context, scenario):
    cond = context.condition
    if cond['enable'] and cond['kind'] == 'scenario' and scenario.status == _text_to_status(cond['cond']):
        cond['next'] = _scenario_name_to_index(context.feature.scenarios, cond['to'])
    else:
        _clear_condition(context)

def after_step(context, step):
    if step.hook_failed:
        step.hook_failed = False

    cond = context.condition
    if cond['enable']:
        if step.status != _text_to_status(cond['cond']) and step.status != Status.skipped:
            _clear_condition(context)

def _clear_condition(context):
    context.condition['enable'] = False
    context.condition['cond'] = False
    context.condition['kind'] = None
    context.condition['to'] = None

def _text_to_status(text):
    if text == 'Failed':
        return Status.failed
    elif text == 'Passed':
        return Status.passed
    else:
        return None

def _scenario_name_to_index(scenarios, name):
    for i, s in enumerate(scenarios):
        if s.name == name:
            return int(i)
    return None

サンプルテストの実行結果



実行結果の解説

  • コマンドライン引数にcontinue_after_failed_stepを渡しています。
  • 文字色ですが、緑が成功、赤が失敗、水がスキップされたステップです。
  • 今回の実装ではscenariosのindexを指定してジャンプしているため、シナリオ自体が実行されず、ジャンプしたシナリオはログには表示されないです。ただし、ステップはスキップしているだけのため表示されています。

最後に

  • 以上、behave にて繰り返し処理を実現する方法を紹介いたしました。更なる改善案として、Failed判定以外でのジャンプ処理の実装や、飛び先へのエイリアスの使用、ステップのジャンプ実装などが考えられます。
  • 今回ご紹介した方法にて、テストケース上からも条件分岐と繰り返し処理を指定することができるようになり、より複雑なテストをより汎用的な文法で記述することが可能となります。