Pruebas de unidades locales para Python 2

La prueba de unidades te permite verificar la calidad del código después de escribirlo. También puedes usarla para mejorar el proceso de desarrollo a medida que avanzas. En lugar de escribir pruebas luego de desarrollar tu aplicación, considera escribirlas conforme avanzas. Esto te ayuda a diseñar unidades de código pequeñas que se pueden mantener y volver a usar. También te facilita probar el código con rapidez y profundidad.

Cuando realices la prueba de unidades local, ejecuta las pruebas que están dentro de tu propio entorno de desarrollo sin involucrar componentes remotos. App Engine proporciona utilidades de prueba que usan implementaciones locales de Datastore y otros servicios de App Engine. Esto significa que puedes usar el código de estos servicios de manera local, sin implementarlo en App Engine, mediante el uso de stubs de servicio.

Un stub de servicio es un método que simula el comportamiento de un servicio. Por ejemplo, el stub de servicio de Datastore que se muestra en Escribe pruebas de Datastore y Memcache te permite probar el código de Datastore sin enviar solicitudes al real. Cualquier entidad almacenada durante una prueba de unidades de Datastore se aloja en la memoria, no en Datastore, y se borra después de ejecutar la prueba. Puedes ejecutar pruebas pequeñas y rápidas sin depender del almacén de datos.

En este documento, se detalla cómo escribir pruebas de unidades en varios servicios de App Engine y se proporciona información sobre cómo configurar un marco de trabajo de pruebas.

Presentación de las utilidades de prueba para Python 2

Un módulo de Python en App Engine denominado testbed permite que los stubs de servicio estén disponibles para las pruebas de unidades.

Los stubs de servicio están disponibles para los siguientes servicios:

• Identidad de la app init_app_identity_stub
• Blobstore (usa init_blobstore_stub)
• Capacidad (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 electrónico (usa init_mail_stub)
• Memcache (usa init_memcache_stub)
• Lista de tareas en cola (usa init_taskqueue_stub)
• Recuperación de URL (usa init_urlfetch_stub)
• Servicio de usuarios (usa init_user_stub)

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

Nota: En algunos casos, los stubs de servicio se comportan de manera diferente a los servicios reales. Incluso si se aprueban las pruebas, es posible que el código falle cuando se ejecute en un entorno de producción.

Cómo escribir pruebas de Datastore y Memcache

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

Asegúrate de que tu ejecutor de pruebas tenga las bibliotecas adecuadas en la ruta de carga de Python, entre ellas, las bibliotecas de App Engine, yaml (que se incluyen en el SDK de App Engine), la raíz de la aplicación y cualquier otra modificación en la ruta de acceso de la biblioteca que el código de la aplicación espera (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 unittest de Python y los módulos de App Engine pertinentes a los servicios que se prueban, en este caso memcache y ndb, que usan Datastore y Memcache. 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

Luego, crea una clase TestModel. En este ejemplo, una función verifica si se almacenó una entidad en Memcache. Si no se encontró ninguna entidad, se verifica que haya una en Datastore. A menudo, esto puede ser redundante en la realidad, ya que ndb usa Memcache en segundo plano, pero sigue siendo un patrón aceptable 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. No importa qué servicios estén en prueba, el caso de prueba debe crear una instancia Testbed y activarla. También debe inicializar los stubs de servicio pertinentes mediante init_datastore_v3_stub y init_memcache_stub. Los métodos cuyo propósito es inicializar otros stubs de servicio de App Engine se muestran en Presentación de las utilidades de pruebas para 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 la memoria que al principio está vacío. Si quieres probar una entidad de almacén de datos existente, incluye su nombre de ruta como argumento para init_datastore_v3_stub().

Nota: El método setUp() usa un ID del proyecto predeterminado para ejecutar pruebas. La prueba solo puede acceder a las entidades del almacén de datos que tengan el mismo ID del proyecto. Para cambiar el ID del proyecto de una prueba, usa setup_env(), como se describe en Cambia las variables de entorno predeterminadas.

Además de setUp(), incluye un método tearDown() que desactive testbed. Esto restablece los stubs originales para que las pruebas no interfieran entre sí.

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

Luego, 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 de Datastore o Memcache en lugar de usar los servicios reales.

Por ejemplo, el método que se muestra a continuación crea dos entidades: la primera entidad usa el valor predeterminado para el atributo number (42) y la segunda, un valor no predeterminado para number (17). El método compila una consulta para entidades de TestModel, pero solo en el caso de las que tienen el valor predeterminado de number.

Después de recuperar todas las entidades que coinciden, el método prueba que se encontró exactamente una entidad y que el valor del atributo number de esa entidad es el 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 en el que se crea una entidad y se la recupera mediante la función GetEntityViaMemcache() que creamos antes. El método prueba que se mostró una entidad y que su valor de number es el mismo que el de la entidad antes creada.

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)

Y, por último, invoca unittest.main().

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

Para ejecutar las pruebas, consulta Cómo ejecutar pruebas.

Cómo escribir pruebas de Cloud Datastore

Si tu app usa Cloud Datastore, recomendamos escribir pruebas que verifiquen el comportamiento de tu aplicación ante una coherencia eventual. db.testbed expone las opciones que facilitan este proceso:

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 una principal). Mediante la configuración de la probabilidad en un 0%, se le indica al stub del almacenamiento de datos que opere con la cantidad máxima de coherencia eventual. La coherencia eventual máxima significa que las operaciones de escritura se confirmarán, pero su aplicación siempre fallará. Por lo tanto, las consultas globales (no la principal) no verán los cambios constantemente. Por supuesto, esto no representa la cantidad de coherencia eventual que verá tu aplicación cuando se ejecute en producción, pero configurar el almacén de datos local con el fin de que siempre se comporte de esta manera es muy útil para realizar pruebas. Si usas una probabilidad distinta de cero, PseudoRandomHRConsistencyPolicy realiza una secuencia determinista de decisiones de coherencia para que los resultados de la prueba 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 API de pruebas son útiles para verificar que tu aplicación se comporte de manera correcta en el caso de una coherencia eventual, 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, realizar get() de una Entity que pertenece a un grupo de entidades con una operación de escritura sin aplicar siempre hará que se vean los resultados de esta operación para las consultas globales posteriores. En producción, este no el caso.

