# Escribe pruebas con Testcontainers


Para probar el listener de Kafka, necesitas un broker de Kafka en ejecución y una base de datos MySQL, además de un contexto de Spring iniciado. Testcontainers levanta ambos servicios en contenedores de Docker y `@DynamicPropertySource` los conecta a Spring.

## Escribe la prueba

Crea `ProductPriceChangedEventHandlerTest.java`:

```java
package com.testcontainers.demo;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Optional;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.springframework.test.context.TestPropertySource;
import org.testcontainers.kafka.ConfluentKafkaContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

@SpringBootTest
@TestPropertySource(
  properties = {
    "spring.kafka.consumer.auto-offset-reset=earliest",
    "spring.datasource.url=jdbc:tc:mysql:8.0.32:///db",
  }
)
@Testcontainers
class ProductPriceChangedEventHandlerTest {

  @Container
  static final ConfluentKafkaContainer kafka =
    new ConfluentKafkaContainer("confluentinc/cp-kafka:7.8.0");

  @DynamicPropertySource
  static void overrideProperties(DynamicPropertyRegistry registry) {
    registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
  }

  @Autowired
  private KafkaTemplate<String, Object> kafkaTemplate;

  @Autowired
  private ProductRepository productRepository;

  @BeforeEach
  void setUp() {
    Product product = new Product(null, "P100", "Product One", BigDecimal.TEN);
    productRepository.save(product);
  }

  @Test
  void shouldHandleProductPriceChangedEvent() {
    ProductPriceChangedEvent event = new ProductPriceChangedEvent(
      "P100",
      new BigDecimal("14.50")
    );

    kafkaTemplate.send("product-price-changes", event.productCode(), event);

    await()
      .pollInterval(Duration.ofSeconds(3))
      .atMost(10, SECONDS)
      .untilAsserted(() -> {
        Optional<Product> optionalProduct = productRepository.findByCode(
          "P100"
        );
        assertThat(optionalProduct).isPresent();
        assertThat(optionalProduct.get().getCode()).isEqualTo("P100");
        assertThat(optionalProduct.get().getPrice())
          .isEqualTo(new BigDecimal("14.50"));
      });
  }
}
```

Esto es lo que hace la prueba:

- `@SpringBootTest` inicia el contexto completo de la aplicación Spring.
- La URL JDBC especial de Testcontainers (`jdbc:tc:mysql:8.0.32:///db`) en
  `@TestPropertySource` levanta un contenedor de MySQL y lo configura como el
  origen de datos (datasource) automáticamente.
- `@Testcontainers` y `@Container` gestionan el ciclo de vida del contenedor de
  Kafka. `@DynamicPropertySource` registra los servidores de bootstrap de Kafka
  con Spring para que el productor y el consumidor se conecten al contenedor de prueba.
- `@BeforeEach` crea un registro de `Product` en la base de datos antes de cada prueba.
- La prueba envía un `ProductPriceChangedEvent` al tema `product-price-changes`
  utilizando `KafkaTemplate`. Spring Boot convierte el objeto a JSON utilizando
  `JsonSerializer`.
- Debido a que el procesamiento de mensajes de Kafka es asíncrono, la prueba utiliza
  [Awaitility](http://www.awaitility.org/) para realizar consultas (polling) cada 3 segundos (hasta un
  máximo de 10 segundos) hasta que el precio del producto en la base de datos coincida con el
  valor esperado.
- La propiedad `spring.kafka.consumer.auto-offset-reset` se establece en `earliest`
  para que el listener consuma los mensajes incluso si se envían al tema
  antes de que el listener esté listo. Esta configuración es útil cuando se ejecutan pruebas.