Pruebas unitarias locales para Python 2

Las pruebas unitarias te permiten comprobar la calidad del código después de escribirlo, pero también puedes usarlas para mejorar el proceso de desarrollo a medida que avanzas. En lugar de escribir pruebas después de terminar de desarrollar tu aplicación, te recomendamos que las escribas a medida que avanzas. De esta forma, podrás diseñar unidades de código pequeñas, fáciles de mantener y reutilizables. También te permite probar tu código de forma exhaustiva y rápida.

Cuando haces pruebas unitarias locales, ejecutas pruebas que se quedan en tu entorno de desarrollo sin implicar componentes remotos. App Engine proporciona utilidades de prueba que usan implementaciones locales de Datastore y otros servicios de App Engine. Esto significa que puedes probar el uso que hace tu código de estos servicios de forma local, sin desplegarlo en App Engine, mediante stubs de servicio.

A través del código auxiliar de un servicio determinado, se puede simular el comportamiento de ese servicio. Por ejemplo, el stub del servicio de Datastore que se muestra en Escribir pruebas de Datastore y Memcache te permite probar tu código de Datastore sin enviar ninguna solicitud al Datastore real. Las entidades almacenadas durante una prueba unitaria del almacén de datos se guardan en la memoria, no en el almacén de datos, y se eliminan después de la prueba. Puedes realizar pruebas pequeñas y rápidas sin depender de Datastore.

En este documento se describe cómo escribir pruebas unitarias para varios servicios locales de App Engine y, a continuación, se ofrece información sobre cómo configurar un framework de pruebas.

Presentamos las utilidades de prueba de Python 2

Un módulo de Python de App Engine llamado testbed hace que los stubs de servicio estén disponibles para las pruebas unitarias.

Existen códigos auxiliares de los siguientes servicios:

  • Identidad de la aplicación init_app_identity_stub
  • Almacén de blobs (usa init_blobstore_stub)
  • Función (usa init_capability_stub)
  • Datastore (usa init_datastore_v3_stub)
  • Archivos (usa init_files_stub)
  • Imágenes (solo para dev_appserver; usa init_images_stub)
  • LogService (usa init_logservice_stub)
  • Correo (usa init_mail_stub)
  • Memcache (usa init_memcache_stub)
  • Cola de tareas (usa init_taskqueue_stub)
  • Obtención de URL (usa init_urlfetch_stub)
  • Servicio de usuario (usa init_user_stub)

Para inicializar todos los stubs al mismo tiempo, puedes usar init_all_stubs.

Escribir pruebas de Datastore y memcache

En esta sección se muestra un ejemplo de cómo escribir código que pruebe el uso de los servicios datastore y memcache.

Asegúrate de que tu ejecutor de pruebas tenga las bibliotecas adecuadas en la ruta de carga de Python, incluidas las bibliotecas de App Engine, yaml (incluida en el SDK de App Engine), la raíz de la aplicación y cualquier otra modificación en la ruta de la biblioteca que espere el código de la aplicación (como un directorio ./lib local, si tienes uno). Por ejemplo:

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

Importa el módulo de Python unittest y los módulos de App Engine que sean relevantes para los servicios que se van a probar. En este caso, memcache y ndb, que usan tanto el almacén de datos como la memoria caché. También importa el módulo testbed.

import unittest

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

A continuación, crea una clase TestModel. En este ejemplo, una función comprueba si una entidad está almacenada en memcache. Si no se encuentra ninguna entidad, busca una entidad en el almacén de datos. A menudo, esto puede ser redundante en la vida real, ya que ndb usa memcache en segundo plano, pero sigue siendo un patrón adecuado para una prueba.

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

A continuación, crea un caso de prueba. Independientemente de los servicios que estés probando, el caso de prueba debe crear una instancia de Testbed y activarla. El caso de prueba también debe inicializar los stubs de servicio pertinentes, en este caso, mediante init_datastore_v3_stub y init_memcache_stub. Los métodos para inicializar otros stubs de servicio de App Engine se indican en Presentación de las utilidades de pruebas de 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()

El método init_datastore_v3_stub() sin argumentos usa un almacén de datos en memoria que está vacío al principio. Si quieres probar una entidad de almacén de datos, incluye su nombre de ruta como argumento de init_datastore_v3_stub().

Además de setUp(), incluye un método tearDown() que desactive el entorno de pruebas. De esta forma, se restauran los stubs originales para que las pruebas no interfieran entre sí.

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

A continuación, implementa las pruebas.

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

Ahora puedes usar TestModel para escribir pruebas que usen los stubs de servicio del almacén de datos o de memcache en lugar de los servicios reales.

Por ejemplo, el método que se muestra a continuación crea dos entidades: la primera usa el valor predeterminado del atributo number (42) y la segunda usa un valor no predeterminado para number (17). A continuación, el método crea una consulta para las entidades TestModel, pero solo para aquellas que tengan el valor predeterminado number.

Después de recuperar todas las entidades coincidentes, el método comprueba que se ha encontrado exactamente una entidad y que el valor del atributo number de esa entidad es el valor predeterminado.

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)

Otro ejemplo es el siguiente método, que crea una entidad y la recupera mediante la función GetEntityViaMemcache() que hemos creado anteriormente. A continuación, el método comprueba que se ha devuelto una entidad y que su valor number es el mismo que el de la entidad creada 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, invoca unittest.main().

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

Para ejecutar las pruebas, consulta Ejecutar pruebas.

Escribir pruebas de Cloud Datastore

