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..00ae6c3435 --- /dev/null +++ b/core-tests/e2e-tests/e2e-tests-utils/src/test/java/org/evomaster/e2etests/utils/AsyncApiTestBase.java @@ -0,0 +1,80 @@ +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; +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()); + } + + /** + * 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()) + .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..773d76b379 --- /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,136 @@ +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.problem.enterprise.ExperimentalFaultCategory; +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"); + + //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); + } +} 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