This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 72882e7734 add volume tracking to http actions, transforms and extra
tests, fixes #6819 (#6820)
72882e7734 is described below
commit 72882e7734a99117ed6be5ebc965b93be42710b4
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Wed Mar 18 20:23:29 2026 +0100
add volume tracking to http actions, transforms and extra tests, fixes
#6819 (#6820)
---
.../hop/workflow/actions/http/ActionHttp.java | 48 ++-
.../actions/http/ActionHttpExecutionTest.java | 298 +++++++++++++++++
.../apache/hop/pipeline/transforms/http/Http.java | 23 +-
.../transforms/http/HttpExecutionTest.java | 330 +++++++++++++++++++
.../hop/pipeline/transforms/http/HttpTest.java | 139 ++++++++
.../hop/pipeline/transforms/httppost/HttpPost.java | 72 ++++-
.../transforms/httppost/HttpPostExecutionTest.java | 356 +++++++++++++++++++++
.../pipeline/transforms/httppost/HttpPostTest.java | 175 +++++++++-
.../apache/hop/pipeline/transforms/rest/Rest.java | 62 ++++
.../hop/pipeline/transforms/rest/RestTest.java | 88 +++++
10 files changed, 1559 insertions(+), 32 deletions(-)
diff --git
a/plugins/actions/http/src/main/java/org/apache/hop/workflow/actions/http/ActionHttp.java
b/plugins/actions/http/src/main/java/org/apache/hop/workflow/actions/http/ActionHttp.java
index 9f55a72b94..51773ce24c 100644
---
a/plugins/actions/http/src/main/java/org/apache/hop/workflow/actions/http/ActionHttp.java
+++
b/plugins/actions/http/src/main/java/org/apache/hop/workflow/actions/http/ActionHttp.java
@@ -43,6 +43,8 @@ import org.apache.hop.core.RowMetaAndData;
import org.apache.hop.core.annotations.Action;
import org.apache.hop.core.encryption.Encr;
import org.apache.hop.core.exception.HopXmlException;
+import org.apache.hop.core.io.CountingInputStream;
+import org.apache.hop.core.io.CountingOutputStream;
import org.apache.hop.core.row.value.ValueMetaString;
import org.apache.hop.core.util.HttpClientManager;
import org.apache.hop.core.util.Utils;
@@ -251,8 +253,10 @@ public class ActionHttp extends ActionBase {
OutputStream outputFile = null;
OutputStream uploadStream = null;
- BufferedInputStream fileStream = null;
+ InputStream fileStream = null;
InputStream input = null;
+ long bytesReadThisRow = 0L;
+ long bytesWrittenThisRow = 0L;
try {
String urlToUse = resolve(row.getString(urlFieldnameToUse, ""));
@@ -301,7 +305,7 @@ public class ActionHttp extends ActionBase {
}
// Create the output File...
- outputFile = HopVfs.getOutputStream(realTargetFile, fileAppended);
+ outputFile = new
CountingOutputStream(HopVfs.getOutputStream(realTargetFile, fileAppended));
// Get a stream for the specified URL
server = new URL(urlToUse);
@@ -345,14 +349,23 @@ public class ActionHttp extends ActionBase {
}
// Grab an output stream to upload data to web server
- uploadStream = connection.getOutputStream();
- fileStream = new BufferedInputStream(new FileInputStream(new
File(realUploadFile)));
+ uploadStream = new
CountingOutputStream(connection.getOutputStream());
+ fileStream =
+ new CountingInputStream(
+ new BufferedInputStream(new FileInputStream(new
File(realUploadFile))));
try {
- int c;
- while ((c = fileStream.read()) >= 0) {
- uploadStream.write(c);
+ byte[] buffer = new byte[8192];
+ int bytesRead;
+ while ((bytesRead = fileStream.read(buffer)) >= 0) {
+ uploadStream.write(buffer, 0, bytesRead);
}
} finally {
+ if (fileStream instanceof CountingInputStream countingInputStream)
{
+ bytesReadThisRow += countingInputStream.getCount();
+ }
+ if (uploadStream instanceof CountingOutputStream
countingOutputStream) {
+ bytesWrittenThisRow += countingOutputStream.getCount();
+ }
// Close upload and file
if (uploadStream != null) {
uploadStream.close();
@@ -373,7 +386,7 @@ public class ActionHttp extends ActionBase {
}
// Read the result from the server...
- input = connection.getInputStream();
+ input = new CountingInputStream(connection.getInputStream());
Date date = new Date(connection.getLastModified());
if (isBasic()) {
logBasic(
@@ -381,17 +394,21 @@ public class ActionHttp extends ActionBase {
PKG, "ActionHTTP.Log.ReplayInfo",
connection.getContentType(), date));
}
- int oneChar;
- long bytesRead = 0L;
- while ((oneChar = input.read()) != -1) {
- outputFile.write(oneChar);
- bytesRead++;
+ byte[] buffer = new byte[8192];
+ int bytesRead;
+ while ((bytesRead = input.read(buffer)) != -1) {
+ outputFile.write(buffer, 0, bytesRead);
}
+ bytesReadThisRow += ((CountingInputStream) input).getCount();
+ bytesWrittenThisRow += ((CountingOutputStream) outputFile).getCount();
if (isBasic()) {
logBasic(
BaseMessages.getString(
- PKG, "ActionHTTP.Log.FinisedWritingReply", bytesRead,
realTargetFile));
+ PKG,
+ "ActionHTTP.Log.FinisedWritingReply",
+ ((CountingInputStream) input).getCount(),
+ realTargetFile));
}
if (addFilenameResult) {
@@ -449,6 +466,9 @@ public class ActionHttp extends ActionBase {
System.setProperty(CONST_HTTPS_PROXY_PORT,
Const.NVL(beforeHttpsProxyPort, ""));
System.setProperty(CONST_HTTP_NON_PROXY_HOSTS,
Const.NVL(beforeNonProxyHosts, ""));
}
+
+ result.setBytesReadThisAction(result.getBytesReadThisAction() +
bytesReadThisRow);
+ result.setBytesWrittenThisAction(result.getBytesWrittenThisAction() +
bytesWrittenThisRow);
}
return result;
diff --git
a/plugins/actions/http/src/test/java/org/apache/hop/workflow/actions/http/ActionHttpExecutionTest.java
b/plugins/actions/http/src/test/java/org/apache/hop/workflow/actions/http/ActionHttpExecutionTest.java
new file mode 100644
index 0000000000..7c64cf3101
--- /dev/null
+++
b/plugins/actions/http/src/test/java/org/apache/hop/workflow/actions/http/ActionHttpExecutionTest.java
@@ -0,0 +1,298 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.workflow.actions.http;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.sun.net.httpserver.Headers;
+import com.sun.net.httpserver.HttpServer;
+import com.sun.net.httpserver.HttpsConfigurator;
+import com.sun.net.httpserver.HttpsServer;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyStore;
+import java.security.SecureRandom;
+import java.util.List;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import org.apache.commons.io.FileUtils;
+import org.apache.hop.core.HopClientEnvironment;
+import org.apache.hop.core.Result;
+import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
+import org.apache.hop.workflow.engines.local.LocalWorkflowEngine;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+class ActionHttpExecutionTest {
+ private static final String HTTP_HOST = "localhost";
+ private static final String HTTPS_HOST = "localhost";
+ private static final String HTTPS_KEYSTORE_PASSWORD = "changeit";
+ private static final String HEADER_NAME = "X-Test-Header";
+ private static final String HEADER_VALUE = "header-value";
+
+ @RegisterExtension
+ static RestoreHopEngineEnvironmentExtension env = new
RestoreHopEngineEnvironmentExtension();
+
+ private static HttpServer httpServer;
+ private static HttpsServer httpsServer;
+ private static Path httpsKeyStorePath;
+
+ @BeforeAll
+ static void setupBeforeClass() throws Exception {
+ HopClientEnvironment.init();
+ ActionHttpExecutionTest.startHttpServer();
+ ActionHttpExecutionTest.startHttpsServer();
+ }
+
+ @AfterAll
+ static void tearDown() throws IOException {
+ ActionHttpExecutionTest.stopHttpServer();
+ ActionHttpExecutionTest.stopHttpsServer();
+ }
+
+ @Test
+ void testHttpConfiguredSingleRunTracksDownloadAndAddsResultFile() throws
Exception {
+ String payload = "download only payload";
+ File tempFileForDownload = File.createTempFile("downloadedFile3", ".tmp");
+ tempFileForDownload.deleteOnExit();
+
+ ActionHttp http = new ActionHttp();
+ http.setParentWorkflow(new LocalWorkflowEngine());
+ http.setRunForEveryRow(false);
+ http.setAddFilenameToResult(true);
+ http.setUrl(httpBaseUrl() + "/downloadFile");
+ http.setTargetFilename(tempFileForDownload.getCanonicalPath());
+
+ Result result = http.execute(new Result(), 0);
+
+ assertTrue(result.getResult());
+ assertEquals(0, result.getNrErrors());
+ assertFalse(result.getResultFiles().isEmpty());
+ assertEquals(payload, FileUtils.readFileToString(tempFileForDownload,
UTF_8));
+ assertEquals(payload.getBytes(UTF_8).length,
result.getBytesReadThisAction());
+ assertEquals(payload.getBytes(UTF_8).length,
result.getBytesWrittenThisAction());
+ }
+
+ @Test
+ void testHttpConfiguredSingleRunSetsHeaders() throws Exception {
+ String payload = "header ok";
+ File tempFileForDownload = File.createTempFile("downloadedHeader", ".tmp");
+ tempFileForDownload.deleteOnExit();
+
+ ActionHttp.Header header = new ActionHttp.Header();
+ header.setHeaderName(HEADER_NAME);
+ header.setHeaderValue(HEADER_VALUE);
+
+ ActionHttp http = new ActionHttp();
+ http.setParentWorkflow(new LocalWorkflowEngine());
+ http.setRunForEveryRow(false);
+ http.setAddFilenameToResult(false);
+ http.setHeaders(List.of(header));
+ http.setUrl(httpBaseUrl() + "/headerCheck");
+ http.setTargetFilename(tempFileForDownload.getCanonicalPath());
+
+ Result result = http.execute(new Result(), 0);
+
+ assertTrue(result.getResult());
+ assertEquals(0, result.getNrErrors());
+ assertEquals(payload, FileUtils.readFileToString(tempFileForDownload,
UTF_8));
+ }
+
+ @Test
+ void testHttpConfiguredSingleRunWithMalformedUrlSetsError() throws Exception
{
+ File tempFileForDownload = File.createTempFile("downloadedMalformed",
".tmp");
+ tempFileForDownload.deleteOnExit();
+
+ ActionHttp http = new ActionHttp();
+ http.setParentWorkflow(new LocalWorkflowEngine());
+ http.setRunForEveryRow(false);
+ http.setUrl("ht!tp://bad-url");
+ http.setTargetFilename(tempFileForDownload.getCanonicalPath());
+
+ Result result = http.execute(new Result(), 0);
+
+ assertFalse(result.getResult());
+ assertEquals(1, result.getNrErrors());
+ assertEquals(0, result.getBytesReadThisAction());
+ assertEquals(0, result.getBytesWrittenThisAction());
+ }
+
+ @Test
+ void testHttpConfiguredSingleRunWithIgnoreSslAcceptsSelfSignedCertificate()
throws Exception {
+ String payload = "https payload";
+ File tempFileForDownload = File.createTempFile("downloadedHttps", ".tmp");
+ tempFileForDownload.deleteOnExit();
+
+ ActionHttp http = new ActionHttp();
+ http.setParentWorkflow(new LocalWorkflowEngine());
+ http.setRunForEveryRow(false);
+ http.setAddFilenameToResult(false);
+ http.setIgnoreSsl(true);
+ http.setUrl(httpsBaseUrl() + "/downloadFile");
+ http.setTargetFilename(tempFileForDownload.getCanonicalPath());
+
+ Result result = http.execute(new Result(), 0);
+
+ assertTrue(result.getResult());
+ assertEquals(0, result.getNrErrors());
+ assertEquals(payload, FileUtils.readFileToString(tempFileForDownload,
UTF_8));
+ assertEquals(payload.getBytes(UTF_8).length,
result.getBytesReadThisAction());
+ assertEquals(payload.getBytes(UTF_8).length,
result.getBytesWrittenThisAction());
+ }
+
+ private static void startHttpServer() throws IOException {
+ httpServer = HttpServer.create(new
InetSocketAddress(ActionHttpExecutionTest.HTTP_HOST, 0), 10);
+ httpServer.createContext(
+ "/downloadFile",
+ httpExchange -> {
+ byte[] response = "download only payload".getBytes(UTF_8);
+ Headers h = httpExchange.getResponseHeaders();
+ h.add("Content-Type", "application/octet-stream");
+ httpExchange.sendResponseHeaders(200, response.length);
+ try (OutputStream os = httpExchange.getResponseBody()) {
+ os.write(response);
+ os.flush();
+ }
+ httpExchange.close();
+ });
+ httpServer.createContext(
+ "/headerCheck",
+ httpExchange -> {
+ String headerValue =
httpExchange.getRequestHeaders().getFirst(HEADER_NAME);
+ byte[] response =
+ (HEADER_VALUE.equals(headerValue) ? "header ok" : "missing
header").getBytes(UTF_8);
+ Headers h = httpExchange.getResponseHeaders();
+ h.add("Content-Type", "text/plain");
+ httpExchange.sendResponseHeaders(
+ HEADER_VALUE.equals(headerValue) ? 200 : 400, response.length);
+ try (OutputStream os = httpExchange.getResponseBody()) {
+ os.write(response);
+ os.flush();
+ }
+ httpExchange.close();
+ });
+ httpServer.start();
+ }
+
+ private static void startHttpsServer() throws Exception {
+ httpsKeyStorePath = Files.createTempFile("action-http-execution-test",
".p12");
+ Files.deleteIfExists(httpsKeyStorePath);
+ generateHttpsKeyStore(httpsKeyStorePath);
+
+ KeyStore keyStore = KeyStore.getInstance("PKCS12");
+ try (InputStream inputStream = Files.newInputStream(httpsKeyStorePath)) {
+ keyStore.load(inputStream, HTTPS_KEYSTORE_PASSWORD.toCharArray());
+ }
+
+ KeyManagerFactory keyManagerFactory =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ keyManagerFactory.init(keyStore, HTTPS_KEYSTORE_PASSWORD.toCharArray());
+
+ SSLContext sslContext = SSLContext.getInstance("TLS");
+ sslContext.init(keyManagerFactory.getKeyManagers(), null, new
SecureRandom());
+
+ httpsServer = HttpsServer.create(new InetSocketAddress(HTTPS_HOST, 0), 10);
+ httpsServer.setHttpsConfigurator(new HttpsConfigurator(sslContext));
+ httpsServer.createContext(
+ "/downloadFile",
+ httpExchange -> {
+ byte[] response = "https payload".getBytes(UTF_8);
+ Headers h = httpExchange.getResponseHeaders();
+ h.add("Content-Type", "application/octet-stream");
+ httpExchange.sendResponseHeaders(200, response.length);
+ try (OutputStream os = httpExchange.getResponseBody()) {
+ os.write(response);
+ os.flush();
+ }
+ httpExchange.close();
+ });
+ httpsServer.start();
+ }
+
+ private static void generateHttpsKeyStore(Path keyStorePath)
+ throws IOException, InterruptedException {
+ Process process =
+ new ProcessBuilder(
+ keytoolExecutable(),
+ "-genkeypair",
+ "-alias",
+ "hop-test",
+ "-keyalg",
+ "RSA",
+ "-keysize",
+ "2048",
+ "-validity",
+ "1",
+ "-storetype",
+ "PKCS12",
+ "-keystore",
+ keyStorePath.toString(),
+ "-storepass",
+ HTTPS_KEYSTORE_PASSWORD,
+ "-keypass",
+ HTTPS_KEYSTORE_PASSWORD,
+ "-dname",
+ "CN=localhost, OU=Hop, O=Apache, L=Test, S=Test, C=US",
+ "-noprompt")
+ .redirectErrorStream(true)
+ .start();
+
+ int exitCode = process.waitFor();
+ if (exitCode != 0) {
+ String output = new String(process.getInputStream().readAllBytes(),
UTF_8);
+ throw new IOException("Unable to generate HTTPS test keystore: " +
output);
+ }
+ }
+
+ private static String keytoolExecutable() {
+ return Path.of(System.getProperty("java.home"), "bin",
"keytool").toString();
+ }
+
+ private static String httpBaseUrl() {
+ return "http://"; + HTTP_HOST + ":" + httpServer.getAddress().getPort();
+ }
+
+ private static String httpsBaseUrl() {
+ return "https://"; + HTTPS_HOST + ":" + httpsServer.getAddress().getPort();
+ }
+
+ private static void stopHttpServer() {
+ if (httpServer != null) {
+ httpServer.stop(2);
+ }
+ }
+
+ private static void stopHttpsServer() throws IOException {
+ if (httpsServer != null) {
+ httpsServer.stop(2);
+ }
+ if (httpsKeyStorePath != null) {
+ Files.deleteIfExists(httpsKeyStorePath);
+ }
+ }
+}
diff --git
a/plugins/transforms/http/src/main/java/org/apache/hop/pipeline/transforms/http/Http.java
b/plugins/transforms/http/src/main/java/org/apache/hop/pipeline/transforms/http/Http.java
index 0f961084eb..c2b37d6494 100644
---
a/plugins/transforms/http/src/main/java/org/apache/hop/pipeline/transforms/http/Http.java
+++
b/plugins/transforms/http/src/main/java/org/apache/hop/pipeline/transforms/http/Http.java
@@ -49,6 +49,7 @@ import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.protocol.HttpClientContext;
import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.impl.auth.BasicScheme;
import org.apache.http.impl.client.BasicAuthCache;
import org.apache.http.impl.client.CloseableHttpClient;
@@ -217,10 +218,7 @@ public class Http extends BaseTransform<HttpMeta,
HttpData> {
default:
HttpEntity entity = httpResponse.getEntity();
if (entity != null) {
- body =
- StringUtils.isEmpty(meta.getEncoding())
- ? EntityUtils.toString(entity)
- : EntityUtils.toString(entity, meta.getEncoding());
+ body = readResponseBody(entity);
} else {
body = "";
}
@@ -229,6 +227,23 @@ public class Http extends BaseTransform<HttpMeta,
HttpData> {
return body;
}
+ private String readResponseBody(HttpEntity entity) throws IOException {
+ byte[] bodyBytes = EntityUtils.toByteArray(entity);
+ dataVolumeIn = (dataVolumeIn != null ? dataVolumeIn : 0L) +
bodyBytes.length;
+
+ ByteArrayEntity countedEntity = new ByteArrayEntity(bodyBytes);
+ if (entity.getContentType() != null) {
+ countedEntity.setContentType(entity.getContentType());
+ }
+ if (entity.getContentEncoding() != null) {
+ countedEntity.setContentEncoding(entity.getContentEncoding());
+ }
+
+ return StringUtils.isEmpty(meta.getEncoding())
+ ? EntityUtils.toString(countedEntity)
+ : EntityUtils.toString(countedEntity, meta.getEncoding());
+ }
+
private static @NotNull HttpClientContext buildtHttpClientContext(HttpHost
target) {
// Create AuthCache instance
AuthCache authCache = new BasicAuthCache();
diff --git
a/plugins/transforms/http/src/test/java/org/apache/hop/pipeline/transforms/http/HttpExecutionTest.java
b/plugins/transforms/http/src/test/java/org/apache/hop/pipeline/transforms/http/HttpExecutionTest.java
new file mode 100644
index 0000000000..773f7d6fb5
--- /dev/null
+++
b/plugins/transforms/http/src/test/java/org/apache/hop/pipeline/transforms/http/HttpExecutionTest.java
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.http;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.sun.net.httpserver.HttpServer;
+import java.io.OutputStream;
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hop.core.HopClientEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.BaseTransform;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.http.message.BasicNameValuePair;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Integration-style tests for {@link Http#callHttpService} using an embedded
HTTP server. */
+class HttpExecutionTest {
+
+ private static HttpServer server;
+ private static int port;
+
+ // Capture values written by server handlers so assertions can read them
+ private static final AtomicReference<String> lastRequestUri = new
AtomicReference<>();
+ private static final AtomicReference<String> lastRequestHeader = new
AtomicReference<>();
+
+ @BeforeAll
+ static void startServer() throws Exception {
+ HopClientEnvironment.init();
+
+ server = HttpServer.create(new InetSocketAddress("localhost", 0), 10);
+ port = server.getAddress().getPort();
+
+ // Returns a fixed payload
+ server.createContext(
+ "/body",
+ exchange -> {
+ byte[] response = "hello world".getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type", "text/plain;
charset=UTF-8");
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(response);
+ }
+ exchange.close();
+ });
+
+ // Records query string so tests can assert on it
+ server.createContext(
+ "/query",
+ exchange -> {
+ lastRequestUri.set(exchange.getRequestURI().toString());
+ byte[] response = "ok".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(response);
+ }
+ exchange.close();
+ });
+
+ // Echoes back the custom request header
+ server.createContext(
+ "/headercheck",
+ exchange -> {
+
lastRequestHeader.set(exchange.getRequestHeaders().getFirst("X-Custom-Header"));
+ byte[] response = "header ok".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(response);
+ }
+ exchange.close();
+ });
+
+ // Returns a custom response header so tests can verify header capture
+ server.createContext(
+ "/withheader",
+ exchange -> {
+ exchange.getResponseHeaders().add("X-Custom-Response", "my-value");
+ byte[] response = "ok".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(response);
+ }
+ exchange.close();
+ });
+
+ // Returns 200 but no entity
+ server.createContext(
+ "/nobody",
+ exchange -> {
+ exchange.sendResponseHeaders(200, 0);
+ exchange.close();
+ });
+
+ // Returns 204 No Content
+ server.createContext(
+ "/nocontent",
+ exchange -> {
+ exchange.sendResponseHeaders(204, -1);
+ exchange.close();
+ });
+
+ // Returns 401 Unauthorized
+ server.createContext(
+ "/unauthorized",
+ exchange -> {
+ exchange.sendResponseHeaders(401, -1);
+ exchange.close();
+ });
+
+ server.start();
+ }
+
+ @AfterAll
+ static void stopServer() {
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ @BeforeEach
+ void resetCaptures() {
+ lastRequestUri.set(null);
+ lastRequestHeader.set(null);
+ }
+
+ // ---- helper factories ----
+
+ private String baseUrl() {
+ return "http://localhost:"; + port;
+ }
+
+ private Http newHttp(HttpMeta meta, HttpData data) {
+ TransformMeta transformMeta = new TransformMeta();
+ transformMeta.setName("HttpExecTest");
+ PipelineMeta pipelineMeta = new PipelineMeta();
+ pipelineMeta.setName("HttpExecTest");
+ pipelineMeta.addTransform(transformMeta);
+ return new Http(transformMeta, meta, data, 0, pipelineMeta, new
LocalPipelineEngine());
+ }
+
+ /** Minimal HttpMeta + HttpData for a no-params GET returning just the body
field. */
+ private Http minimalGet(String url) {
+ HttpMeta meta = new HttpMeta(); // resultFields.fieldName = "result" by
default
+ HttpData data = new HttpData();
+ data.realUrl = url;
+ data.argNrs = new int[0];
+ return newHttp(meta, data);
+ }
+
+ // ---- tests ----
+
+ @Test
+ void callHttpServiceReturnsBodyAndTracksDataVolumeIn() throws Exception {
+ Http http = minimalGet(baseUrl() + "/body");
+
+ Object[] result = http.callHttpService(new RowMeta(), new Object[0]);
+
+ assertEquals("hello world", result[0]);
+ assertEquals(
+ (long) "hello world".getBytes(StandardCharsets.UTF_8).length,
getDataVolumeIn(http));
+ }
+
+ @Test
+ void callHttpServiceWith204ReturnsEmptyBody() throws Exception {
+ Http http = minimalGet(baseUrl() + "/nocontent");
+
+ Object[] result = http.callHttpService(new RowMeta(), new Object[0]);
+
+ assertEquals("", result[0]);
+ assertNull(getDataVolumeIn(http)); // nothing was read
+ }
+
+ @Test
+ void callHttpServiceWithEmptyEntityReturnsEmptyBody() throws Exception {
+ Http http = minimalGet(baseUrl() + "/nobody");
+
+ Object[] result = http.callHttpService(new RowMeta(), new Object[0]);
+
+ assertEquals("", result[0]);
+ }
+
+ @Test
+ void callHttpServiceWith401ThrowsHopException() {
+ Http http = minimalGet(baseUrl() + "/unauthorized");
+
+ assertThrows(HopException.class, () -> http.callHttpService(new RowMeta(),
new Object[0]));
+ }
+
+ @Test
+ void callHttpServiceWithQueryParameterAppendsToUrl() throws Exception {
+ HttpMeta meta = new HttpMeta();
+ meta.getLookupParameters().getQueryParameters().add(new
HttpMeta.QueryParameter("qField", "q"));
+
+ HttpData data = new HttpData();
+ data.realUrl = baseUrl() + "/query";
+ data.argNrs = new int[] {0};
+
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("qField"));
+
+ Http http = newHttp(meta, data);
+ http.callHttpService(rowMeta, new Object[] {"searchterm"});
+
+ assertNotNull(lastRequestUri.get());
+ assertTrue(lastRequestUri.get().contains("q=searchterm"));
+ }
+
+ @Test
+ void callHttpServiceSendsRequestHeaderFromRow() throws Exception {
+ HttpMeta meta = new HttpMeta();
+ HttpData data = new HttpData();
+ data.realUrl = baseUrl() + "/headercheck";
+ data.argNrs = new int[0];
+ data.useHeaderParameters = true;
+ data.headerParameters =
+ new org.apache.http.NameValuePair[] {
+ new BasicNameValuePair("X-Custom-Header", "" /* value comes from row
*/)
+ };
+ data.headerParametersNrs = new int[] {0};
+
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("headerValueField"));
+ data.inputRowMeta = rowMeta;
+
+ Http http = newHttp(meta, data);
+ http.callHttpService(rowMeta, new Object[] {"testvalue"});
+
+ assertEquals("testvalue", lastRequestHeader.get());
+ }
+
+ @Test
+ void callHttpServiceStatusCodeAppearsInResultWhenConfigured() throws
Exception {
+ HttpMeta meta = new HttpMeta();
+ meta.getResultFields().setResultCodeFieldName("statusCode");
+
+ HttpData data = new HttpData();
+ data.realUrl = baseUrl() + "/body";
+ data.argNrs = new int[0];
+
+ Http http = newHttp(meta, data);
+ Object[] result = http.callHttpService(new RowMeta(), new Object[0]);
+
+ // result[0] = body (fieldName="result" by default), result[1] = status
code
+ assertEquals("hello world", result[0]);
+ assertEquals(200L, result[1]);
+ }
+
+ @Test
+ void callHttpServiceResponseHeadersIncludedInResult() throws Exception {
+ HttpMeta meta = new HttpMeta();
+ meta.getResultFields().setFieldName(""); // no body field
+ meta.getResultFields().setResponseHeaderFieldName("responseHeaders");
+
+ HttpData data = new HttpData();
+ data.realUrl = baseUrl() + "/withheader";
+ data.argNrs = new int[0];
+
+ Http http = newHttp(meta, data);
+ Object[] result = http.callHttpService(new RowMeta(), new Object[0]);
+
+ // result[0] = header JSON (body field empty → skipped)
+ String headerJson = (String) result[0];
+ assertNotNull(headerJson);
+ // Header JSON must contain our custom value (header name casing may vary
by HTTP stack)
+ assertTrue(
+ headerJson.contains("my-value"),
+ "Expected headers JSON to contain 'my-value' but got: " + headerJson);
+ }
+
+ @Test
+ void callHttpServiceWithUnknownHostThrowsHopException() {
+ Http http = minimalGet("http://this.host.does.not.exist.invalid";);
+
+ assertThrows(HopException.class, () -> http.callHttpService(new RowMeta(),
new Object[0]));
+ }
+
+ @Test
+ void callHttpServiceResponseTimeAppearsInResultWhenConfigured() throws
Exception {
+ HttpMeta meta = new HttpMeta();
+ meta.getResultFields().setResponseTimeFieldName("responseTime");
+
+ HttpData data = new HttpData();
+ data.realUrl = baseUrl() + "/body";
+ data.argNrs = new int[0];
+
+ Http http = newHttp(meta, data);
+ Object[] result = http.callHttpService(new RowMeta(), new Object[0]);
+
+ // result[0] = body, result[1] = responseTime (Long >= 0)
+ assertTrue(result[1] instanceof Long);
+ assertTrue((Long) result[1] >= 0L);
+ }
+
+ // ---- reflection helpers ----
+
+ private static Long getDataVolumeIn(Http http) throws Exception {
+ Field field = BaseTransform.class.getDeclaredField("dataVolumeIn");
+ field.setAccessible(true);
+ return (Long) field.get(http);
+ }
+}
diff --git
a/plugins/transforms/http/src/test/java/org/apache/hop/pipeline/transforms/http/HttpTest.java
b/plugins/transforms/http/src/test/java/org/apache/hop/pipeline/transforms/http/HttpTest.java
new file mode 100644
index 0000000000..e34f75ad59
--- /dev/null
+++
b/plugins/transforms/http/src/test/java/org/apache/hop/pipeline/transforms/http/HttpTest.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.http;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.net.HttpURLConnection;
+import java.nio.charset.StandardCharsets;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.BaseTransform;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.http.Header;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.message.BasicHeader;
+import org.json.simple.JSONArray;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.junit.jupiter.api.Test;
+
+class HttpTest {
+
+ private Http newTransform(String encoding) {
+ TransformMeta transformMeta = new TransformMeta();
+ transformMeta.setName("HttpTest");
+ PipelineMeta pipelineMeta = new PipelineMeta();
+ pipelineMeta.setName("HttpTest");
+ pipelineMeta.addTransform(transformMeta);
+
+ HttpMeta meta = new HttpMeta();
+ meta.setEncoding(encoding);
+
+ HttpData data = new HttpData();
+ data.realUrl = "http://localhost/test";;
+
+ return new Http(transformMeta, meta, data, 0, pipelineMeta, new
LocalPipelineEngine());
+ }
+
+ @Test
+ void extractHeaderStringAggregatesDuplicateHeaders() throws Exception {
+ Header[] headers = {
+ new BasicHeader("X-Test", "first"),
+ new BasicHeader("X-Test", "second"),
+ new BasicHeader("Content-Type", "application/json")
+ };
+
+ JSONObject headerJson = (JSONObject) new
JSONParser().parse(invokeExtractHeaderString(headers));
+
+ assertEquals("application/json", headerJson.get("Content-Type"));
+ assertTrue(headerJson.get("X-Test") instanceof JSONArray);
+ assertEquals("first", ((JSONArray) headerJson.get("X-Test")).get(0));
+ assertEquals("second", ((JSONArray) headerJson.get("X-Test")).get(1));
+ }
+
+ @Test
+ void handResponseReadsBodyAndTracksDataVolumeIn() throws Exception {
+ Http http = newTransform("UTF-8");
+ CloseableHttpResponse response = mock(CloseableHttpResponse.class);
+ String payload = "hello http";
+ doReturn(new StringEntity(payload,
StandardCharsets.UTF_8)).when(response).getEntity();
+
+ String body = invokeHandResponse(http, HttpURLConnection.HTTP_OK,
response);
+
+ assertEquals(payload, body);
+ assertEquals(
+ (long) payload.getBytes(StandardCharsets.UTF_8).length,
getLongField(http, "dataVolumeIn"));
+ }
+
+ @Test
+ void handResponseReturnsEmptyStringForNoContent() throws Exception {
+ Http http = newTransform("UTF-8");
+ CloseableHttpResponse response = mock(CloseableHttpResponse.class);
+
+ String body = invokeHandResponse(http, HttpURLConnection.HTTP_NO_CONTENT,
response);
+
+ assertEquals("", body);
+ }
+
+ @Test
+ void handResponseThrowsForUnauthorizedStatus() {
+ Http http = newTransform("UTF-8");
+ CloseableHttpResponse response = mock(CloseableHttpResponse.class);
+
+ assertThrows(
+ HopTransformException.class,
+ () -> invokeHandResponse(http, HttpURLConnection.HTTP_UNAUTHORIZED,
response));
+ }
+
+ private static String invokeHandResponse(
+ Http http, int statusCode, CloseableHttpResponse response) throws
Exception {
+ Method method =
+ http.getClass().getDeclaredMethod("handResponse", int.class,
CloseableHttpResponse.class);
+ method.setAccessible(true);
+ try {
+ return (String) method.invoke(http, statusCode, response);
+ } catch (InvocationTargetException e) {
+ if (e.getCause() instanceof Exception exception) {
+ throw exception;
+ }
+ throw e;
+ }
+ }
+
+ private static String invokeExtractHeaderString(Header[] headers) throws
Exception {
+ Method method = Http.class.getDeclaredMethod("extractHeaderString",
Header[].class);
+ method.setAccessible(true);
+ return (String) method.invoke(null, (Object) headers);
+ }
+
+ private static Long getLongField(Object target, String fieldName) throws
Exception {
+ Field field = BaseTransform.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return (Long) field.get(target);
+ }
+}
diff --git
a/plugins/transforms/httppost/src/main/java/org/apache/hop/pipeline/transforms/httppost/HttpPost.java
b/plugins/transforms/httppost/src/main/java/org/apache/hop/pipeline/transforms/httppost/HttpPost.java
index 3a8b759d59..4c2b3a6341 100644
---
a/plugins/transforms/httppost/src/main/java/org/apache/hop/pipeline/transforms/httppost/HttpPost.java
+++
b/plugins/transforms/httppost/src/main/java/org/apache/hop/pipeline/transforms/httppost/HttpPost.java
@@ -21,6 +21,7 @@ import static
org.apache.hop.pipeline.transforms.httppost.HttpPostMeta.DEFAULT_E
import com.google.common.annotations.VisibleForTesting;
import java.io.InputStreamReader;
+import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.net.HttpURLConnection;
import java.net.URI;
@@ -35,6 +36,7 @@ import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.exception.HopFileException;
import org.apache.hop.core.exception.HopTransformException;
import org.apache.hop.core.exception.HopValueException;
+import org.apache.hop.core.io.CountingOutputStream;
import org.apache.hop.core.row.RowDataUtil;
import org.apache.hop.core.util.HttpClientManager;
import org.apache.hop.core.util.StringUtil;
@@ -56,6 +58,7 @@ import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.protocol.HttpClientContext;
import org.apache.http.client.utils.URIBuilder;
import org.apache.http.client.utils.URLEncodedUtils;
+import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.http.entity.mime.MultipartEntityBuilder;
@@ -86,7 +89,8 @@ public class HttpPost extends BaseTransform<HttpPostMeta,
HttpPostData> {
super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline);
}
- private Object[] callHttpPOST(Object[] rowData) throws HopException {
+ @VisibleForTesting
+ Object[] callHttpPOST(Object[] rowData) throws HopException {
HttpClientManager.HttpClientBuilderFacade clientBuilder =
HttpClientManager.getInstance().createBuilder();
@@ -135,8 +139,9 @@ public class HttpPost extends BaseTransform<HttpPostMeta,
HttpPostData> {
// BODY PARAMETERS
addBodyParams(post, rowData, multipart);
// Set request entity?
- addBodyFileParam(rowData, multipart);
+ addBodyFileParam(post, rowData, multipart);
addBodyParamsAfter(post, multipart);
+ collectRequestBytes(post.getEntity());
// Execute request
Object[] newRow = null;
@@ -196,7 +201,7 @@ public class HttpPost extends BaseTransform<HttpPostMeta,
HttpPostData> {
default:
HttpEntity entity = httpResponse.getEntity();
if (entity != null) {
- body = EntityUtils.toString(entity);
+ body = readResponseBody(entity);
} else {
body = "";
}
@@ -259,6 +264,45 @@ public class HttpPost extends BaseTransform<HttpPostMeta,
HttpPostData> {
}
}
+ void collectRequestBytes(HttpEntity entity) {
+ if (entity == null) {
+ return;
+ }
+
+ long contentLength = entity.getContentLength();
+ if (contentLength < 0) {
+ try (CountingOutputStream counting =
+ new CountingOutputStream(OutputStream.nullOutputStream())) {
+ entity.writeTo(counting);
+ contentLength = counting.getCount();
+ } catch (Exception e) {
+ if (isDebug()) {
+ logDebug("Unable to determine HTTP POST request size", e);
+ }
+ return;
+ }
+ }
+
+ if (contentLength > 0) {
+ dataVolumeOut = (dataVolumeOut != null ? dataVolumeOut : 0L) +
contentLength;
+ }
+ }
+
+ private String readResponseBody(HttpEntity entity) throws Exception {
+ byte[] bodyBytes = EntityUtils.toByteArray(entity);
+ dataVolumeIn = (dataVolumeIn != null ? dataVolumeIn : 0L) +
bodyBytes.length;
+
+ ByteArrayEntity countedEntity = new ByteArrayEntity(bodyBytes);
+ if (entity.getContentType() != null) {
+ countedEntity.setContentType(entity.getContentType());
+ }
+ if (entity.getContentEncoding() != null) {
+ countedEntity.setContentEncoding(entity.getContentEncoding());
+ }
+
+ return EntityUtils.toString(countedEntity);
+ }
+
protected int requestStatusCode(HttpResponse httpResponse) {
return httpResponse.getStatusLine().getStatusCode();
}
@@ -604,7 +648,10 @@ public class HttpPost extends BaseTransform<HttpPostMeta,
HttpPostData> {
*
* @param rowData row data
*/
- private void addBodyFileParam(Object[] rowData, MultipartEntityBuilder
multipart)
+ private void addBodyFileParam(
+ org.apache.http.client.methods.HttpPost post,
+ Object[] rowData,
+ MultipartEntityBuilder multipart)
throws UnsupportedEncodingException, HopFileException, HopValueException
{
if (data.indexOfRequestEntity < 0) {
return;
@@ -635,7 +682,24 @@ public class HttpPost extends BaseTransform<HttpPostMeta,
HttpPostData> {
if (meta.isMultipartupload()) {
multipart.addBinaryBody("file", bytes);
+ return;
}
+
+ attachRawRequestEntityIfNeeded(post, bytes);
+ }
+
+ @VisibleForTesting
+ void attachRawRequestEntityIfNeeded(org.apache.http.client.methods.HttpPost
post, byte[] bytes) {
+ // In the non-multipart case the request entity field is the raw POST body
when no other body
+ // has been configured.
+ if (post.getEntity() == null) {
+ post.setEntity(new ByteArrayEntity(bytes));
+ }
+ }
+
+ @VisibleForTesting
+ Long getTrackedDataVolumeOut() {
+ return dataVolumeOut;
}
/**
diff --git
a/plugins/transforms/httppost/src/test/java/org/apache/hop/pipeline/transforms/httppost/HttpPostExecutionTest.java
b/plugins/transforms/httppost/src/test/java/org/apache/hop/pipeline/transforms/httppost/HttpPostExecutionTest.java
new file mode 100644
index 0000000000..0a608c686d
--- /dev/null
+++
b/plugins/transforms/httppost/src/test/java/org/apache/hop/pipeline/transforms/httppost/HttpPostExecutionTest.java
@@ -0,0 +1,356 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.httppost;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.sun.net.httpserver.HttpServer;
+import java.io.OutputStream;
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hop.core.HopClientEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.BaseTransform;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.http.message.BasicNameValuePair;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Integration-style tests for {@link HttpPost#callHttpPOST} using an
embedded HTTP server. */
+class HttpPostExecutionTest {
+
+ private static HttpServer server;
+ private static int port;
+
+ private static final AtomicReference<String> lastRequestBody = new
AtomicReference<>();
+ private static final AtomicReference<String> lastRequestContentType = new
AtomicReference<>();
+
+ @BeforeAll
+ static void startServer() throws Exception {
+ HopClientEnvironment.init();
+
+ server = HttpServer.create(new InetSocketAddress("localhost", 0), 10);
+ port = server.getAddress().getPort();
+
+ // Captures request body and echoes it back
+ server.createContext(
+ "/echo",
+ exchange -> {
+ byte[] body = exchange.getRequestBody().readAllBytes();
+ lastRequestBody.set(new String(body, StandardCharsets.UTF_8));
+
lastRequestContentType.set(exchange.getRequestHeaders().getFirst("Content-type"));
+ exchange.sendResponseHeaders(200, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ exchange.close();
+ });
+
+ // Returns a fixed body payload
+ server.createContext(
+ "/body",
+ exchange -> {
+ byte[] response = "post response".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(response);
+ }
+ exchange.close();
+ });
+
+ // Returns two values for the same response header
+ server.createContext(
+ "/multiheader",
+ exchange -> {
+ exchange.getResponseHeaders().add("X-Resp", "val1");
+ exchange.getResponseHeaders().add("X-Resp", "val2");
+ byte[] response = "ok".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(response);
+ }
+ exchange.close();
+ });
+
+ // Returns 204 No Content
+ server.createContext(
+ "/nocontent",
+ exchange -> {
+ exchange.sendResponseHeaders(204, -1);
+ exchange.close();
+ });
+
+ // Returns 401 Unauthorized
+ server.createContext(
+ "/unauthorized",
+ exchange -> {
+ exchange.sendResponseHeaders(401, -1);
+ exchange.close();
+ });
+
+ server.start();
+ }
+
+ @AfterAll
+ static void stopServer() {
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ @BeforeEach
+ void resetCaptures() {
+ lastRequestBody.set(null);
+ lastRequestContentType.set(null);
+ }
+
+ // ---- helpers ----
+
+ private String baseUrl() {
+ return "http://localhost:"; + port;
+ }
+
+ private HttpPost newPost(HttpPostMeta meta, HttpPostData data) {
+ TransformMeta transformMeta = new TransformMeta();
+ transformMeta.setName("HttpPostExecTest");
+ PipelineMeta pipelineMeta = new PipelineMeta();
+ pipelineMeta.setName("HttpPostExecTest");
+ pipelineMeta.addTransform(transformMeta);
+ return new HttpPost(transformMeta, meta, data, 0, pipelineMeta, new
LocalPipelineEngine());
+ }
+
+ /**
+ * Minimal meta with a named body result field and no status/time/header
fields. The default
+ * constructor sets code="result", so we clear that to avoid an extra index
in the output.
+ */
+ private HttpPostMeta bodyOnlyMeta() {
+ HttpPostMeta meta = new HttpPostMeta();
+ meta.setDefault();
+ meta.getResultFields().get(0).setName("response");
+ meta.getResultFields().get(0).setCode(null);
+ return meta;
+ }
+
+ /** Minimal data pointing at the given URL with an empty input row. */
+ private HttpPostData minimalData(String url) {
+ HttpPostData data = new HttpPostData();
+ data.realUrl = url;
+ data.inputRowMeta = new RowMeta();
+ data.outputRowMeta = new RowMeta();
+ return data;
+ }
+
+ // ---- tests ----
+
+ @Test
+ void callHttpPOSTReturnsBodyAndTracksDataVolumeIn() throws Exception {
+ HttpPost http = newPost(bodyOnlyMeta(), minimalData(baseUrl() + "/body"));
+
+ Object[] result = http.callHttpPOST(new Object[0]);
+
+ assertEquals("post response", result[0]);
+ assertEquals(
+ (long) "post response".getBytes(StandardCharsets.UTF_8).length,
getDataVolumeIn(http));
+ }
+
+ @Test
+ void callHttpPOSTWith204ReturnsEmptyBody() throws Exception {
+ HttpPost http = newPost(bodyOnlyMeta(), minimalData(baseUrl() +
"/nocontent"));
+
+ Object[] result = http.callHttpPOST(new Object[0]);
+
+ assertEquals("", result[0]);
+ assertNull(getDataVolumeIn(http)); // nothing was read
+ }
+
+ @Test
+ void callHttpPOSTWith401ThrowsHopException() {
+ HttpPost http = newPost(bodyOnlyMeta(), minimalData(baseUrl() +
"/unauthorized"));
+
+ assertThrows(HopException.class, () -> http.callHttpPOST(new Object[0]));
+ }
+
+ @Test
+ void callHttpPOSTWithBodyParamsSendsFormEncodedPayload() throws Exception {
+ HttpPostMeta meta = bodyOnlyMeta();
+ HttpPostData data = minimalData(baseUrl() + "/echo");
+ data.useBodyParameters = true;
+ data.body_parameters_nrs = new int[] {0};
+ data.bodyParameters =
+ new org.apache.http.NameValuePair[] {
+ new BasicNameValuePair("city", "") // name fixed; value comes from
row
+ };
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("cityField"));
+ data.inputRowMeta = rowMeta;
+ data.outputRowMeta = rowMeta;
+
+ HttpPost http = newPost(meta, data);
+ http.callHttpPOST(new Object[] {"Berlin"});
+
+ assertNotNull(lastRequestBody.get());
+ assertTrue(lastRequestBody.get().contains("city=Berlin"));
+ }
+
+ @Test
+ void callHttpPOSTWithQueryParamsSendsFormEncodedPayload() throws Exception {
+ // In HttpPost "query params" are sent as URL-encoded form body
(UrlEncodedFormEntity)
+ HttpPostMeta meta = bodyOnlyMeta();
+ HttpPostData data = minimalData(baseUrl() + "/echo");
+ data.useQueryParameters = true;
+ data.query_parameters_nrs = new int[] {0};
+ data.queryParameters = new org.apache.http.NameValuePair[] {new
BasicNameValuePair("q", "")};
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("qField"));
+ data.inputRowMeta = rowMeta;
+ data.outputRowMeta = rowMeta;
+
+ HttpPost http = newPost(meta, data);
+ http.callHttpPOST(new Object[] {"test"});
+
+ assertNotNull(lastRequestBody.get());
+ assertTrue(lastRequestBody.get().contains("q=test"));
+ }
+
+ @Test
+ void callHttpPOSTWithRawRequestEntitySendsBytes() throws Exception {
+ HttpPostMeta meta = bodyOnlyMeta();
+ HttpPostData data = minimalData(baseUrl() + "/echo");
+ data.indexOfRequestEntity = 0;
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("bodyField"));
+ data.inputRowMeta = rowMeta;
+ data.outputRowMeta = rowMeta;
+
+ HttpPost http = newPost(meta, data);
+ http.callHttpPOST(new Object[] {"raw body content"});
+
+ assertEquals("raw body content", lastRequestBody.get());
+ assertEquals(
+ (long) "raw body content".getBytes(StandardCharsets.UTF_8).length,
getDataVolumeOut(http));
+ }
+
+ @Test
+ void callHttpPOSTStatusCodeAppearsInResultWhenConfigured() throws Exception {
+ HttpPostMeta meta = new HttpPostMeta();
+ meta.setDefault();
+ meta.getResultFields().get(0).setName("response");
+ meta.getResultFields().get(0).setCode("status");
+
+ HttpPost http = newPost(meta, minimalData(baseUrl() + "/body"));
+ Object[] result = http.callHttpPOST(new Object[0]);
+
+ assertEquals("post response", result[0]);
+ assertEquals(200L, result[1]);
+ }
+
+ @Test
+ void callHttpPOSTWithDuplicateResponseHeadersAggregatesIntoArray() throws
Exception {
+ HttpPostMeta meta = new HttpPostMeta();
+ meta.setDefault();
+ meta.getResultFields().get(0).setName(null);
+ meta.getResultFields().get(0).setCode(null);
+ meta.getResultFields().get(0).setResponseHeaderFieldName("headers");
+
+ HttpPost http = newPost(meta, minimalData(baseUrl() + "/multiheader"));
+ Object[] result = http.callHttpPOST(new Object[0]);
+
+ // result[0] = headers JSON (name/code both empty → skipped)
+ String headersJson = (String) result[0];
+ assertNotNull(headersJson);
+ // Both header values must appear (name casing/folding varies by HTTP
stack)
+ assertTrue(
+ headersJson.contains("val1"),
+ "Expected headers JSON to contain 'val1' but got: " + headersJson);
+ assertTrue(
+ headersJson.contains("val2"),
+ "Expected headers JSON to contain 'val2' but got: " + headersJson);
+ }
+
+ @Test
+ void callHttpPOSTWithUnknownHostThrowsHopException() {
+ HttpPost http = newPost(bodyOnlyMeta(),
minimalData("http://this.host.does.not.exist.invalid";));
+
+ assertThrows(HopException.class, () -> http.callHttpPOST(new Object[0]));
+ }
+
+ @Test
+ void callHttpPOSTResponseTimeAppearsInResultWhenConfigured() throws
Exception {
+ HttpPostMeta meta = new HttpPostMeta();
+ meta.setDefault();
+ meta.getResultFields().get(0).setName("response");
+ meta.getResultFields().get(0).setCode(null);
+ meta.getResultFields().get(0).setResponseTimeFieldName("time");
+
+ HttpPost http = newPost(meta, minimalData(baseUrl() + "/body"));
+ Object[] result = http.callHttpPOST(new Object[0]);
+
+ // result[0] = body, result[1] = response time
+ assertTrue(result[1] instanceof Long);
+ assertTrue((Long) result[1] >= 0L);
+ }
+
+ @Test
+ void callHttpPOSTSetsDefaultContentTypeHeader() throws Exception {
+ HttpPost http = newPost(bodyOnlyMeta(), minimalData(baseUrl() + "/echo"));
+ http.callHttpPOST(new Object[0]);
+
+ // addHeadersContentType sets text/xml when no encoding and no override
+ assertNotNull(lastRequestContentType.get());
+ assertTrue(lastRequestContentType.get().contains("text/xml"));
+ }
+
+ @Test
+ void callHttpPOSTWithEncodingSetsContentTypeWithCharset() throws Exception {
+ HttpPostMeta meta = bodyOnlyMeta();
+ HttpPostData data = minimalData(baseUrl() + "/echo");
+ data.realEncoding = "UTF-8";
+
+ HttpPost http = newPost(meta, data);
+ http.callHttpPOST(new Object[0]);
+
+ assertNotNull(lastRequestContentType.get());
+ assertTrue(lastRequestContentType.get().contains("UTF-8"));
+ }
+
+ // ---- reflection helpers ----
+
+ private static Long getDataVolumeIn(HttpPost http) throws Exception {
+ Field field = BaseTransform.class.getDeclaredField("dataVolumeIn");
+ field.setAccessible(true);
+ return (Long) field.get(http);
+ }
+
+ private static Long getDataVolumeOut(HttpPost http) throws Exception {
+ Field field = BaseTransform.class.getDeclaredField("dataVolumeOut");
+ field.setAccessible(true);
+ return (Long) field.get(http);
+ }
+}
diff --git
a/plugins/transforms/httppost/src/test/java/org/apache/hop/pipeline/transforms/httppost/HttpPostTest.java
b/plugins/transforms/httppost/src/test/java/org/apache/hop/pipeline/transforms/httppost/HttpPostTest.java
index a18b5f5582..06e648db1b 100644
---
a/plugins/transforms/httppost/src/test/java/org/apache/hop/pipeline/transforms/httppost/HttpPostTest.java
+++
b/plugins/transforms/httppost/src/test/java/org/apache/hop/pipeline/transforms/httppost/HttpPostTest.java
@@ -18,24 +18,48 @@
package org.apache.hop.pipeline.transforms.httppost;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.nullable;
-import static org.mockito.Mockito.doCallRealMethod;
-import static org.mockito.Mockito.mock;
+import static org.junit.jupiter.api.Assertions.assertSame;
-import org.apache.hop.core.exception.HopException;
+import java.io.ByteArrayInputStream;
+import java.io.InputStreamReader;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.BaseTransform;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpVersion;
import org.apache.http.NameValuePair;
+import org.apache.http.entity.InputStreamEntity;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.message.BasicHttpResponse;
import org.apache.http.message.BasicNameValuePair;
+import org.apache.http.message.BasicStatusLine;
import org.junit.jupiter.api.Test;
class HttpPostTest {
+ private HttpPost newTransform() {
+ TransformMeta transformMeta = new TransformMeta();
+ transformMeta.setName("HttpPostTest");
+ PipelineMeta pipelineMeta = new PipelineMeta();
+ pipelineMeta.setName("HttpPostTest");
+ pipelineMeta.addTransform(transformMeta);
+ return new HttpPost(
+ transformMeta,
+ new HttpPostMeta(),
+ new HttpPostData(),
+ 0,
+ pipelineMeta,
+ new LocalPipelineEngine());
+ }
+
@Test
- void getRequestBodyParametersAsStringWithNullEncoding() throws HopException {
- HttpPost http = mock(HttpPost.class);
- doCallRealMethod()
- .when(http)
- .getRequestBodyParamsAsStr(any(NameValuePair[].class),
nullable(String.class));
+ void getRequestBodyParametersAsStringWithNullEncoding() {
+ HttpPost http = newTransform();
NameValuePair[] pairs =
new NameValuePair[] {
@@ -44,4 +68,135 @@ class HttpPostTest {
assertEquals("u=usr&p=pass", http.getRequestBodyParamsAsStr(pairs, null));
}
+
+ @Test
+ void getRequestBodyParametersAsStringWithEncodingEscapesContent() {
+ HttpPost http = newTransform();
+
+ NameValuePair[] pairs = {
+ new BasicNameValuePair("city", "Sao Paulo"), new BasicNameValuePair("q",
"a+b")
+ };
+
+ assertEquals("city=Sao+Paulo&q=a%2Bb",
http.getRequestBodyParamsAsStr(pairs, "UTF-8"));
+ }
+
+ @Test
+ void attachRawRequestEntityIfNeededSetsEntityWhenMissing() {
+ HttpPost http = newTransform();
+
+ org.apache.http.client.methods.HttpPost post =
+ new org.apache.http.client.methods.HttpPost("http://localhost";);
+
+ http.attachRawRequestEntityIfNeeded(post, "payload".getBytes());
+
+ assertEquals(7, post.getEntity().getContentLength());
+ }
+
+ @Test
+ void attachRawRequestEntityIfNeededDoesNotOverrideExistingEntity() throws
Exception {
+ HttpPost http = newTransform();
+
+ org.apache.http.client.methods.HttpPost post =
+ new org.apache.http.client.methods.HttpPost("http://localhost";);
+ HttpEntity existing = new StringEntity("existing");
+ post.setEntity(existing);
+
+ http.attachRawRequestEntityIfNeeded(post, "payload".getBytes());
+
+ assertSame(existing, post.getEntity());
+ }
+
+ @Test
+ void collectRequestBytesCountsUnknownLengthEntity() {
+ HttpPost http = newTransform();
+
+ HttpEntity entity = new InputStreamEntity(new
ByteArrayInputStream("payload".getBytes()), -1);
+
+ http.collectRequestBytes(entity);
+
+ assertEquals(7L, http.getTrackedDataVolumeOut());
+ }
+
+ @Test
+ void collectRequestBytesCountsKnownLengthEntity() {
+ HttpPost http = newTransform();
+
+ HttpEntity entity = new StringEntity("payload", StandardCharsets.UTF_8);
+
+ http.collectRequestBytes(entity);
+
+ assertEquals(7L, http.getTrackedDataVolumeOut());
+ }
+
+ @Test
+ void readResponseBodyTracksDataVolumeIn() throws Exception {
+ HttpPost http = newTransform();
+
+ String body =
+ (String)
+ invokePrivate(
+ http, "readResponseBody", new StringEntity("response",
StandardCharsets.UTF_8));
+
+ assertEquals("response", body);
+ assertEquals(8L, getLongField(http, "dataVolumeIn"));
+ }
+
+ @Test
+ void openStreamWithEncodingReturnsReaderDecodedInThatEncoding() throws
Exception {
+ HttpPost http = newTransform();
+ HttpResponse response = responseWithBody("hello");
+
+ try (InputStreamReader reader = http.openStream("UTF-8", response)) {
+ char[] buf = new char[5];
+ int read = reader.read(buf);
+ assertEquals(5, read);
+ assertEquals("hello", new String(buf));
+ }
+ }
+
+ @Test
+ void openStreamWithoutEncodingReturnsDefaultReader() throws Exception {
+ HttpPost http = newTransform();
+ HttpResponse response = responseWithBody("world");
+
+ try (InputStreamReader reader = http.openStream(null, response)) {
+ char[] buf = new char[5];
+ int read = reader.read(buf);
+ assertEquals(5, read);
+ assertEquals("world", new String(buf));
+ }
+ }
+
+ @Test
+ void openStreamWithEmptyEncodingReturnsDefaultReader() throws Exception {
+ HttpPost http = newTransform();
+ HttpResponse response = responseWithBody("hop");
+
+ try (InputStreamReader reader = http.openStream("", response)) {
+ char[] buf = new char[3];
+ int read = reader.read(buf);
+ assertEquals(3, read);
+ assertEquals("hop", new String(buf));
+ }
+ }
+
+ private static HttpResponse responseWithBody(String body) {
+ BasicHttpResponse response =
+ new BasicHttpResponse(new BasicStatusLine(HttpVersion.HTTP_1_1, 200,
"OK"));
+ response.setEntity(new StringEntity(body, StandardCharsets.UTF_8));
+ return response;
+ }
+
+ private static Object invokePrivate(Object target, String methodName,
Object... args)
+ throws Exception {
+ Method method = target.getClass().getDeclaredMethod(methodName,
HttpEntity.class);
+ method.setAccessible(true);
+ return method.invoke(target, args);
+ }
+
+ private static Long getLongField(Object target, String fieldName) throws
Exception {
+ Field field = BaseTransform.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return (Long) field.get(target);
+ }
}
diff --git
a/plugins/transforms/rest/src/main/java/org/apache/hop/pipeline/transforms/rest/Rest.java
b/plugins/transforms/rest/src/main/java/org/apache/hop/pipeline/transforms/rest/Rest.java
index 7948ed22d2..709dd4d40c 100644
---
a/plugins/transforms/rest/src/main/java/org/apache/hop/pipeline/transforms/rest/Rest.java
+++
b/plugins/transforms/rest/src/main/java/org/apache/hop/pipeline/transforms/rest/Rest.java
@@ -34,12 +34,14 @@ import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.CertificateException;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.Supplier;
import javax.net.ssl.SSLContext;
@@ -333,6 +335,7 @@ public class Rest extends BaseTransform<RestMeta, RestData>
{
}
}
}
+ trackResponseBytes(response, body);
// get Header
MultivaluedMap<String, Object> headers = searchForHeaders(response);
JSONObject json = new JSONObject();
@@ -491,6 +494,9 @@ public class Rest extends BaseTransform<RestMeta, RestData>
{
return invocationBuilder.get(Response.class);
}
case RestMeta.HTTP_METHOD_POST -> {
+ trackRequestBytes(
+ entityString,
+ contentType != null ? resolveCharset(contentType) :
resolveCharset(data.mediaType));
if (null != contentType) {
return invocationBuilder.post(Entity.entity(entityString,
contentType));
} else {
@@ -498,6 +504,9 @@ public class Rest extends BaseTransform<RestMeta, RestData>
{
}
}
case RestMeta.HTTP_METHOD_PUT -> {
+ trackRequestBytes(
+ entityString,
+ contentType != null ? resolveCharset(contentType) :
resolveCharset(data.mediaType));
if (null != contentType) {
return invocationBuilder.put(Entity.entity(entityString,
contentType));
} else {
@@ -505,6 +514,9 @@ public class Rest extends BaseTransform<RestMeta, RestData>
{
}
}
case RestMeta.HTTP_METHOD_DELETE -> {
+ trackRequestBytes(
+ entityString,
+ contentType != null ? resolveCharset(contentType) :
resolveCharset(data.mediaType));
Invocation invocation =
invocationBuilder.build("DELETE", Entity.entity(entityString,
data.mediaType));
return invocation.invoke();
@@ -516,6 +528,9 @@ public class Rest extends BaseTransform<RestMeta, RestData>
{
return invocationBuilder.options();
}
case RestMeta.HTTP_METHOD_PATCH -> {
+ trackRequestBytes(
+ entityString,
+ contentType != null ? resolveCharset(contentType) :
resolveCharset(data.mediaType));
if (null != contentType) {
return invocationBuilder.method(
RestMeta.HTTP_METHOD_PATCH, Entity.entity(entityString,
contentType));
@@ -533,6 +548,53 @@ public class Rest extends BaseTransform<RestMeta,
RestData> {
}
}
+ private void trackRequestBytes(String entityString, Charset charset) {
+ if (entityString == null) {
+ return;
+ }
+
+ byte[] requestBytes = entityString.getBytes(charset);
+ if (requestBytes.length > 0) {
+ dataVolumeOut = (dataVolumeOut != null ? dataVolumeOut : 0L) +
requestBytes.length;
+ }
+ }
+
+ private void trackResponseBytes(Response response, String body) {
+ long responseBytes = response.getLength();
+ if (responseBytes < 0 && body != null) {
+ responseBytes =
body.getBytes(resolveCharset(response.getMediaType())).length;
+ }
+ if (responseBytes > 0) {
+ dataVolumeIn = (dataVolumeIn != null ? dataVolumeIn : 0L) +
responseBytes;
+ }
+ }
+
+ private Charset resolveCharset(String mediaTypeValue) {
+ if (!Utils.isEmpty(mediaTypeValue)) {
+ try {
+ return resolveCharset(MediaType.valueOf(mediaTypeValue));
+ } catch (Exception ignored) {
+ // Fall back to UTF-8 below if the header value is malformed.
+ }
+ }
+ return StandardCharsets.UTF_8;
+ }
+
+ private Charset resolveCharset(MediaType mediaType) {
+ if (mediaType != null) {
+ Map<String, String> parameters = mediaType.getParameters();
+ String charsetName = parameters.get("charset");
+ if (!Utils.isEmpty(charsetName)) {
+ try {
+ return Charset.forName(charsetName);
+ } catch (Exception ignored) {
+ // Fall back to UTF-8 below if the charset is unknown.
+ }
+ }
+ }
+ return StandardCharsets.UTF_8;
+ }
+
/**
* set the Authentication/Authorization header from the connection first, if
available. this
* transform's headers will override this value if available.
diff --git
a/plugins/transforms/rest/src/test/java/org/apache/hop/pipeline/transforms/rest/RestTest.java
b/plugins/transforms/rest/src/test/java/org/apache/hop/pipeline/transforms/rest/RestTest.java
index fe3b697c91..98ad1e2036 100644
---
a/plugins/transforms/rest/src/test/java/org/apache/hop/pipeline/transforms/rest/RestTest.java
+++
b/plugins/transforms/rest/src/test/java/org/apache/hop/pipeline/transforms/rest/RestTest.java
@@ -27,11 +27,17 @@ import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.spy;
import jakarta.ws.rs.client.Client;
+import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.MultivaluedHashMap;
import jakarta.ws.rs.core.MultivaluedMap;
import jakarta.ws.rs.core.Response;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+import org.apache.hop.metadata.rest.RestConnection;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.TransformMeta;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -135,4 +141,86 @@ class RestTest {
assertNull(data.indexOfHeaderFields);
assertNull(data.paramNames);
}
+
+ @Test
+ void testTrackRequestBytesAddsBytesForCharset() throws Exception {
+ Rest rest = newRest();
+
+ invokePrivate(rest, "trackRequestBytes", "hello",
StandardCharsets.UTF_16LE);
+
+ assertEquals(10L, getLongField(rest, "dataVolumeOut"));
+ }
+
+ @Test
+ void testTrackResponseBytesFallsBackToBodyLength() throws Exception {
+ Rest rest = newRest();
+ Response response = mock(Response.class);
+ doReturn(-1).when(response).getLength();
+ doReturn(MediaType.valueOf("text/plain;
charset=UTF-16LE")).when(response).getMediaType();
+
+ invokePrivate(rest, "trackResponseBytes", response, "ok");
+
+ assertEquals(4L, getLongField(rest, "dataVolumeIn"));
+ }
+
+ @Test
+ void testAddApiKeyHeaderIfAbsentAddsPrefixedHeaderWithoutOverriding() throws
Exception {
+ Rest rest = newRest();
+ RestConnection connection = mock(RestConnection.class);
+ doReturn("API Key").when(connection).getAuthType();
+ doReturn("Authorization").when(connection).getAuthorizationHeaderName();
+ doReturn("secret").when(connection).getAuthorizationHeaderValue();
+ doReturn("Bearer").when(connection).getAuthorizationPrefix();
+
+ Field connectionField = Rest.class.getDeclaredField("connection");
+ connectionField.setAccessible(true);
+ connectionField.set(rest, connection);
+
+ MultivaluedMap<String, Object> headers = new MultivaluedHashMap<>();
+ invokePrivate(rest, "addApiKeyHeaderIfAbsent", headers);
+ assertEquals("Bearer secret", headers.getFirst("Authorization"));
+
+ headers.putSingle("Authorization", "existing");
+ invokePrivate(rest, "addApiKeyHeaderIfAbsent", headers);
+ assertEquals("existing", headers.getFirst("Authorization"));
+ }
+
+ private Rest newRest() {
+ TransformMeta transformMeta = new TransformMeta();
+ transformMeta.setName("TestRest");
+ PipelineMeta pipelineMeta = new PipelineMeta();
+ pipelineMeta.setName("TestRest");
+ pipelineMeta.addTransform(transformMeta);
+ return new Rest(
+ transformMeta,
+ mock(RestMeta.class),
+ new RestData(),
+ 1,
+ pipelineMeta,
+ spy(new LocalPipelineEngine()));
+ }
+
+ private static Object invokePrivate(Object target, String methodName,
Object... args)
+ throws Exception {
+ Method method =
+ switch (methodName) {
+ case "trackRequestBytes" ->
+ target
+ .getClass()
+ .getDeclaredMethod(methodName, String.class,
java.nio.charset.Charset.class);
+ case "trackResponseBytes" ->
+ target.getClass().getDeclaredMethod(methodName, Response.class,
String.class);
+ case "addApiKeyHeaderIfAbsent" ->
+ target.getClass().getDeclaredMethod(methodName,
MultivaluedMap.class);
+ default -> throw new NoSuchMethodException(methodName);
+ };
+ method.setAccessible(true);
+ return method.invoke(target, args);
+ }
+
+ private static Long getLongField(Object target, String fieldName) throws
Exception {
+ Field field = BaseTransform.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return (Long) field.get(target);
+ }
}