Si tu aplicación usa Cloud Datastore, te recomendamos que escribas pruebas que verifiquen el comportamiento de tu aplicación ante la coherencia final. db.testbed ofrece opciones que facilitan esta tarea:

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

La clase PseudoRandomHRConsistencyPolicy te permite controlar la probabilidad de que se aplique una escritura antes de cada consulta global (no antecesora). Si asignamos el valor 0 % a la probabilidad, indicamos al stub del almacén de datos que opere con la máxima coherencia final. La coherencia final máxima significa que las escrituras se completarán, pero siempre fallarán al aplicarse, por lo que las consultas globales (no de ancestros) no podrán ver los cambios. Por supuesto, esto no representa la cantidad de coherencia final que verá tu aplicación cuando se ejecute en producción, pero, a efectos de prueba, es muy útil poder configurar el almacén de datos local para que se comporte de esta forma cada vez. Si usas una probabilidad distinta de cero, PseudoRandomHRConsistencyPolicy toma una secuencia determinista de decisiones de coherencia para que los resultados de las pruebas sean coherentes:

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

Las APIs de prueba son útiles para verificar que tu aplicación se comporta correctamente en caso de que se produzca una coherencia final, pero ten en cuenta que el modelo de coherencia de lectura de alta replicación local es una aproximación del modelo de coherencia de lectura de alta replicación de producción, no una réplica exacta. En el entorno local, al realizar una get() de un Entity que pertenece a un grupo de entidades con una escritura no aplicada, los resultados de la escritura no aplicada siempre serán visibles para las consultas globales posteriores. En cambio, esto no sucede en un entorno de producción.

Escribir pruebas de correo

Puedes usar el stub del servicio de correo para probar el servicio mail. Al igual que con otros servicios compatibles con testbed, primero inicializas el stub, luego invocas el código que usa la API de correo y, por último, compruebas si se han enviado los mensajes correctos.

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)

Escribir pruebas de colas de tareas

Puedes usar el stub taskqueue para escribir pruebas que usen el servicio taskqueue. Al igual que con otros servicios compatibles con el banco de pruebas, primero inicializas el stub, luego invocas el código que usa la API Task Queue y, por último, compruebas si las tareas se han añadido correctamente a la cola.

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

Configurar el archivo queue.yaml

Si quieres ejecutar pruebas en código que interactúa con una cola no predeterminada, tendrás que crear y especificar un archivo queue.yaml para que lo use tu aplicación. A continuación, se muestra un ejemplo de queue.yaml:

Para obtener más información sobre las opciones disponibles de queue.yaml, consulta la configuración de colas de tareas.

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

La ubicación de queue.yaml se especifica al inicializar el stub:

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

En el ejemplo, queue.yaml está en el mismo directorio que las pruebas. Si estuviera en otra carpeta, habría que especificar esa ruta en root_path.

Filtrar tareas

El stub de taskqueue get_filtered_tasks te permite filtrar las tareas en cola. De esta forma, es más fácil escribir pruebas que necesiten verificar código que ponga en cola varias tareas.

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)

Escribir pruebas de tareas diferidas

Si el código de tu aplicación usa la biblioteca diferida, puedes usar el stub taskqueue junto con deferred para verificar que las funciones diferidas se ponen en cola y se ejecutan correctamente.

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)

Cambiar las variables de entorno predeterminadas

Los servicios de App Engine suelen depender de variables de entorno. El método activate() de la clase testbed.Testbed usa valores predeterminados para estos elementos, pero puedes definir valores personalizados según tus necesidades de prueba con el método setup_env de la clase testbed.Testbed.

Por ejemplo, supongamos que tienes una prueba que almacena varias entidades en Datastore, todas ellas vinculadas al mismo ID de aplicación. Ahora quieres volver a ejecutar las mismas pruebas, pero con un ID de aplicación diferente del que está vinculado a las entidades almacenadas. Para ello, transfiere el nuevo valor a self.setup_env() como app_id.

Por ejemplo:

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

Simulando el inicio de sesión

Otro uso frecuente de setup_env es simular que un usuario ha iniciado sesión, ya sea con o sin privilegios de administrador, para comprobar si tus controladores funcionan correctamente en 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())

Ahora, tus métodos de prueba pueden llamar, por ejemplo, a self.loginUser('', '') para simular que ningún usuario ha iniciado sesión, a self.loginUser('test@example.com', '123') para simular que un usuario que no es administrador ha iniciado sesión o a self.loginUser('test@example.com', '123', is_admin=True) para simular que un usuario administrador ha iniciado sesión.

Configurar un marco de pruebas

Las utilidades de prueba del SDK no están vinculadas a un framework específico. Puedes ejecutar tus pruebas unitarias con cualquier testrunner de App Engine disponible, como nose-gae o ferrisnose. También puedes escribir tu propio testrunner sencillo o usar el que se muestra a continuación.

Las siguientes secuencias de comandos usan el módulo unittest de Python.

Puedes darle el nombre que quieras a la secuencia de comandos. Cuando lo ejecutes, proporciona la ruta a tu instalación de la CLI de Google Cloud o del SDK de Google App Engine, así como la ruta a tus módulos de prueba. La secuencia de comandos detectará todas las pruebas de la ruta proporcionada e imprimirá los resultados en el flujo de errores estándar. Los archivos de prueba siguen la convención de tener el prefijo test en su nombre.

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

Ejecutar las pruebas

Para ejecutar estas pruebas, solo tienes que ejecutar la secuencia de comandos runner.py, que se describe en detalle en la sección Configurar un marco de pruebas:

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