Teste de unidade local para Python 2

O teste de unidade permitirá verificar a qualidade do código depois que você escrevê-lo, mas também é possível usá-lo para melhorar o processo de desenvolvimento à medida que avança. Em vez de gravar testes depois de concluir o desenvolvimento do aplicativo, considere a possibilidade de gravar os testes conforme você avança. Isso ajuda você a projetar unidades de código pequenas, de fácil manutenção e reutilizáveis. Isso também facilita o teste rápido e completo do código.

Ao fazer o teste de unidade local, você executa testes que permanecem dentro do ambiente de desenvolvimento sem envolver componentes remotos. O App Engine oferece utilitários de teste que usam implementações locais de armazenamento de dados e outros serviços do App Engine. Isso significa ser possível fazer o código usar esses serviços localmente, sem implantá-lo no App Engine, usando-se stubs de serviço.

Stub de serviço é um método que simula o comportamento do serviço. Por exemplo, o stub de serviço do armazenamento de dados mostrado em Como escrever testes de Datastore e Memcache permite testar o código do armazenamento de dados sem fazer solicitações para o armazenamento de dados real. Qualquer entidade armazenada durante um teste de unidade do armazenamento de dados é mantida na memória, e não no armazenamento de dados, e será excluída depois da execução do teste. É possível executar testes pequenos e rápidos sem dependência do próprio armazenamento de dados.

Neste documento, descrevemos como criar testes de unidade em vários serviços locais do Google App Engine e, em seguida, fornecemos algumas informações sobre como configurar uma estrutura de teste.

Apresentação dos utilitários de teste para Python 2

Um módulo Python do App Engine chamado testbed disponibiliza stubs de serviço para testes de unidade.

Os stubs de serviço estão disponíveis para os seguintes serviços:

  • App Identity init_app_identity_stub
  • Blobstore (use init_blobstore_stub)
  • Capacidade (use init_capability_stub)
  • Armazenamento de dados (use init_datastore_v3_stub)
  • Arquivos (use init_files_stub)
  • Imagens (somente para dev_appserver. Use init_images_stub)
  • LogService (use init_logservice_stub)
  • E-mail (use init_mail_stub)
  • Memcache (use init_memcache_stub)
  • Task Queue (use init_taskqueue_stub)
  • Busca de URL (use init_urlfetch_stub)
  • Serviço de usuário (use init_user_stub)

Para inicializar todos os stubs ao mesmo tempo, é possível usar init_all_stubs.

Como escrever testes de armazenamento de dados e memcache

Nesta seção, você verá um exemplo de como gravar um código que testa o uso dos serviços de armazenamento de dados e Memcache.

Verifique se o executor de teste tem as bibliotecas apropriadas no caminho de carregamento do Python, incluindo as bibliotecas do App Engine, yaml (incluídas no SDK do App Engine), a raiz do aplicativo e outras modificações esperadas pelo código do aplicativo (como um diretório ./lib local, se você tiver um). Exemplo:

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')

Importe o módulo Python unittest e os módulos do App Engine que são relevantes para os serviços testados, neste caso memcache e ndb, que usa o armazenamento de dados e o memcache. Importe também o módulo testbed.

import unittest

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

Em seguida, crie uma classe TestModel. Neste exemplo, uma função verifica se uma entidade está armazenada no memcache. Se não for encontrada uma entidade, ela verificará se há uma entidade no armazenamento de dados. Muitas vezes, isso pode ser redundante na vida real, já que ndb usa o próprio Memcache por trás das cortinas, mas ainda é um padrão aceitável para um teste.

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

Em seguida, crie um caso de teste. Não importa os serviços testados, o caso de teste precisa criar uma instância de Testbed e ativá-la. O caso de teste também precisa inicializar os stubs de serviço relevantes, neste caso usando init_datastore_v3_stub e init_memcache_stub. Os métodos para inicializar outros stubs de serviço do App Engine estão listados em Introdução aos utilitários de teste do 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()

O método init_datastore_v3_stub() sem argumento usa um armazenamento de dados na memória que é inicialmente vazio. Se você quiser testar uma entidade atual do armazenamento de dados, inclua seu nome de caminho como argumento para init_datastore_v3_stub().

Além de setUp(), inclua um método tearDown() para desativar o testbed. Isso restaura os stubs originais para que os testes não interfiram uns nos outros.

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

Em seguida, implemente os testes.

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

Agora é possível usar TestModel para criar testes que usam stubs de serviço do armazenamento de dados ou do memcache em vez de usar serviços reais.

Por exemplo, o método mostrado abaixo cria duas entidades: a primeira entidade usa o valor padrão para o atributo number (42) e a segunda usa o valor não padrão para number (17). O método então constrói uma consulta para entidades TestModel, mas apenas para as que têm o valor padrão de number.

Depois de recuperar todas as entidades correspondentes, o método testa se exatamente uma entidade foi encontrada e se o valor do atributo number dessa entidade é o valor padrão.

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)

Como outro exemplo, o método a seguir cria uma entidade e a recupera usando a função GetEntityViaMemcache() que criamos acima. O método testa se uma entidade foi retornada e se o valor number dela é o mesmo da entidade criada anteriormente.

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)

Por fim, invoque unittest.main().

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

Para executar os testes, consulte Como executar testes.

