StackStorm の Workflow テスト開発事始め

はじめに

StackStorm(以下st2) の Workflow を開発する機会があるのですが、テストを開発する方法を知る機会があったので紹介します。

前提

pack の action や Workflow をテストするためには、事前に StackStorm がインストールされ、st2 コマンドが実行できるようになっている必要があります。

• StackStorm がインストールできていること
• st2 関連コマンドが使える状態になっていること
• pack が開発できるようになっていること

本記事では事始めということで簡単に前提をクリアする方法を紹介します。
既に前提をクリアできている場合は読み飛ばしてください。

StackStorm がインストールできていること

方法としては StackStorm 公式ドキュメントの Installation の項目 に書かれています。

開発が必要な時だけ動かす場合を考えると、docker-compose を使う方法がよいでしょう。
Usage に書かれた 5 step を実行すればインストールと起動は完了です。

起動後、https://localhost にアクセスすると StackStorm のログイン画面が表示されますので、conf/stackstorm.env に書かれたパスワード(st2-admin)を入力することでログインできるようになります。

ログイン画面	ログイン後画面(実行履歴画面)

st2 関連コマンドが使える状態になっていること

インストールした方法に応じてそれぞれ異なります。

• st2-docker の StackStorm コンテナ内のシェルが使えていること
• st2 用の virtualenv (PyPI package) が activate されていること

st2-docker を使ってインストールした場合は以下の方法で st2 コマンドが使えます。

StackStormコンテナのシェルを使う

$ docker-compose exec stackstorm bash
# 以降、`st2～` といったコマンドはこのシェル上で実行する
root@0abeff4b43bf:/# st2 --version
st2 3.1.0, on Python 2.7.6

pack が開発できるようになっていること

pack をテストするコードは pack の中に含まれるため、テストコードを書くためには pack が開発できるようになっている必要があります。

st2-docker を使ってインストールした場合は st2-docker の README.md にあるとおり packs.dev ディレクトリが作成されます。

このディレクトリは st2 にインストールされた pack を指すディレクトリとして設定されているため、packs.dev 配下に pack のソースコードを配置して開発することができます。

尚、Workflow の内容(spec) を変更した場合は最新の状態になりますが、Workflow の meta ファイルを変更した場合は更新された内容を読み込ませる必要があるため、 st2ctl reload --register-all コマンドを実行する必要があるので注意してください。

st2 のテスト環境について

st2 の pack に含まれる action や Workflow をテストする環境として、st2-run-pack-tests コマンドが用意されています。

st2-run-pack-tests -p ${PATH_TO_PACK} ※ を実行すると ${PATH_TO_PACK}/tests 配下にある全テストが実行されます。
※ ${PATH_TO_PACK} は適宜 pack へのパスに置き換えて下さい

実行されるテストは Python の unittest フレームワークに即したメソッドです。(参考)

• test から始まる Python ファイル (例: test_XXX.py)
• unittest が提供する TestCase クラスを継承している
• test から始まるインスタンスメソッド

例えば以下のテストコードがあった場合、テストを実行すると文字列 "hello world" が表示され、テストは成功します。

tests/test_orquesta.py

import unittest

class TestOrquesta(unittest.TestCase):
    def test_hello_world(self):
        s = 'hello world'
        self.assertEqual(s, 'hello world')

root@0abeff4b43bf:/opt/stackstorm/packs.dev# st2-run-pack-tests -j -p tutorial-example -f tests/test_orquesta.py
Running tests for pack: tutorial-example
Activating virtualenv in /tmp/st2-pack-tests-virtualenvs/tutorial-example...
Running tests...
test_hello_world (test_orquesta.TestOrquesta) ... passed





TEST RESULT OUTPUT:

-----------------------------------------------------------------------------
1 test run in 0.002 seconds (1 test passed)

以上のように、テストを作る場合は TestCase を基底クラスとする Python クラスを <pack_name>/tests ディレクトリ配下に保存していくことになります。

