Testes de unidades locais para Python 2

Os testes de unidades permitem-lhe verificar a qualidade do código depois de o escrever, mas também pode usar os testes de unidades para melhorar o processo de desenvolvimento à medida que avança. Em vez de escrever testes depois de terminar o desenvolvimento da aplicação, considere escrever os testes à medida que avança. Isto ajuda a criar unidades de código pequenas, fáceis de manter e reutilizáveis. Também lhe permite testar o código de forma exaustiva e rápida.

Quando faz testes unitários locais, executa testes que permanecem no seu próprio ambiente de desenvolvimento sem envolver componentes remotos. O App Engine fornece utilitários de teste que usam implementações locais do armazenamento de dados e outros serviços do App Engine. Isto significa que pode usar o seu código destes serviços localmente, sem implementar o seu código no App Engine, usando stubs de serviço.

Um stub de serviço é um método que simula o comportamento do serviço. Por exemplo, o stub do serviço de armazenamento de dados apresentado em Escrever testes de armazenamento de dados e Memcache permite-lhe testar o seu código de armazenamento de dados sem fazer pedidos ao armazenamento de dados real. Qualquer entidade armazenada durante um teste de unidade da base de dados é mantida na memória, não na base de dados, e é eliminada após a execução do teste. Pode executar testes pequenos e rápidos sem qualquer dependência do próprio arquivo de dados.

Este documento descreve como escrever testes unitários em vários serviços locais do App Engine e, em seguida, fornece algumas informações sobre a configuração de uma framework de testes.

Apresentamos as utilidades de teste do Python 2

Um módulo do Python do App Engine denominado testbed disponibiliza stubs de serviços para testes de unidades.

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

  • Identidade da app init_app_identity_stub
  • Blobstore (use init_blobstore_stub)
  • Capacidade (use init_capability_stub)
  • Armazenamento de dados (use init_datastore_v3_stub)
  • Ficheiros (use init_files_stub)
  • Imagens (apenas para dev_appserver; use init_images_stub)
  • LogService (use init_logservice_stub)
  • Email (use init_mail_stub)
  • Memcache (use init_memcache_stub)
  • Fila de tarefas (use init_taskqueue_stub)
  • Obtenção de URL (use init_urlfetch_stub)
  • Serviço de utilizadores (use init_user_stub)

Para inicializar todos os stubs ao mesmo tempo, pode usar init_all_stubs.

Escrever testes de Datastore e memcache

Esta secção mostra um exemplo de como escrever código que testa a utilização dos serviços datastore e memcache.

Certifique-se de que o seu test runner tem as bibliotecas adequadas no caminho de carregamento do Python, incluindo as bibliotecas do App Engine, yaml (incluídas no SDK do App Engine), a raiz da aplicação e quaisquer outras modificações ao caminho da biblioteca esperadas pelo código da aplicação (como um diretório ./lib local, se tiver um). Por 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 unittest do Python e os módulos do App Engine relevantes para os serviços que estão a ser testados. Neste caso, memcache e ndb, que usam o datastore e a 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 turma TestModel. Neste exemplo, uma função verifica se uma entidade está armazenada na cache de memória. Se não for encontrada nenhuma entidade, procura uma entidade no arquivo de dados. Isto pode ser frequentemente redundante na vida real, uma vez que o ndb usa o próprio memcache nos bastidores, mas continua a ser um padrão adequado 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 exemplo de teste. Independentemente dos serviços que estiver a testar, o exemplo de teste tem de criar uma instância Testbed e ativá-la. O exemplo de teste também tem de inicializar os stubs de serviço relevantes, neste caso, através de init_datastore_v3_stub e init_memcache_stub. Os métodos para inicializar outros stubs de serviços do App Engine estão listados no artigo Apresentamos as utilidades 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 arquivo de dados na memória que está inicialmente vazio. Se quiser testar uma entidade de arquivo de dados existente, inclua o respetivo caminho como um argumento para init_datastore_v3_stub().

Além de setUp(), inclua um método tearDown() que desative o testbed. Isto restaura os stubs originais para que os testes não interfiram uns com os 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, pode usar TestModel para escrever testes que usam os stubs de serviço do datastore ou memcache em vez de usar os serviços reais.

Por exemplo, o método apresentado abaixo cria duas entidades: a primeira entidade usa o valor predefinido para o atributo number (42) e a segunda usa um valor não predefinido para number (17). Em seguida, o método cria uma consulta para entidades TestModel, mas apenas para aquelas com o valor predefinido de number.

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

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 seguinte cria uma entidade e obtém-na através da função GetEntityViaMemcache() que criámos acima. Em seguida, o método testa se foi devolvida uma entidade e se o respetivo valor number é igual ao 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 último, invoque unittest.main().

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

Para executar os testes, consulte o artigo Executar testes.

Escrever testes do Cloud Datastore

