ローカルでの Python 2 の単体テスト

単体テストを行うと、作成したコードの品質を検証するだけでなく、コードの作成を進めながら開発プロセスの改善を行えます。アプリケーションの開発が終わった後にテストを作成するのではなく、開発プロセスと並行して単体テストを作成することを検討してください。これにより、保守が簡単で再利用できる単体コードを作成できます。また、コードのテストをすばやく、徹底的に行うことができます。

ローカルで単体テストを行う場合、ユーザーの開発環境内ですべてのテストを行います。リモートのコンポーネントは使用しません。App Engine のテスト ユーティリティを使用すると、データストアや他の App Engine サービスをローカルに実装できます。コードを App Engine にデプロイする必要はありません。サービススタブにより、これらのサービスをローカルで利用できるようにします。

サービススタブはサービスの動作をシミュレートします。たとえば、データストアと Memcache のテストを作成するで説明するデータストアのサービススタブでは、実際のデータストアにリクエストを送信することなく、データストアのコードをテストできます。データストアの単体テストで生成されるエンティティは、データストアではなくメモリ上に格納され、テストの実行後に削除されます。データストア自体に依存することがないため、短時間で実行できる小規模なテストを実行できます。

このドキュメントでは複数のローカル App Engine サービスに対する単体テストを作成する方法を説明し、テスト用フレームワークを設定するうえでの詳細情報を提供します。

Python 2 テスト用ユーティリティの概要

App Engine の testbed という Python モジュールでは、単体テスト用のサービススタブを使用できます。

以下のサービスのサービススタブが用意されています。

  • App Identity（init_app_identity_stub）
  • Blobstore（init_blobstore_stub を使用）
  • Capability（init_capability_stub を使用）
  • Datastore（init_datastore_v3_stub を使用）
  • Files（init_files_stub を使用）
  • Images（dev_appserver のみ、init_images_stub を使用）
  • LogService（init_logservice_stub を使用）
  • Mail（init_mail_stub を使用）
  • Memcache（init_memcache_stub を使用）
  • Task Queue（init_taskqueue_stub を使用）
  • URL fetch（init_urlfetch_stub を使用）
  • User service（init_user_stub を使用）

すべてのスタブを同時に初期化するには、init_all_stubs を使用します。

データストアと Memcache のテストを作成する

ここでは、データストア サービスと Memcache サービスの使用をテストするコードの作成例を示します。

テストランナーに、Python ロードパスに App Engine ライブラリ yaml（App Engine SDK に同梱）を含む適切なライブラリとアプリケーション ルートが設定されていること、アプリケーション コードで前提とされているライブラリパス（ローカル ./lib ディレクトリがある場合は、そのディレクトリなど）に対する変更がすべて適用されていることを確認します。例:

import sys
sys.path.insert(1, 'google-cloud-sdk/platform/google_appengine')
sys.path.insert(1, 'google-cloud-sdk/platform/google_appengine/lib/yaml/lib')
sys.path.insert(1, 'myapp/lib')

まず、テスト対象のサービスに関連する Python unittest モジュールと App Engine モジュールをインポートします。この場合は、memcache と ndb（データストアおよび Memcache を使用）です。また、testbed モジュールもインポートします。

import unittest

from google.appengine.api import memcache
from google.appengine.ext import ndb
from google.appengine.ext import testbed

次に、TestModel クラスを作成します。この例では、エンティティが Memcache に格納されるかどうかをチェックする関数を使用します。この関数では、エンティティが見つからない場合、データストア内でエンティティの有無をチェックします。ndb は Memcache 自体をバックグラウンドで使用するため、実際にはこのチェックは冗長であることが多いですが、それでもテストの OK パターンです。

class TestModel(ndb.Model):
    """A model class used for testing."""
    number = ndb.IntegerProperty(default=42)
    text = ndb.StringProperty()


class TestEntityGroupRoot(ndb.Model):
    """Entity group root"""
    pass


def GetEntityViaMemcache(entity_key):
    """Get entity from memcache if available, from datastore if not."""
    entity = memcache.get(entity_key)
    if entity is not None:
        return entity
    key = ndb.Key(urlsafe=entity_key)
    entity = key.get()
    if entity is not None:
        memcache.set(entity_key, entity)
    return entity

