eclipse-sailing-analytics/sailing-analytics
1
0
Fork 0
mirror of https://github.com/eclipse-sailing-analytics/sailing-analytics.git synced 2026-09-20 04:35:32 +00:00
Code Activity

added test for sending replication operation to master through HTTP POST

Browse Source 
Change-Id: Ie16a2abb308dc3ae860cf729374362d9b5d21516
This commit is contained in:
Axel Uhl
2019-01-22 10:50:19 +01:00
parent 78764ac04a
commit bb07391659
3 changed files with 132 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
java/com.sap.sse.replication.testsupport/META-INF/MANIFEST.MF
+7
View File
@@ -13,3 +13,10 @@ Require-Bundle: org.junit;bundle-version="4.8.2",
com.sap.sse.replication.interfaces;bundle-version="1.0.0"
Export-Package: com.sap.sse.replication.testsupport
Automatic-Module-Name: com.sap.sse.replication.testsupport
Import-Package: javax.servlet;version="3.1.0",
javax.servlet.http;version="3.1.0",
org.mockito;version="1.10.14",
org.mockito.invocation;version="1.10.14",
org.mockito.mock;version="1.10.14",
org.mockito.stubbing;version="1.10.14",
org.osgi.util.tracker;version="1.5.1"
java/com.sap.sse.replication.testsupport/src/com/sap/sse/replication/testsupport/AbstractServerReplicationTestSetUp.java
+107 -29
View File
@@ -2,9 +2,9 @@ package com.sap.sse.replication.testsupport;
import static org.junit.Assert.assertFalse;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStreamWriter;
@@ -24,8 +24,18 @@ import java.util.UUID;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.servlet.ReadListener;
import javax.servlet.ServletException;
import javax.servlet.ServletInputStream;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.junit.Rule;
import org.junit.rules.Timeout;
import org.mockito.Matchers;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
@@ -42,6 +52,7 @@ import com.sap.sse.replication.impl.ReplicationInstancesManager;
import com.sap.sse.replication.impl.ReplicationMasterDescriptorImpl;
import com.sap.sse.replication.impl.ReplicationReceiverImpl;
import com.sap.sse.replication.impl.ReplicationServiceImpl;
import com.sap.sse.replication.impl.ReplicationServlet;
import com.sap.sse.replication.impl.SingletonReplicablesProvider;
import net.jpountz.lz4.LZ4BlockOutputStream;
@@ -277,41 +288,97 @@ public abstract class AbstractServerReplicationTestSetUp<ReplicableInterface ext
boolean stop = false;
while (!stop) {
Socket s = ss.accept();
String request = new BufferedReader(new InputStreamReader(s.getInputStream())).readLine();
final InputStream inputStream = s.getInputStream();
String request = readLine(inputStream);
logger.info("received request "+request);
PrintWriter pw = new PrintWriter(new OutputStreamWriter(s.getOutputStream()));
pw.println("HTTP/1.1 200 OK");
pw.println("Content-Type: text/plain");
pw.println();
pw.flush();
if (request.contains("DEREGISTER")) {
// assuming that it is safe to unregister all replicas for tests
for (ReplicaDescriptor descriptor : getReplicaInfo()) {
unregisterReplica(descriptor);
if (request.startsWith("POST /replication/replication")) {
final ReplicationServlet servlet = new ReplicationServlet(new SingletonReplicablesProvider(master), /* replicationServiceTracker */ null);
final HttpServletRequest requestMock = Mockito.mock(HttpServletRequest.class);
Mockito.when(requestMock.getInputStream()).thenReturn(new ServletInputStream() {
@Override
public boolean isFinished() {
try {
return inputStream.available() <= 0;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public boolean isReady() {
try {
return inputStream.available() > 0;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public void setReadListener(ReadListener readListener) {
}
@Override
public int read() throws IOException {
return inputStream.read();
}
});
final HttpServletResponse responseMock = Mockito.mock(HttpServletResponse.class);
final boolean[] error = { false };
Mockito.doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
pw.println("HTTP/1.1 500 Bad Request");
pw.println("Content-Type: text/plain");
pw.println();
pw.println(invocation.getArgumentAt(1, String.class));
pw.flush();
error[0] = true;
return null;
}
}).when(responseMock).sendError(Matchers.anyInt(), Matchers.isA(String.class));
while (!readLine(inputStream).isEmpty());
servlet.doPost(requestMock, responseMock);
if (!error[0]) {
pw.println("HTTP/1.1 200 OK");
pw.println("Content-Type: text/plain");
pw.println();
pw.flush();
}
} else {
pw.println("HTTP/1.1 200 OK");
pw.println("Content-Type: text/plain");
pw.println();
pw.flush();
if (request.contains("DEREGISTER")) {
// assuming that it is safe to unregister all replicas for tests
for (ReplicaDescriptor descriptor : getReplicaInfo()) {
unregisterReplica(descriptor);
}
} else if (request.contains("REGISTER")) {
final String uuid = UUID.randomUUID().toString();
registerReplicaUuidForMaster(uuid, masterDescriptor);
masterReplicationService.registerReplica(replicaDescriptor);
pw.print(uuid);
} else if (request.contains("INITIAL_LOAD")) {
Channel channel = masterReplicationService.createMasterChannel();
RabbitOutputStream ros = new RabbitOutputStream(INITIAL_LOAD_PACKAGE_SIZE, channel,
/* queueName */ "initial-load-for-TestClient-"+UUID.randomUUID(), /* syncAfterTimeout */ false);
pw.println(ros.getQueueName());
final LZ4BlockOutputStream compressingOutputStream = new LZ4BlockOutputStream(ros);
master.serializeForInitialReplication(compressingOutputStream);
compressingOutputStream.finish();
ros.close();
} else if (request.contains("STOP")) {
stop = true;
logger.info("received STOP request");
}
} else if (request.contains("REGISTER")) {
final String uuid = UUID.randomUUID().toString();
registerReplicaUuidForMaster(uuid, masterDescriptor);
masterReplicationService.registerReplica(replicaDescriptor);
pw.print(uuid);
} else if (request.contains("INITIAL_LOAD")) {
Channel channel = masterReplicationService.createMasterChannel();
RabbitOutputStream ros = new RabbitOutputStream(INITIAL_LOAD_PACKAGE_SIZE, channel,
/* queueName */ "initial-load-for-TestClient-"+UUID.randomUUID(), /* syncAfterTimeout */ false);
pw.println(ros.getQueueName());
final LZ4BlockOutputStream compressingOutputStream = new LZ4BlockOutputStream(ros);
master.serializeForInitialReplication(compressingOutputStream);
compressingOutputStream.finish();
ros.close();
} else if (request.contains("STOP")) {
stop = true;
logger.info("received STOP request");
}
pw.close();
s.close();
logger.info("Request handled successfully.");
}
} catch (IOException e) {
} catch (IOException | ServletException e) {
throw new RuntimeException(e);
} finally {
try {
@@ -323,6 +390,17 @@ public abstract class AbstractServerReplicationTestSetUp<ReplicableInterface ext
}
}
}
private String readLine(final InputStream inputStream) throws IOException {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
int b;
while ((b=inputStream.read()) != -1 && b != '\n') {
if (b != '\r') { // ignore CR
bos.write(b);
}
}
return new String(bos.toByteArray());
}
};
initialLoadTestServerThread.start();
synchronized (listening) {