Pedro López Mareque

Software Engineer

Caso real de flaky test por acoplamiento

Una cosa que echo mucho en falta cuando leo posts sobre programación, es la falta de casos reales. Es algo que me aplico a mí mismo, ya que muchas veces es complicado tener algo que mostrar sin problemas de privacidad. Hoy por suerte os puedo traer una situación real que me ha tocado resolver, hacer unos días en el trabajo.

Historia de un caso real

Todo empieza una mañana cuando un test que empieza a fallar sin motivo aparente. Mirando los últimos cambios en el proyecto no se ve nada que tenga con este test ni con la clase que está testeando. Extraño, ¿no?

El test está fallando de manera aleatoria, a veces pasa y a veces no, así que hemos encontrado un flaky test.

Aclarar, ya que como muy bien comenta aquí Albert, este error se produce por qué todos los tests se ejecutan en paralelo con un orden aleatorio y así la suite completa sea lo más rápida posible.

Ok, ¿pero qué es un flaky test?

Podríamos definir un test como flaky cuando su comportamiento no es consistente, es decir, sin cambiar las condiciones del mismo una vez falla y otras no.

Sin un motivo claro, el test puede fallar de manera repentina, lo que hace que tu red de tests no sea fiable (la peor pesadilla para alguien que desarrolla) y todo se tambalee a tu alrededor.

Si quieres saber un poco más sobre el tema, este video de Jason Palmer me parece indespensable.

Investigación, ¿qué está pasando?

Para entender qué está ocurriendo, lo primero que hacemos es leer el error que tenemos. En este caso nos dice algo relativo a any-url:

AssertionError: 
expected: 'any-url' to equal 'https://app.datadoghq.eu/...'

=========================== short test summary info ============================
FAILED tests/unit/infrastructure/test_datadog_helper.py::TestDatadogHelper::test_generates_url_from_pod_name
======================= 1 failed, 1335 passed in 17.77s ========================

Vamos, por tanto, a revisar el test que está fallando. En este caso el test es test_generates_url_from_pod_name:

import time

from doublex import method_returning
from expects import expect, equal

from infrastructure.datadog.datadog_helper import DatadogHelper, FIVE_MINUTES, MILLISECONDS

FIXED_TIME = 1000


class TestDatadogHelper:
    def test_generates_url_from_pod_name(self) -> None:
        time.time = method_returning(FIXED_TIME)
        from_timestamp = (FIXED_TIME * MILLISECONDS) - FIVE_MINUTES
        to_timestamp = FIXED_TIME * MILLISECONDS

        pod_name = "a-pod-name"
        expect(DatadogHelper.generate_url(pod_name)).to( # <---- HERE
            equal(
                f"https://app.datadoghq.eu/logs?query=pod_name%3A{pod_name}%2A&from_ts={from_timestamp}&to_ts={to_timestamp}&live=false"
            )
        )

Uhm…qué raro…ni rastro de any-url

Revisando el test que está fallando, podemos ver que en no hay ninguna referencia a nada como any-url, por tanto, tiene que ser otro test es el que está cambiando el comportamiento del método DatadogHelper.generate_url.

Si alguna vez te ves en esta situación donde se cambian comportamientos a nivel global… preocupate.

Si hacemos una búsqueda rápida en el resto del proyecto, podemos ver que hay otro test (sin ningún tipo de relación con el test qué está fallando) que está llamando también al método DatadogHelper.generate_url y que efectivamente está devolviendo el famoso any-url:

def test_raise_error_updating_one_schema(self) -> None:
    any_url = "any-url" # <---- HERE
    argo_helper: ArgoHelper = Mimic(Stub, ArgoHelper)
    with Mimic(Spy, ArgoApiClient) as argo_client:
        argo_client.get_workflow_status(ANY_ARG).returns("Failed")
    DatadogHelper.generate_url = method_returning(any_url) # <---- HERE
    schema_updater = SchemaUpdater(argo_helper, argo_client)

    expect(lambda: schema_updater.update_schema("any-user-name", "any-environment")).to(
        raise_error(
            SchemaUpdateFailedException,
            f"The {database_type.value.title()} schema update failed.\nYou can find more details in the <{any_url}|logs>",
        )
    )