次に、テストケースを作成します。どのサービスをテストする場合でも、テストケースでは Testbed インスタンスを作成してアクティブ化する必要があります。また、関連するサービススタブを初期化する必要もあります。この場合は、init_datastore_v3_stub と init_memcache_stub を使用します（他の App Engine サービススタブを初期化するメソッドについては、Python テスト用ユーティリティの概要をご覧ください）。

class DatastoreTestCase(unittest.TestCase):

    def setUp(self):
        # First, create an instance of the Testbed class.
        self.testbed = testbed.Testbed()
        # Then activate the testbed, which prepares the service stubs for use.
        self.testbed.activate()
        # Next, declare which service stubs you want to use.
        self.testbed.init_datastore_v3_stub()
        self.testbed.init_memcache_stub()
        # Clear ndb's in-context cache between tests.
        # This prevents data from leaking between tests.
        # Alternatively, you could disable caching by
        # using ndb.get_context().set_cache_policy(False)
        ndb.get_context().clear_cache()

init_datastore_v3_stub() メソッドに引数が指定されていない場合は、初期状態では空となっているメモリ内データストアが使用されます。既存のデータストア エンティティをテストするには、そのパス名を init_datastore_v3_stub() の引数として指定します。

setUp() に加えて、testbed を無効にする tearDown() メソッドを追加します。これにより、元のスタブが復元されるため、テストが互いに干渉することがなくなります。

def tearDown(self):
    self.testbed.deactivate()

次に、テストを実装します。

def testInsertEntity(self):
    TestModel().put()
    self.assertEqual(1, len(TestModel.query().fetch(2)))

以上で、TestModel を使用して、実際のデータストア サービスや Memcache サービスの代わりにサービススタブを使用するテストを作成できます。

たとえば、下に示すメソッドでは 2 つのエンティティが作成されます。1 つ目のエンティティは number 属性のデフォルト値（42）を使用し、2 つ目のエンティティは number のデフォルトではない値（17）を使用します。次に、このメソッドは TestModel エンティティに対するクエリを作成しますが、そのクエリで対象とするのは、number にデフォルト値が設定されているエンティティのみです。

一致するすべてのエンティティの取得後、このメソッドでは、エンティティが 1 つだけ見つかったこと、そのエンティティの number 属性値がデフォルト値であることを検証します。

def testFilterByNumber(self):
    root = TestEntityGroupRoot(id="root")
    TestModel(parent=root.key).put()
    TestModel(number=17, parent=root.key).put()
    query = TestModel.query(ancestor=root.key).filter(
        TestModel.number == 42)
    results = query.fetch(2)
    self.assertEqual(1, len(results))
    self.assertEqual(42, results[0].number)

別の例として、以下のメソッドはエンティティを作成し、上記で作成した GetEntityViaMemcache() 関数を使用してそのエンティティを取得しています。次に、このメソッドでは、エンティティが返されたこと、その number の値が、以前作成したエンティティと同じであることを検証します。

def testGetEntityViaMemcache(self):
    entity_key = TestModel(number=18).put().urlsafe()
    retrieved_entity = GetEntityViaMemcache(entity_key)
    self.assertNotEqual(None, retrieved_entity)
    self.assertEqual(18, retrieved_entity.number)

最後に、unittest.main() を呼び出します。

if __name__ == '__main__':
    unittest.main()

テストを実行するには、テストを実行するをご覧ください。

Cloud Datastore のテストを作成する

アプリで Cloud Datastore を使用する場合は、結果整合性に即してアプリケーションの動作を検証するテストを作成することをおすすめします。これは db.testbed のオプションにより簡単になります。

from google.appengine.datastore import datastore_stub_util  # noqa


class HighReplicationTestCaseOne(unittest.TestCase):

    def setUp(self):
        # First, create an instance of the Testbed class.
        self.testbed = testbed.Testbed()
        # Then activate the testbed, which prepares the service stubs for use.
        self.testbed.activate()
        # Create a consistency policy that will simulate the High Replication
        # consistency model.
        self.policy = datastore_stub_util.PseudoRandomHRConsistencyPolicy(
            probability=0)
        # Initialize the datastore stub with this policy.
        self.testbed.init_datastore_v3_stub(consistency_policy=self.policy)
        # Initialize memcache stub too, since ndb also uses memcache
        self.testbed.init_memcache_stub()
        # Clear in-context cache before each test.
        ndb.get_context().clear_cache()

    def tearDown(self):
        self.testbed.deactivate()

    def testEventuallyConsistentGlobalQueryResult(self):
        class TestModel(ndb.Model):
            pass

        user_key = ndb.Key('User', 'ryan')

        # Put two entities
        ndb.put_multi([
            TestModel(parent=user_key),
            TestModel(parent=user_key)
        ])

        # Global query doesn't see the data.
        self.assertEqual(0, TestModel.query().count(3))
        # Ancestor query does see the data.
        self.assertEqual(2, TestModel.query(ancestor=user_key).count(3))