Cómo escribir pruebas de correo electrónico

Para probar el servicio de correo electrónico, puedes usar el stub de este servicio. Al igual que con otros servicios que testbed admite, primero inicializa el stub, luego invoca el código que usa la API de correo electrónico y, por último, realiza una prueba para ver si se enviaron 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='[e-mail]',
                       subject='This is a test',
                       sender='[e-mail]',
                       body='This is a test e-mail')
        messages = self.mail_stub.get_sent_messages(to='[e-mail]')
        self.assertEqual(1, len(messages))
        self.assertEqual('[e-mail]', messages[0].to)

Escribe pruebas de lista de tareas en cola

Para escribir pruebas que usen el servicio de taskqueue, puedes usar el stub de taskqueue. Al igual que con otros servicios que testbed admite, primero inicializa el stub, luego invoca el código que usa la API de taskqueue y, por último, realiza una prueba para ver si se agregaron correctamente las tareas 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')

Configura el archivo de configuración queue.yaml

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

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

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

La ubicación del archivo queue.yaml se especifica cuando inicializas el stub:

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

En la muestra, queue.yaml está en el mismo directorio que las pruebas. Si estuviera en otra carpeta, la ruta de acceso tendría que especificarse en root_path.

Filtra tareas

get_filtered_tasks del stub de taskqueue te permite filtrar las tareas en cola. Esto facilita escribir pruebas que deben verificar el código que agrega varias tareas a la cola.

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)

Escribe pruebas de tareas diferidas

Si el código de la aplicación usa la biblioteca diferida, puedes usar el stub de taskqueue junto con deferred para verificar que las funciones diferidas se agreguen a la cola y se ejecuten de forma correcta.

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)

Cambia las variables de entorno predeterminadas

Los servicios de App Engine suelen depender de las variables de entorno. El método activate() de clase testbed.Testbed usa valores predeterminados para estas, pero puedes establecer valores personalizados basados en tus necesidades de prueba con el método setup_env de clase testbed.Testbed.

Nota: Asegúrate de que cualquier llamada a setup_env se realice después de la llamada activate, a fin de que la deactivate final restablezca el entorno original de forma correcta. Esto es importante para separar las pruebas de unidades entre sí.

Por ejemplo, supongamos que tienes una prueba que almacena varias entidades en Datastore, todas ellas están vinculadas al mismo ID de aplicación. Ahora, deseas volver a ejecutar las mismas pruebas, pero con un ID de aplicación diferente al que está vinculado a las entidades almacenadas. Para hacerlo, pasa el valor nuevo 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')

Simula el acceso

Otro uso frecuente de setup_env es simular el acceso de un usuario, con o sin privilegios de administrador, para verificar si los controladores operan de forma correcta 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='[e-mail]', 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(), '[e-mail]')
        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 accedió, a self.loginUser('[e-mail]', '123') a fin de simular que un usuario no administrador accedió y a self.loginUser('[e-mail]', '123', is_admin=True) para simular que un usuario administrador accedió.

Configura un framework de pruebas

Las utilidades de prueba del SDK no están vinculadas a un marco de trabajo específico. Puedes ejecutar las pruebas de unidades con cualquier ejecutor de pruebas disponible de App Engine, por ejemplo, nose-gae o ferrisnose. También puedes escribir tu propio ejecutor de pruebas o usar el que se muestra a continuación.

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

Puedes nombrar la secuencia de comandos como desees. Cuando la ejecutes, proporciona la ruta a la instalación de Google Cloud CLI o el SDK de Google App Engine y la ruta a los módulos de prueba. La secuencia de comandos descubrirá todas las pruebas en la ruta proporcionada y mostrará los resultados en la transmisión 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)

Ejecuta las pruebas

Puedes realizar estas pruebas con solo ejecutar la secuencia de comandos runner.py, que se describe en detalle en Configura un framework de pruebas:

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