Debido a que el método #generate_url es estático el helper method_returning() de doublex está cambiando su comportamiento a nivel global.

Pero para mí hay algo más extraño aún en este test, se está cambiando el comportamiento de la clase DatadogHelper pese a que en el propio test esta clase no se está usando.

Aquí está el primer acoplamiento que hemos detectado. La clase ArgoHelper está usando interna y directamente nuestro DatadogHelper en lugar de recibirlo como una dependencia, lo que nos obliga a hacer esa magia en el test:

class SchemaUpdater:
    def __init__(self, argo_helper: ArgoHelper, argo_client: ArgoClient):
        self.argo_helper = argo_helper
        self.argo_client = argo_client

    def update_schema(self, user_name: str, environment: str, schema: str) -> None:
        ...
        if self.argo_client.get_workflow_status(workflow_name) != "Succeeded":
            datadog_url = DatadogHelper.generate_url(workflow_name) # <---- HERE
            ...

Y digo magia cuando podría decir mala práctica la verdad. Este tipo de código deberíamos evitarlo a toda costa, ya que tarde o temprano nos va a dar muchos problemas.

Encontrado el primer problema vamos, por tanto, a pensar como nos gustaría que fuera nuestro test sin tener dicho acoplamiento, para después adaptar nuestro código de producción y solucionarlo:

    def test_raise_error_updating_one_schema(self) -> None:
        argo_helper: ArgoHelper = Mimic(Stub, ArgoHelper)
        any_url = "any-url"
        with Mimic(Spy, ArgoApiClient) as argo_client:
            argo_client.get_workflow_status(ANY_ARG).returns("Failed")
        with Mimic(Stub, DatadogHelper) as datadog_helper: # <---- HERE
            datadog_helper.generate_url(ANY_ARG).returns(any_url)
        schema_updater = SchemaUpdater(argo_helper, argo_client, datadog_helper) # <---- HERE

        expect(lambda: schema_updater.update_schema(database_type, "any-user-name", "any-environment")).to(
            raise_error(
                SchemaUpdateFailedException,
                f"The {database_type.value.title()} schema update failed.\nYou can find more details in the <{any_url}|logs>",
            )
        )

Como podemos ver la clase DatadogHelper ahora es una dependencia que se inyecta desde el exterior vía constructor. De esta forma podemos cambiar el comportamiento de la clase DatadogHelper en nuestro test sin afectar a otros tests.

En nuestro caso el beneficio principal es poder pasarle un objeto que responde a la llamada al método #generate_url con el valor que nosotros queramos.

Para volver a pasar nuestro test necesitamos adaptar nuestra clase SchemaUpdater para que acepte la nueva dependencia y haga uso de la misma:

class SchemaUpdater:
    def __init__(
        self,
        argo_helper: ArgoHelper,
        argo_client: ArgoClient, 
        datadog_helper: DatadogHelper # <---- HERE
    ) -> None:
        self.argo_helper = argo_helper
        self.argo_client = argo_client
        self.datadog_helper = datadog_helper # <---- HERE

    def update_schema(self, user_name: str, environment: str, schema: str) -> None:
        ...
        if self.argo_client.get_workflow_status(workflow_name) != "Succeeded":
            datadog_url = self.datadog_helper.generate_url(workflow_name) # <---- HERE
            ...

Espera, espera, que aún hay más acoplamiento

Con los cambios anteriores hemos eliminado el acoplamiento entre SchemaUpdater y DatadogHelper, ya no necesitamos modificar el método estático que afectaba a otros tests y, por tanto, nuestro flaky test ha dejado de serlo.

Pero si prestamos un poco más de atención a la implementación interna de nuestra clase DatadogHelper, podemos ver un par de cosas muy mejorables:

  • Se está usando directamente módulo time (otro acoplamiento).
  • El método generate_url es un método estático que no tiene acceso a las variables de la instancia.
import time

MILLISECONDS = 1000
FIVE_MINUTES = 300 * MILLISECONDS


