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

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

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

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

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

Python 測試公用程式簡介

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

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

  • 應用程式識別 init_app_identity_stub
  • Blobstore (使用 init_blobstore_stub)
  • Capability (使用 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

撰寫 Datastore 和 Memcache 測試

本節會示範如何撰寫 DatastoreMemcache 服務使用測試的程式碼。

請確保您的測試執行器在 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 模組,本案例中就是 memcachendb,這兩個模組包含 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_stubinit_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% 時,代表我們指示資料儲存庫服務虛設常式儘可能提高最終一致性的程度來運作。採用最高程度的最終一致性表示即使寫入也一律無法套用,導致全域 (非祖系) 查詢永遠都無法找出變更。當然,這不代表您的應用程式在生產中執行時會面臨的最終一致性程度,但就測試目的而言,能夠每次都以這種方式設定本機資料儲存庫的行為相當實用。如果機率並非設為 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='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 服務的測試。與其他測試平台支援的服務相同,首先需要將服務虛設常式初始化,然後叫用使用 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 中指定該路徑。

篩選工作

taskqueue 服務虛設常式的 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 服務經常依賴環境變數activate() 方法 (屬於 testbed.Testbed 類別) 使用預設的環境變數。但您可以依自己的測試需求設定自訂值,使用 setup_env 方法 (屬於 testbed.Testbed 類別) 即可。

舉例而言,假設您的測試將許多實體儲存在資料儲存庫之中,這些實體皆連結到相同的應用程式 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-gaeferrisnose。您也可以自行撰寫簡單的測試執行程式,或使用以下測試執行程式。

下列指令碼使用 Python 的 unittest 模組。

您可以任意命名指令碼。執行時,請將該路徑提供給您的 Google Cloud SDK 或 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> .
本頁內容對您是否有任何幫助?請提供意見:

傳送您對下列選項的寶貴意見...

這個網頁
Python 2 適用的 App Engine 標準環境