PseudoRandomHRConsistencyPolicy クラスでは、（祖先クエリでない）各グローバル クエリの前に書き込みが適用される可能性を制御できます。確率を 0% に設定すると、データストア スタブに対し、最大限の結果整合性で動作するよう指示することになります。最大限の結果整合性とは、書き込みが commit されても常に適用に失敗することを意味します。そのため、グローバル（祖先でない）クエリには一貫して、変更が反映されません。これはもちろん、本番環境での実行時にアプリケーションに反映される結果整合性の程度を表すものではありませんが、テストのためにローカル データストアを毎回このように動作するよう構成できれば非常に便利です。可能性をゼロ以外に設定すると、テストが一貫した結果になるように、PseudoRandomHRConsistencyPolicy によって整合性の決定論的順序が決まります:

def testDeterministicOutcome(self):
    # 50% chance to apply.
    self.policy.SetProbability(.5)
    # Use the pseudo random sequence derived from seed=2.
    self.policy.SetSeed(2)

    class TestModel(ndb.Model):
        pass

    TestModel().put()

    self.assertEqual(0, TestModel.query().count(3))
    self.assertEqual(0, TestModel.query().count(3))
    # Will always be applied before the third query.
    self.assertEqual(1, TestModel.query().count(3))

テスト用 API は、アプリケーションが結果整合性に即して適切に動作することを検証するために有用ですが、ローカルの高レプリケーション読み取り整合性モデルは、本番環境の高レプリケーション読み取り整合性モデルの近似モデルであり、厳密なレプリカではないことに注意してください。ローカル環境では、適用されない書き込みがあるエンティティ グループに属する Entity の get() を実行すると常に、その適用されない書き込みの結果が、以降のグローバル クエリで認識されます。本番環境の場合、これは該当しません。

メールテストを作成する

メール サービス スタブを使用して、メール サービスをテストできます。testbed でサポートされている他のサービスと同様に、まずスタブを初期化してから、Mail API を使用するコードを呼び出し、正しいメッセージが送信されたかどうかを検証します。

import unittest

from google.appengine.api import mail
from google.appengine.ext import testbed


class MailTestCase(unittest.TestCase):

    def setUp(self):
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.testbed.init_mail_stub()
        self.mail_stub = self.testbed.get_stub(testbed.MAIL_SERVICE_NAME)

    def tearDown(self):
        self.testbed.deactivate()

    def testMailSent(self):
        mail.send_mail(to='alice@example.com',
                       subject='This is a test',
                       sender='bob@example.com',
                       body='This is a test e-mail')
        messages = self.mail_stub.get_sent_messages(to='alice@example.com')
        self.assertEqual(1, len(messages))
        self.assertEqual('alice@example.com', messages[0].to)

タスクキューのテストを作成する

タスクキュー スタブを使用して、タスクキュー サービスを使用したテストを作成できます。testbed でサポートされている他のサービスと同様に、まずスタブを初期化してからタスクキュー API を使用するコードを呼び出し、最後にタスクがキューに正しく追加されたかどうかを検証します。

import operator
import os
import unittest

from google.appengine.api import taskqueue
from google.appengine.ext import deferred
from google.appengine.ext import testbed


class TaskQueueTestCase(unittest.TestCase):
    def setUp(self):
        self.testbed = testbed.Testbed()
        self.testbed.activate()

        # root_path must be set the the location of queue.yaml.
        # Otherwise, only the 'default' queue will be available.
        self.testbed.init_taskqueue_stub(
            root_path=os.path.join(os.path.dirname(__file__), 'resources'))
        self.taskqueue_stub = self.testbed.get_stub(
            testbed.TASKQUEUE_SERVICE_NAME)

    def tearDown(self):
        self.testbed.deactivate()

    def testTaskAddedToQueue(self):
        taskqueue.Task(name='my_task', url='/url/of/my/task/').add()
        tasks = self.taskqueue_stub.get_filtered_tasks()
        self.assertEqual(len(tasks), 1)
        self.assertEqual(tasks[0].name, 'my_task')

