add support for incoming calls to OcppJsonChargePoint

thus far, test cases supported only request/response interactions
initiated by the _station_. any request/call initiated by steve was not
taken into account by station. with this change, station test cases
can express requests that are expected (initiated by steve) and
corresponding responses to send back to steve.

see OcppJsonChargePoint#expectRequest

Author: goekay

src/test/java/de/rwth/idsg/steve/utils/OcppJsonChargePoint.java

@@ -19,9 +19,14 @@
 package de.rwth.idsg.steve.utils;
 
 import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.TreeNode;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.NullNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import de.rwth.idsg.ocpp.jaxb.RequestType;
 import de.rwth.idsg.ocpp.jaxb.ResponseType;
 import de.rwth.idsg.steve.SteveException;
@@ -36,6 +41,9 @@
 import de.rwth.idsg.steve.ocpp.ws.data.OcppJsonResponse;
 import de.rwth.idsg.steve.ocpp.ws.data.OcppJsonResult;
 import de.rwth.idsg.steve.ocpp.ws.pipeline.Serializer;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.Value;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.StatusCode;
@@ -50,8 +58,10 @@
 import java.io.IOException;
 import java.net.URI;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
@@ -69,6 +79,7 @@ public class OcppJsonChargePoint {
     private final String chargeBoxId;
     private final String connectionPath;
     private final Map<String, ResponseContext> responseContextMap;
+    private final Map<String, RequestContext> requestContextMap;
     private final MessageDeserializer deserializer;
     private final WebSocketClient client;
     private final CountDownLatch closeHappenedSignal;
@@ -88,6 +99,7 @@ public OcppJsonChargePoint(String ocppVersion, String chargeBoxId, String pathPr
         this.chargeBoxId = chargeBoxId;
         this.connectionPath = pathPrefix + chargeBoxId;
         this.responseContextMap = new LinkedHashMap<>(); // because we want to keep the insertion order of test cases
+        this.requestContextMap = new HashMap<>();
         this.deserializer = new MessageDeserializer();
         this.client = new WebSocketClient();
         this.closeHappenedSignal = new CountDownLatch(1);
@@ -121,6 +133,8 @@ public void onMessage(Session session, String msg) {
             } else if (ocppMsg instanceof OcppJsonError) {
                 ResponseContext ctx = responseContextMap.remove(ocppMsg.getMessageId());
                 ctx.errorHandler.accept((OcppJsonError) ocppMsg);
+            } else if (ocppMsg instanceof OcppJsonCallForTesting) {
+                handleCall((OcppJsonCallForTesting) ocppMsg);
             }
         } catch (Exception e) {
             log.error("Exception", e);
@@ -171,9 +185,25 @@ public <T extends ResponseType> void prepare(RequestType payload, String action,
         responseContextMap.put(messageId, resCtx);
     }
 
+    public void expectRequest(RequestType expectedRequest, ResponseType plannedResponse) {
+        String requestPayload;
+        JsonNode responsePayload;
+        try {
+            ObjectMapper mapper = JsonObjectMapper.INSTANCE.getMapper();
+            requestPayload = mapper.writeValueAsString(expectedRequest);
+            responsePayload = mapper.valueToTree(plannedResponse);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+
+        String action = getOperationName(expectedRequest);
+        requestContextMap.put(action, new RequestContext(requestPayload, responsePayload));
+    }
+
     public void process() {
+        int requestCount = requestContextMap.values().size();
         int responseCount = responseContextMap.values().size();
-        receivedMessagesSignal = new CountDownLatch(responseCount);
+        receivedMessagesSignal = new CountDownLatch(requestCount + responseCount);
 
         // copy the values in a new list to be iterated over, because otherwise we get a ConcurrentModificationException,
         // since the onMessage(..) uses the same responseContextMap to remove an item while looping over its items here.
@@ -231,6 +261,27 @@ private static String getOperationName(RequestType requestType) {
         return s;
     }
 
+    private void handleCall(OcppJsonCallForTesting call) {
+        try {
+            ArrayNode node = JsonObjectMapper.INSTANCE.getMapper()
+                .createArrayNode()
+                .add(MessageType.CALL_RESULT.getTypeNr())
+                .add(call.getMessageId())
+                .add(call.getContext().getResponsePayload());
+
+            String str = JsonObjectMapper.INSTANCE.getMapper().writeValueAsString(node);
+            session.getRemote().sendString(str);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Value
+    private static class RequestContext {
+        String requestPayload;
+        JsonNode responsePayload;
+    }
+
     private static class ResponseContext {
         private final String outgoingMessage;
         private final Class<ResponseType> responseClass;
@@ -269,6 +320,8 @@ private OcppJsonMessage extract(String msg) throws Exception {
                         return handleResult(messageId, parser);
                     case CALL_ERROR:
                         return handleError(messageId, parser);
+                    case CALL:
+                        return handleCall(messageId, parser);
                     default:
                         throw new SteveException("Unknown enum type");
                 }
@@ -311,6 +364,54 @@ private OcppJsonResponse handleError(String messageId, JsonParser parser) throws
             error.setErrorDetails(details);
             return error;
         }
+
+        private OcppJsonCall handleCall(String messageId, JsonParser parser) {
+            // parse action
+            String action;
+            try {
+                parser.nextToken();
+                action = parser.getText();
+            } catch (IOException e) {
+                throw new RuntimeException();
+            }
+
+            // parse request payload
+            String req;
+            try {
+                parser.nextToken();
+                JsonNode requestPayload = parser.readValueAsTree();
+
+                // https://github.com/steve-community/steve/issues/1109
+                if (requestPayload instanceof NullNode) {
+                    requestPayload = new ObjectNode(JsonNodeFactory.instance);
+                }
+
+                req = requestPayload.toString();
+            } catch (IOException e) {
+                log.error("Exception occurred", e);
+                throw new RuntimeException();
+            }
+
+            RequestContext context = requestContextMap.get(action);
+            if (context == null) {
+                testerThreadInterruptReason = new RuntimeException("Unexpected message arrived: " + req);
+                testerThread.interrupt();
+            } else if (Objects.equals(context.requestPayload, req)) {
+                requestContextMap.remove(action);
+            }
+
+            OcppJsonCallForTesting call = new OcppJsonCallForTesting();
+            call.setAction(action);
+            call.setMessageId(messageId);
+            call.setContext(context);
+            return call;
+        }
+    }
+
+    @Setter
+    @Getter
+    private static class OcppJsonCallForTesting extends OcppJsonCall {
+        private RequestContext context;
     }
 
 }