diff --git a/.gitignore b/.gitignore index e18133e336..ad386984d7 100644 --- a/.gitignore +++ b/.gitignore @@ -173,6 +173,7 @@ Migrations/ /core-tests/e2e-tests/spring/spring-rest-rsa/target/ /core-tests/e2e-tests/spring/spring-rest-dynamodb/target/ /core-tests/e2e-tests/spring/spring-asyncapi-kafka/target/ +/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/target/ /core-tests/e2e-tests/spring/spring-rest-h2-v1/em.yaml /core-tests/integration-tests/core-it/target/ /core-tests/integration-tests/core-it/em.yaml diff --git a/client-java/pom.xml b/client-java/pom.xml index 978c382859..f7dea6fa4f 100644 --- a/client-java/pom.xml +++ b/client-java/pom.xml @@ -45,8 +45,6 @@ 2.48 2.5.4 - - 2.7.6 diff --git a/core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/AsyncApiTestBase.java b/core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/AsyncApiTestBase.java index 81e7bd25fd..a99f893db0 100644 --- a/core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/AsyncApiTestBase.java +++ b/core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/AsyncApiTestBase.java @@ -1,17 +1,22 @@ package org.evomaster.e2etests.utils; +import com.google.inject.Injector; import com.webfuzzing.commons.faults.FaultCategory; +import kotlin.Pair; import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiReplyDto; +import org.evomaster.client.java.instrumentation.shared.ObjectiveNaming; import org.evomaster.core.Main; import org.evomaster.core.problem.asyncapi.data.AsyncApiCallResult; import org.evomaster.core.problem.asyncapi.data.AsyncApiIndividual; import org.evomaster.core.problem.asyncapi.data.AsyncApiOutcome; import org.evomaster.core.problem.enterprise.DetectedFault; import org.evomaster.core.search.Solution; +import org.evomaster.core.search.service.IdMapper; import java.util.List; import java.util.Objects; import java.util.Set; +import java.util.TreeSet; import java.util.stream.Collectors; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -26,6 +31,34 @@ protected Solution initAndRun(List args) { return (Solution) Main.initAndRun(args.toArray(new String[0])); } + /** + * Run the search, and keep the injector too, to read the names of the targets afterwards. + */ + protected Pair> initAndDebug(List args) { + Pair> run = Main.initAndDebug(args.toArray(new String[0])); + return new Pair<>(run.getFirst(), (Solution) run.getSecond()); + } + + /** + * That some test in the solution ran code of {@code className}, a class of the SUT. + * + * A white-box assertion, for messages nothing answers. A whole class rather than a line keeps + * the test independent of how the SUT's sources are laid out. + * + * @param run what {@link #initAndDebug} returned + * @param className the class, with dots, e.g. {@code com.foo.Bar} + */ + protected void assertClassCovered(Pair> run, String className) { + IdMapper idMapper = run.getFirst().getInstance(IdMapper.class); + String target = ObjectiveNaming.classObjectiveName(className); + Set coveredClasses = run.getSecond().getOverall().coveredTargets().stream() + .map(idMapper::getDescriptiveId) + .filter(id -> id.startsWith(ObjectiveNaming.CLASS)) + .collect(Collectors.toCollection(TreeSet::new)); + assertTrue(coveredClasses.contains(target), + "With seed " + defaultSeed + ": no test ran any code of " + className + ", only of " + coveredClasses); + } + /** * The result of every message published, across the whole solution. */ diff --git a/core-tests/e2e-tests/spring/pom.xml b/core-tests/e2e-tests/spring/pom.xml index 4d5f71808b..85d0f0d222 100644 --- a/core-tests/e2e-tests/spring/pom.xml +++ b/core-tests/e2e-tests/spring/pom.xml @@ -38,6 +38,7 @@ spring-rpc-thrift spring-mcp-bb spring-asyncapi-kafka + spring-asyncapi-kafka-listener diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/pom.xml b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/pom.xml new file mode 100644 index 0000000000..8f4543893c --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/pom.xml @@ -0,0 +1,131 @@ + + + + evomaster-e2e-tests-spring + org.evomaster + 6.2.1-SNAPSHOT + + 4.0.0 + + + evomaster-e2e-tests-spring-asyncapi-kafka-listener + jar + + + ${skipAsyncApi} + + + + + + javax.validation + validation-api + 2.0.1.Final + + + org.evomaster + evomaster-e2e-tests-utils + test-jar + + + org.evomaster + evomaster-client-java-controller + + + org.evomaster + evomaster-core + test + + + org.evomaster + evomaster-client-java-instrumentation + test-jar + + + + org.springframework.boot + spring-boot-starter + ${springboot.version} + + + + org.springframework.boot + spring-boot-starter-jdbc + ${springboot.version} + + + org.springframework.kafka + spring-kafka + ${spring-kafka.version} + + + com.h2database + h2 + + + com.fasterxml.jackson.core + jackson-databind + + + + org.apache.kafka + kafka-clients + + + org.testcontainers + kafka + + + + org.hamcrest + hamcrest-all + + + org.junit.jupiter + junit-jupiter-api + test + + + org.junit.jupiter + junit-jupiter-engine + test + + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + false + + + + + + diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/ExpressDispatch.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/ExpressDispatch.java new file mode 100644 index 0000000000..17975c07e2 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/ExpressDispatch.java @@ -0,0 +1,21 @@ +package com.foo.asyncapi.shop; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Books a courier for an order shipped express. Only ever reached from + * {@link OrderListener#onOrderPlaced}, when the shipping is {@link OrderListener#EXPRESS}, so + * whether any of its code is covered says whether a search got there. + */ +public final class ExpressDispatch { + + private static final Logger log = LoggerFactory.getLogger(ExpressDispatch.class); + + private ExpressDispatch() { + } + + static void book(OrderPlaced order) { + log.debug("Courier booked for order {}", order.getOrderId()); + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceListener.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceListener.java new file mode 100644 index 0000000000..451306e4d0 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceListener.java @@ -0,0 +1,46 @@ +package com.foo.asyncapi.shop; + +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.stereotype.Component; + +import java.util.concurrent.ExecutorService; + +/** + * Consumes {@link InvoiceRequested}, and leaves the rendering to another thread: the listener + * returns at once, and the invoice is rendered a while later. Neither the listener returning nor + * the record being committed says the service is done; only waiting long enough does. + */ +@Component +public class InvoiceListener { + + /** + * How long rendering an invoice takes, in milliseconds. Long enough for the listener to have + * returned, and the offset to have been committed, well before the invoice is rendered. + */ + static final long RENDERING_MS = 150; + + private static final String VALUE_TYPE = + JsonDeserializer.VALUE_DEFAULT_TYPE + "=com.foo.asyncapi.shop.InvoiceRequested"; + + private final ExecutorService invoiceExecutor; + + public InvoiceListener(ExecutorService invoiceExecutor) { + this.invoiceExecutor = invoiceExecutor; + } + + @KafkaListener(id = ShopTopics.INVOICES_LISTENER, topics = ShopTopics.INVOICES_REQUESTED, properties = VALUE_TYPE) + public void onInvoiceRequested(InvoiceRequested request) { + + invoiceExecutor.submit(() -> { + try { + Thread.sleep(RENDERING_MS); + } catch (InterruptedException e) { + //the service is shutting down + Thread.currentThread().interrupt(); + return; + } + InvoiceRenderer.render(request); + }); + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceRenderer.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceRenderer.java new file mode 100644 index 0000000000..323068b399 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceRenderer.java @@ -0,0 +1,23 @@ +package com.foo.asyncapi.shop; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Renders an invoice. Only ever run on the thread {@link InvoiceListener} hands its work to, after + * the listener has returned, so whether any of its code is covered says whether the work done + * after the handler was still credited to the message. + */ +public final class InvoiceRenderer { + + private static final Logger log = LoggerFactory.getLogger(InvoiceRenderer.class); + + private InvoiceRenderer() { + } + + static void render(InvoiceRequested request) { + Integer cents = request.getAmountCents(); + String total = cents == null ? "?" : String.valueOf(cents); + log.debug("Invoice for order {} rendered, total {}", request.getOrderId(), total); + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceRequested.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceRequested.java new file mode 100644 index 0000000000..fe3f47004d --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/InvoiceRequested.java @@ -0,0 +1,28 @@ +package com.foo.asyncapi.shop; + +/** + * A request to render the invoice of an order, as published on + * {@link ShopTopics#INVOICES_REQUESTED}. Every field is null when the message left it out. + */ +public class InvoiceRequested { + + private String orderId; + + private Integer amountCents; + + public String getOrderId() { + return orderId; + } + + public void setOrderId(String orderId) { + this.orderId = orderId; + } + + public Integer getAmountCents() { + return amountCents; + } + + public void setAmountCents(Integer amountCents) { + this.amountCents = amountCents; + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/OrderListener.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/OrderListener.java new file mode 100644 index 0000000000..e80770cb61 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/OrderListener.java @@ -0,0 +1,62 @@ +package com.foo.asyncapi.shop; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.stereotype.Component; + +/** + * Consumes {@link OrderPlaced}. Two things it does are behind a string value that the contract + * does not list, so that a search drawing random strings does not get there, but one that sees + * what the service compares the value with does: express shipping, and an urgent priority. + */ +@Component +public class OrderListener { + + private static final Logger log = LoggerFactory.getLogger(OrderListener.class); + + /** + * Compared with {@code equals}, the way a string-keyed rule usually is. + */ + static final String EXPRESS = "EXPRESS"; + + /** + * Which class the record value is read into, for this listener only: each topic carries a + * type of its own, and the publisher sends no type header. + */ + private static final String VALUE_TYPE = + JsonDeserializer.VALUE_DEFAULT_TYPE + "=com.foo.asyncapi.shop.OrderPlaced"; + + @KafkaListener(id = ShopTopics.ORDERS_LISTENER, topics = ShopTopics.ORDERS_PLACED, properties = VALUE_TYPE) + public void onOrderPlaced(OrderPlaced order) { + + if (EXPRESS.equals(order.getShipping())) { + ExpressDispatch.book(order); + } + + Priority priority = priorityOf(order); + if (priority == Priority.URGENT) { + UrgentEscalation.escalate(order); + } + + log.debug("Order {} placed", order.getOrderId()); + } + + /** + * The order's priority, or null when it names none of them: an order is still taken when its + * priority is not understood. + */ + private static Priority priorityOf(OrderPlaced order) { + + if (order.getPriority() == null) { + return null; + } + try { + return Priority.valueOf(order.getPriority()); + } catch (IllegalArgumentException e) { + log.debug("Unknown priority {}", order.getPriority()); + return null; + } + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/OrderPlaced.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/OrderPlaced.java new file mode 100644 index 0000000000..69a6306c73 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/OrderPlaced.java @@ -0,0 +1,55 @@ +package com.foo.asyncapi.shop; + +/** + * A new order, as published on {@link ShopTopics#ORDERS_PLACED}. Every field is null when the + * message left it out. + */ +public class OrderPlaced { + + private String orderId; + + /** + * How the order is to be shipped. Free text in the contract; only {@code EXPRESS} means + * anything to the service. + */ + private String shipping; + + /** + * The name of a {@link Priority}, or anything else, which the service ignores. + */ + private String priority; + + private Integer quantity; + + public String getOrderId() { + return orderId; + } + + public void setOrderId(String orderId) { + this.orderId = orderId; + } + + public String getShipping() { + return shipping; + } + + public void setShipping(String shipping) { + this.shipping = shipping; + } + + public String getPriority() { + return priority; + } + + public void setPriority(String priority) { + this.priority = priority; + } + + public Integer getQuantity() { + return quantity; + } + + public void setQuantity(Integer quantity) { + this.quantity = quantity; + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/OversoldStock.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/OversoldStock.java new file mode 100644 index 0000000000..81a8fad478 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/OversoldStock.java @@ -0,0 +1,21 @@ +package com.foo.asyncapi.shop; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Records that more items of a product were taken than there were in stock. Only ever reached + * from {@link StockLedger#adjust}, for a product an earlier message recorded, so whether any of + * its code is covered says whether a test sent the messages that lead there. + */ +public final class OversoldStock { + + private static final Logger log = LoggerFactory.getLogger(OversoldStock.class); + + private OversoldStock() { + } + + static void record(String sku, int quantity) { + log.debug("Product {} oversold, stock at {}", sku, quantity); + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/Priority.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/Priority.java new file mode 100644 index 0000000000..70dd143665 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/Priority.java @@ -0,0 +1,11 @@ +package com.foo.asyncapi.shop; + +/** + * How urgent an order is. The contract carries it as a plain string, which the service reads + * with {@link #valueOf(String)}. + */ +public enum Priority { + LOW, + NORMAL, + URGENT +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/ShopKafkaApplication.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/ShopKafkaApplication.java new file mode 100644 index 0000000000..844646e59a --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/ShopKafkaApplication.java @@ -0,0 +1,46 @@ +package com.foo.asyncapi.shop; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.autoconfigure.validation.ValidationAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.context.annotation.Bean; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * A shop back end that is driven only by Kafka messages, and never answers any of them: an order + * placed, an invoice requested, a stock adjustment. Each is consumed by a {@code @KafkaListener}, + * with Spring Kafka's {@code JsonDeserializer} turning the record into a typed object, as + * described by {@code asyncapi/shop-kafka.yaml}. + * + * Written for the EvoMaster E2E tests. Each listener is there for one thing a search has to cope + * with when nothing comes back from the service: {@link OrderListener} hides code behind string + * values random data does not produce, {@link InvoiceListener} leaves its work to another thread, + * and {@link StockListener} keeps state in a database from one message to the next. + */ +/* + Bean validation is excluded: this service validates nothing, and the validation API that the + EvoMaster client puts on the classpath would otherwise make Spring look for an EL + implementation that is not there. + */ +@SpringBootApplication(exclude = ValidationAutoConfiguration.class) +public class ShopKafkaApplication { + + /** + * Where {@link InvoiceListener} hands its work to. One thread, so that invoices are rendered + * in the order they were requested. + */ + @Bean(destroyMethod = "shutdownNow") + public ExecutorService invoiceExecutor() { + return Executors.newSingleThreadExecutor(); + } + + public static void main(String[] args) { + new SpringApplicationBuilder(ShopKafkaApplication.class) + .web(WebApplicationType.NONE) + .run(args); + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/ShopTopics.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/ShopTopics.java new file mode 100644 index 0000000000..efb14abbd8 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/ShopTopics.java @@ -0,0 +1,33 @@ +package com.foo.asyncapi.shop; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +/** + * The topics the service consumes, as named in {@code asyncapi/shop-kafka.yaml}, and the ids of + * the listeners that consume them. Each id is also the listener's consumer group. + */ +public final class ShopTopics { + + public static final String ORDERS_PLACED = "shop.orders.placed"; + + public static final String INVOICES_REQUESTED = "shop.invoices.requested"; + + public static final String STOCK_ADJUSTED = "shop.stock.adjusted"; + + public static final String ORDERS_LISTENER = "shop-orders"; + + public static final String INVOICES_LISTENER = "shop-invoices"; + + public static final String STOCK_LISTENER = "shop-stock"; + + /** + * Every topic above, for whoever has to create them before the service starts. + */ + public static final List ALL = + Collections.unmodifiableList(Arrays.asList(ORDERS_PLACED, INVOICES_REQUESTED, STOCK_ADJUSTED)); + + private ShopTopics() { + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/StockAdjusted.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/StockAdjusted.java new file mode 100644 index 0000000000..f666048c1c --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/StockAdjusted.java @@ -0,0 +1,31 @@ +package com.foo.asyncapi.shop; + +/** + * A change to how many items of a product are in stock, as published on + * {@link ShopTopics#STOCK_ADJUSTED}. Every field is null when the message left it out. + */ +public class StockAdjusted { + + private String sku; + + /** + * How many items were added, or taken away when negative. + */ + private Integer delta; + + public String getSku() { + return sku; + } + + public void setSku(String sku) { + this.sku = sku; + } + + public Integer getDelta() { + return delta; + } + + public void setDelta(Integer delta) { + this.delta = delta; + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/StockLedger.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/StockLedger.java new file mode 100644 index 0000000000..2e841936a1 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/StockLedger.java @@ -0,0 +1,49 @@ +package com.foo.asyncapi.shop; + +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; + +import java.util.List; + +/** + * How many items of each product are in stock, kept in the {@code stock} table of an in-memory + * database. + * + * The first adjustment of a product records it as it comes; the ones after it add to what is + * there. Selling more than what is in stock can therefore only happen on a product an earlier + * message already recorded, which takes two messages, in order, in the same test. + */ +@Component +public class StockLedger { + + static final String TABLE = "stock"; + + private static final String SELECT = "SELECT quantity FROM " + TABLE + " WHERE sku = ?"; + + private static final String INSERT = "INSERT INTO " + TABLE + " (sku, quantity) VALUES (?, ?)"; + + private static final String UPDATE = "UPDATE " + TABLE + " SET quantity = ? WHERE sku = ?"; + + private final JdbcTemplate jdbc; + + public StockLedger(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + void adjust(String sku, int delta) { + + List current = jdbc.queryForList(SELECT, Integer.class, sku); + + if (current.isEmpty()) { + jdbc.update(INSERT, sku, delta); + return; + } + + int quantity = current.get(0) + delta; + jdbc.update(UPDATE, quantity, sku); + + if (quantity < 0) { + OversoldStock.record(sku, quantity); + } + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/StockListener.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/StockListener.java new file mode 100644 index 0000000000..6d6baa8db9 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/StockListener.java @@ -0,0 +1,32 @@ +package com.foo.asyncapi.shop; + +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.stereotype.Component; + +/** + * Consumes {@link StockAdjusted}, and keeps the stock of each product in the database through + * {@link StockLedger}. + */ +@Component +public class StockListener { + + private static final String VALUE_TYPE = + JsonDeserializer.VALUE_DEFAULT_TYPE + "=com.foo.asyncapi.shop.StockAdjusted"; + + private final StockLedger ledger; + + public StockListener(StockLedger ledger) { + this.ledger = ledger; + } + + @KafkaListener(id = ShopTopics.STOCK_LISTENER, topics = ShopTopics.STOCK_ADJUSTED, properties = VALUE_TYPE) + public void onStockAdjusted(StockAdjusted adjustment) { + + if (adjustment.getSku() == null || adjustment.getDelta() == null) { + //nothing to adjust + return; + } + ledger.adjust(adjustment.getSku(), adjustment.getDelta()); + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/UrgentEscalation.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/UrgentEscalation.java new file mode 100644 index 0000000000..fd3837ff55 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/java/com/foo/asyncapi/shop/UrgentEscalation.java @@ -0,0 +1,21 @@ +package com.foo.asyncapi.shop; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Pages whoever is on call about an urgent order. Only ever reached from + * {@link OrderListener#onOrderPlaced}, for {@link Priority#URGENT}, so whether any of its code is + * covered says whether a search got there. + */ +public final class UrgentEscalation { + + private static final Logger log = LoggerFactory.getLogger(UrgentEscalation.class); + + private UrgentEscalation() { + } + + static void escalate(OrderPlaced order) { + log.debug("On-call paged for order {}", order.getOrderId()); + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/resources/application.properties b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/resources/application.properties new file mode 100644 index 0000000000..27ca3628a2 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/resources/application.properties @@ -0,0 +1,19 @@ +# The broker's address is given by whoever starts the service: spring.kafka.bootstrap-servers + +# Every listener reads its topic from the start the first time it joins. The topics are created +# empty right before the service starts, so nothing published to them is ever skipped, however +# long joining the group takes. +spring.kafka.consumer.auto-offset-reset=earliest +spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer + +# Record values are read into typed objects by JsonDeserializer, each listener naming its own type. +# Wrapped in ErrorHandlingDeserializer, so that a record that cannot be read is logged and skipped, +# rather than failing every poll that would return it. +spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer +spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=org.springframework.kafka.support.serializer.JsonDeserializer +spring.kafka.consumer.properties.spring.json.use.type.headers=false + +# Short, so that the listeners stop quickly when the service does. +spring.kafka.listener.poll-timeout=500 + +logging.level.org.apache.kafka=WARN diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/resources/asyncapi/shop-kafka.yaml b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/resources/asyncapi/shop-kafka.yaml new file mode 100644 index 0000000000..7694764177 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/resources/asyncapi/shop-kafka.yaml @@ -0,0 +1,111 @@ +# Artificial: written for the EvoMaster E2E tests, describing the shop service in +# com.foo.asyncapi.shop. Not taken from a third party. +asyncapi: 3.0.0 +info: + title: Shop events over Kafka + version: 1.0.0 + description: | + A shop back end that reacts to events published on **Kafka**, and never answers them. + Every operation is fire-and-forget: a client publishes a message on the operation's + topic, and that is the end of it as far as the client can see. + + What the service does with each message is only visible from inside it: it books + couriers, pages whoever is on call, renders invoices on a thread of its own, and keeps + the stock of each product in a database. +defaultContentType: application/json + +servers: + kafka: + host: localhost:9092 + protocol: kafka + description: Kafka broker carrying the shop topics. + +channels: + ordersPlaced: + address: shop.orders.placed + messages: + orderPlaced: + $ref: '#/components/messages/orderPlaced' + invoicesRequested: + address: shop.invoices.requested + messages: + invoiceRequested: + $ref: '#/components/messages/invoiceRequested' + stockAdjusted: + address: shop.stock.adjusted + messages: + stockAdjusted: + $ref: '#/components/messages/stockAdjusted' + +operations: + placeOrder: + action: receive + summary: An order was placed. + channel: + $ref: '#/channels/ordersPlaced' + requestInvoice: + action: receive + summary: The invoice of an order is wanted; it is rendered in the background. + channel: + $ref: '#/channels/invoicesRequested' + adjustStock: + action: receive + summary: Items of a product were added to, or taken from, the stock. + channel: + $ref: '#/channels/stockAdjusted' + +components: + messages: + orderPlaced: + name: OrderPlaced + contentType: application/json + payload: + $ref: '#/components/schemas/OrderPlaced' + invoiceRequested: + name: InvoiceRequested + contentType: application/json + payload: + $ref: '#/components/schemas/InvoiceRequested' + stockAdjusted: + name: StockAdjusted + contentType: application/json + payload: + $ref: '#/components/schemas/StockAdjusted' + + schemas: + OrderPlaced: + type: object + required: [orderId, shipping, priority, quantity] + properties: + orderId: { type: string } + # Deliberately free text, with no enum: which values the service acts on is only + # visible from inside it. A real document would often list them. + shipping: + type: string + description: How the order is shipped. + priority: + type: string + description: How urgent the order is. + quantity: { type: integer, format: int32, minimum: 1, maximum: 100 } + InvoiceRequested: + type: object + required: [orderId, amountCents] + properties: + orderId: { type: string } + amountCents: { type: integer, format: int32, minimum: 0 } + StockAdjusted: + type: object + required: [sku, delta] + properties: + # The whole catalogue, so that two adjustments of the same product are a matter of + # course rather than of luck. + sku: + type: string + enum: [A-100, B-200] + # Mostly sales: more can be taken away at once than added. + delta: + type: integer + format: int32 + minimum: -50 + maximum: 20 + description: Items added, or taken away when negative. diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/resources/schema.sql b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/resources/schema.sql new file mode 100644 index 0000000000..576a6d4800 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/main/resources/schema.sql @@ -0,0 +1,4 @@ +CREATE TABLE stock ( + sku VARCHAR(255) PRIMARY KEY, + quantity INT NOT NULL +); diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/test/java/com/foo/asyncapi/ShopKafkaController.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/test/java/com/foo/asyncapi/ShopKafkaController.java new file mode 100644 index 0000000000..f9d070ffba --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/test/java/com/foo/asyncapi/ShopKafkaController.java @@ -0,0 +1,418 @@ +package com.foo.asyncapi; + +import com.foo.asyncapi.shop.ShopKafkaApplication; +import com.foo.asyncapi.shop.ShopTopics; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.ListOffsetsResult; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.admin.OffsetSpec; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.TopicExistsException; +import org.apache.kafka.common.serialization.StringSerializer; +import org.evomaster.client.java.controller.EmbeddedSutController; +import org.evomaster.client.java.controller.api.dto.SutInfoDto; +import org.evomaster.client.java.controller.api.dto.auth.AuthenticationDto; +import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiActionDto; +import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiReplyDto; +import org.evomaster.client.java.controller.problem.AsyncApiProblem; +import org.evomaster.client.java.controller.problem.ProblemInfo; +import org.evomaster.client.java.sql.DbSpecification; +import org.evomaster.client.java.utils.SimpleLogger; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.kafka.config.KafkaListenerEndpointRegistry; +import org.springframework.kafka.listener.MessageListenerContainer; +import org.testcontainers.kafka.KafkaContainer; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.function.BooleanSupplier; +import java.util.stream.Collectors; + +/** + * The driver for the shop service, whose operations are all fire-and-forget: it owns the broker, + * publishes what the core asks it to, and returns as soon as the broker has the message. Nothing + * ever comes back, so what the service did with it is only known from its coverage. + * + * Outside {@code com.foo.asyncapi.shop}, the package whose code is covered, so that the driver's + * own lines are neither targets of the search nor activity that a wait for the service's + * coverage to settle would mistake for the service's. + */ +public class ShopKafkaController extends EmbeddedSutController { + + private static final String KAFKA_IMAGE = "apache/kafka:3.8.0"; + + private static final String DOCUMENT = "/asyncapi/shop-kafka.yaml"; + + private static final String BOOTSTRAP_PROPERTY = "spring.kafka.bootstrap-servers"; + + private static final String DELETE_STOCK = "DELETE FROM stock"; + + /** + * The bean name of the executor the service renders invoices on. + */ + private static final String INVOICE_EXECUTOR = "invoiceExecutor"; + + private static final long PUBLISH_TIMEOUT_SECONDS = 10; + + /** + * How long to wait, at start-up, for every listener to be given its topic. + */ + private static final long ASSIGNMENT_TIMEOUT_MS = 60_000; + + /** + * How long to wait, between two tests, for the service to have consumed everything published + * so far, and then for the invoices it was asked for to be rendered. Far more than it ever + * needs; reaching it is reported, not fatal. + */ + private static final long CONSUMED_TIMEOUT_MS = 10_000; + + /** + * How often the two waits above check again. + */ + private static final long CHECK_EVERY_MS = 20; + + private static final Duration CLOSE_TIMEOUT = Duration.ofSeconds(5); + + private final KafkaContainer kafka = new KafkaContainer(KAFKA_IMAGE); + + private ConfigurableApplicationContext ctx; + + private KafkaProducer producer; + + private AdminClient admin; + + public ShopKafkaController() { + super.setControllerPort(0); + } + + @Override + public String startSut() { + + try { + return start(); + } catch (RuntimeException e) { + /* + Nothing else will: isSutRunning() answers on the Spring context, which is not + assigned until the service is up, so a failure here would leave the broker running + for the rest of the fork. + */ + stopSut(); + throw e; + } + } + + private String start() { + + kafka.start(); + String bootstrap = kafka.getBootstrapServers(); + + Properties adminProps = new Properties(); + adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap); + admin = AdminClient.create(adminProps); + + createTopics(); + + ctx = new SpringApplicationBuilder(ShopKafkaApplication.class) + .web(WebApplicationType.NONE) + .properties(BOOTSTRAP_PROPERTY + "=" + bootstrap) + .run(); + + Properties props = new Properties(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap); + props.put(ProducerConfig.ACKS_CONFIG, "all"); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producer = new KafkaProducer<>(props); + + awaitListenersAssigned(); + + //this service has no URL: what there is to know is where its broker listens + return bootstrap; + } + + private void createTopics() { + + List topics = ShopTopics.ALL.stream() + .map(name -> new NewTopic(name, 1, (short) 1)) + .collect(Collectors.toList()); + try { + admin.createTopics(topics).all().get(PUBLISH_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (ExecutionException e) { + if (!(e.getCause() instanceof TopicExistsException)) { + throw new RuntimeException("Could not create the shop topics", e); + } + } catch (InterruptedException | TimeoutException e) { + throw new RuntimeException("Could not create the shop topics", e); + } + } + + /** + * Return once every listener has been given its topic. No message would be lost before that, + * as the topics are new and read from the start, but one handled only after its listener + * joined, which can take seconds, would have its coverage credited to a later message, or to + * none. + */ + private void awaitListenersAssigned() { + + boolean assigned = await(ASSIGNMENT_TIMEOUT_MS, () -> listeners().stream().allMatch(c -> { + Collection partitions = c.getAssignedPartitions(); + return partitions != null && !partitions.isEmpty(); + })); + + if (!assigned) { + throw new IllegalStateException( + "The shop listeners were not assigned their topics within " + ASSIGNMENT_TIMEOUT_MS + " ms"); + } + } + + private Collection listeners() { + return ctx.getBean(KafkaListenerEndpointRegistry.class).getListenerContainers(); + } + + @Override + public void stopSut() { + + /* + Every close is given a deadline and is allowed to fail: a Kafka client that will not + shut down must not keep the whole controller from stopping. + */ + if (producer != null) { + quietly(() -> producer.close(CLOSE_TIMEOUT)); + producer = null; + } + if (admin != null) { + quietly(() -> admin.close(CLOSE_TIMEOUT)); + admin = null; + } + if (ctx != null) { + quietly(ctx::close); + ctx = null; + } + quietly(kafka::stop); + } + + private static void quietly(Runnable action) { + try { + action.run(); + } catch (RuntimeException e) { + SimpleLogger.warn("Ignored while stopping the SUT: " + e.getMessage()); + } + } + + @Override + public boolean isSutRunning() { + return ctx != null && ctx.isRunning(); + } + + @Override + public String getPackagePrefixesToCover() { + return "com.foo.asyncapi.shop."; + } + + @Override + public void resetStateOfSUT() { + + /* + The stock is the service's only state. Before emptying it, wait for everything + published to be consumed, or a message of the previous test could write to the stock + after the reset; then for that test's invoices, or they would be covered in this one. + */ + if (!await(CONSUMED_TIMEOUT_MS, this::everythingConsumed)) { + SimpleLogger.warn("The shop service had not consumed everything published within " + + CONSUMED_TIMEOUT_MS + " ms"); + } + awaitInvoicesRendered(); + + ctx.getBean(JdbcTemplate.class).update(DELETE_STOCK); + } + + /** + * Wait until the invoice executor has run everything handed to it so far. It has one thread + * and runs tasks in order, so an empty task finishing means every earlier one has. + */ + private void awaitInvoicesRendered() { + + try { + Future done = ctx.getBean(INVOICE_EXECUTOR, ExecutorService.class).submit(() -> { + }); + done.get(CONSUMED_TIMEOUT_MS, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (ExecutionException | TimeoutException | RejectedExecutionException e) { + SimpleLogger.warn("The shop service had not rendered every invoice within " + + CONSUMED_TIMEOUT_MS + " ms: " + e.getMessage()); + } + } + + /** + * Whether every listener's group has committed up to the end of the partitions it was + * assigned, i.e., whether every message published so far has been handled. + */ + private boolean everythingConsumed() { + + try { + for (MessageListenerContainer listener : listeners()) { + + Collection assigned = listener.getAssignedPartitions(); + if (assigned == null) { + //the listener has not joined its group yet, so it has not consumed anything + return false; + } + Map latest = new LinkedHashMap<>(); + assigned.forEach(p -> latest.put(p, OffsetSpec.latest())); + + Map ends = + admin.listOffsets(latest).all().get(PUBLISH_TIMEOUT_SECONDS, TimeUnit.SECONDS); + Map committed = + admin.listConsumerGroupOffsets(listener.getGroupId()) + .partitionsToOffsetAndMetadata() + .get(PUBLISH_TIMEOUT_SECONDS, TimeUnit.SECONDS); + + for (Map.Entry end : ends.entrySet()) { + OffsetAndMetadata done = committed.get(end.getKey()); + long position = done == null ? 0 : done.offset(); + if (position < end.getValue().offset()) { + return false; + } + } + } + return true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } catch (ExecutionException | TimeoutException e) { + SimpleLogger.warn("Could not read the shop consumer groups: " + e.getMessage()); + return false; + } + } + + /** + * Check {@code condition} every {@link #CHECK_EVERY_MS} until it holds or {@code timeoutMs} + * has passed. + * + * @return whether it held + */ + private static boolean await(long timeoutMs, BooleanSupplier condition) { + + long deadline = System.currentTimeMillis() + timeoutMs; + while (true) { + if (condition.getAsBoolean()) { + return true; + } + if (System.currentTimeMillis() >= deadline) { + return false; + } + try { + Thread.sleep(CHECK_EVERY_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + } + + @Override + public List getInfoForAuthentication() { + return null; + } + + /** + * None, deliberately: {@link #resetStateOfSUT()} empties the stock table, and this E2E is not + * about SQL heuristics or data insertion. + */ + @Override + public List getDbSpecifications() { + return null; + } + + /** + * The container picks a port when it starts, so the address in the document is never where + * this broker actually is. A generated test asks for this instead of trusting the document. + * + * Asking before the container is up is answered with null, as the contract says to, rather + * than with the exception testcontainers raises for a port that has not been mapped yet. + */ + @Override + public String getAsyncApiServerAddress(String serverName) { + return kafka.isRunning() ? kafka.getBootstrapServers() : null; + } + + @Override + public ProblemInfo getProblemInfo() { + return AsyncApiProblem.fromSchemaText(readDocument()); + } + + @Override + public SutInfoDto.OutputFormat getPreferredOutputFormat() { + return SutInfoDto.OutputFormat.JAVA_JUNIT_5; + } + + private String readDocument() { + try (InputStream in = getClass().getResourceAsStream(DOCUMENT)) { + if (in == null) { + throw new IllegalStateException("Missing resource " + DOCUMENT); + } + return new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8)) + .lines() + .collect(Collectors.joining("\n")); + } catch (IOException e) { + throw new IllegalStateException("Cannot read " + DOCUMENT, e); + } + } + + /** + * Publish, and return once the broker has the message. The document declares no reply for + * any operation, so none is waited for; what the service does next is for the controller's + * settle window to wait on, if the core asks for one. + */ + @Override + public void executeAsyncApiAction(AsyncApiActionDto dto, AsyncApiReplyDto reply) { + + if (dto.replyAddress != null) { + throw new IllegalArgumentException("The shop service declares no replies, yet one was asked for on " + + dto.replyAddress); + } + + //no key: the generated tests publish none + ProducerRecord record = new ProducerRecord<>(dto.address, null, dto.payload); + if (dto.headers != null) { + dto.headers.forEach((name, value) -> record.headers().add(name, bytes(value))); + } + + try { + producer.send(record).get(PUBLISH_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (InterruptedException | ExecutionException | TimeoutException e) { + throw new RuntimeException("Could not publish to " + dto.address + ": " + e.getMessage(), e); + } + + reply.published = true; + reply.replyExpected = false; + } + + private static byte[] bytes(String value) { + return value == null ? null : value.getBytes(StandardCharsets.UTF_8); + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/test/java/org/evomaster/e2etests/spring/asyncapi/listener/ShopKafkaEMTest.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/test/java/org/evomaster/e2etests/spring/asyncapi/listener/ShopKafkaEMTest.java new file mode 100644 index 0000000000..34f62cd145 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka-listener/src/test/java/org/evomaster/e2etests/spring/asyncapi/listener/ShopKafkaEMTest.java @@ -0,0 +1,202 @@ +package org.evomaster.e2etests.spring.asyncapi.listener; + +import com.foo.asyncapi.ShopKafkaController; +import com.google.inject.Injector; +import kotlin.Pair; +import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiReplyDto; +import org.evomaster.client.java.instrumentation.shared.ObjectiveNaming; +import org.evomaster.core.problem.asyncapi.data.AsyncApiIndividual; +import org.evomaster.core.problem.asyncapi.data.AsyncApiOutcome; +import org.evomaster.core.search.EvaluatedIndividual; +import org.evomaster.core.search.FitnessValue; +import org.evomaster.core.search.Heuristics; +import org.evomaster.core.search.Solution; +import org.evomaster.core.search.action.EvaluatedAction; +import org.evomaster.core.search.service.IdMapper; +import org.evomaster.e2etests.utils.AsyncApiTestBase; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A white-box search over a service that never answers: every operation is fire-and-forget, so + * what the service did with a message is only known from its coverage, and that is what these + * tests assert on. + */ +public class ShopKafkaEMTest extends AsyncApiTestBase { + + private static final String REQUEST_INVOICE = "requestInvoice"; + + private static final String ADJUST_STOCK = "adjustStock"; + + private static final String SUT_PACKAGE = "com.foo.asyncapi.shop."; + + private static final String EXPRESS_DISPATCH = SUT_PACKAGE + "ExpressDispatch"; + + private static final String URGENT_ESCALATION = SUT_PACKAGE + "UrgentEscalation"; + + private static final String INVOICE_RENDERER = SUT_PACKAGE + "InvoiceRenderer"; + + private static final String OVERSOLD_STOCK = SUT_PACKAGE + "OversoldStock"; + + @BeforeAll + public static void initClass() throws Exception { + AsyncApiTestBase.initClass(new ShopKafkaController()); + } + + /** + * That some test in the solution credits the coverage of {@code className} to the action on + * {@code operation} that caused it. The core records, for each target, the index of the + * action open when the SUT first reached it in that test. + * + * {@code causesBefore} is how many actions on {@code operation} it takes before one can lead + * to the class; a credit counts only when the test has no more than that before it, so none + * of them could have landed late. Only the same test matters: the driver waits between tests. + * + * Some test, not every one: on a slow enough machine the work outlasts the window. + */ + private void assertClassCreditedTo( + Pair> run, String className, String operation, int causesBefore) { + + IdMapper idMapper = run.getFirst().getInstance(IdMapper.class); + String target = ObjectiveNaming.classObjectiveName(className); + + //the operations of each test that covered the class, with the one credited in brackets + List credits = new ArrayList<>(); + boolean credited = false; + for (EvaluatedIndividual test : run.getSecond().getIndividuals()) { + List operations = new ArrayList<>(); + for (EvaluatedAction action : test.evaluatedMainActions()) { + operations.add(action.getAction().getName()); + } + for (Map.Entry reached : test.getFitness().getViewOfData().entrySet()) { + if (!FitnessValue.Companion.isMaxValue(reached.getValue().getScore()) + || !target.equals(idMapper.getDescriptiveId(reached.getKey()))) { + continue; + } + int index = reached.getValue().getActionIndex(); + if (index < 0 || index >= operations.size()) { + credits.add(operations + " [none: index " + index + "]"); + continue; + } + credits.add(operations + " [" + index + "]"); + long earlier = operations.subList(0, index).stream().filter(operation::equals).count(); + if (operation.equals(operations.get(index)) && earlier <= causesBefore) { + credited = true; + } + } + } + + assertTrue(credited, "With seed " + defaultSeed + ": " + className + " was never credited to the " + + operation + " that caused it: " + credits); + } + + /** + * Express shipping and an urgent priority are both behind a string the document leaves free, + * compared with {@code equals} for the one and read with {@code Priority.valueOf} for the + * other. The search gets there by taint analysis, which is on by default in white-box mode: + * the payload carries a tainted string, the service reports what it was compared with, and + * the next mutation uses that value. + */ + @Test + public void testRunEM() throws Throwable { + + String outputFolder = "ShopKafkaEM"; + String className = "org.foo.asyncapi.ShopKafkaEM"; + + /* + The generated suite is compiled and run too. Its tests publish and move on, as the + service gives them nothing to wait for. + */ + runTestHandlingFlakyAndCompilation( + outputFolder, + className, + null, + 300, + true, + false, + (args) -> { + + /* + Without a window, a test's results are collected as soon as its last + message is on the broker, often before the service read it, and what it + compared that message's strings with is lost. A handler here is done in + milliseconds, so most waits end on stable coverage. + */ + setOption(args, "asyncApiSettleMs", "2000"); + setOption(args, "asyncApiCompletion", "COVERAGE_STABLE"); + setOption(args, "asyncApiStableMs", "100"); + + //more, shorter tests: each is another chance to use a value the service compared with + setOption(args, "maxTestSize", "3"); + + Pair> run = initAndDebug(args); + Solution solution = run.getSecond(); + + assertTrue(countOutcome(solution, AsyncApiOutcome.PUBLISHED) > 0, "nothing was published"); + assertEquals(0, countOutcome(solution, AsyncApiOutcome.PUBLISH_FAILED), "a message never left"); + + //stable coverage is what let messages go, not only the deadline + assertTrue(countSettleEnd(solution, AsyncApiReplyDto.SettleEndReason.COVERAGE_STABLE) > 0, + "With seed " + defaultSeed + ": no wait ended on stable coverage"); + + assertClassCovered(run, EXPRESS_DISPATCH); + assertClassCovered(run, URGENT_ESCALATION); + }, + 5); + + //the tests that reach those classes are kept, with the values that got there + assertTextInTests(outputFolder, className, "EXPRESS"); + assertTextInTests(outputFolder, className, "URGENT"); + } + + /** + * Two things a message leads to that the handler returning says nothing about: an invoice + * rendered on another thread a while after the listener has returned, and a product sold + * beyond its stock, which needs an earlier message of the same test to have been handled + * first. With a fixed window longer than both take, they are covered, and credited to the + * messages that caused them. + */ + @Test + public void testRunEMWithASettleWindow() throws Throwable { + + runTestHandlingFlaky( + "ShopKafkaSettleEM", + "org.foo.asyncapi.ShopKafkaSettleEM", + 150, + false, + (args) -> { + + /* + Rendering an invoice takes 150 ms after the listener returns. A window + more than three times that leaves room for a loaded machine, while each + test still costs little. + */ + setOption(args, "asyncApiSettleMs", "500"); + + Pair> run = initAndDebug(args); + Solution solution = run.getSecond(); + + assertEquals(0, countOutcome(solution, AsyncApiOutcome.PUBLISH_FAILED), "a message never left"); + assertEverySettled(solution); + + assertClassCovered(run, INVOICE_RENDERER); + assertClassCovered(run, OVERSOLD_STOCK); + + /* + Covered at all could be late coverage counted on whatever action was open. + An invoice is rendered for any request; overselling needs one earlier + adjustment. + */ + assertClassCreditedTo(run, INVOICE_RENDERER, REQUEST_INVOICE, 0); + assertClassCreditedTo(run, OVERSOLD_STOCK, ADJUST_STOCK, 1); + }, + 5); + } +} diff --git a/pom.xml b/pom.xml index e554db8e56..bdb21656db 100644 --- a/pom.xml +++ b/pom.xml @@ -297,6 +297,11 @@ 9.9 1.21.4 3.7.2 + + 2.7.6 1.17.2 5.13.2