queue.yaml 構成ファイルの設定

デフォルト以外のキューと対話するコードに対してテストを実行するには、アプリケーションで使用する queue.yaml ファイルを作成し、そのファイルを指定する必要があります。下記の例 queue.yaml をご覧ください。

使用できる queue.yaml オプションの詳細については、タスクキューの構成をご覧ください。

queue:
- name: default
  rate: 5/s
- name: queue-1
  rate: 5/s
- name: queue-2
  rate: 5/s

queue.yaml の場所は、スタブの初期化時に指定されます。

self.testbed.init_taskqueue_stub(root_path='.')

この例では queue.yaml はテストと同じディレクトリにあります。別のフォルダ内にあるとしたら、そのパスを root_path で指定する必要があります。

タスクをフィルタリングする

タスクキュー スタブの get_filtered_tasks を使用すると、キューに入ったタスクをフィルタリングできます。これにより、複数のタスクをエンキューするコードを検証しなければならないテストを簡単に作成できます。

def testFiltering(self):
    taskqueue.Task(name='task_one', url='/url/of/task/1/').add('queue-1')
    taskqueue.Task(name='task_two', url='/url/of/task/2/').add('queue-2')

    # All tasks
    tasks = self.taskqueue_stub.get_filtered_tasks()
    self.assertEqual(len(tasks), 2)

    # Filter by name
    tasks = self.taskqueue_stub.get_filtered_tasks(name='task_one')
    self.assertEqual(len(tasks), 1)
    self.assertEqual(tasks[0].name, 'task_one')

    # Filter by URL
    tasks = self.taskqueue_stub.get_filtered_tasks(url='/url/of/task/1/')
    self.assertEqual(len(tasks), 1)
    self.assertEqual(tasks[0].name, 'task_one')

    # Filter by queue
    tasks = self.taskqueue_stub.get_filtered_tasks(queue_names='queue-1')
    self.assertEqual(len(tasks), 1)
    self.assertEqual(tasks[0].name, 'task_one')

    # Multiple queues
    tasks = self.taskqueue_stub.get_filtered_tasks(
        queue_names=['queue-1', 'queue-2'])
    self.assertEqual(len(tasks), 2)

遅延タスクのテストを作成する

アプリケーション コードで遅延ライブラリを使用する場合は、taskqueue スタブを deferred と一緒に使用すると、遅延関数がキューに入り、正しく実行されることを確認できます。

def testTaskAddedByDeferred(self):
    deferred.defer(operator.add, 1, 2)

    tasks = self.taskqueue_stub.get_filtered_tasks()
    self.assertEqual(len(tasks), 1)

    result = deferred.run(tasks[0].payload)
    self.assertEqual(result, 3)

デフォルトの環境変数を変更する

App Engine サービスは多くの場合、環境変数に依存しています。testbed.Testbed クラスの activate() メソッドでは、環境変数にデフォルト値を使用しますが、テストの必要性に応じて、testbed.Testbed クラスの setup_env メソッドでカスタム値を設定できます。

たとえば、データストアに複数のエンティティを保管するテストを実行する場合を考えてみましょう。このテストで保管するエンティティは、すべて同一のアプリケーション ID に関連付けられているものです。次に、保管されたエンティティに関連付けられているものとは異なるアプリケーション ID を使用して、同じテストをもう一度実行するとします。それには、新しい値を app_id として self.setup_env() に渡します。

次に例を示します。

import os
import unittest

from google.appengine.ext import testbed


class EnvVarsTestCase(unittest.TestCase):
    def setUp(self):
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.testbed.setup_env(
            app_id='your-app-id',
            my_config_setting='example',
            overwrite=True)

    def tearDown(self):
        self.testbed.deactivate()

    def testEnvVars(self):
        self.assertEqual(os.environ['APPLICATION_ID'], 'your-app-id')
        self.assertEqual(os.environ['MY_CONFIG_SETTING'], 'example')

ログインをシミュレーションする

別の一般的な setup_env の使用例として、管理権限を持つユーザー、または持たないユーザーのログインをシミュレーションし、いずれの場合もハンドラが適切に動作するかどうかを確認します。

import unittest

