mizzsugar’s blog

日々感じていることや学んだことを書きます。エンジニアリング以外にも書くかもしれません。

【Python】GoogleAPIを利用したメソッドのテストでモックを使う

Python

外部APIを利用したメソッドの単体テスト書きたいなあ・・・　

単体テストで何回も外部APIに接続するのはお行儀悪いって聞くけどどう書けば良いのだろう？と思い。

簡単な機能のテストを書きました。

ちょっと変更しましたが、Google Calendar APIのquick startでやっていることをやります。

developers.google.com


一つのメソッドが長いので下記のような役割のメソッドに分割しました。

  1. pickleに保存した認証情報をロードする

  2. 認証情報が存在しないor有効期限切れならば認証情報を再生成する

  3. イベント情報を取得する役割のオブジェクトを生成する

  4. イベント情報を取得する


今回テストするプロダクトコード↓

event.py

from dataclasses import dataclass
from typing import Iterable, Optional, Dict
import datetime
import pickle
import os.path
import pathlib

import googleapiclient.discovery
import google_auth_oauthlib.flow
from google.auth.transport.requests import Request
import google.oauth2.credentials
import iso8601


# 環境変数にした方がよいですが、ブログではわかりやすいように表示しています
SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']


class ServiceBuilder:
    @classmethod
    def renew_token(cls, credentials: google.oauth2.credentials.Credentials) -> google.oauth2.credentials.Credentials:
        # google.oauth2.credentials.Credentialsは外部APIのクラス
        if credentials.expired and credentials.refresh_token:
            # 外部APIに接続
            credentials.refresh(Request())
        else:
            # from_client_secrets_file returns google_auth_oauthlib.flow.Flow object
            flow = google_auth_oauthlib.flow.InstalledAppFlow.from_client_secrets_file(
                'credentials.json', SCOPES)
            credentials = flow.run_local_server()
        return credentials

    @classmethod
    def build(cls, credentials: google.oauth2.credentials.Credentials) -> googleapiclient.discovery.Resource:
        """トークンを元にServiceを生成します
        """

        # If there are no (valid) credentials available, let the user log in.
        # TODO: 42-43はrenew_tokenの先頭の方がよい
        if not credentials.valid:
            credentials = cls.renew_token(credentials)
        
        # 外部APIに接続
        return googleapiclient.discovery.build('calendar', 'v3', credentials=credentials)


@dataclass
class Event:
    summary: str
    start: datetime.datetime
    end: datetime.datetime


class EventAdapter:
    @classmethod
    def from_google_api(cls, google_event: Dict) -> Event:
        """GoogleAPIから取得したDict型のイベント情報をEventオブジェクトに変換します

        Dict型のままだと利用しない情報も含まれている、かつネストが深くて使いづらいため

        iso8601について
        https://pypi.org/project/iso8601/
        """
        return Event(
            summary=google_event.get('summary'),
            start=iso8601.parse_date(google_event.get('start').get('dateTime')),
            end=iso8601.parse_date(google_event.get('start').get('dateTime'))
        )


class EventManager:
    @classmethod
    def fetch_events(cls, number: int, service: googleapiclient.discovery.Resource) -> Iterable[Event]:
        """Google Calendar APIからイベントを取得します。

        number: int 取得するイベントの最大数
        service: イベントを取得するResourceオブジェクト
        """
        # service = ServiceBuilder.build()
        now = datetime.datetime.utcnow().isoformat() + 'Z' # 'Z' indicates UTC time
        print('Getting the upcoming 10 events')
        events_result = service.events().list(calendarId='primary', timeMin=now,
                                            maxResults=10, singleEvents=True,
                                            orderBy='startTime').execute()
        events = events_result.get('items', [])

        if not events:
            return []
        return [EventAdapter.from_google_api(event) for event in events]

main.py

import pathlib
from typing import Optional
import pickle

import google.oauth2.credentials

import event