この際、StackStorm テスト用に TestCase クラスを基底とするクラスや、Action Runner 等のモックが st2tests にて定義されているため、import してから継承して利用します。(参考)

何を import すればよいか分からない場合は StackStorm におけるコンポーネントの名前が理解できていない可能性があります。参考情報として st2 におけるコンポーネントの名前 をまとめましたので参考にしてみて下さい。

Orquesta Workflow 内タスクのユニットテストを作成する

Pack のテストコードを記述していくことにします。

今回、Orquesta Workflow 内タスクの YAQL expression に対するユニットテストを作成することにします。

※ 本記事を執筆した '20/01/12 現在は Orquesta の Workflow をテストする方法は st2 の公式ドキュメントには記載されておらず、st2 のソースコードを読み進めながら解釈した内容です。
そのため、内容が最適ではない可能性や、将来的に公式の方法が公開される可能性があるので、公式ドキュメントは必ず参考にするようにしてください。

テスト対象となる Workflow

まずはテストする対象となる Workflow のコードを紹介します。
指定された任意の Linux コマンドを実行し、その結果として stdout に 1 行以上出力されたかどうかを確認するものです。

actions/orquesta-basic.meta.yaml

---
name: orquesta-basic
pack: tutorial-example
description: Run a local linux command
runner_type: orquesta
entry_point: workflows/orquesta-basic.yaml
enabled: true
parameters:
    cmd:
        required: true
        type: string

actions/workflows/orquesta-basic.yaml

version: 1.0

description: Execute an arbitrary linux command and confirm at least one line is outputed.

input:
  - cmd

tasks:
  task1:
    action: core.local cmd=<% ctx(cmd) %>
    next:
      - when: <% succeeded() %>
        publish:
          - cmd_result: <% result().stdout %>
          - num_cmd_result_lines: <% (ctx().cmd_result.split("\n").where($.len() > 0).len()) %>
        do: task2

  task2:
    action: core.noop
    next:
      - when: <% ctx().num_cmd_result_lines > 0 %>
        do:
          - noop
      - when: <% ctx().num_cmd_result_lines = 0 %>
        do:
          - fail

output:
  - cmd_result: <% ctx().cmd_result %>
  - num_cmd_result_lines: <% ctx().num_cmd_result_lines %>

Workflow を実行すると次のように、与えられたコマンドの内容によって成功・失敗します。

• cmd=echo workflow will successfully finish を指定した場合は、stdout に 1 行表示されるため "成功" します
• cmd=echo workflow will fail > /dev/null を指定した場合は stdout に何も表示されないため "失敗" します

成功時	失敗時

テストコードを記述する

上記のように、stdout の結果に応じて成功・失敗できるかテストコードを書いてみましょう。

テストする方針は次のとおりとなります。

1. テスト対象の task を含む Workflow ファイル全体を読み込む (actions/workflows/orquesta-basic.yaml)
2. Workflow からテスト対象の YAQL 記述を取り出す (tasks.task1.next[0].publish[1].num_cmd_result_lines)
3. YAQL を評価する (<% (ctx().cmd_result.split("\n").where($.len() > 0).len()) %>)
4. 評価結果をテストする

Workflow を読み込むためには orquesta.tests.fixtures モジュールの loader が利用できます。
また、YAQL を評価するためには orquesta.utils モジュールの plugin が利用できます。

まずはテストする方針に従ってべた書きしてみます。

tests/test_orquesta.py

import unittest

import os

from orquesta.tests.fixtures import loader as fixture_loader
from orquesta.utils import plugin as plugin_util