Se a sua app usar o Cloud Datastore, é recomendável escrever testes que validem o comportamento da sua aplicação perante a consistência final. O db.testbed expõe opções que facilitam esta tarefa:

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-lhe controlar a probabilidade de uma gravação ser aplicada antes de cada consulta global (não antecessora). Ao definir a probabilidade para 0%, estamos a dar instruções ao stub da base de dados para funcionar com a quantidade máxima de consistência eventual. A consistência eventual máxima significa que as escritas são confirmadas, mas falham sempre na aplicação, pelo que as consultas globais (não antecessoras) falham sempre na deteção de alterações. Claro que isto não é representativo da quantidade de consistência eventual que a sua aplicação vai ver quando estiver em produção, mas, para fins de teste, é muito útil poder configurar o armazenamento de dados local para se comportar desta forma sempre. Se usar uma probabilidade diferente de zero, o PseudoRandomHRConsistencyPolicy toma uma sequência determinística de decisões de consistência para que os resultados dos testes sejam 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 a sua aplicação se comporta corretamente perante a consistência eventual, mas tenha em atenção 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 e não uma réplica exata. No ambiente local, a execução de uma get() de um Entity que pertence a um grupo de entidades com uma gravação não aplicada torna sempre os resultados da gravação não aplicada visíveis para consultas globais subsequentes. Na produção, não é o caso.

Escrever testes de email

Pode usar o stub do serviço de correio para testar o serviço de correio. À semelhança de outros serviços suportados pelo testbed, primeiro inicializa o stub, depois invoca o código que usa a API Mail e, por último, testa se foram enviadas as mensagens corretas.

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)

Escrever testes de filas de tarefas

Pode usar o stub taskqueue para escrever testes que usam o serviço taskqueue. Semelhante a outros serviços suportados pelo testbed, primeiro inicializa o stub, depois invoca o código que usa a API taskqueue e, finalmente, testa se as tarefas foram adicionadas corretamente à 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')

Definir o ficheiro de configuração queue.yaml

Se quiser executar testes em código que interage com uma fila não predefinida, tem de criar e especificar um ficheiro queue.yaml para a sua aplicação usar. Segue-se um exemplo queue.yaml:

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

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

A localização do queue.yaml é especificada quando o stub é inicializado:

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

No exemplo, queue.yaml está no mesmo diretório que os testes. Se estivesse noutra pasta, esse caminho teria de ser especificado em root_path.

Filtrar tarefas

O stub taskqueue get_filtered_tasks permite-lhe filtrar tarefas em fila. Isto facilita a escrita de testes que precisam de validar o código que coloca várias tarefas em fila.

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)

Escrever testes de tarefas diferidas

Se o código da sua aplicação usar a biblioteca diferida, pode usar o stub taskqueue juntamente com deferred para verificar se as funções diferidas são colocadas em 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)

Alterar as variáveis de ambiente predefinidas

Os serviços do App Engine dependem frequentemente de variáveis de ambiente. O método activate() da classe testbed.Testbed usa valores predefinidos para estes, mas pode definir valores personalizados com base nas suas necessidades de teste com o método setup_env da classe testbed.Testbed.

Por exemplo, suponhamos que tem um teste que armazena várias entidades no datastore, todas elas associadas ao mesmo ID da aplicação. Agora, quer executar os mesmos testes novamente, mas usando um ID da aplicação diferente do que está associado às entidades armazenadas. Para tal, transmita o novo valor para self.setup_env() como app_id.

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

Simulação de início de sessão

Outra utilização frequente de setup_env é simular a sessão iniciada de um utilizador, com ou sem privilégios de administrador, para verificar se os seus controladores funcionam 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, os seus métodos de teste podem chamar, por exemplo, self.loginUser('', '') para simular que nenhum utilizador tem sessão iniciada, self.loginUser('test@example.com', '123') para simular que um utilizador não administrador tem sessão iniciada e self.loginUser('test@example.com', '123', is_admin=True) para simular que um utilizador administrador tem sessão iniciada.

Configurar uma framework de testes

As utilidades de teste do SDK não estão associadas a uma framework específica. Pode executar os testes unitários com qualquer testrunner do App Engine disponível, por exemplo, nose- gae ou ferrisnose. Também pode escrever um testrunner simples ou usar o apresentado abaixo.

Os scripts seguintes usam o módulo unittest do Python.

Pode atribuir o nome que quiser ao guião. Quando o executar, indique o caminho para a instalação da CLI Google Cloud ou do SDK Google App Engine e o caminho para os módulos de teste. O script vai descobrir todos os testes no caminho fornecido e vai imprimir os resultados no fluxo de erros padrão. Os ficheiros de teste seguem a convenção de ter test prefixado no respetivo 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)

Executar os testes

Pode executar estes testes simplesmente executando o script runner.py, que é descrito detalhadamente em Configurar uma framework de testes:

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