Python 2용 로컬 단위 테스트

코드 작성 후 단위 테스트를 사용하면 코드의 품질을 확인할 수 있으며 개발을 진행하면서 개발 프로세스를 개선하는 데도 단위 테스트를 사용할 수 있습니다. 애플리케이션 개발을 마친 후에 테스트를 작성하는 대신 개발을 진행하면서 테스트를 작성하는 것이 좋습니다. 그러면 작고 유지관리 및 재사용이 가능한 코드 단위를 설계하는 데 도움이 됩니다. 또한 코드를 완벽하고 신속하게 테스트하기도 더 쉽습니다.

로컬 단위 테스트를 수행할 때는 원격 구성요소를 사용하지 않고 자체 개발 환경 내에서 테스트를 실행합니다. App Engine에서는 데이터 저장소와 기타 App Engine 서비스의 로컬 구현을 사용하는 테스트 유틸리티를 제공합니다. 따라서 코드를 App Engine에 배포하지 않고 서비스 스텁을 사용하여 코드에서 이러한 서비스를 사용하는 방식을 로컬에서 테스트해 볼 수 있습니다.

서비스 스텁은 서비스의 동작을 시뮬레이션하는 수단입니다. 예를 들어 Datastore 및 Memcache 테스트 작성에 표시된 데이터 저장소 서비스 스텁을 사용하면 실제 데이터 저장소에 어떠한 요청도 하지 않고 데이터 저장소 코드를 테스트할 수 있습니다. 데이터 저장소 단위 테스트 중에 저장된 모든 항목은 데이터 저장소가 아니라 메모리에 보관되었다가 테스트 실행 후에 삭제됩니다. 데이터 저장소 자체에 어떠한 종속성도 없이 소규모 테스트를 빠르게 실행할 수 있습니다.

이 문서에서는 몇 개의 로컬 App Engine 서비스에 대한 단위 테스트를 작성하는 방법을 설명하고 테스트 프레임워크 설정에 대한 몇 가지 정보를 제공합니다.

Python 2 테스트 유틸리티 소개

testbed라는 App Engine Python 모듈은 단위 테스트에 서비스 스텁을 사용할 수 있도록 합니다.

서비스 스텁은 다음 서비스에 사용할 수 있습니다.

  • 앱 ID 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 사용)
  • 태스크 큐(init_taskqueue_stub 사용)
  • URL Fetch(init_urlfetch_stub 사용)
  • 사용자 서비스(init_user_stub 사용)

모든 스텁을 동시에 초기화하려면 init_all_stubs를 사용합니다.

Datastore 및 memcache 테스트 작성

이 섹션에서는 datastore 및 memcache 서비스의 사용을 테스트하는 코드의 작성 방법을 예시로 보여줍니다.

App Engine 라이브러리, yaml(App Engine SDK에 포함), 애플리케이션 루트, 애플리케이션 코드에 필요한 라이브러리 경로의 기타 모든 수정사항(예: 있는 경우 로컬 ./lib 디렉터리)을 비롯하여 Python 로드 경로에 있는 적절한 라이브러리를 테스트 실행기에 제공해야 합니다. 예를 들면 다음과 같습니다.

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 모듈을 가져옵니다. 이 경우에는 Datastore와 Memcache를 모두 사용하는 memcache와 ndb가 해당됩니다. 또한 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를 사용하지만 여전히 테스트에서는 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() 메서드는 처음에 비어 있는 메모리 내 Datastore를 사용합니다. 기존 Datastore 항목을 테스트하려면 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을 사용하여 실제 서비스 대신 Datastore 또는 Memcache 서비스 스텁을 사용하는 테스트를 작성할 수 있습니다.

예를 들어 아래 표시된 메서드는 두 개의 항목을 만듭니다. 첫 번째 항목은 number 속성에 기본값(42)을 사용하고 두 번째 항목은 number에 기본값이 아닌 값(17)을 사용합니다. 그런 다음 이 메서드는 number에 기본값을 사용하는 항목만을 대상으로 TestModel 항목에 대한 쿼리를 빌드합니다.

일치하는 모든 항목을 검색한 후 이 메서드는 정확히 하나의 항목이 발견되었고 해당 항목의 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를 사용하는 경우 eventual consistency에 직면하여 애플리케이션의 동작을 확인하는 테스트를 작성할 수 있습니다. 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%로 설정하면 Datastore 스텁이 최대한의 eventual consistency 상태에서 작동하게 됩니다. 최대한의 최종 일관성 상태에서는 쓰기가 커밋되지만 항상 적용에 실패하므로 전역(비상위) 쿼리가 일관되게 변경사항을 확인하지 못하게 됩니다. 물론 이것이 프로덕션 환경에서 실행 시 애플리케이션에서 실제로 확인되는 eventual consistency 수준을 나타내지는 않지만, 테스트 목적으로는 로컬 Datastore를 매번 이런 식으로 작동하도록 구성하는 것이 매우 유용합니다. 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))

Testing API는 eventual consistency에 직면하여 애플리케이션이 제대로 작동하는지 확인하는 데 유용하지만, 로컬 고성능 복제 읽기 일관성 모델은 프로덕션 고성능 복제 읽기 일관성 모델의 정확한 복제본이 아니라 근사치임에 유의해야 합니다. 로컬 환경에서 적용되지 않은 쓰기가 있는 항목 그룹에 속하는 Entity에 대해 get()을 수행하면 후속 전역 쿼리에서는 적용되지 않은 쓰기의 결과가 항상 표시됩니다. 이는 프로덕션 환경에서는 해당되지 않습니다.

메일 테스트 작성

mail 서비스 스텁을 사용하여 mail 서비스를 테스트할 수 있습니다. 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)

태스크 큐 테스트 작성

taskqueue 스텁을 사용하여 taskqueue 서비스를 사용하는 테스트를 작성할 수 있습니다. testbed에서 지원되는 다른 서비스와 마찬가지로 먼저 스텁을 초기화한 다음 Task Queue 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 메서드를 사용하여 테스트 요구사항에 따라 커스텀 값을 설정할 수도 있습니다.

예를 들어 여러 항목을 Datastore에 저장하는 테스트가 있으며 해당 항목들이 모두 동일한 애플리케이션 ID에 연결되어 있다고 가정해 보겠습니다. 이제 동일한 테스트를 다시 실행하되 저장된 항목에 연결된 애플리케이션 ID와는 다른 애플리케이션 ID를 사용하려고 합니다. 이렇게 하려면 새 값을 self.setup_env()에 app_id로 전달합니다.

예를 들면 다음과 같습니다.

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> .