def load_token(file: pathlib.Path) -> Optional[google.oauth2.credentials.Credentials]:
    """トークンをロードします。

    pickleの中にtokenがなければNoneを返します。
    """
    try:
        # pathlib.Path('token.pickle').read_bytes()をモックしたテスト
        with file.open('rb') as f:
            return pickle.load(f)
    except FileNotFoundError:
        return None


def main():
    token_file = pathlib.Path('token.pickle')
    credentials = event.ServiceBuilder.renew_token(credentials=load_token(token_file))

    with token_file.open('wb') as f:
        pickle.dump(credentials, f)
    
    service = event.ServiceBuilder.build() # missing one argument
    events = event.EventManager.fetch_events(number=2, service=service)
    print(events)

if __name__ == '__main__':
    main()


テストしたいこと

  1. 認証情報が有効切れor認証情報がpickleにない場合は新たにトークンが生成されるかどうか

  2. fetch_eventsでイベントが取得されるかどうか。

  3. fetch_eventsでGoogle Calendar APIから取得されたイベントが0件の場合は空の配列が返されるかどうか


from_google_apiとload_tokenに関しては

ちゃんとインスタンスができているか？　例外を発生するか？

といったようにPythonの仕様の確認になってしまうのでテストしないことにしました。


1. 認証情報が有効切れor認証情報がpickleにない場合は新たにトークンが生成されるかどうか

まず、どこが外部APIに接続しているか調べます。

class ServiceBuilder:
    @classmethod
    def renew_token(cls, credentials: google.oauth2.credentials.Credentials) -> google.oauth2.credentials.Credentials:
        # google.oauth2.credentials.Credentialsは外部APIのクラス
        if credentials.expired and credentials.refresh_token:
            credentials.refresh(Request())  # 外部APIに接続
        else:
            # from_client_secrets_file returns google_auth_oauthlib.flow.Flow object
            flow = google_auth_oauthlib.flow.InstalledAppFlow.from_client_secrets_file(
                'credentials.json', SCOPES)   # 外部APIに接続
            credentials = flow.run_local_server()   # 外部APIに接続
        return credentials


「外部APIに接続」の部分をモックすれば人様のAPIに迷惑をかけない！

renew_token 自体をテストしようとしました。

でも、renew_token自体をテストしても良いけれども、 renew_tokenをまるっとモックしてServiceBuilder.buildで確認しちゃった方が楽だなあと。

credentailsの状態によってbuildの振る舞いが変わるので credentialsのモックを作り、credentialsモックの状態がそれぞれ異なる場合のテストを作ります。

test.py

@unittest.mock.patch('event.ServiceBuilder.renew_token')
@unittest.mock.patch('googleapiclient.discovery.build')
def test_valid_token(mock_build, mock_renew_token):
    credentials = unittest.mock.Mock(valid=True)

    event.ServiceBuilder.build(credentials=credentials)
    assert not mock_renew_token.called

@unittest.mock.patch('event.ServiceBuilder.renew_token')
@unittest.mock.patch('googleapiclient.discovery.build')
def test_expired_token_has_refresh_token(mock_build, mock_renew_token):
    credentials = unittest.mock.Mock(valid=False, expired=True, refresh_token='I am refresh token')

    event.ServiceBuilder.build(credentials=credentials)
    assert  mock_renew_token.called

credentailsの状態が妥当でない場合に認証情報が再生成されるメソッドが呼ばれるかどうかを確認できました！


2. fetch_eventsでイベントが取得されるかどうか。

3. fetch_eventsでGoogle Calendar APIから取得されたイベントが0件の場合は空の配列が返されるかどうか

イベントの取得のメソッドテストでは、イベントをちゃんと取得できるかのみを確認したいです。

serviceにGoogle Calendar APIからイベントを取得させていますが、serviceに利用されているcredentialsが有効期間切れでrefresh_tokenが実行されてしまう・・・

なんてことがあったら面倒です。