class DatadogHelper:
    @staticmethod # <---- HERE
    def generate_url(pod_name: str) -> str:
        now = int(time.time()) * MILLISECONDS # <---- HERE
        from_timestamp = now - FIVE_MINUTES
        to_timestamp = now
        return f"https://app.datadoghq.eu/logs?query=pod_name%3A{pod_name}%2A&from_ts={from_timestamp}&to_ts={to_timestamp}&live=false"

Idealmente, deberíamos inyectar el módulo time como dependencia y hacer que el método generate_url sea de instancia en lugar de estático para poder hacer uso de la nueva dependencia a través de self.

Una vez más vamos a pensar en como nos gustaría que fuera nuestro test sin tener acoplamiento para después hacer los cambios necesarios en nuestro código de producción:

from doublex import Stub
from expects import expect, equal

from infrastructure.datadog.datadog_helper import DatadogHelper


class TestDatadogHelper:

    def test_generates_url_from_pod_name(self) -> None:
        any_fixed_time = 200
        from_timestamp = (any_fixed_time * DatadogHelper.MILLISECONDS) - DatadogHelper.FIVE_MINUTES
        to_timestamp = any_fixed_time * DatadogHelper.MILLISECONDS
        with Stub() as time_module: # <---- HERE
            time_module.time().returns(any_fixed_time)
        datadog_helper = DatadogHelper(time_module) # <---- HERE

        pod_name = "a-pod-name"
        generate_url = datadog_helper.generate_url(pod_name) # <---- HERE

        expected_url = f"https://app.datadoghq.eu/logs?query=pod_name%3A{pod_name}%2A&from_ts={from_timestamp}&to_ts={to_timestamp}&live=false"
        expect(generate_url).to(equal(expected_url))

Tres cosas han cambiado en nuestro test:

  • Hemos creado un Stub para el módulo time que nos permite controlar el valor que devuelve el método time().
  • Hemos creado una instancia de DatadogHelper pasándole el módulo time como dependencia.
  • Hemos llamado al método generate_url de la instancia en lugar de hacerlo de manera estática.

Además de esta forma habríamos eliminado el acoplamiento por segunda vez, evitando usar el método method_returning que estaba modificando el comportamiento del módulo time cuando lanzamos nuestra suite de tests.

Vamos, por tanto, a hacer los cambios necesarios en nuestro código de producción para que pase nuestro test:

from urllib import parse


class DatadogHelper:
    BASE_URL = "https://app.datadoghq.eu/logs"
    MILLISECONDS = 1000
    FIVE_MINUTES = 300 * MILLISECONDS

    def __init__(self, time_module: ModuleType) -> None: # <---- HERE
        self.time_module = time_module # <---- HERE

    def generate_url(self, pod_name: str) -> str:
        return f"{self.BASE_URL}?{self._build_params(pod_name)}"

    def _build_params(self, pod_name: str) -> str:
        current_time_in_seconds = int(self.time_module.time()) # <---- HERE
        now_ts = current_time_in_seconds * self.MILLISECONDS
        from_ts = now_ts - self.FIVE_MINUTES
        params = {
            "query": f"pod_name:{pod_name}*",
            "from_ts": from_ts,
            "to_ts": now_ts,
            "live": "false",
        }
        return parse.urlencode(params)

Hemos aprovechado para refactorizar el método generate_url y extraer la lógica de construcción de parámetros a un método, lo que da un poco más de semántica al método principal.

Conclusión

Pese a que tener un mínimo de acoplamiento es inevitable, hay que intentar siempre minimizarlo lo máximo posible. En este caso hemos visto cómo usar otras clases directamente dentro de nuestras clases en lugar de inyectar es una mala práctica que nos puede dar muchos dolores de cabeza.

Por tanto, yo recomiendo siempre hacer uso de la inyección de dependencias, ya que nos permite cambiar el comportamiento de nuestras clases en nuestros tests, además de eliminar el acoplamiento de las mismas.

Espero que os haya gustado el post, y que os haya servido para entender un poco mejor qué es el acoplamiento y cómo minimizarlo.