class TestOrquesta(unittest.TestCase):
    def test_task1(self):
        # 1. Read whole workflow file
        wf_spec_file_path = os.path.join(os.path.dirname(__file__), "..", "actions", "workflows", 'orquesta-basic.yaml')
        wf_def = fixture_loader.get_fixture_content(
            wf_spec_file_path,
            'workflows'
        )
        
        # 2. Read YAQL expression in workflow
        cmd_result_expr = wf_def['tasks']['task1']['next'][0]['publish'][0]['cmd_result']
        num_cmd_result_lines_expr = wf_def['tasks']['task1']['next'][0]['publish'][1]['num_cmd_result_lines']
        
        # 3-a. Prepare YAQL Evaluator
        # ref. https://github.com/StackStorm/orquesta/blob/9706dc1f13a44deadb6cefc8e92c83caad591127/orquesta/tests/unit/expressions/test_facade_yaql_evaluate.py
        evaluator = plugin_util.get_module(
            'orquesta.expressions.evaluators',
            'yaql'
        )
        
        # -- Test case of success --
        # When result has one line, num_cmd_result_lines equal to 1
        # 
        # 3-b. Evaluate YAQL expression
        # 4. Assert result of YAQL expression
        one_line_result = {
            '__current_task': {
                'result': {
                    'stdout': 'workflow will successfully finish'
                }
            }
        }
        ctx = {
            'cmd_result': evaluator.evaluate(
                cmd_result_expr,
                one_line_result
            )
        }
        expected_num_cmd_result_lines = 1
        self.assertEqual(
            expected_num_cmd_result_lines,
            evaluator.evaluate(
                num_cmd_result_lines_expr, ctx
            )
        )
        
        # -- Test case of fail --
        # When result empty line, num_cmd_result_lines equal to 0
        # 
        # 3-b. Evaluate YAQL
        # 4. Assert result of YAQL expression
        empty_result = {
            '__current_task': {
                'result': {
                    'stdout': ''
                }
            }
        }
        ctx = {
            'cmd_result': evaluator.evaluate(
                cmd_result_expr,
                empty_result
            )
        }
        expected_num_cmd_result_lines = 0
        self.assertEqual(
            expected_num_cmd_result_lines,
            evaluator.evaluate(
                num_cmd_result_lines_expr,
                ctx
            )
        )

テストコードを実行すると、テストは期待通りの結果となりパスすることが分かります。

root@0abeff4b43bf:/opt/stackstorm/packs.dev# st2-run-pack-tests -j -p ./tutorial-example/ -f tests/test_orquesta.py
Running tests for pack: tutorial-example
Activating virtualenv in /tmp/st2-pack-tests-virtualenvs/tutorial-example...
Running tests...
test_task2 (test_orquesta.TestOrquesta) ... /opt/stackstorm/st2/local/lib/python2.7/site-packages/cryptography/hazmat/primitives/constant_time.py:26: CryptographyDeprecationWarning: Support for your Python version is deprecated. The next version of cryptography will remove support. Please upgrade to a 2.7.x release that supports hmac.compare_digest as soon as possible.
  utils.PersistentlyDeprecated2018,
passed





TEST RESULT OUTPUT:

-----------------------------------------------------------------------------
1 test run in 1.185 seconds (1 test passed)

今回テストしたいコードは以上で完成ですが、テストコードが冗長にならず再利用が可能になるようにリファクタしてみます。

リファクタする方針としては、以下の基本的なものは base.py としてまとめ、テストする時に継承できるようにします。

• unittest の基底クラスを継承する
• Workflow を読み込む処理
• YAQL を評価する処理

そして task のテストは単体で実行する private メソッドを作成して、テストするケース毎に INPUT データと期待される値を指定して単体テストを実行することとします。

結果次のようになります。

tests/base.py

import unittest

import os

from orquesta.tests.fixtures import loader as fixture_loader
from orquesta.utils import plugin as plugin_util

class TestBase(unittest.TestCase):
    # Read whole workflow file
    def get_wf_def(self, rel_wf_path, raw=False):
        wf_spec_file_path = os.path.join(
            os.path.dirname(__file__),
            "..",
            "actions",
            "workflows",
            rel_wf_path)
        return fixture_loader.get_fixture_content(
            wf_spec_file_path,
            'workflows',
            raw=raw
        )

    # Evaluate YAQL expression
    def evaluate_yaql(self, yaql_expr, ctx):
        evaluator = plugin_util.get_module(
            'orquesta.expressions.evaluators',
            'yaql'
        )
        return evaluator.evaluate(yaql_expr, ctx)