なので、fetch_eventsの外でserviceを生成させ、serviceをfetch_eventsの引数にします。

fetch_eventsのテストでは、serviceをモックし、serviceが正常に動くことを前提とすることにしました。

Google Calendar APIのドキュメントを確認すると、 イベント取得メソッドは下記のようになっています。

service.events().list(calendarId='primary', timeMin=now,
                                        maxResults=10, singleEvents=True,
                                        orderBy='startTime').execute()

https://developers.google.com/calendar/quickstart/python


serviceって何クラスのオブジェクトだと調べました。

from googleapiclient.discovery import build

service = build('calendar', 'v3', credentials=creds)


serviceはbuildによって生成されたオブジェクトらしい・・・

googleapiclient.discoveryのbuildはResourceオブジェクトを返すそうです。

https://github.com/googleapis/google-api-python-client/blob/master/googleapiclient/discovery.py#L170


Resourceを探したところ、

googleapiclient.discoveryのResourceだとわかりました。

https://github.com/googleapis/google-api-python-client/blob/master/googleapiclient/discovery.py#L995


ということでserviceはResourceオブジェクトをモックしよう。

service.events().list(calendarId='primary', timeMin=now,
                                        maxResults=10, singleEvents=True,
                                        orderBy='startTime').execute()


上記をもってGoogle Calendar APIからイベント情報のDictが返ってきます。

ということで、serviceをモックしてこんな感じでイベント情報を返すとします。

実際のAPIではもっとたくさんの情報が返されますが

テストに必要な情報のみにしました。

https://developers.google.com/calendar/v3/reference/events/list

mock_resource.events.return_value.list.return_value.execute.return_value = {
        "items": [
                    {
                        "summary": "test",
                        "start": {
                            "dateTime": "2019-06-03T02:00:00+09:00"
                        },
                        "end": {
                            "dateTime": "2019-06-03T02:45:00+09:00"
                        },
                    },
                        {
                        "summary": "test",
                        "start": {
                            "dateTime": "2019-06-03T02:00:00+09:00"
                        },
                        "end": {
                            "dateTime": "2019-06-03T02:45:00+09:00"
                        },
                    },
        ]
    }

そして2と3のテストを

@unittest.mock.patch('googleapiclient.discovery.Resource')
def test_fetch_events(mock_resource):
    mock_resource.events.return_value.list.return_value.execute.return_value = {
        "items": [
                    {
                        "summary": "test",
                        "start": {
                            "dateTime": "2019-06-03T02:00:00+09:00"
                        },
                        "end": {
                            "dateTime": "2019-06-03T02:45:00+09:00"
                        },
                    },
                        {
                        "summary": "test",
                        "start": {
                            "dateTime": "2019-06-03T02:00:00+09:00"
                        },
                        "end": {
                            "dateTime": "2019-06-03T02:45:00+09:00"
                        },
                    },
        ]
    }
    actual = event.EventManager.fetch_events(number=2, service=mock_resource)
    assert 2 == len(actual)

    for item in actual:
        assert isinstance(item, event.Event)


@unittest.mock.patch('googleapiclient.discovery.Resource')
def test_fetch_no_events(mock_resource):
    mock_resource.events.return_value.list.return_value.execute.return_value = {}

    actual = event.EventManager.fetch_events(number=2, service=mock_resource)
    assert [] == actual


ポイントは

  • 外部APIのオブジェクトは何かを調べてそのクラスのオブジェクトをモックすること

  • return_valueをどう実行するか

だと思いました。

外部APIのソースをある程度読むことになり、少し大変でしたが 勉強になりました。


今回のソースコードはこちら

github.com


また、NaritoさんがDjangoを利用してGoogle Calendar APIと連携したアプリケーションを作成されていました。

https://torina.top/detail/466/


近いうちにDjangoとGoogle Calendar APIのテストの記事も書こうかなと思います。

（内容はあまり変わらないと思いますが）