From f4e257b821828942326ff02a44591a1b833e9ef9 Mon Sep 17 00:00:00 2001 From: Lautaro Petaccio Date: Mon, 14 Sep 2026 08:56:41 -0300 Subject: [PATCH 1/2] 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 cccecf55ff..049e42cb1d 100644 --- a/.gitignore +++ b/.gitignore @@ -170,6 +170,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 98683efea5..cdda55a395 100644 --- a/core-tests/e2e-tests/spring/pom.xml +++ b/core-tests/e2e-tests/spring/pom.xml @@ -34,6 +34,7 @@ spring-rpc-grpc spring-rpc-thrift spring-mcp-bb + spring-asyncapi-kafka @@ -44,6 +45,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 052f17d17d..79d4f1f633 100644 --- a/pom.xml +++ b/pom.xml @@ -243,6 +243,7 @@ 1.33 9.9 1.21.4 + 3.7.2 1.17.2 5.13.2 @@ -897,6 +898,19 @@ ${testcontainers.version} test + + + org.testcontainers + kafka + ${testcontainers.version} + test + + + + org.apache.kafka + kafka-clients + ${kafka-clients.version} + org.testcontainers selenium From df9519498a87ef2a46b7e4334971b8685fb3f27f Mon Sep 17 00:00:00 2001 From: LautaroPetaccio Date: Fri, 18 Sep 2026 18:03:01 -0300 Subject: [PATCH 2/2] 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); + } }