Python 2 本地单元测试

您可以在编写代码后使用单元测试来检查代码的质量，同时也可以使用该功能来改进开发过程。建议您一边开发一边编写测试，而不是在完成应用的开发之后才编写测试。这样做有助于您设计出可维护、可重复使用的小型代码单元，也方便您迅速而彻底地测试您的代码。

在进行本地单元测试时，您的测试将限制于自己的开发环境中，而不会涉及远程组件。App Engine 提供的测试实用程序使用数据存储区和其他 App Engine 服务的本地实现。这意味着，您可以借助服务存根，在本地运行并测试您的代码对这些服务的使用情况，而无需将您的代码部署到 App Engine。

服务桩是模拟服务行为的一种方法。例如，凭借编写数据存储区和 Memcache 测试中所示的数据存储区服务桩，您可以对您的数据存储区代码进行测试，而不需要向实际数据存储区发送任何请求。在数据存储区单元测试过程中存储的任何实体都会保存在内存中，而不是保存在数据存储区中，并会在测试运行结束后删除。您无需依赖数据存储区本身便可快速运行小型测试。

本文档介绍了如何为几项本地 App Engine 服务编写单元测试，并提供了一些关于设置测试框架的信息。

Python 2 测试实用程序简介

App Engine Python testbed 模块为单元测试提供了服务存根。

服务存根适用于以下服务：

  • 应用身份 init_app_identity_stub
  • Blobstore（使用 init_blobstore_stub）
  • 容量（使用 init_capability_stub）
  • Datastore（使用 init_datastore_v3_stub）
  • 文件（使用 init_files_stub）
  • 映像（仅适用于 dev_appserver；使用 init_images_stub）
  • LogService（使用 init_logservice_stub）
  • 邮件（使用 init_mail_stub）
  • Memcache（使用 init_memcache_stub）
  • 任务队列（使用 init_taskqueue_stub）
  • 网址提取（使用 init_urlfetch_stub）
  • 用户服务（使用 init_user_stub）

要同时初始化所有存根，您可以使用 init_all_stubs。

编写数据存储区和 memcache 测试

本部分介绍如何编写代码以测试 datastore 和 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，但其仍然不失为一种合适的测试模式。

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() 外，还应提供取消激活测试台的 tearDown() 方法。该方法可以恢复原始存根，使测试互不干扰。

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

然后执行测试。

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

现在，您可以利用 TestModel 编写使用数据存储区或 Memcache 服务存根而不是使用实际服务的测试。

例如，下面所示方法创建了两个实体：第一个实体使用默认值作为 number 特性 (42)，第二个实体则使用非默认值作为 number (17)。然后，该方法会为 TestModel 实体构建查询，但仅针对那些使用 number 默认值的实体。

在检索所有匹配的实体之后，该方法测试出只找出一个实体，且该实体的 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%，则表示我们在指示数据存储区存根以最高的最终一致性进行操作。最高的最终一致性意味着，写入内容将被提交，但始终无法应用，因此，全局（非祖先实体）查询将始终无法看到变化。当然，这并不代表应用在生产环境中运行时也会面临这种程度的最终一致性，但在测试中，如果能够将本地数据存储区配置为每次都采取这种行为方式，是非常有用的。如果使用非零可能性，那么 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()，而其属于一个带有未应用写入操作的实体组，则未应用写入的结果将对后续全局查询始终可见。在生产环境中，不会出现这种情况。

编写邮件测试

您可以使用邮件服务桩来测试邮件服务。与测试平台支持的其他服务类似，首先初始化存根，然后调用使用邮件 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)

编写任务队列测试

您可以使用任务队列存根来编写使用taskqueue服务的测试。与测试平台支持的其他服务类似，首先初始化存根，然后调用使用任务队列 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)

编写延迟任务测试

如果您的应用代码使用延迟库，则您可以搭配使用任务队列存根与 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 不同的应用 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> .