Como escrever testes do Cloud Datastore

Se o app usar o Cloud Datastore, será possível escrever testes que verifiquem o comportamento do app diante de uma consistência eventual. db.testbed expõe opções que tornam isso fácil:

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

A classe PseudoRandomHRConsistencyPolicy permite que você controle a probabilidade de uma gravação se aplicar antes de cada consulta global (não ancestral). Ao definir a probabilidade como 0%, instruímos o stub do armazenamento de dados a operar com a quantidade máxima de consistência eventual. Consistência eventual máxima significa que haverá commit das gravações, mas elas sempre deixarão de ser aplicadas. Dessa maneira, consultas globais (não de ancestral) sempre deixarão de ver alterações. É claro que isso não representa a quantidade de consistência eventual que seu aplicativo vê ao ser executado na produção, mas para fins de teste é muito útil poder configurar o armazenamento de dados local para se comportar sempre dessa maneira. Se você usa uma probabilidade diferente de zero, PseudoRandomHRConsistencyPolicy faz uma sequência determinística de decisões de consistência, então os resultados do teste são consistentes:

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

As APIs de teste são úteis para verificar se seu aplicativo se comporta adequadamente diante de consistência eventual, mas lembre-se que o modelo de consistência de leitura de Alta replicação local é uma aproximação do modelo de consistência de leitura de Alta replicação de produção, não uma réplica exata. No ambiente local, executar um get() de um Entity que pertence a um grupo de entidades com uma gravação não aplicada sempre tornará os resultados da gravação não aplicada visíveis para as consultas globais subsequentes. Isso não ocorre na produção.

Como escrever testes de e-mail

Você pode usar o stub de serviço de e-mail para testar o serviço de e-mail. Semelhante a outros serviços compatíveis com testbed, primeiro você inicializa o stub, invoca o código que usa a API de e-mail e finalmente testa se as mensagens corretas foram enviadas.

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)

Como escrever testes de fila de tarefas

Você pode usar o stub de fila de tarefas para gravar testes que usam o serviço de fila de tarefas. Semelhante a outros serviços compatíveis com testbed, primeiro você inicializa o stub, invoca o código que usa a API de fila de tarefas e, finalmente, testa se as tarefas foram adequadamente adicionadas à fila.

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')

Como definir o arquivo de configuração queue.yaml

Se você quiser executar testes no código que interage com uma fila não padrão, será necessário criar e especificar um arquivo queue.yaml para o aplicativo usar. Veja um exemplo queue.yaml abaixo.

Para mais informações sobre as opções disponíveis de queue.yaml, consulte configuração da fila de tarefas.

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

O local do queue.yaml é especificado ao inicializar o stub:

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

Na amostra, queue.yaml está no mesmo diretório que os testes. Se ele estivesse em outra pasta, esse caminho precisaria ser especificado em root_path.

Como filtrar tarefas

O stub da fila de tarefas get_filtered_tasks permite que você filtre tarefas enfileiradas. Isso facilita a gravação de testes que precisam verificar o código que enfileira várias tarefas.

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)

Como escrever testes de tarefa adiada

Se o código do aplicativo usar a biblioteca adiada, será possível usar o stub taskqueue junto com deferred para verificar se as funções adiadas estão na fila e executadas corretamente.

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)

Como alterar as variáveis de ambiente padrão

Os serviços do App Engine geralmente dependem de variáveis de ambiente. O método activate() de classe testbed.Testbed usa valores padrão para eles, mas é possível definir valores personalizados com base nas necessidades de teste com o método setup_env de classe testbed.Testbed.

Por exemplo, digamos que você tem um teste que armazena diversas entidades no armazenamento de dados, todas elas ligadas ao mesmo ID de aplicativo. Agora você quer executar os mesmos testes novamente, mas usando um ID de aplicativo diferente daquele ligado às entidades armazenadas. Para fazer isso, passe o novo valor para self.setup_env() como app_id.

Exemplo:

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')

Como simular o login

Outro uso frequente para setup_env é simular um usuário que está sendo conectado, com ou sem privilégios de administrador, para verificar se seus manipuladores operam corretamente em cada caso.

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())

Agora, seus métodos de teste podem chamar, por exemplo, self.loginUser('', '') para simular que nenhum usuário está conectado, self.loginUser('test@example.com', '123') para simular que um usuário não administrador está conectado, self.loginUser('test@example.com', '123', is_admin=True) para simular que um usuário administrador está conectado.

Como configurar um framework de testes

Os utilitários de teste do SDK não estão associados a uma biblioteca específica. É possível executar os testes de unidade com qualquer executor disponível no App Engine, como nose- gae ou ferrisnose. Você também pode criar um executor de testes simples por conta própria ou usar o que é mostrado abaixo.

Os scripts a seguir usam o módulo unittest do Python.

Você pode dar o nome que quiser ao script. Ao executá-lo, forneça o caminho para a Google Cloud CLI ou a instalação do Google App Engine SDK e o caminho para os módulos de teste. O script descobrirá todos os testes no caminho fornecido e imprimirá os resultados no fluxo de erros padrão. Os arquivos de teste seguem a convenção de ter o prefixo test no nome.

"""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)

Como executar os testes

É possível executar esses testes simplesmente executando o script runner.py, é descrito em detalhes em Como configurar um framework de teste:

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