既存の解析システムに対して pytest-mock と pydantic を活用してクイックに総合テストを実装した話

## はじめに
この記事は 2020 年の RevComm アドベントカレンダー 18 日目の記事です。 17 日目は @enotesupa さんの 「SOQLでSELECT * FROM SOME-TABLEっぽいことする」 でした。

@zomaphone と申します。普段は NLP 系を仕事を主にしておりますが、音声認識周辺の整備とかもしています。

今回は既存のシステムに対して pytest-mock と pydantic を活用してどのようにして短時間で総合テストのテストコード作成したかについて紹介します。

コードが修正しにくい、テストコード書く時間がないという人にとっては意味のある記事かもしれません。

## やりたかったこと

revcomm で開発している解析システムは音声データを入力し、他のシステムと通信をしながら処理をし、最後に解析結果を POST (送信) するというような構成となっております。解析システムではこれまで音声認識処理、要約生成、句読点挿入、スコアリング等といった様々な処理を行っておりそれぞれに対するテストは自動化されていたものの、総合テストはこれまで自動化されておらず、常にマニュアルで確認をしていました。人数も少なかったというのもあり開発に関わる全員がコードをよく理解していたため、総合テストを実装することはそれほど重要ではないと認識していましたが、今後人数が増えることを考えると自動化する必要があると思い、着手しました。
また、将来的にリファクタリングすることも想定しています。その際、改めて何が必要かを考えると、既存の I/O に対して変化があるかを確認する基盤（テスト）ではないかと思いました。そのような基盤があれば、リファクタリングもやりやすくなると思い、着手しました。


## 課題

今回のテストコードを実装するにあたり、3つの課題がありました。

1つは時間が足りなかったことです。従来、品質の高いテストを作成するためにはコードのリファクタリングとテストコードの実装の両方が必要です。しかし、対象となるコードにおいてリファクタリングの要素が多く、かつテストコードの実装にそれほど時間がかけられなかったため、リファクタリングを捨てる必要がありました。

2つ目はそれぞれの送信情報が複雑な形を取っていたことでした。音声データや動画データによってバックエンドに送信する情報が異なり、かつ送信する結果もそれなりの情報量が含まれていたことから、通常の isinstance や演算子を使って assert で実装すると、逆に可読性の悪いテストコードを作成することになり、かえって運用保守がしづらくなると思いました。

3つ目は全ての値を保証できなかったことです。実際にテスト対象に複数のディープラーニングのモデルがあり、モデルの状態によって結果が変わることがあります。結果がモデルに依存している以上、特定の値を出力する保証はないです。また、既存の解析機能は常時改良しているため、出力自体も変わる可能性があります。

## ゴール設定

改めてどのような総合テストが（最低限）必要かを考えた時、「これまで通りリクエストする JSON の形式が意図通りになるかを確認するテスト」がいいのではと思いました。web アプリケーションでやるテストのように 1 vs 1 でフィールドの値を確認するというテストが理想的かもしれませんが、先ほど述べた課題が前提になるとなかなか厳しいです。また、一旦「これまで通りの挙動」をゴールとすることで、例えリファクタリングをしても、リファクタリングによるI/Oの変化は避けられるのではと思いました。

## 実施したこと

今回自動テストを実行するためのライブラリとして **pytest** を使用しました。

- **pytest** は python のテストライブラリの一つで、paramterize や fixture 等で複数のテスト変数を定義することができ、それらを使ってテストすることができます。CI/CDの自動テストでよく使うテストツールの一つです。

また、上記の課題を克服するために、**pytest-mock** を **pydantic** を使いました。

- **pytest-mock** とは `unittest.mock` の薄いラッパーで、既存ライブラリや関数等をモック（既存品の模型）することで pytest 実行時にモックされたオブジェクトの実挙動を回避することができます。

- **pydantic** はデータバリデーションのためのライブラリです。https://pydantic-docs.helpmanual.io/
こちらのライブラリを使用することで、オブジェクトのフィールドの型を確認することができ、また 値自体に制約があった場合にそれらの制約を満たしているかどうかを確認することができます。よくある使い方として、django の restframework のserializer クラスみたいにデータを確認することが一般的です。

今回やりたかったテストに対しては、これらのツールの通常のユースケースとは異なり、 pytest-mock で POST する関数をモックし、pydantic で 渡されたデータが正しい形になっているかを確認しました。解析処理の過程で実行したダウンロード、 GET、 POST に関しては通常の pytest-mock の使い方をしました。このように実装することで、短時間で自動化するための総合テストを実装することができました。

