From 1473048642719820875e3ee93c11181dafb5bd04 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Tue, 29 Sep 2026 23:19:21 -0300 Subject: [PATCH 1/6] AsyncAPI 3.x: an E2E test, NCS driven through Kafka The first search against a real broker. The SUT is the NCS case study re-expressed as a Kafka service: six request topics answered on six reply topics with a result or an error, as ncs-kafka.yaml describes. It speaks only Kafka; there is no HTTP endpoint. Its routines are ported from EMB's NCS and cited as such. The driver owns the broker, a testcontainers apache/kafka, and is the reference implementation of SutController.executeAsyncApiAction: publish with the correlation id where the document says it goes, wait for the reply that carries it back, report and do not judge. kafka-clients and testcontainers-kafka are used only here, in the test tree, and are managed in the root pom like everything else. The test asserts that every operation answers and that expint and gammq each reach both their declared replies. Those two are the operations whose rejected inputs the schema does not rule out; bessj's and remainder's bounds are in the schema, so their errors need invalid data, which the search does not publish yet. 662 targets in 26 seconds, on two runs. --- .gitignore | 1 + .../e2etests/utils/AsyncApiTestBase.java | 65 +++ core-tests/e2e-tests/spring/pom.xml | 6 + .../spring/spring-asyncapi-kafka/pom.xml | 83 ++++ .../foo/asyncapi/ncs/NcsKafkaApplication.java | 28 ++ .../foo/asyncapi/ncs/NcsRequestConsumer.java | 127 +++++ .../java/com/foo/asyncapi/ncs/NcsService.java | 206 ++++++++ .../java/com/foo/asyncapi/ncs/imp/Bessj.java | 111 +++++ .../java/com/foo/asyncapi/ncs/imp/Expint.java | 71 +++ .../java/com/foo/asyncapi/ncs/imp/Fisher.java | 62 +++ .../java/com/foo/asyncapi/ncs/imp/Gammq.java | 89 ++++ .../com/foo/asyncapi/ncs/imp/Remainder.java | 44 ++ .../ncs/imp/TriangleClassification.java | 29 ++ .../main/resources/asyncapi/ncs-kafka.yaml | 288 +++++++++++ .../foo/asyncapi/ncs/NcsKafkaController.java | 452 ++++++++++++++++++ .../spring/asyncapi/kafka/NcsKafkaEMTest.java | 88 ++++ pom.xml | 14 + 17 files changed, 1764 insertions(+) create mode 100644 core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/AsyncApiTestBase.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsKafkaApplication.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsRequestConsumer.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsService.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Bessj.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Expint.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Fisher.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Gammq.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Remainder.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/TriangleClassification.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java create mode 100644 core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java diff --git a/.gitignore b/.gitignore index 55de0442a7..edb0b2a2e5 100644 --- a/.gitignore +++ b/.gitignore @@ -171,6 +171,7 @@ Migrations/ /core-tests/e2e-tests/spring/spring-rest-h2-v2/target/ /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-rest-h2-v1/em.yaml /core-tests/integration-tests/core-it/target/ /core-tests/integration-tests/core-it/em.yaml 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 new file mode 100644 index 0000000000..cf45e00398 --- /dev/null +++ b/core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/AsyncApiTestBase.java @@ -0,0 +1,65 @@ +package org.evomaster.e2etests.utils; + +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.search.Solution; + +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * What an E2E test over an AsyncAPI service needs: to run the search, and to read what + * publishing to each operation was seen to do. + */ +public class AsyncApiTestBase extends EnterpriseTestBase { + + protected Solution initAndRun(List args) { + return (Solution) Main.initAndRun(args.toArray(new String[0])); + } + + /** + * The result of every message published to [operation], across the whole solution. + */ + protected List resultsOf(Solution solution, String operation) { + return solution.getIndividuals().stream() + .flatMap(ind -> ind.evaluatedMainActions().stream()) + .filter(e -> e.getAction().getName().equals(operation)) + .map(e -> (AsyncApiCallResult) e.getResult()) + .collect(Collectors.toList()); + } + + /** + * The declared messages the replies to [operation] were recognised as. + */ + protected Set repliesOf(Solution solution, String operation) { + return resultsOf(solution, operation).stream() + .map(AsyncApiCallResult::getReplyMessage) + .filter(Objects::nonNull) + .collect(Collectors.toSet()); + } + + protected long countOutcome(Solution solution, AsyncApiOutcome outcome) { + return solution.getIndividuals().stream() + .flatMap(ind -> ind.evaluatedMainActions().stream()) + .map(e -> (AsyncApiCallResult) e.getResult()) + .filter(r -> r.getOutcome() == outcome) + .count(); + } + + protected void assertReplied(Solution solution, String operation) { + boolean ok = resultsOf(solution, operation).stream().anyMatch(r -> r.getOutcome() == AsyncApiOutcome.REPLIED); + assertTrue(ok, "With seed " + defaultSeed + ": no reply to '" + operation + "' was ever received"); + } + + protected void assertReplyReached(Solution solution, String operation, String messageId) { + Set replies = repliesOf(solution, operation); + assertTrue(replies.contains(messageId), + "With seed " + defaultSeed + ": '" + operation + "' never replied with '" + messageId + "', only with " + replies); + } +} diff --git a/core-tests/e2e-tests/spring/pom.xml b/core-tests/e2e-tests/spring/pom.xml index 2ce26087e4..7855d2c019 100644 --- a/core-tests/e2e-tests/spring/pom.xml +++ b/core-tests/e2e-tests/spring/pom.xml @@ -35,6 +35,7 @@ spring-rpc-grpc spring-rpc-thrift spring-mcp-bb + spring-asyncapi-kafka @@ -45,6 +46,11 @@ spring-boot ${springboot.version} + + org.springframework.boot + spring-boot-starter + ${springboot.version} + org.springframework.boot spring-boot-starter-web diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml new file mode 100644 index 0000000000..ee98ad4989 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml @@ -0,0 +1,83 @@ + + + + evomaster-e2e-tests-spring + org.evomaster + 6.2.1-SNAPSHOT + + 4.0.0 + + + evomaster-e2e-tests-spring-asyncapi-kafka + jar + + + + + javax.validation + validation-api + 2.0.1.Final + + + javax.ws.rs + javax.ws.rs-api + + + 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 + + + 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 + + + diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsKafkaApplication.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsKafkaApplication.java new file mode 100644 index 0000000000..a58bb7aa39 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsKafkaApplication.java @@ -0,0 +1,28 @@ +package com.foo.asyncapi.ncs; + +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; + +/** + * NCS over Kafka: the six numerical operations of the NCS case study, each consuming a request + * topic and answering on a reply topic, as described by {@code asyncapi/ncs-kafka.yaml}. + * + * The service speaks only Kafka. Its one component is {@link NcsRequestConsumer}. + */ +/* + 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 NcsKafkaApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(NcsKafkaApplication.class) + .web(WebApplicationType.NONE) + .run(args); + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsRequestConsumer.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsRequestConsumer.java new file mode 100644 index 0000000000..e53d248d3d --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsRequestConsumer.java @@ -0,0 +1,127 @@ +package com.foo.asyncapi.ncs; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +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.errors.WakeupException; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.SmartLifecycle; +import org.springframework.stereotype.Component; + +import java.time.Duration; +import java.util.Properties; + +/** + * Reads every request topic, answers each request on the matching reply topic, and copies the + * correlation id header over so that the requester can pair the two. + */ +@Component +public class NcsRequestConsumer implements SmartLifecycle { + + /** + * The header a request carries its correlation id in, as the document declares. + */ + static final String CORRELATION_HEADER = "correlationId"; + + private static final Duration POLL = Duration.ofMillis(200); + + private final String bootstrapServers; + + private final NcsService service; + + private volatile boolean running; + + private Thread loop; + + private KafkaConsumer consumer; + + private KafkaProducer producer; + + public NcsRequestConsumer(@Value("${ncs.kafka.bootstrap}") String bootstrapServers, NcsService service) { + this.bootstrapServers = bootstrapServers; + this.service = service; + } + + @Override + public void start() { + + Properties consumerProps = new Properties(); + consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + //a fresh group each start, so that a restarted service does not resume old offsets + consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "ncs-" + System.nanoTime()); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); + consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + consumer = new KafkaConsumer<>(consumerProps); + consumer.subscribe(service.requestTopics()); + + Properties producerProps = new Properties(); + producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + producerProps.put(ProducerConfig.ACKS_CONFIG, "all"); + producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producer = new KafkaProducer<>(producerProps); + + running = true; + loop = new Thread(this::consume, "ncs-kafka-consumer"); + loop.start(); + } + + private void consume() { + try { + while (running) { + ConsumerRecords records = consumer.poll(POLL); + for (ConsumerRecord request : records) { + answer(request); + } + } + } catch (WakeupException e) { + //asked to stop + } finally { + consumer.close(); + producer.close(); + } + } + + private void answer(ConsumerRecord request) { + + NcsService.Reply reply = service.handle(request.topic(), request.value()); + + ProducerRecord record = new ProducerRecord<>(reply.topic, request.key(), reply.body); + + Header correlation = request.headers().lastHeader(CORRELATION_HEADER); + if (correlation != null) { + record.headers().add(CORRELATION_HEADER, correlation.value()); + } + + producer.send(record); + } + + @Override + public void stop() { + running = false; + if (consumer != null) { + consumer.wakeup(); + } + if (loop != null) { + try { + loop.join(5000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + + @Override + public boolean isRunning() { + return running; + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsService.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsService.java new file mode 100644 index 0000000000..40622de4d6 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsService.java @@ -0,0 +1,206 @@ +package com.foo.asyncapi.ncs; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.foo.asyncapi.ncs.imp.Bessj; +import com.foo.asyncapi.ncs.imp.Expint; +import com.foo.asyncapi.ncs.imp.Fisher; +import com.foo.asyncapi.ncs.imp.Gammq; +import com.foo.asyncapi.ncs.imp.Remainder; +import com.foo.asyncapi.ncs.imp.TriangleClassification; +import org.springframework.stereotype.Service; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; + +/** + * The NCS operations, keyed by the topic their requests arrive on. Each answers with a result + * message or, for the inputs the REST version answers with a 400, an error message. + * + * The checks mirror {@code NcsRest} in EMB: bessj rejects an order outside 3..1000, fisher + * degrees of freedom above 1000, remainder operands beyond 10000 in either direction, and + * expint and gammq whatever their routines throw on. A request that is not the JSON the + * contract describes is an error too, as it would be over HTTP. + */ +@Service +public class NcsService { + + /** + * What goes back: on which topic, and the JSON body. + */ + public static final class Reply { + + public final String topic; + + public final String body; + + Reply(String topic, String body) { + this.topic = topic; + this.body = body; + } + } + + static final String TRIANGLE_REQUEST = "ncs.triangle.request"; + static final String TRIANGLE_REPLY = "ncs.triangle.reply"; + static final String BESSJ_REQUEST = "ncs.bessj.request"; + static final String BESSJ_REPLY = "ncs.bessj.reply"; + static final String EXPINT_REQUEST = "ncs.expint.request"; + static final String EXPINT_REPLY = "ncs.expint.reply"; + static final String FISHER_REQUEST = "ncs.fisher.request"; + static final String FISHER_REPLY = "ncs.fisher.reply"; + static final String GAMMQ_REQUEST = "ncs.gammq.request"; + static final String GAMMQ_REPLY = "ncs.gammq.reply"; + static final String REMAINDER_REQUEST = "ncs.remainder.request"; + static final String REMAINDER_REPLY = "ncs.remainder.reply"; + + private static final String RESULT_AS_INT = "resultAsInt"; + private static final String RESULT_AS_DOUBLE = "resultAsDouble"; + private static final String ERROR = "error"; + private static final String CODE = "code"; + private static final String MESSAGE = "message"; + + private static final int BAD_REQUEST = 400; + + private static final int REMAINDER_LIMIT = 10_000; + private static final int MAX_DEGREES_OF_FREEDOM = 1000; + + private final ObjectMapper mapper = new ObjectMapper(); + + /** + * A request the service could not use, answered as the REST version answers with a 400. + */ + private static final class Rejected extends RuntimeException { + Rejected(String message) { + super(message); + } + } + + public List requestTopics() { + return Arrays.asList( + TRIANGLE_REQUEST, BESSJ_REQUEST, EXPINT_REQUEST, FISHER_REQUEST, GAMMQ_REQUEST, REMAINDER_REQUEST); + } + + public Reply handle(String requestTopic, String json) { + + String replyTopic = replyTopicOf(requestTopic); + + try { + JsonNode request = parse(json); + return new Reply(replyTopic, compute(requestTopic, request)); + } catch (Rejected e) { + return new Reply(replyTopic, error(e.getMessage())); + } catch (RuntimeException e) { + //what the numerical routines throw on inputs they cannot handle + return new Reply(replyTopic, error(e.getMessage())); + } + } + + private String compute(String requestTopic, JsonNode request) { + + switch (requestTopic) { + + case TRIANGLE_REQUEST: + return intResult(TriangleClassification.classify(integer(request, "a"), integer(request, "b"), integer(request, "c"))); + + case BESSJ_REQUEST: { + int n = integer(request, "n"); + if (n <= 2 || n > MAX_DEGREES_OF_FREEDOM) { + throw new Rejected("n must be in 3..1000"); + } + return doubleResult(new Bessj().bessj(n, number(request, "x"))); + } + + case EXPINT_REQUEST: + return doubleResult(Expint.exe(integer(request, "n"), number(request, "x"))); + + case FISHER_REQUEST: { + int m = integer(request, "m"); + int n = integer(request, "n"); + if (m > MAX_DEGREES_OF_FREEDOM || n > MAX_DEGREES_OF_FREEDOM) { + throw new Rejected("m and n must not exceed 1000"); + } + return doubleResult(Fisher.exe(m, n, number(request, "x"))); + } + + case GAMMQ_REQUEST: + return doubleResult(new Gammq().exe(number(request, "a"), number(request, "x"))); + + case REMAINDER_REQUEST: { + int a = integer(request, "a"); + int b = integer(request, "b"); + if (a > REMAINDER_LIMIT || a < -REMAINDER_LIMIT || b > REMAINDER_LIMIT || b < -REMAINDER_LIMIT) { + throw new Rejected("a and b must be within -10000..10000"); + } + return intResult(Remainder.exe(a, b)); + } + + default: + throw new IllegalArgumentException("Not a request topic: " + requestTopic); + } + } + + private String replyTopicOf(String requestTopic) { + switch (requestTopic) { + case TRIANGLE_REQUEST: return TRIANGLE_REPLY; + case BESSJ_REQUEST: return BESSJ_REPLY; + case EXPINT_REQUEST: return EXPINT_REPLY; + case FISHER_REQUEST: return FISHER_REPLY; + case GAMMQ_REQUEST: return GAMMQ_REPLY; + case REMAINDER_REQUEST: return REMAINDER_REPLY; + default: throw new IllegalArgumentException("Not a request topic: " + requestTopic); + } + } + + private JsonNode parse(String json) { + try { + JsonNode node = json == null ? null : mapper.readTree(json); + if (node == null || !node.isObject()) { + throw new Rejected("the request must be a JSON object"); + } + return node; + } catch (IOException e) { + throw new Rejected("the request is not JSON: " + e.getMessage()); + } + } + + private static int integer(JsonNode request, String field) { + JsonNode value = request.get(field); + if (value == null || !value.canConvertToInt()) { + throw new Rejected("'" + field + "' must be an integer"); + } + return value.intValue(); + } + + private static double number(JsonNode request, String field) { + JsonNode value = request.get(field); + if (value == null || !value.isNumber()) { + throw new Rejected("'" + field + "' must be a number"); + } + return value.doubleValue(); + } + + private String intResult(int value) { + return mapper.createObjectNode().put(RESULT_AS_INT, value).toString(); + } + + private String doubleResult(double value) { + /* + JSON has no NaN or infinity. Rather than emit a token no reader accepts, or a string + where the contract promises a number, a result the routine could not compute is an + error, as the request was one the service cannot serve. + */ + if (Double.isNaN(value) || Double.isInfinite(value)) { + throw new Rejected("the result is not a finite number"); + } + return mapper.createObjectNode().put(RESULT_AS_DOUBLE, value).toString(); + } + + private String error(String message) { + ObjectNode error = mapper.createObjectNode(); + error.put(CODE, BAD_REQUEST); + error.put(MESSAGE, message == null ? "rejected" : message); + return mapper.createObjectNode().set(ERROR, error).toString(); + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Bessj.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Bessj.java new file mode 100644 index 0000000000..d07340f418 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Bessj.java @@ -0,0 +1,111 @@ +package com.foo.asyncapi.ncs.imp; + +/* + Ported unchanged, apart from the package, from the NCS case study of EMB (WebFuzzing/EMB): + jdk_8_maven/cs/rest/artificial/ncs, org.restncs.imp. The routines there follow Numerical + Recipes. + */ +public class Bessj { + + private final double ACC = 40.0; + private final double BIGNO = 1.0e10; + private final double BIGNI = 1.0e-10; + + public double bessj(int n, double x) { + int j, jsum, m; + double ax, bj, bjm, bjp, sum, tox, ans; + + if (n < 2) + throw new IllegalArgumentException("Index n less than 2 in bessj"); + ax = Math.abs(x); + if (ax == 0.0) + return 0.0; + else if (ax > n) { + tox = 2.0 / ax; + bjm = bessj0(ax); + bj = bessj1(ax); + for (j = 1; j < n; j++) { + bjp = j * tox * bj - bjm; + bjm = bj; + bj = bjp; + } + ans = bj; + } else { + tox = 2.0 / ax; + m = 2 * ((n + (int) Math.round(Math.sqrt(ACC * n))) / 2); + jsum = 0; + bjp = ans = sum = 0.0; + bj = 1.0; + for (j = m; j > 0; j--) { + bjm = j * tox * bj - bjp; + bjp = bj; + bj = bjm; + if (Math.abs(bj) > BIGNO) { + bj *= BIGNI; + bjp *= BIGNI; + ans *= BIGNI; + sum *= BIGNI; + } + if (jsum != 0) + sum += bj; + jsum = (jsum != 0) ? 0 : 1; + if (j == n) + ans = bjp; + } + sum = 2.0 * sum - bj; + ans /= sum; + } + return x < 0.0 && (n & 1) != 0 ? -ans : ans; + } + + private static double bessj0(double x) { + double ax, z; + double xx, y, ans, ans1, ans2; + + if ((ax = Math.abs(x)) < 8.0) { + y = x * x; + ans1 = 57568490574.0 + y * (-13362590354.0 + y * (651619640.7 + + y * (-11214424.18 + y * (77392.33017 + y * (-184.9052456))))); + ans2 = 57568490411.0 + y * (1029532985.0 + y * (9494680.718 + + y * (59272.64853 + y * (267.8532712 + y * 1.0)))); + ans = ans1 / ans2; + } else { + z = 8.0 / ax; + y = z * z; + xx = ax - 0.785398164; + ans1 = 1.0 + y * (-0.1098628627e-2 + y * (0.2734510407e-4 + + y * (-0.2073370639e-5 + y * 0.2093887211e-6))); + ans2 = -0.1562499995e-1 + y * (0.1430488765e-3 + + y * (-0.6911147651e-5 + y * (0.7621095161e-6 + - y * 0.934935152e-7))); + ans = Math.sqrt(0.636619772 / ax) * (Math.cos(xx) * ans1 - z * Math.sin(xx) * ans2); + } + return ans; + } + + private static double bessj1(double x) { + double ax, z; + double xx, y, ans, ans1, ans2; + + if ((ax = Math.abs(x)) < 8.0) { + y = x * x; + ans1 = x * (72362614232.0 + y * (-7895059235.0 + y * (242396853.1 + + y * (-2972611.439 + y * (15704.48260 + y * (-30.16036606)))))); + ans2 = 144725228442.0 + y * (2300535178.0 + y * (18583304.74 + + y * (99447.43394 + y * (376.9991397 + y * 1.0)))); + ans = ans1 / ans2; + } else { + z = 8.0 / ax; + y = z * z; + xx = ax - 2.356194491; + ans1 = 1.0 + y * (0.183105e-2 + y * (-0.3516396496e-4 + + y * (0.2457520174e-5 + y * (-0.240337019e-6)))); + ans2 = 0.04687499995 + y * (-0.2002690873e-3 + + y * (0.8449199096e-5 + y * (-0.88228987e-6 + + y * 0.105787412e-6))); + ans = Math.sqrt(0.636619772 / ax) * (Math.cos(xx) * ans1 - z * Math.sin(xx) * ans2); + if (x < 0.0) ans = -ans; + } + return ans; + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Expint.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Expint.java new file mode 100644 index 0000000000..2ae02f4796 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Expint.java @@ -0,0 +1,71 @@ +package com.foo.asyncapi.ncs.imp; + +/* + Ported unchanged, apart from the package, from the NCS case study of EMB (WebFuzzing/EMB): + jdk_8_maven/cs/rest/artificial/ncs, org.restncs.imp. The routines there follow Numerical + Recipes. + */ +public class Expint { + + private static final double MAXIT = 100; + private static final double EULER = 0.5772156649; + private static final double FPMIN = 1.0e-30; + private static final double EPS = 1.0e-7; + + public static double exe(int n, double x) { + int i, ii, nm1; + double a, b, c, d, del, fact, h, psi, ans; + + nm1 = n - 1; + if (n < 0 || x < 0.0 || (x == 0.0 && (n == 0 || n == 1))) + throw new RuntimeException("error: n < 0 or x < 0"); + else { + if (n == 0) + ans = Math.exp(-x) / x; + else { + if (x == 0.0) + ans = 1.0 / nm1; + else { + if (x > 1.0) { + b = x + n; + c = 1.0 / FPMIN; + d = 1.0 / b; + h = d; + for (i = 1; i <= MAXIT; i++) { + a = -i * (nm1 + i); + b += 2.0; + d = 1.0 / (a * d + b); + c = b + a / c; + del = c * d; + h *= del; + if (Math.abs(del - 1.0) < EPS) { + return h * Math.exp(-x); + } + } + throw new RuntimeException("continued fraction failed in expint"); + } else { + ans = (nm1 != 0 ? 1.0 / nm1 : -Math.log(x) - EULER); + fact = 1.0; + for (i = 1; i <= MAXIT; i++) { + fact *= -x / i; + if (i != nm1) + del = -fact / (i - nm1); + else { + psi = -EULER; + for (ii = 1; ii <= nm1; ii++) + psi += 1.0 / ii; + del = fact * (-Math.log(x) + psi); + } + ans += del; + if (Math.abs(del) < Math.abs(ans) * EPS) { + return ans; + } + } + throw new RuntimeException("series failed in expint"); + } + } + } + } + return ans; + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Fisher.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Fisher.java new file mode 100644 index 0000000000..f9d2eb9c18 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Fisher.java @@ -0,0 +1,62 @@ +package com.foo.asyncapi.ncs.imp; + +/* + Ported unchanged, apart from the package, from the NCS case study of EMB (WebFuzzing/EMB): + jdk_8_maven/cs/rest/artificial/ncs, org.restncs.imp. The routines there follow Numerical + Recipes. + */ +public class Fisher { + + public static double exe(int m, int n, double x) { + int a, b, i, j; + double w, y, z, zk, d, p; + + a = 2 * (m / 2) - m + 2; + b = 2 * (n / 2) - n + 2; + w = (x * m) / n; + z = 1.0 / (1.0 + w); + if (a == 1) { + if (b == 1) { + p = Math.sqrt(w); + y = 0.3183098862; + d = y * z / p; + p = 2.0 * y * Math.atan(p); + } else { + p = Math.sqrt(w * z); + d = 0.5 * p * z / w; + } + } else if (b == 1) { + p = Math.sqrt(z); + d = 0.5 * z * p; + p = 1.0 - p; + } else { + d = z * z; + p = w * z; + } + y = 2.0 * w / z; + if (a == 1) + for (j = b + 2; j <= n; j += 2) { + d *= (1.0 + 1.0 / (j - 2)) * z; + p += d * y / (j - 1); + } + else { + zk = Math.pow(z, (double) ((n - 1) / 2)); + d *= (zk * n) / b; + p = p * zk + w * z * (zk - 1.0) / (z - 1.0); + } + y = w * z; + z = 2.0 / z; + b = n - 2; + for (i = a + 2; i <= m; i += 2) { + j = i + b; + d *= (y * j) / (i - 2); + p -= z * d / j; + } + if (p < 0.0) + return 0.0; + else if (p > 1.0) + return 1.0; + else + return p; + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Gammq.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Gammq.java new file mode 100644 index 0000000000..a7886428d9 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Gammq.java @@ -0,0 +1,89 @@ +package com.foo.asyncapi.ncs.imp; + +/* + Ported unchanged, apart from the package, from the NCS case study of EMB (WebFuzzing/EMB): + jdk_8_maven/cs/rest/artificial/ncs, org.restncs.imp. The routines there follow Numerical + Recipes. + */ +public class Gammq { + + private static final int ITMAX = 100; + private static final double EPS = 3.0e-7; + private static final double FPMIN = 1.0e-30; + + private double gamser, gammcf, gln; + + private double gammln(double xx) { + double x, y, tmp, ser; + double cof[] = {76.18009172947146, -86.50532032941677, 24.01409824083091, -1.231739572450155, 0.1208650973866179e-2, -0.5395239384953e-5}; + int j; + + y = x = xx; + tmp = x + 5.5; + tmp -= (x + 0.5) * Math.log(tmp); + ser = 1.000000000190015; + for (j = 0; j <= 5; j++) ser += cof[j] / ++y; + return -tmp + Math.log(2.5066282746310005 * ser / x); + } + + private void gcf(double a, double x) { + int i; + double an, b, c, d, del, h; + + gln = gammln(a); + b = x + 1.0 - a; + c = 1.0 / FPMIN; + d = 1.0 / b; + h = d; + for (i = 1; i <= ITMAX; i++) { + an = -i * (i - a); + b += 2.0; + d = an * d + b; + if (Math.abs(d) < FPMIN) d = FPMIN; + c = b + an / c; + if (Math.abs(c) < FPMIN) c = FPMIN; + d = 1.0 / d; + del = d * c; + h *= del; + if (Math.abs(del - 1.0) < EPS) break; + } + if (i > ITMAX) throw new RuntimeException("a too large, ITMAX too small in gcf"); + gammcf = Math.exp(-x + a * Math.log(x) - gln) * h; + } + + private void gser(double a, double x) { + int n; + double sum, del, ap; + + gln = gammln(a); + if (x <= 0.0) { + if (x < 0.0) throw new RuntimeException("x less than 0 in routine gser"); + gamser = 0.0; + return; + } else { + ap = a; + del = sum = 1.0 / a; + for (n = 1; n <= ITMAX; n++) { + ++ap; + del *= x / ap; + sum += del; + if (Math.abs(del) < Math.abs(sum) * EPS) { + gamser = sum * Math.exp(-x + a * Math.log(x) - gln); + return; + } + } + throw new RuntimeException("a too large, ITMAX too small in routine gser"); + } + } + + public double exe(double a, double x) { + if (x < 0.0 || a <= 0.0) throw new RuntimeException("Invalid arguments in routine gammq"); + if (x < (a + 1.0)) { + gser(a, x); + return 1 - gamser; + } else { + gcf(a, x); + return gammcf; + } + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Remainder.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Remainder.java new file mode 100644 index 0000000000..3c7468f108 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/Remainder.java @@ -0,0 +1,44 @@ +package com.foo.asyncapi.ncs.imp; + +/* + Ported unchanged, apart from the package, from the NCS case study of EMB (WebFuzzing/EMB): + jdk_8_maven/cs/rest/artificial/ncs, org.restncs.imp. The routines there follow Numerical + Recipes. + */ +public class Remainder { + + public static int exe(int a, int b) { + int r = 0 - 1; + int cy = 0; + int ny = 0; + if (a == 0) ; + else if (b == 0) ; + else if (a > 0) + if (b > 0) + while ((a - ny) >= b) { + ny = ny + b; + r = a - ny; + cy = cy + 1; + } + else // b<0 + while ((a + ny) >= ((b >= 0) ? b : -b)) { + ny = ny + b; + r = a + ny; + cy = cy - 1; + } + else // a<0 + if (b > 0) + while (((a + ny) >= 0 ? (a + ny) : -(a + ny)) >= b) { + ny = ny + b; + r = a + ny; + cy = cy - 1; + } + else + while (b >= (a - ny)) { + ny = ny + b; + r = ((a - ny) >= 0 ? (a - ny) : -(a - ny)); + cy = cy + 1; + } + return r; + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/TriangleClassification.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/TriangleClassification.java new file mode 100644 index 0000000000..4aee2d468d --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/imp/TriangleClassification.java @@ -0,0 +1,29 @@ +package com.foo.asyncapi.ncs.imp; + +/* + Ported unchanged, apart from the package, from the NCS case study of EMB (WebFuzzing/EMB): + jdk_8_maven/cs/rest/artificial/ncs, org.restncs.imp. The routines there follow Numerical + Recipes. + */ +public class TriangleClassification { + + public static int classify(int a, int b, int c) { + if (a <= 0 || b <= 0 || c <= 0) { + return 0; + } + if (a == b && b == c) { + return 3; + } + int max = Math.max(a, Math.max(b, c)); + if ((max == a && max - b - c >= 0) || + (max == b && max - a - c >= 0) || + (max == c && max - a - b >= 0)) { + return 0; + } + if (a == b || b == c || a == c) { + return 2; + } else { + return 1; + } + } +} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml new file mode 100644 index 0000000000..9ef0139005 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml @@ -0,0 +1,288 @@ +# The NCS numerical service, re-expressed as an event-driven API over Kafka. +# Written for EvoMaster; not taken from a third party. +asyncapi: 3.0.0 +info: + title: NCS over Kafka + version: 1.0.0 + description: | + Request/reply model of the **NCS** (Numerical Case Study) SUT from the EvoMaster + Dataset (`jdk_8_maven/cs/rest/artificial/ncs`), re-expressed as an event-driven API + over **Kafka**. + + NCS is a stateless numerical service: each operation takes a few numbers and returns a + computed result. The six original REST endpoints map one-to-one to request/reply + operations here — a client publishes a request on the operation's request topic, and the + service (the SUT) replies on the operation's reply topic. Error replies mirror the REST + 4xx responses (e.g. an out-of-range argument). + + Correlation between a request and its reply travels in a **Kafka message header** + (`correlationId`). +defaultContentType: application/json + +servers: + kafka: + host: localhost:9092 + protocol: kafka + description: Kafka broker carrying the NCS request and reply topics. + +channels: + triangleRequest: + address: ncs.triangle.request + messages: + triangleRequest: + $ref: '#/components/messages/triangleRequest' + triangleReply: + address: ncs.triangle.reply + messages: + result: + $ref: '#/components/messages/intResult' + + bessjRequest: + address: ncs.bessj.request + messages: + bessjRequest: + $ref: '#/components/messages/bessjRequest' + bessjReply: + address: ncs.bessj.reply + messages: + result: + $ref: '#/components/messages/doubleResult' + error: + $ref: '#/components/messages/error' + + expintRequest: + address: ncs.expint.request + messages: + expintRequest: + $ref: '#/components/messages/expintRequest' + expintReply: + address: ncs.expint.reply + messages: + result: + $ref: '#/components/messages/doubleResult' + error: + $ref: '#/components/messages/error' + + fisherRequest: + address: ncs.fisher.request + messages: + fisherRequest: + $ref: '#/components/messages/fisherRequest' + fisherReply: + address: ncs.fisher.reply + messages: + result: + $ref: '#/components/messages/doubleResult' + error: + $ref: '#/components/messages/error' + + gammqRequest: + address: ncs.gammq.request + messages: + gammqRequest: + $ref: '#/components/messages/gammqRequest' + gammqReply: + address: ncs.gammq.reply + messages: + result: + $ref: '#/components/messages/doubleResult' + error: + $ref: '#/components/messages/error' + + remainderRequest: + address: ncs.remainder.request + messages: + remainderRequest: + $ref: '#/components/messages/remainderRequest' + remainderReply: + address: ncs.remainder.reply + messages: + result: + $ref: '#/components/messages/intResult' + error: + $ref: '#/components/messages/error' + +operations: + checkTriangle: + action: receive + summary: Classify a triangle from three integer edges (REST GET /api/triangle/{a}/{b}/{c}). + channel: + $ref: '#/channels/triangleRequest' + reply: + channel: + $ref: '#/channels/triangleReply' + bessj: + action: receive + summary: Bessel function J_n(x) (REST GET /api/bessj/{n}/{x}). + channel: + $ref: '#/channels/bessjRequest' + reply: + channel: + $ref: '#/channels/bessjReply' + expint: + action: receive + summary: Exponential integral E_n(x) (REST GET /api/expint/{n}/{x}). + channel: + $ref: '#/channels/expintRequest' + reply: + channel: + $ref: '#/channels/expintReply' + fisher: + action: receive + summary: Fisher F-distribution value (REST GET /api/fisher/{m}/{n}/{x}). + channel: + $ref: '#/channels/fisherRequest' + reply: + channel: + $ref: '#/channels/fisherReply' + gammq: + action: receive + summary: Incomplete gamma function Q(a, x) (REST GET /api/gammq/{a}/{x}). + channel: + $ref: '#/channels/gammqRequest' + reply: + channel: + $ref: '#/channels/gammqReply' + remainder: + action: receive + summary: Integer remainder of a / b (REST GET /api/remainder/{a}/{b}). + channel: + $ref: '#/channels/remainderRequest' + reply: + channel: + $ref: '#/channels/remainderReply' + +components: + messages: + triangleRequest: + name: TriangleRequest + title: Triangle classification request + contentType: application/json + correlationId: + location: '$message.header#/correlationId' + payload: + $ref: '#/components/schemas/TriangleRequest' + bessjRequest: + name: BessjRequest + contentType: application/json + correlationId: + location: '$message.header#/correlationId' + payload: + $ref: '#/components/schemas/BessjRequest' + expintRequest: + name: ExpintRequest + contentType: application/json + correlationId: + location: '$message.header#/correlationId' + payload: + $ref: '#/components/schemas/ExpintRequest' + fisherRequest: + name: FisherRequest + contentType: application/json + correlationId: + location: '$message.header#/correlationId' + payload: + $ref: '#/components/schemas/FisherRequest' + gammqRequest: + name: GammqRequest + contentType: application/json + correlationId: + location: '$message.header#/correlationId' + payload: + $ref: '#/components/schemas/GammqRequest' + remainderRequest: + name: RemainderRequest + contentType: application/json + correlationId: + location: '$message.header#/correlationId' + payload: + $ref: '#/components/schemas/RemainderRequest' + intResult: + name: IntResult + title: Integer result reply (Dto.resultAsInt) + contentType: application/json + correlationId: + location: '$message.header#/correlationId' + payload: + $ref: '#/components/schemas/IntResult' + doubleResult: + name: DoubleResult + title: Floating-point result reply (Dto.resultAsDouble) + contentType: application/json + correlationId: + location: '$message.header#/correlationId' + payload: + $ref: '#/components/schemas/DoubleResult' + error: + name: Error + title: Error reply (mirrors a REST 4xx response) + contentType: application/json + correlationId: + location: '$message.header#/correlationId' + payload: + $ref: '#/components/schemas/Error' + + schemas: + TriangleRequest: + type: object + required: [a, b, c] + properties: + a: { type: integer, format: int32, description: First edge } + b: { type: integer, format: int32, description: Second edge } + c: { type: integer, format: int32, description: Third edge } + BessjRequest: + type: object + required: [n, x] + properties: + n: + type: integer + format: int32 + minimum: 3 + maximum: 1000 + description: Order; the service replies with an error outside 3..1000. + x: { type: number, format: double } + ExpintRequest: + type: object + required: [n, x] + properties: + n: { type: integer, format: int32, minimum: 0 } + x: { type: number, format: double, description: x >= 0; the service errors otherwise. } + FisherRequest: + type: object + required: [m, n, x] + properties: + m: { type: integer, format: int32, minimum: 1, maximum: 1000 } + n: { type: integer, format: int32, minimum: 1, maximum: 1000 } + x: { type: number, format: double } + GammqRequest: + type: object + required: [a, x] + properties: + a: { type: number, format: double, description: a > 0; the service errors otherwise. } + x: { type: number, format: double, description: x >= 0. } + RemainderRequest: + type: object + required: [a, b] + properties: + a: { type: integer, format: int32, minimum: -10000, maximum: 10000 } + b: { type: integer, format: int32, minimum: -10000, maximum: 10000 } + IntResult: + type: object + required: [resultAsInt] + properties: + resultAsInt: { type: integer, format: int32 } + DoubleResult: + type: object + required: [resultAsDouble] + properties: + resultAsDouble: { type: number, format: double } + Error: + type: object + required: [error] + properties: + error: + type: object + required: [code, message] + properties: + code: { type: integer, description: 'Mirrors the REST status (e.g. 400).' } + message: { type: string } diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java new file mode 100644 index 0000000000..7fb039c0fc --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java @@ -0,0 +1,452 @@ +package com.foo.asyncapi.ncs; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +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.header.Header; +import org.apache.kafka.common.serialization.StringDeserializer; +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.utils.SimpleLogger; +import org.evomaster.client.java.sql.DbSpecification; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.context.ConfigurableApplicationContext; +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.Arrays; +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.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.stream.Collectors; + +/** + * The driver for NCS over Kafka, and the reference implementation of + * {@link #executeAsyncApiAction}: it owns the broker, publishes what the core asks it to, waits + * for the reply that answers it, and reports what came back without judging it. + */ +public class NcsKafkaController extends EmbeddedSutController { + + private static final String KAFKA_IMAGE = "apache/kafka:3.8.0"; + + private static final String DOCUMENT = "/asyncapi/ncs-kafka.yaml"; + + private static final String BOOTSTRAP_PROPERTY = "ncs.kafka.bootstrap"; + + /** + * The header a correlation id travels in when the document names no location. Kafka has + * no native correlation, so the driver has to pick one. + */ + private static final String DEFAULT_CORRELATION_HEADER = "correlationId"; + + private static final long DEFAULT_REPLY_TIMEOUT_MS = 5_000; + + private static final long PUBLISH_TIMEOUT_SECONDS = 10; + + private static final long POLL_MS = 100; + + /** + * How long to wait, at start-up, for the service to answer a probe request. + */ + private static final long READY_TIMEOUT_MS = 60_000; + + /** + * How long to wait for one probe to come back before publishing another. + */ + private static final long READY_POLL_MS = 2_000; + + private static final Duration CLOSE_TIMEOUT = Duration.ofSeconds(5); + + /** + * A triangle request that is valid, so the service answers with a result rather than an error. + */ + private static final String PROBE_REQUEST = "{\"a\":3,\"b\":4,\"c\":5}"; + + /** + * Every topic the document names, created up front so that no first request or reply has + * to wait for auto-creation. + */ + private static final List TOPICS = Arrays.asList( + "ncs.triangle.request", "ncs.triangle.reply", + "ncs.bessj.request", "ncs.bessj.reply", + "ncs.expint.request", "ncs.expint.reply", + "ncs.fisher.request", "ncs.fisher.reply", + "ncs.gammq.request", "ncs.gammq.reply", + "ncs.remainder.request", "ncs.remainder.reply"); + + private final KafkaContainer kafka = new KafkaContainer(KAFKA_IMAGE); + + private final ObjectMapper mapper = new ObjectMapper(); + + /** + * Reply address -> the consumer positioned on it, created the first time a reply is + * awaited there and kept for the rest of the run. + */ + private final Map> replyConsumers = new LinkedHashMap<>(); + + private ConfigurableApplicationContext ctx; + + private KafkaProducer producer; + + public NcsKafkaController() { + super.setControllerPort(0); + } + + @Override + public String startSut() { + + kafka.start(); + String bootstrap = kafka.getBootstrapServers(); + + createTopics(bootstrap); + + ctx = new SpringApplicationBuilder(NcsKafkaApplication.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); + + awaitServiceConsuming(); + + //this service has no URL: what there is to know is where its broker listens + return bootstrap; + } + + /** + * Return only once the service is answering, rather than as soon as it has been started. + * + * Joining a consumer group takes a moment, and a request published before that happens is + * answered late, after whoever sent it has stopped waiting. For a search, that would look + * like a service that does not reply, and the first dozen messages would each cost the + * whole reply timeout. + */ + private void awaitServiceConsuming() { + + long deadline = System.currentTimeMillis() + READY_TIMEOUT_MS; + + try (KafkaConsumer replies = consumerAt(NcsService.TRIANGLE_REPLY)) { + + while (System.currentTimeMillis() < deadline) { + + String probe = "ready-" + System.nanoTime(); + ProducerRecord record = + new ProducerRecord<>(NcsService.TRIANGLE_REQUEST, probe, PROBE_REQUEST); + record.headers().add(DEFAULT_CORRELATION_HEADER, bytes(probe)); + producer.send(record); + + long roundTrip = System.currentTimeMillis() + READY_POLL_MS; + while (System.currentTimeMillis() < roundTrip) { + for (ConsumerRecord reply : replies.poll(Duration.ofMillis(POLL_MS))) { + Header correlation = reply.headers().lastHeader(DEFAULT_CORRELATION_HEADER); + if (correlation != null && probe.equals(new String(correlation.value(), StandardCharsets.UTF_8))) { + return; + } + } + } + } + } + + throw new IllegalStateException( + "The NCS service did not answer a probe within " + READY_TIMEOUT_MS + " ms"); + } + + private void createTopics(String bootstrap) { + + Properties props = new Properties(); + props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap); + + try (AdminClient admin = AdminClient.create(props)) { + List topics = TOPICS.stream() + .map(name -> new NewTopic(name, 1, (short) 1)) + .collect(Collectors.toList()); + 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 NCS topics", e); + } + } catch (InterruptedException | TimeoutException e) { + throw new RuntimeException("Could not create the NCS topics", e); + } + } + + @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, which the E2E asserts. + */ + replyConsumers.values().forEach(c -> quietly(() -> c.close(CLOSE_TIMEOUT))); + replyConsumers.clear(); + + if (producer != null) { + quietly(() -> producer.close(CLOSE_TIMEOUT)); + producer = 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.ncs."; + } + + @Override + public void resetStateOfSUT() { + //stateless: every request is answered from its own content + } + + @Override + public List getInfoForAuthentication() { + return null; + } + + @Override + public List getDbSpecifications() { + return 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); + } + } + + @Override + public void executeAsyncApiAction(AsyncApiActionDto dto, AsyncApiReplyDto reply) { + + //positioned before publishing, so that the reply cannot slip past + KafkaConsumer replies = dto.replyAddress == null ? null : consumerOn(dto.replyAddress); + + ProducerRecord record = new ProducerRecord<>(dto.address, dto.correlationId, stamped(dto)); + dto.headers.forEach((name, value) -> record.headers().add(name, bytes(value))); + if (AsyncApiActionDto.CORRELATION_IN_HEADER.equals(dto.correlationLocation) || dto.correlationLocation == null) { + record.headers().add(correlationHeaderName(dto), bytes(dto.correlationId)); + } + + 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; + + if (replies == null) { + reply.replyExpected = false; + return; + } + + reply.replyExpected = true; + awaitReply(replies, dto, reply); + } + + private void awaitReply(KafkaConsumer replies, AsyncApiActionDto dto, AsyncApiReplyDto reply) { + + long timeout = dto.replyTimeoutMs != null ? dto.replyTimeoutMs : DEFAULT_REPLY_TIMEOUT_MS; + long start = System.currentTimeMillis(); + long deadline = start + timeout; + + while (true) { + long remaining = deadline - System.currentTimeMillis(); + if (remaining <= 0) { + break; + } + + ConsumerRecords polled = replies.poll(Duration.ofMillis(Math.min(POLL_MS, remaining))); + + for (ConsumerRecord candidate : polled) { + String id = correlationOf(candidate, dto); + if (id != null && !id.equals(dto.correlationId)) { + //an answer to some other request + continue; + } + reply.replyReceived = true; + reply.replyPayload = candidate.value(); + reply.replyHeaders = headersOf(candidate); + reply.correlationMatched = id != null; + reply.waitedMs = System.currentTimeMillis() - start; + return; + } + } + + reply.waitedMs = timeout; + } + + /** + * The payload with the correlation id written into it, when that is where the document + * says it goes; otherwise the payload as it is. + */ + private String stamped(AsyncApiActionDto dto) { + + if (!AsyncApiActionDto.CORRELATION_IN_PAYLOAD.equals(dto.correlationLocation) + || dto.payload == null || dto.correlationPointer == null) { + return dto.payload; + } + + try { + JsonNode root = mapper.readTree(dto.payload); + if (!root.isObject()) { + return dto.payload; + } + List segments = segmentsOf(dto.correlationPointer); + ObjectNode holder = (ObjectNode) root; + for (int i = 0; i < segments.size() - 1; i++) { + JsonNode next = holder.get(segments.get(i)); + holder = next != null && next.isObject() ? (ObjectNode) next : holder.putObject(segments.get(i)); + } + holder.put(segments.get(segments.size() - 1), dto.correlationId); + return root.toString(); + } catch (IOException e) { + return dto.payload; + } + } + + /** + * The correlation id a reply carries, read from wherever the request's was written, or null + * when it carries none. + */ + private String correlationOf(ConsumerRecord record, AsyncApiActionDto dto) { + + if (AsyncApiActionDto.CORRELATION_IN_PAYLOAD.equals(dto.correlationLocation)) { + try { + JsonNode node = record.value() == null ? null : mapper.readTree(record.value()); + for (String segment : segmentsOf(dto.correlationPointer)) { + node = node == null ? null : node.get(segment); + } + return node == null || !node.isValueNode() ? null : node.asText(); + } catch (IOException e) { + return null; + } + } + + Header header = record.headers().lastHeader(correlationHeaderName(dto)); + return header == null ? null : new String(header.value(), StandardCharsets.UTF_8); + } + + private static String correlationHeaderName(AsyncApiActionDto dto) { + List segments = segmentsOf(dto.correlationPointer); + return segments.isEmpty() ? DEFAULT_CORRELATION_HEADER : segments.get(segments.size() - 1); + } + + private static List segmentsOf(String pointer) { + if (pointer == null) { + return Arrays.asList(); + } + return Arrays.stream(pointer.split("/")) + .filter(s -> !s.isEmpty()) + .map(s -> s.replace("~1", "/").replace("~0", "~")) + .collect(Collectors.toList()); + } + + private KafkaConsumer consumerOn(String address) { + + return replyConsumers.computeIfAbsent(address, this::consumerAt); + } + + /** + * A consumer positioned at the end of [address], so that it sees only what is published + * from now on. + */ + private KafkaConsumer consumerAt(String address) { + + Properties props = new Properties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers()); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + + KafkaConsumer consumer = new KafkaConsumer<>(props); + + //assigned rather than subscribed: no group, no rebalance to wait for + List partitions = consumer.partitionsFor(address).stream() + .map(p -> new TopicPartition(address, p.partition())) + .collect(Collectors.toList()); + consumer.assign(partitions); + consumer.seekToEnd(partitions); + //seekToEnd is lazy; asking for the position makes it take effect now + partitions.forEach(consumer::position); + + return consumer; + } + + private static Map headersOf(ConsumerRecord record) { + Map headers = new LinkedHashMap<>(); + for (Header header : record.headers()) { + headers.put(header.key(), header.value() == null ? null : new String(header.value(), StandardCharsets.UTF_8)); + } + return headers; + } + + 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/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java new file mode 100644 index 0000000000..d1b5f071b8 --- /dev/null +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java @@ -0,0 +1,88 @@ +package org.evomaster.e2etests.spring.asyncapi.kafka; + +import com.foo.asyncapi.ncs.NcsKafkaController; +import org.evomaster.core.problem.asyncapi.data.AsyncApiIndividual; +import org.evomaster.core.problem.asyncapi.data.AsyncApiOutcome; +import org.evomaster.core.search.Solution; +import org.evomaster.e2etests.utils.AsyncApiTestBase; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A search over NCS driven through Kafka: every operation answers, and where the contract + * declares a result and an error, the search reaches both. + */ +public class NcsKafkaEMTest extends AsyncApiTestBase { + + private static final List OPERATIONS = + Arrays.asList("checkTriangle", "bessj", "expint", "fisher", "gammq", "remainder"); + + @BeforeAll + public static void initClass() throws Exception { + AsyncApiTestBase.initClass(new NcsKafkaController()); + } + + @Test + public void testRunEM() throws Throwable { + + //no test writer for AsyncAPI yet, so the search alone + runTestHandlingFlaky( + "NcsKafkaEM", + "org.foo.asyncapi.NcsKafkaEM", + 300, + false, + (args) -> { + + /* + The broker is local and the service answers in milliseconds, so a reply + that has not arrived in a second is not coming. Left at its default, one + stalled message would eat a large share of the budget for this test. + */ + args.add("--asyncApiReplyTimeoutMs"); + args.add("1000"); + + /* + The kill switch stops SUT code that is still running once an individual + has been evaluated. It assumes a request handled on its own thread: here + the service consumes on one long-lived thread, so killing it stops the + service for good, and every later message goes unanswered. Any + message-driven SUT has this shape. + */ + args.add("--killSwitch"); + args.add("false"); + + Solution solution = initAndRun(args); + + assertTrue(solution.getIndividuals().size() >= 1); + + for (String operation : OPERATIONS) { + assertReplied(solution, operation); + } + + assertReplyReached(solution, "checkTriangle", "intResult"); + assertReplyReached(solution, "bessj", "doubleResult"); + assertReplyReached(solution, "remainder", "intResult"); + + /* + Of the rejections NCS makes, only these two lie within what the schema + allows: expint rejects a negative x, gammq a non-positive a or a negative + x. bessj's order and remainder's operands are bounded by the schema, so + their error replies cannot be reached without publishing invalid data. + */ + assertReplyReached(solution, "expint", "doubleResult"); + assertReplyReached(solution, "expint", "error"); + assertReplyReached(solution, "gammq", "doubleResult"); + assertReplyReached(solution, "gammq", "error"); + + assertEquals(0, countOutcome(solution, AsyncApiOutcome.NO_REPLY), "a promised reply never came"); + assertEquals(0, countOutcome(solution, AsyncApiOutcome.PUBLISH_FAILED), "a message never left"); + }, + 5); + } +} diff --git a/pom.xml b/pom.xml index 270907ece2..31a15ee3b6 100644 --- a/pom.xml +++ b/pom.xml @@ -266,6 +266,7 @@ 1.33 9.9 1.21.4 + 3.7.2 1.17.2 5.13.2 @@ -921,6 +922,19 @@ ${testcontainers.version} test + + + org.testcontainers + kafka + ${testcontainers.version} + test + + + + org.apache.kafka + kafka-clients + ${kafka-clients.version} + org.testcontainers selenium From 06e2b39bad6c774c6e216acfaf9dec3ce758cf5b Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Tue, 29 Sep 2026 23:19:21 -0300 Subject: [PATCH 2/6] AsyncAPI 3.x: the no-reply oracle, end to end Neither side of the experimental oracle gate was exercised against a live broker. A second search now runs with the oracles on and a reply deadline no round trip can meet, so every message goes unanswered and the fault is reported, while the existing search asserts that a service which answers correctly has nothing reported against it. Nothing in the service is broken on purpose for this. The driver ignores replies whose correlation id does not match the request it is waiting on, so the late answers piling up on the reply topics cannot be taken for fresh ones. --- .../e2etests/utils/AsyncApiTestBase.java | 15 ++++++ .../spring/asyncapi/kafka/NcsKafkaEMTest.java | 48 +++++++++++++++++++ 2 files changed, 63 insertions(+) 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 cf45e00398..00ae6c3435 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,9 +1,11 @@ package org.evomaster.e2etests.utils; +import com.webfuzzing.commons.faults.FaultCategory; 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 java.util.List; @@ -44,6 +46,19 @@ protected Set repliesOf(Solution solution, String op .collect(Collectors.toSet()); } + /** + * The fault categories reported across the whole solution. Read off the action results, + * which is where the reports count faults from, rather than off the covered targets. + */ + protected Set faultsOf(Solution solution) { + return solution.getIndividuals().stream() + .flatMap(ind -> ind.evaluatedMainActions().stream()) + .map(e -> (AsyncApiCallResult) e.getResult()) + .flatMap(r -> r.getFaults().stream()) + .map(DetectedFault::getCategory) + .collect(Collectors.toSet()); + } + protected long countOutcome(Solution solution, AsyncApiOutcome outcome) { return solution.getIndividuals().stream() .flatMap(ind -> ind.evaluatedMainActions().stream()) diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java index d1b5f071b8..773d76b379 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java @@ -3,6 +3,7 @@ import com.foo.asyncapi.ncs.NcsKafkaController; import org.evomaster.core.problem.asyncapi.data.AsyncApiIndividual; import org.evomaster.core.problem.asyncapi.data.AsyncApiOutcome; +import org.evomaster.core.problem.enterprise.ExperimentalFaultCategory; import org.evomaster.core.search.Solution; import org.evomaster.e2etests.utils.AsyncApiTestBase; import org.junit.jupiter.api.BeforeAll; @@ -82,7 +83,54 @@ public void testRunEM() throws Throwable { assertEquals(0, countOutcome(solution, AsyncApiOutcome.NO_REPLY), "a promised reply never came"); assertEquals(0, countOutcome(solution, AsyncApiOutcome.PUBLISH_FAILED), "a message never left"); + + //the oracles are off here, so a well-behaved service must report nothing + assertTrue(faultsOf(solution).isEmpty(), "faults were reported: " + faultsOf(solution)); }, 5); } + + /** + * The no-reply oracle, end to end. Given a deadline no round trip through a broker can meet, + * every message goes unanswered, and that is reported as a fault once experimental oracles + * are asked for. + */ + @Test + public void testAPromisedReplyThatNeverArrivesIsAFault() throws Throwable { + + runTestHandlingFlaky( + "NcsKafkaNoReplyEM", + "org.foo.asyncapi.NcsKafkaNoReplyEM", + 30, + false, + (args) -> { + + /* + Shorter than any round trip through a broker. The service answers as it + always does, just never in time, which is how an unanswered message tends + to look in practice. Nothing has to be broken on purpose for it. + */ + args.add("--asyncApiReplyTimeoutMs"); + args.add("1"); + + //as in testRunEM: the kill switch would stop the consumer thread for good + args.add("--killSwitch"); + args.add("false"); + + /* + Both AsyncAPI categories are experimental, so without this the outcome is + still reached and still a target, but nothing is reported as a fault. + */ + args.add("--useExperimentalOracles"); + args.add("true"); + + Solution solution = initAndRun(args); + + assertTrue(countOutcome(solution, AsyncApiOutcome.NO_REPLY) > 0, + "every message was answered in time, so the oracle had nothing to find"); + assertTrue(faultsOf(solution).contains(ExperimentalFaultCategory.ASYNCAPI_NO_REPLY), + "the unanswered messages were not reported as a fault: " + faultsOf(solution)); + }, + 3); + } } From 3349b3deb9325f6e826afb87357ee9dd198cd619 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Tue, 29 Sep 2026 23:19:52 -0300 Subject: [PATCH 3/6] AsyncAPI 3.x: the E2E writes the suite, compiles it and runs it The search was covered by a run against a real broker; what the run wrote was not. Both output languages are now generated, compiled and executed against the same broker: Kotlin in the existing run, Java in a second one, because a Java half that never compiles is invisible to a Kotlin-only check. The service gains one deliberate defect, marked as such where it is written: a non-positive triangle edge is answered with a payload the document declares nowhere for that reply. It correlates, it arrives, and a client written against the contract still cannot read it, so it is the one fault only the reply classifier finds. The suite is expected to report it. The two replies whose errors no conforming client can reach say so in the document, beside the reply itself, so the next reader does not take them for a gap in the search. The driver compared the correlation location against the two string constants that have since become an enum, and the shared compile step assumed Kotlin sources. --- .../e2etests/utils/AsyncApiTestBase.java | 11 ++++ .../e2etests/utils/EnterpriseTestBase.java | 32 +++++++-- .../java/com/foo/asyncapi/ncs/NcsService.java | 27 +++++++- .../main/resources/asyncapi/ncs-kafka.yaml | 8 +++ .../foo/asyncapi/ncs/NcsKafkaController.java | 15 ++++- .../spring/asyncapi/kafka/NcsKafkaEMTest.java | 65 +++++++++++++++++-- 6 files changed, 142 insertions(+), 16 deletions(-) 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 00ae6c3435..5c488e0b9a 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 @@ -72,6 +72,17 @@ protected void assertReplied(Solution solution, String opera assertTrue(ok, "With seed " + defaultSeed + ": no reply to '" + operation + "' was ever received"); } + /** + * That some reply to [operation] arrived and matched none of the messages the contract + * declares for it, which is the one fault only a reply classifier can find. + */ + protected void assertRepliedWithUndeclaredMessage(Solution solution, String operation) { + boolean found = resultsOf(solution, operation).stream() + .anyMatch(r -> r.getOutcome() == AsyncApiOutcome.REPLIED && r.getReplyMessage() == null); + assertTrue(found, "With seed " + defaultSeed + ": every reply to '" + operation + + "' matched a declared message, so the undeclared-reply path was never reached"); + } + protected void assertReplyReached(Solution solution, String operation, String messageId) { Set replies = repliesOf(solution, operation); assertTrue(replies.contains(messageId), diff --git a/core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/EnterpriseTestBase.java b/core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/EnterpriseTestBase.java index 50a48b47f1..d5f5ee5194 100644 --- a/core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/EnterpriseTestBase.java +++ b/core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/EnterpriseTestBase.java @@ -430,15 +430,37 @@ protected Class loadClass(ClassName className){ } } + private static boolean containsSourceWithExtension(File folder, String extension){ + File[] children = folder.listFiles(); + if (children == null) { + return false; + } + for (File f : children) { + if (f.isDirectory() && containsSourceWithExtension(f, extension)) { + return true; + } + if (f.isFile() && f.getName().endsWith(extension)) { + return true; + } + } + return false; + } + protected void compile(String outputFolderName){ long start = System.currentTimeMillis(); - CompilerForTestGenerated.INSTANCE.compile( - OutputFormat.KOTLIN_JUNIT_5, - new File(outputFolderPath(outputFolderName)), - new File("target/test-classes") - ); + File folder = new File(outputFolderPath(outputFolderName)); + + /* + Which language was generated is not passed down here, but the folder says: a suite is + written in one of them, so whichever is on disk is what has to be compiled. + */ + OutputFormat format = containsSourceWithExtension(folder, ".java") + ? OutputFormat.JAVA_JUNIT_5 + : OutputFormat.KOTLIN_JUNIT_5; + + CompilerForTestGenerated.INSTANCE.compile(format, folder, new File("target/test-classes")); int passed = (int)(System.currentTimeMillis() - start) / 1000; System.out.println("Folder compiled in " + passed + " seconds: " + outputFolderName); diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsService.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsService.java index 40622de4d6..a4a78b897f 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsService.java +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsService.java @@ -101,8 +101,24 @@ private String compute(String requestTopic, JsonNode request) { switch (requestTopic) { - case TRIANGLE_REQUEST: - return intResult(TriangleClassification.classify(integer(request, "a"), integer(request, "b"), integer(request, "c"))); + case TRIANGLE_REQUEST: { + int a = integer(request, "a"); + int b = integer(request, "b"); + int c = integer(request, "c"); + /* + DELIBERATE DEFECT, not a faithful port of the NCS routine: the original + classifies a non-positive edge as 0 and answers with the declared result. + + This answers it with a shape the document never declares for this reply, so + that the E2E has a service which commits the one fault only a reply + classifier can find: the message arrives, it correlates, and a client written + against the contract still cannot read it. Nothing else in the run notices. + */ + if (a <= 0 || b <= 0 || c <= 0) { + return undeclared("not a triangle"); + } + return intResult(TriangleClassification.classify(a, b, c)); + } case BESSJ_REQUEST: { int n = integer(request, "n"); @@ -197,6 +213,13 @@ private String doubleResult(double value) { return mapper.createObjectNode().put(RESULT_AS_DOUBLE, value).toString(); } + /** + * A payload matching none of the messages the document declares for the reply. + */ + private String undeclared(String reason) { + return mapper.createObjectNode().put("classification", reason).toString(); + } + private String error(String message) { ObjectNode error = mapper.createObjectNode(); error.put(CODE, BAD_REQUEST); diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml index 9ef0139005..f9ae4518f1 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml @@ -43,6 +43,10 @@ channels: bessjRequest: $ref: '#/components/messages/bessjRequest' bessjReply: + # The error reply is declared because the service can produce one, but no client + # obeying this document can make it: the request's bounds are exactly the range the + # service checks. Kept so the contract describes the service, not only what a + # conforming client can reach. address: ncs.bessj.reply messages: result: @@ -95,6 +99,10 @@ channels: remainderRequest: $ref: '#/components/messages/remainderRequest' remainderReply: + # The error reply is declared because the service can produce one, but no client + # obeying this document can make it: the request's bounds are exactly the range the + # service checks. Kept so the contract describes the service, not only what a + # conforming client can reach. address: ncs.remainder.reply messages: result: diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java index 7fb039c0fc..6b1eff7b26 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java @@ -257,6 +257,15 @@ 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. + */ + @Override + public String getAsyncApiServerAddress(String serverName) { + return kafka == null ? null : kafka.getBootstrapServers(); + } + @Override public ProblemInfo getProblemInfo() { return AsyncApiProblem.fromSchemaText(readDocument()); @@ -288,7 +297,7 @@ public void executeAsyncApiAction(AsyncApiActionDto dto, AsyncApiReplyDto reply) ProducerRecord record = new ProducerRecord<>(dto.address, dto.correlationId, stamped(dto)); dto.headers.forEach((name, value) -> record.headers().add(name, bytes(value))); - if (AsyncApiActionDto.CORRELATION_IN_HEADER.equals(dto.correlationLocation) || dto.correlationLocation == null) { + if (dto.correlationLocation == AsyncApiActionDto.CorrelationLocation.HEADER || dto.correlationLocation == null) { record.headers().add(correlationHeaderName(dto), bytes(dto.correlationId)); } @@ -346,7 +355,7 @@ private void awaitReply(KafkaConsumer replies, AsyncApiActionDto */ private String stamped(AsyncApiActionDto dto) { - if (!AsyncApiActionDto.CORRELATION_IN_PAYLOAD.equals(dto.correlationLocation) + if (dto.correlationLocation != AsyncApiActionDto.CorrelationLocation.PAYLOAD || dto.payload == null || dto.correlationPointer == null) { return dto.payload; } @@ -375,7 +384,7 @@ private String stamped(AsyncApiActionDto dto) { */ private String correlationOf(ConsumerRecord record, AsyncApiActionDto dto) { - if (AsyncApiActionDto.CORRELATION_IN_PAYLOAD.equals(dto.correlationLocation)) { + if (dto.correlationLocation == AsyncApiActionDto.CorrelationLocation.PAYLOAD) { try { JsonNode node = record.value() == null ? null : mapper.readTree(record.value()); for (String segment : segmentsOf(dto.correlationPointer)) { diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java index 773d76b379..8489c7e179 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java @@ -32,11 +32,17 @@ public static void initClass() throws Exception { @Test public void testRunEM() throws Throwable { - //no test writer for AsyncAPI yet, so the search alone - runTestHandlingFlaky( + /* + The generated suite is compiled and run, so this covers the writer as well as the + search: the tests publish with a Kafka client of their own, written from the + contract, and never go through the driver. + */ + runTestHandlingFlakyAndCompilation( "NcsKafkaEM", "org.foo.asyncapi.NcsKafkaEM", + null, 300, + true, false, (args) -> { @@ -71,15 +77,30 @@ public void testRunEM() throws Throwable { assertReplyReached(solution, "remainder", "intResult"); /* - Of the rejections NCS makes, only these two lie within what the schema - allows: expint rejects a negative x, gammq a non-positive a or a negative - x. bessj's order and remainder's operands are bounded by the schema, so - their error replies cannot be reached without publishing invalid data. + An error reply is reachable exactly where the schema leaves the input the + service rejects within what it allows. These three describe their + constraint in prose rather than as a bound -- expint's x, gammq's a, and + fisher's x, which its degrees of freedom do not cover -- so the search is + free to produce it. + + bessj and remainder declare an error reply that nothing obeying the + contract can provoke: their bounds are exactly the range the service + checks. That is a property of the document, not a gap in the search. */ assertReplyReached(solution, "expint", "doubleResult"); assertReplyReached(solution, "expint", "error"); assertReplyReached(solution, "gammq", "doubleResult"); assertReplyReached(solution, "gammq", "error"); + assertReplyReached(solution, "fisher", "doubleResult"); + assertReplyReached(solution, "fisher", "error"); + + /* + checkTriangle answers an edge that is not positive with a shape the + contract does not declare. Only the classifier can tell: the message + arrived, was correlated, and still means nothing to a client written + against the document. + */ + assertRepliedWithUndeclaredMessage(solution, "checkTriangle"); assertEquals(0, countOutcome(solution, AsyncApiOutcome.NO_REPLY), "a promised reply never came"); assertEquals(0, countOutcome(solution, AsyncApiOutcome.PUBLISH_FAILED), "a message never left"); @@ -90,6 +111,38 @@ public void testRunEM() throws Throwable { 5); } + @Test + public void testRunEMInJava() throws Throwable { + + /* + The lines that publish are written in the language the suite is generated in, and + only one of the two is exercised by the test above. Compiling and running the other + is what catches a Java half that was never compiled: the Kotlin one did not compile + first time either. + + A short budget: what is under test here is the emitted Java, not the search. + */ + runTestHandlingFlakyAndCompilation( + "NcsKafkaJavaEM", + "org.foo.asyncapi.NcsKafkaJavaEM", + null, + 40, + true, + false, + (args) -> { + + setOption(args, "outputFormat", "JAVA_JUNIT_5"); + setOption(args, "asyncApiReplyTimeoutMs", "1000"); + setOption(args, "killSwitch", "false"); + + Solution solution = initAndRun(args); + + assertTrue(solution.getIndividuals().size() >= 1); + assertTrue(countOutcome(solution, AsyncApiOutcome.REPLIED) > 0, "nothing was answered"); + }, + 5); + } + /** * The no-reply oracle, end to end. Given a deadline no round trip through a broker can meet, * every message goes unanswered, and that is reported as a fault once experimental oracles From 96cdc0032498c3bb397fec203f0102ae34ce2e11 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Tue, 29 Sep 2026 23:20:07 -0300 Subject: [PATCH 4/6] AsyncAPI 3.x: a profile for the E2E that needs a broker The module starts a broker in a container, so it wants Docker and it is slower than the rest. It gets a profile of its own and a flag every other profile turns off, which is how the database E2Es are kept out of the runs that have no business paying for them, and CI gains the job that turns it on. --- .github/workflows/ci.yml | 2 +- .../spring/spring-asyncapi-kafka/pom.xml | 4 +++ pom.xml | 31 +++++++++++++++++++ 3 files changed, 36 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 20c823867f..751e03afe7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -460,7 +460,7 @@ jobs: strategy: fail-fast: false matrix: - profile: [ BaseTests, CoreTests, CoreITTests, OAV3Tests, OAV2Tests, GraphQLTests, MongoTests, H2V1Tests, MiscDB ] + profile: [ BaseTests, CoreTests, CoreITTests, OAV3Tests, OAV2Tests, GraphQLTests, MongoTests, H2V1Tests, MiscDB, AsyncApiTests ] steps: - uses: actions/checkout@v6 - name: Setup JDK ${{env.build-jdk}} diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml index ee98ad4989..52275e009a 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml @@ -17,6 +17,10 @@ evomaster-e2e-tests-spring-asyncapi-kafka jar + + ${skipAsyncApi} + + true true + true @@ -191,6 +198,7 @@ true false true + true @@ -206,6 +214,28 @@ true true false + true + + + + + AsyncApiTests + + true + + true + true + true + true + true + true + true + true + false @@ -315,6 +345,7 @@ false false false + false From 112b4606098371265facb70b3e13853e79106b36 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Fri, 2 Oct 2026 18:30:51 -0300 Subject: [PATCH 5/6] AsyncAPI 3.x: correct what the fixture claims, and what it leaves behind bessj's error reply is reachable after all. Its n is fenced exactly, which is what the comment saw, but its x carries no bound at all, and the routine returns a value that is not a number for a very small one, which the service answers with an error. Checked by running the routine: n=3 with x=1e-100 is one of many. remainder's identical claim does hold -- integer arithmetic, no finiteness check -- so only bessj's is corrected, here and in the E2E that repeats it. fisher was described as stating its constraint in prose; it states none. The driver kept its reply consumers for the whole class, while every test restarts the search under the same seed and so mints the same correlation ids again. A reply left on a topic by one test could then answer a request of the next. They are dropped when the state is reset, and the next one needed is created at the end of its topic. A broker that failed to start was never stopped: isSutRunning answers on the Spring context, which is assigned last, so nothing downstream would have called stopSut. And the address it answers with was guarded on a field that is final and never null, so asking before the container was up threw where the contract says to return null. A reply the service could not publish was dropped in silence, which the run then reports as a reply that never came. Said out loud instead. The triangle reply channel now declares the error message the other five do. The service answers a rejected request the same way everywhere, so leaving it out made an ordinary error a second undeclared payload on that topic, indistinguishable from the deliberate one. A second on a loaded CI machine is not a generous reply deadline when the test asserts that nothing went unanswered, so it is three. The module stops reusing forks, as every other container-backed E2E does, and drops two dependencies nothing in it uses. The ported NCS routines are listed on the re-used code page, which is where the policy says they belong. --- .../e2etests/utils/AsyncApiTestBase.java | 6 +-- .../spring/spring-asyncapi-kafka/pom.xml | 21 ++++++---- .../foo/asyncapi/ncs/NcsKafkaApplication.java | 3 +- .../foo/asyncapi/ncs/NcsRequestConsumer.java | 14 ++++++- .../main/resources/asyncapi/ncs-kafka.yaml | 14 +++++-- .../foo/asyncapi/ncs/NcsKafkaController.java | 40 ++++++++++++++++--- .../spring/asyncapi/kafka/NcsKafkaEMTest.java | 35 +++++++++------- docs/reused_code.md | 5 +++ pom.xml | 6 ++- 9 files changed, 107 insertions(+), 37 deletions(-) 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 5c488e0b9a..3e9713ca96 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 @@ -26,7 +26,7 @@ protected Solution initAndRun(List args) { } /** - * The result of every message published to [operation], across the whole solution. + * The result of every message published to {@code operation}, across the whole solution. */ protected List resultsOf(Solution solution, String operation) { return solution.getIndividuals().stream() @@ -37,7 +37,7 @@ protected List resultsOf(Solution soluti } /** - * The declared messages the replies to [operation] were recognised as. + * The declared messages the replies to {@code operation} were recognised as. */ protected Set repliesOf(Solution solution, String operation) { return resultsOf(solution, operation).stream() @@ -73,7 +73,7 @@ protected void assertReplied(Solution solution, String opera } /** - * That some reply to [operation] arrived and matched none of the messages the contract + * That some reply to {@code operation} arrived and matched none of the messages the contract * declares for it, which is the one fault only a reply classifier can find. */ protected void assertRepliedWithUndeclaredMessage(Solution solution, String operation) { diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml index 52275e009a..c85046d57f 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml @@ -30,10 +30,6 @@ validation-api 2.0.1.Final - - javax.ws.rs - javax.ws.rs-api - org.evomaster evomaster-e2e-tests-utils @@ -69,10 +65,6 @@ org.testcontainers kafka - - org.hamcrest - hamcrest-all - org.junit.jupiter junit-jupiter-api @@ -84,4 +76,17 @@ test + + + + org.apache.maven.plugins + maven-surefire-plugin + + + false + + + + + diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsKafkaApplication.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsKafkaApplication.java index a58bb7aa39..36d2749db3 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsKafkaApplication.java +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsKafkaApplication.java @@ -10,7 +10,8 @@ * NCS over Kafka: the six numerical operations of the NCS case study, each consuming a request * topic and answering on a reply topic, as described by {@code asyncapi/ncs-kafka.yaml}. * - * The service speaks only Kafka. Its one component is {@link NcsRequestConsumer}. + * The service speaks only Kafka. What listens on it is {@link NcsRequestConsumer}, which hands + * every request to {@link NcsService}. */ /* Bean validation is excluded: this service validates nothing, and the validation API that the diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsRequestConsumer.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsRequestConsumer.java index e53d248d3d..039581b107 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsRequestConsumer.java +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/java/com/foo/asyncapi/ncs/NcsRequestConsumer.java @@ -11,6 +11,8 @@ import org.apache.kafka.common.header.Header; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.SmartLifecycle; import org.springframework.stereotype.Component; @@ -25,6 +27,8 @@ @Component public class NcsRequestConsumer implements SmartLifecycle { + private static final Logger log = LoggerFactory.getLogger(NcsRequestConsumer.class); + /** * The header a request carries its correlation id in, as the document declares. */ @@ -102,7 +106,15 @@ private void answer(ConsumerRecord request) { record.headers().add(CORRELATION_HEADER, correlation.value()); } - producer.send(record); + /* + A send that fails is otherwise indistinguishable from a slow one: the run reports a + reply that never came, and nothing says why. + */ + producer.send(record, (metadata, failure) -> { + if (failure != null) { + log.warn("Could not publish a reply to {}", reply.topic, failure); + } + }); } @Override diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml index f9ae4518f1..9a0df4472b 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/main/resources/asyncapi/ncs-kafka.yaml @@ -32,10 +32,16 @@ channels: triangleRequest: $ref: '#/components/messages/triangleRequest' triangleReply: + # The error reply is declared here as on every other channel, because the service answers a + # rejected request the same way everywhere. Keeping it out would make the ordinary error a + # second undeclared payload on this topic, indistinguishable from the deliberate one the + # service publishes for a non-positive edge. address: ncs.triangle.reply messages: result: $ref: '#/components/messages/intResult' + error: + $ref: '#/components/messages/error' bessjRequest: address: ncs.bessj.request @@ -43,10 +49,10 @@ channels: bessjRequest: $ref: '#/components/messages/bessjRequest' bessjReply: - # The error reply is declared because the service can produce one, but no client - # obeying this document can make it: the request's bounds are exactly the range the - # service checks. Kept so the contract describes the service, not only what a - # conforming client can reach. + # 'n' is fenced exactly: the 3..1000 the request allows is the range the service checks, + # so no conforming client can be rejected for it. 'x' is not fenced, and the routine + # returns a value that is not a number for a very small one, which the service answers + # with an error. So this reply is reachable, if rarely. address: ncs.bessj.reply messages: result: diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java index 6b1eff7b26..556b4a5004 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/com/foo/asyncapi/ncs/NcsKafkaController.java @@ -123,6 +123,21 @@ public NcsKafkaController() { @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 end, 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(); @@ -210,8 +225,7 @@ 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, which the E2E asserts. */ - replyConsumers.values().forEach(c -> quietly(() -> c.close(CLOSE_TIMEOUT))); - replyConsumers.clear(); + closeReplyConsumers(); if (producer != null) { quietly(() -> producer.close(CLOSE_TIMEOUT)); @@ -244,7 +258,20 @@ public String getPackagePrefixesToCover() { @Override public void resetStateOfSUT() { - //stateless: every request is answered from its own content + + /* + The service itself is stateless: every request is answered from its own content. + What does carry over is where the reply consumers are positioned, and that matters: + a search restarted under the same seed mints the same correlation ids again, so a + reply left on a topic by the previous test would answer a request of this one. They + are dropped here, and the next one to be needed is created at the end of its topic. + */ + closeReplyConsumers(); + } + + private void closeReplyConsumers() { + replyConsumers.values().forEach(c -> quietly(() -> c.close(CLOSE_TIMEOUT))); + replyConsumers.clear(); } @Override @@ -260,10 +287,13 @@ public List getDbSpecifications() { /** * 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 == null ? null : kafka.getBootstrapServers(); + return kafka.isRunning() ? kafka.getBootstrapServers() : null; } @Override @@ -421,7 +451,7 @@ private KafkaConsumer consumerOn(String address) { } /** - * A consumer positioned at the end of [address], so that it sees only what is published + * A consumer positioned at the end of {@code address}, so that it sees only what is published * from now on. */ private KafkaConsumer consumerAt(String address) { diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java index 8489c7e179..06d54e63d7 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/src/test/java/org/evomaster/e2etests/spring/asyncapi/kafka/NcsKafkaEMTest.java @@ -47,12 +47,14 @@ public void testRunEM() throws Throwable { (args) -> { /* - The broker is local and the service answers in milliseconds, so a reply - that has not arrived in a second is not coming. Left at its default, one - stalled message would eat a large share of the budget for this test. + The broker is local and the service answers in milliseconds. Left at its + default, one stalled message would eat a large share of the budget for + this test; a second is generous for the round trip, but not for a loaded + CI machine, where a garbage collection in the SUT can cost more than + that, and this test asserts that no reply was missed. */ args.add("--asyncApiReplyTimeoutMs"); - args.add("1000"); + args.add("3000"); /* The kill switch stops SUT code that is still running once an individual @@ -78,14 +80,15 @@ public void testRunEM() throws Throwable { /* An error reply is reachable exactly where the schema leaves the input the - service rejects within what it allows. These three describe their - constraint in prose rather than as a bound -- expint's x, gammq's a, and - fisher's x, which its degrees of freedom do not cover -- so the search is - free to produce it. - - bessj and remainder declare an error reply that nothing obeying the - contract can provoke: their bounds are exactly the range the service - checks. That is a property of the document, not a gap in the search. + service rejects within what it allows. expint's x and gammq's a state + their constraint in prose rather than as a bound, and fisher's x carries + none at all, so the search is free to produce one. + + remainder declares an error reply that nothing obeying the contract can + provoke: its bounds are exactly the range the service checks. That is a + property of the document, not a gap in the search. bessj's n is fenced + the same way, but its x is not, so an error there is reachable too -- + rarely enough that this test does not ask for it. */ assertReplyReached(solution, "expint", "doubleResult"); assertReplyReached(solution, "expint", "error"); @@ -105,7 +108,11 @@ public void testRunEM() throws Throwable { assertEquals(0, countOutcome(solution, AsyncApiOutcome.NO_REPLY), "a promised reply never came"); assertEquals(0, countOutcome(solution, AsyncApiOutcome.PUBLISH_FAILED), "a message never left"); - //the oracles are off here, so a well-behaved service must report nothing + /* + Both AsyncAPI categories are experimental and the flag is off here, so + even the undeclared reply asserted above goes unreported. The other run + turns the flag on and checks the other side of that. + */ assertTrue(faultsOf(solution).isEmpty(), "faults were reported: " + faultsOf(solution)); }, 5); @@ -132,7 +139,7 @@ public void testRunEMInJava() throws Throwable { (args) -> { setOption(args, "outputFormat", "JAVA_JUNIT_5"); - setOption(args, "asyncApiReplyTimeoutMs", "1000"); + setOption(args, "asyncApiReplyTimeoutMs", "3000"); setOption(args, "killSwitch", "false"); Solution solution = initAndRun(args); diff --git a/docs/reused_code.md b/docs/reused_code.md index ea915bcaf8..a79cf91dc8 100644 --- a/docs/reused_code.md +++ b/docs/reused_code.md @@ -33,3 +33,8 @@ are listed here: * _asyncapi/sut/scalar.yaml_: AsyncAPI description from [Scalar](https://github.com/scalar/scalar), used unmodified as a test resource of `core`. Released under MIT license. + +* _com/foo/asyncapi/ncs/imp/*.java_: the numerical routines of the NCS case study of + [EMB](https://github.com/WebFuzzing/EMB) (`jdk_8_maven/cs/rest/artificial/ncs`, + `org.restncs.imp`), ported unchanged apart from the package into the artificial SUT of the + AsyncAPI E2E tests. The routines there follow Numerical Recipes. diff --git a/pom.xml b/pom.xml index 1f663c5776..c5f2bf7f27 100644 --- a/pom.xml +++ b/pom.xml @@ -961,7 +961,11 @@ test - + org.apache.kafka kafka-clients ${kafka-clients.version} From ac3b5fe8787f57c88083334dfcc0a5d0c07c3bb2 Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Mon, 5 Oct 2026 19:31:21 -0300 Subject: [PATCH 6/6] AsyncAPI 3.x: put back the dependency the generated suite needs hamcrest was dropped as unused, which it is -- by this module's own sources. The suites EvoMaster generates import org.hamcrest.Matchers, and they are compiled against this module's classpath, so without it the E2E fails at "unresolved reference 'Matchers'" the moment it tries to compile what the run wrote. Said in the pom, so the next reader does not drop it again. --- core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml index c85046d57f..c38724704c 100644 --- a/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml +++ b/core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml @@ -65,6 +65,14 @@ org.testcontainers kafka + + + org.hamcrest + hamcrest-all + org.junit.jupiter junit-jupiter-api