from google.appengine.api import users
from google.appengine.ext import testbed


class LoginTestCase(unittest.TestCase):
    def setUp(self):
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.testbed.init_user_stub()

    def tearDown(self):
        self.testbed.deactivate()

    def loginUser(self, email='user@example.com', id='123', is_admin=False):
        self.testbed.setup_env(
            user_email=email,
            user_id=id,
            user_is_admin='1' if is_admin else '0',
            overwrite=True)

    def testLogin(self):
        self.assertFalse(users.get_current_user())
        self.loginUser()
        self.assertEquals(users.get_current_user().email(), 'user@example.com')
        self.loginUser(is_admin=True)
        self.assertTrue(users.is_current_user_admin())

この場合、たとえばテストメソッドで self.loginUser('', '') を呼び出して、どのユーザーもログインしていないこと、self.loginUser('test@example.com', '123') を呼び出して、管理権限を持たないユーザーがログインしていること、そして self.loginUser('test@example.com', '123', is_admin=True) を呼び出して、管理者ユーザーがログインしていることをシミュレーションできます。

テスト用フレームワークを設定する

SDK のテスト用ユーティリティは、特定のフレームワークとは関連付けられていません。利用可能な App Engine テストランナー（nose-gae、ferrisnose など）をどれでも使用して、単体テストを実行できます。また、シンプルなテストランナーを独自に作成したり、下記のテストランナーを使用したりすることも可能です。

次のスクリプトでは Python の unittest モジュールを使用します。

スクリプト名は自由に設定できます。スクリプトを実行する際は、Google Cloud CLI または Google App Engine SDK のインストール パスと、テスト モジュールのパスを指定します。スクリプトは指定されたパスですべてのテストを検出し、標準的なエラー ストリームに結果を出力します。テストファイルの名前は、命名規則に従い test で始まります。

"""App Engine local test runner example.

This program handles properly importing the App Engine SDK so that test modules
can use google.appengine.* APIs and the Google App Engine testbed.

Example invocation:

    $ python runner.py ~/google-cloud-sdk
"""

import argparse
import os
import sys
import unittest


def fixup_paths(path):
    """Adds GAE SDK path to system path and appends it to the google path
    if that already exists."""
    # Not all Google packages are inside namespace packages, which means
    # there might be another non-namespace package named `google` already on
    # the path and simply appending the App Engine SDK to the path will not
    # work since the other package will get discovered and used first.
    # This emulates namespace packages by first searching if a `google` package
    # exists by importing it, and if so appending to its module search path.
    try:
        import google
        google.__path__.append("{0}/google".format(path))
    except ImportError:
        pass

    sys.path.insert(0, path)


def main(sdk_path, test_path, test_pattern):
    # If the SDK path points to a Google Cloud SDK installation
    # then we should alter it to point to the GAE platform location.
    if os.path.exists(os.path.join(sdk_path, 'platform/google_appengine')):
        sdk_path = os.path.join(sdk_path, 'platform/google_appengine')

    # Make sure google.appengine.* modules are importable.
    fixup_paths(sdk_path)

    # Make sure all bundled third-party packages are available.
    import dev_appserver
    dev_appserver.fix_sys_path()

    # Loading appengine_config from the current project ensures that any
    # changes to configuration there are available to all tests (e.g.
    # sys.path modifications, namespaces, etc.)
    try:
        import appengine_config
        (appengine_config)
    except ImportError:
        print('Note: unable to import appengine_config.')

    # Discover and run tests.
    suite = unittest.loader.TestLoader().discover(test_path, test_pattern)
    return unittest.TextTestRunner(verbosity=2).run(suite)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument(
        'sdk_path',
        help='The path to the Google App Engine SDK or the Google Cloud SDK.')
    parser.add_argument(
        '--test-path',
        help='The path to look for tests, defaults to the current directory.',
        default=os.getcwd())
    parser.add_argument(
        '--test-pattern',
        help='The file pattern for test modules, defaults to *_test.py.',
        default='*_test.py')

    args = parser.parse_args()

    result = main(args.sdk_path, args.test_path, args.test_pattern)

    if not result.wasSuccessful():
        sys.exit(1)

テストを実行する

runner.py スクリプトを実行するだけで、このようなテストを実行できます。このスクリプトの詳細については、テスト用フレームワークを設定するをご覧ください。

python runner.py <path-to-appengine-or-gcloud-SDK> .