実際にサンプルコードを参考にどのようなことをしたか説明するとわかりやすいと思うので、サンプルコードを元にどういうことをしたかを解説します。

## サンプルコード

以下のようなサンプルコードがあったとします。

```
.
├── __init__.py
└── file.py
```

```__init__.py
from . import file
```

```file.py
import requests
import boto3
from scipy.io.wavfile import read

MY_BUCKET = "MY_BUCKET"

def send_info(url,data):
    # 実際はいろいろな処理
    # …….

    # wav info
    data = get_wav_info(data)

    # post result
    requests.post(url, data=data)

def get_wav_info(data):
    # get boto3
    file = data["wav_dir"]
    s3 = boto3.client('s3')
    s3.download_file(MY_BUCKET, file, f'./_{file}')
    _, wav  = read(f'./_{file}')

    # set left right data
    for i, lr_key in enumerate(["left","right"]):
        data[lr_key] = {
        "wav": wav[i].tolist(),
        "avg": wav[i].mean(),
        "max": wav[i].max(),
        "min": wav[i].min(),
    }

    return data
```

上記のコードを解説すると、
`send_info` 関数を実行すると、`get_wav_info` で音声データを S3 から取得し、平均や最大値等を取得します。
その後、それらを data に追加し、`post` (出力) という処理をしています。

ここで、テストしたいものは POST するタイミングで  `data` という変数が期待通りになっているかです。
 `get_wav_info` のテストだけで済むかもしれませんが、問題を簡易化するためにこのようなサンプルとしました。

では、どのように実装したかを解説します。

## pytest 

まず外部のシステムに依存する前提で pytest を作成しました。

```
mock_pydantic
├── __init__.py
├── file.py
└── tests
    ├── __init__.py
    └── test_file.py
```

```tests/__init__.py

```

```tests/test_file.py
from mock_pydantic import file

def test_evaluate(mocker):

    url = "xxxx"
    data = {
        "class_name":"revcomm",
        "user_count":3,
        "wav_dir":"/tmp/abcdef.wav",
        "users":[
            {
                "first":"ben",
                "last":"frank",
                "active":False,
            },
            {
                "first":"mike",
                "last":"faraday",
                "active":False,
            },
            {
                "first":"rich",
                "last":"feinman",
                "active":True,
            },
        ],
    }
    
    file.send_info(url, data)

   
```

## pytest-mock の適用
次にモックを作成しました。

pytest-mock のおさらいですが、特定のモジュールのクラスや関数等を置き換える場合は以下のように実装します。

```python
def test_something(mocker):
    mocker.patch("some_module.function",mock_function)
    mocker.patch.object(some_module,"function",mock_function)
```

また、クラスや関数等を `Mock()` というインスタンスで置き換えることもできます。

```python
# some_module にある MeCab モックの例
def mock_tagger(x):
    return x

obj = mocker.Mock()
obj.Tagger = mock_tagger
mocker.patch.object(some_module,"MeCab",obj)
```


今回のテストファイルに適用すると以下のように実装しました。

```tests/test_file.py
import numpy as np
import file

def mock_post(url, data):
    assert isinstance(data["class_name"],str)
    assert isinstance(data["user_count"],int)
    assert isinstance(len(data["users"]),int)
    for user in data["users"]:
        assert isinstance(user["first"],str)
        assert isinstance(user["last"],str)
        assert isinstance(user["active"],bool)
    for k in ["left","right"]:
        assert data[k]
        assert isinstance(data[k]["wav"],list)
        for wav_i in data[k]["wav"]:
            assert isinstance(wav_i,float)
        assert isinstance(data[k]["avg"],float)
        assert isinstance(data[k]["max"],float) 
        assert isinstance(data[k]["min"],float)

def test_evaluate(mocker):

    # requests
    mock_requests = mocker.Mock()
    mock_requests.post = mock_post
    mocker.patch.object(file, "requests", mock_requests)

    # boto3
    mock_s3 = mocker.Mock()
    mock_s3.download_file = mocker.Mock()
    mocker.patch.object(file.boto3, "client", mock_s3)

    # scipy_read
    mock_read = mocker.Mock(
        return_value=(None, np.random.random((2,32))),
        )
    mocker.patch.object(file, "read", mock_read)

    url = "xxxx"
    data = {
        "class_name":"revcomm",
        "user_count":3,
        "wav_dir":"/tmp/abcdef.wav",
        "users":[
            {
                "first":"ben",
                "last":"frank",
                "active":False,
            },
            {
                "first":"mike",
                "last":"faraday",
                "active":False,
            },
            {
                "first":"rich",
                "last":"feinman",
                "active":True,
            },
        ]
    }
    
    file.send_info(url, data)
```


