AsyncAPI 3.x: an E2E test, NCS driven through Kafka

.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

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<AsyncApiIndividual> initAndRun(List<String> args) {
+        return (Solution<AsyncApiIndividual>) Main.initAndRun(args.toArray(new String[0]));
+    }
+
+    /**
+     * The result of every message published to [operation], across the whole solution.
+     */
+    protected List<AsyncApiCallResult> resultsOf(Solution<AsyncApiIndividual> 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<String> repliesOf(Solution<AsyncApiIndividual> 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<FaultCategory> faultsOf(Solution<AsyncApiIndividual> 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<AsyncApiIndividual> 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<AsyncApiIndividual> 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<AsyncApiIndividual> solution, String operation, String messageId) {
+        Set<String> replies = repliesOf(solution, operation);
+        assertTrue(replies.contains(messageId),
+                "With seed " + defaultSeed + ": '" + operation + "' never replied with '" + messageId + "', only with " + replies);
+    }
+}

core-tests/e2e-tests/spring/pom.xml

@@ -34,6 +34,7 @@
         <module>spring-rpc-grpc</module>
         <module>spring-rpc-thrift</module>
         <module>spring-mcp-bb</module>
+        <module>spring-asyncapi-kafka</module>
     </modules>
 
     <dependencyManagement>
@@ -44,6 +45,11 @@
                 <artifactId>spring-boot</artifactId>
                 <version>${springboot.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-starter</artifactId>
+                <version>${springboot.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-starter-web</artifactId>

core-tests/e2e-tests/spring/spring-asyncapi-kafka/pom.xml

@@ -0,0 +1,83 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>evomaster-e2e-tests-spring</artifactId>
+        <groupId>org.evomaster</groupId>
+        <version>6.2.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <!--
+        NCS, the numerical case study, re-expressed as a service driven by Kafka messages:
+        a request topic and a reply topic per operation, described by an AsyncAPI document.
+        The SUT speaks only Kafka; there is no HTTP endpoint at all.
+    -->
+    <artifactId>evomaster-e2e-tests-spring-asyncapi-kafka</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <!--
+            Issue with Jersey (coming from client-java) using old version
+        -->
+        <dependency>
+            <groupId>javax.validation</groupId>
+            <artifactId>validation-api</artifactId>
+            <version>2.0.1.Final</version>
+        </dependency>
+        <dependency>
+            <groupId>javax.ws.rs</groupId>
+            <artifactId>javax.ws.rs-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.evomaster</groupId>
+            <artifactId>evomaster-e2e-tests-utils</artifactId>
+            <type>test-jar</type>
+        </dependency>
+        <dependency>
+            <groupId>org.evomaster</groupId>
+            <artifactId>evomaster-client-java-controller</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.evomaster</groupId>
+            <artifactId>evomaster-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.evomaster</groupId>
+            <artifactId>evomaster-client-java-instrumentation</artifactId>
+            <type>test-jar</type>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>kafka</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.hamcrest</groupId>
+            <artifactId>hamcrest-all</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

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);
+    }
+}

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<String, String> consumer;
+
+    private KafkaProducer<String, String> 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<String, String> records = consumer.poll(POLL);
+                for (ConsumerRecord<String, String> request : records) {
+                    answer(request);
+                }
+            }
+        } catch (WakeupException e) {
+            //asked to stop
+        } finally {
+            consumer.close();
+            producer.close();
+        }
+    }
+
+    private void answer(ConsumerRecord<String, String> request) {
+
+        NcsService.Reply reply = service.handle(request.topic(), request.value());
+
+        ProducerRecord<String, String> 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;
+    }
+}