tests/test_orquesta.py

from base import TestBase

class TestOrquesta(TestBase):
    def test_task1(self):
        # -- Test case of success --
        # When result has one line, num_cmd_result_lines equal to 1
        one_line_stdout = 'workflow will successfully finish'
        expected_num_cmd_result_lines = 1
        self.__test_task1(expected_num_cmd_result_lines, one_line_stdout)
        
        # -- Test case of fail --
        # When result empty line, num_cmd_result_lines equal to 0
        empty_stdout = ''
        expected_num_cmd_result_lines = 0
        self.__test_task1(expected_num_cmd_result_lines, empty_stdout)

    def __test_task1(self, expected_num_cmd_result_lines, stdout):
        # 1. Read whole workflow file
        wf_def = self.get_wf_def('orquesta-basic.yaml')
        # 2. Read YAQL expression in workflow
        publish_expr = wf_def['tasks']['task1']['next'][0]['publish']
        
        cmd_result_expr = publish_expr[0]['cmd_result']
        num_cmd_result_lines_expr = publish_expr[1]['num_cmd_result_lines']
        
        # 3 Evaluate YAQL expression
        current_task = {
            '__current_task': {
                'result': {
                    'stdout': stdout
                }
            }
        }
        ctx = {
            'cmd_result': self.evaluate_yaql(
                cmd_result_expr,
                current_task
            )
        }
        self.assertEqual(
            expected_num_cmd_result_lines,
            self.evaluate_yaql(
                num_cmd_result_lines_expr,
                ctx
            )
        )

テストしたい内容が分かりやすくなったと思います。

最後に

st2 の Workflow 内の YAQL をテストする方法を紹介しました。

Python の unittest フレームワークに従ってテストを作成することとなり、action をテストする場合はベースとなるクラスが存在するものの、Workflow についてのクラスは用意されていないため、st2 のソースコードから必要な module を import してテストコードを作成する必要がありました。('20/01/12現在)

今回は action は実行せずに task のテストを行いましたが、もし Workflow 内の action を実行した上で、その結果をテスト出来る方法が分かれば別途執筆しようと思います。

参考情報

• st2 express の pack 一覧には test コードが存在するかどうかを示す icon がある
  • 実際に稼働する test のコードを参考にするとよいでしょう

st2 におけるコンポーネントの名前

テストする対象を正しく理会するために、st2 におけるコンポーネントの名前を正しく理解しましょう。(参考)

名前	説明
engine	Orquesta や Mistral 等の Workflow を実行する主体
composer	Workflow モデルにより構成されるグラフと conductor の組み合わせ
conductor	グラフに従って Workflow の実行を指揮する主体

• engine は以下から構成されます
  • 言語仕様毎に作成される Workflow モデル
  • composer
  • conductor

st2 のテストコマンド st2-run-pack-tests

st2-run-pack-testsコマンドのUsage

$ st2 --version
st2 3.1.0, on Python 2.7.6
$ st2-run-pack-tests
Usage: /usr/bin/st2-run-pack-tests  [-c] [-j] [-t] [-v] [-x] -p <path to pack> [-f <test file module / class / method>]
  -f : Run a specific test file or test class method (e.g. -f test_action_download:DownloadGitRepoActionTestCase.test_run_pack_download)
  -c : Run tests with code coverage reporting enabled
  -j : Just run tests. Use previously installed dependencies
       and virtualenv, if any, for subsequent test runs.
  -t : Run tests with timing enabled
  -v : Verbose mode (log debug messages to stdout).
  -x : Do not create virtualenv for test for tests, e.g. when running in existing one.

参考: https://docs.stackstorm.com/development/pack_testing.html