`requests`, `boto3`, `scipy` をモックしました。

`mock_post` で出力データが正しい形になっているかを確認しています。
しかし、このままだと `assert` だらけで見通しが悪いです。


## pydanticの適用

最後に pydantic でオブジェクトのフィールドを確認しました。

以下 pydantic を用いたサンプルコードです。

```python
import pytest
from pydantic import BaseModel, StrictStr, StrictInt, StrictFloat, StrictBool

class A(BaseModel):
    a: StrictInt
    b: StrictStr
    c: StrictFloat
    d: StrictBool

good_res = {
    "a":1,
    "b":"hi",
    "c":3.2,
    "d":True,
}

bad_res = {
    "a":"1",
    "b":"hi",
    "c":3.2,
    "d":True,
}

def test_pass():
    A(**good_res)
    print("pass")

def test_not_pass():
    with pytest.raises(Exception) as excinfo:
        A(**bad_res)
    print(excinfo.value.args[0])

```

上記を実行すると以下の結果が得られます。

```bash
>> pytest -v
============================================================================== test session starts ===============================================================================
...
collected 2 items                                                                                                                                                                

test_pydantic.py::test_pass PASSED                                                                                                                                         [ 50%]
test_pydantic.py::test_not_pass PASSED                                                                                                                                     [100%]

=============================================================================== 2 passed in 0.11s ================================================================================
```


総合テストのテストコードでは以下のように実装しました。


```tests/test_file.py
from typing import List, Optional
from pydantic import BaseModel, StrictStr, StrictInt, StrictBool, StrictFloat
import numpy as np
import file


class Name(BaseModel):
    first: StrictStr
    last: StrictStr
    active: Optional[StrictBool]

class WAV_INFO(BaseModel):
    wav: List[StrictFloat]
    avg: StrictFloat
    min: StrictFloat
    max: StrictFloat

class Body(BaseModel):
    class_name: StrictStr
    user_count: StrictInt
    users: List[Name]
    left: WAV_INFO
    right: WAV_INFO

def mock_post(url, data):
    Body(**data)


def test_evaluate(mocker):

    # requests
    mock_requests = mocker.Mock()
    mock_requests.post = mock_post
    mocker.patch.object(file, "requests", mock_requests)

    # boto3
    mock_s3 = mocker.Mock()
    mock_s3.download_file = mocker.Mock()
    mocker.patch.object(file.boto3, "client", mock_s3)

    # scipy_read
    mock_read = mocker.Mock(
        return_value=(None,np.random.random((2,32))),
        )
    mocker.patch.object(file, "read", mock_read)

    url = "xxxx"
    data = {
        "class_name":"revcomm",
        "user_count":3,
        "wav_dir":"/tmp/abcdef.wav",
        "users":[
            {
                "first":"ben",
                "last":"frank",
                "active":False,
            },
            {
                "first":"mike",
                "last":"faraday",
                "active":False,
            },
            {
                "first":"rich",
                "last":"feinman",
                "active":True,
            },
        ]
    }
    
    file.send_info(url, data)
```

assert だらけの箇所がとてもシンプルになりました。

実行すると以下になります。

```bash
>> pytest -v
============================================================================== test session starts ===============================================================================
...
collected 1 item                                                                                                                                                                 

tests/test_file.py::test_evaluate PASSED                                                                                                                                   [100%]

=============================================================================== 1 passed in 0.31s ================================================================================

```
無事動くことも確認できました。


## まとめ
pytest-mock と pydantic を使ってどのように総合テストを実装したかを解説しました。

本来、自動テストを作成する時、リファクタリングとテストコードの実装を同時に行う必要があります。しかし、上記のようにリファクタリングする要素が多すぎたり、時間に制約があったりすると、なかなかテストコードが作成できないという問題があります。今回のようにこれまでのI/Oを担保できるかをテストしたい場合、今回紹介した  pydantic と pytest-mock を用いた手法は有効な手法の一つかもしれません。
もちろん、本来の pytest-mock と pydantic のユースケースに比べると、極めてオーソドックスではない使い方をしています。正しいユースケースについてはそれぞれの公式ドキュメントをご覧ください。