適用於 Python 2 的本機單位測試

單元測試可讓您在撰寫程式碼之後檢查品質，但也可以在開發期間使用單元測試來改進開發流程。您不需在開發完成後才撰寫測試，而可以在開發過程中即撰寫測試。這麼做可協助您設計可管理且可重複使用的小單元程式碼。測試程式碼也可以更輕鬆且快速徹底。

進行本機單元測試時，可在擁有的開發環境中執行測試，而不需遠端元件的輔助。App Engine 提供的測試公用程式採用了資料儲存庫及其他 App Engine 服務的本機實作。換句話說，您可以在本機執行程式碼的服務用途，無需透過服務虛設常式將程式碼部署至 App Engine。

服務虛設常式是一種模擬服務行為的方法。例如，撰寫 Datastore 與 Memcache 測試所示的資料儲存庫服務虛設常式可讓您測試資料儲存庫程式碼，而無須對真實的資料儲存庫提出任何要求。在資料儲存庫單元測試時儲存的實體不會儲存在資料儲存庫中，而是儲存在記憶體中，執行測試後即會遭到刪除。您可以執行快速的小型測試，而完全不需動用資料儲存庫。

本文件說明如何針對多種本機 App Engine 服務撰寫單元測試，並提供一些設定測試架構的相關資訊。

Python 2 測試公用程式簡介

名為 testbed 的 App Engine Python 模組可提供用於單元測試的服務虛設常式。

下列服務可使用服務虛設常式：

  • 應用程式識別資訊 init_app_identity_stub
  • Blobstore (使用 init_blobstore_stub)
  • 功能 (使用 init_capability_stub)
  • 資料儲存庫 (使用 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，這兩者都會使用 Datastore 和 Memcache。也要匯入 testbed 模組。

import unittest

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

接著建立 TestModel 類別。在本例中，會有函式檢查 Memcache 中是否儲存了實體。如果找不到實體，該函式就會在 Datastore 中檢查實體。在實際情況中，這可能會造成重複，因為 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 非常實用。不過，請記住，本機的「高複製」讀取一致性模型只能說相當接近正式作業「高複製」的讀取一致性模型，而非完全相同。在本機環境中，如果執行的 get() 函式所屬 Entity 屬於的實體群組有未套用的寫入，未套用寫入的結果一律可見於後續全域查詢中。但在正式作業環境中就不是這樣。

撰寫郵件測試

您可以使用郵件服務虛設常式測試郵件服務。與其他測試平台支援的服務相同，首先需要將服務虛設常式初始化，然後叫用使用郵件 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='[email protected]',
                       subject='This is a test',
                       sender='[email protected]',
                       body='This is a test e-mail')
        messages = self.mail_stub.get_sent_messages(to='[email protected]')
        self.assertEqual(1, len(messages))
        self.assertEqual('[email protected]', messages[0].to)

撰寫工作佇列測試

您可以使用工作佇列服務虛設常式撰寫使用 taskqueue 服務的測試。與其他測試平台支援的服務相同，首先需要將服務虛設常式初始化，然後叫用使用 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

初始化 Stub 時，會指定 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='[email protected]', 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(), '[email protected]')
        self.loginUser(is_admin=True)
        self.assertTrue(users.is_current_user_admin())

如今，您的測試方法可以呼叫 self.loginUser('', '') 模擬沒有使用者登入的情況、self.loginUser('[email protected]', '123') 模擬非管理員使用者登入的情況，以及 self.loginUser('[email protected]', '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> .