@ -16,13 +16,9 @@
package org.thingsboard.rule.engine.rest ;
import com.datastax.oss.driver.api.core.uuid.Uuids ;
import org.apache.http.HttpException ;
import org.apache.http.HttpRequest ;
import org.apache.http.HttpResponse ;
import org.apache.http.config.SocketConfig ;
import org.apache.http.impl.bootstrap.HttpServer ;
import org.apache.http.impl.bootstrap.ServerBootstrap ;
import org.apache.http.protocol.HttpContext ;
import org.apache.http.protocol.HttpRequestHandler ;
import org.junit.jupiter.api.AfterEach ;
import org.junit.jupiter.api.BeforeEach ;
@ -52,6 +48,8 @@ import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.exception.DataValidationException ;
import java.io.IOException ;
import java.io.InputStream ;
import java.nio.charset.StandardCharsets ;
import java.util.ArrayList ;
import java.util.Collections ;
import java.util.List ;
@ -79,11 +77,11 @@ public class TbRestApiCallNodeTest extends AbstractRuleNodeUpgradeTest {
@Mock
private TbContext ctx ;
private EntityId originator = new DeviceId ( Uuids . timeBased ( ) ) ;
private TbMsgMetaData metaData = new TbMsgMetaData ( ) ;
private final EntityId originator = new DeviceId ( Uuids . timeBased ( ) ) ;
private final TbMsgMetaData metaData = new TbMsgMetaData ( ) ;
private RuleChainId ruleChainId = new RuleChainId ( Uuids . timeBased ( ) ) ;
private RuleNodeId ruleNodeId = new RuleNodeId ( Uuids . timeBased ( ) ) ;
private final RuleChainId ruleChainId = new RuleChainId ( Uuids . timeBased ( ) ) ;
private final RuleNodeId ruleNodeId = new RuleNodeId ( Uuids . timeBased ( ) ) ;
private HttpServer server ;
@ -217,22 +215,17 @@ public class TbRestApiCallNodeTest extends AbstractRuleNodeUpgradeTest {
public void deleteRequestWithoutBody ( ) throws IOException , InterruptedException {
final CountDownLatch latch = new CountDownLatch ( 1 ) ;
final String path = "/path/to/delete" ;
setupServer ( "*" , new HttpRequestHandler ( ) {
@Override
public void handle ( HttpRequest request , HttpResponse response , HttpContext context )
throws HttpException , IOException {
try {
assertEquals ( request . getRequestLine ( ) . getUri ( ) , path , "Request path matches" ) ;
assertTrue ( request . containsHeader ( "Foo" ) , "Custom header included" ) ;
assertEquals ( "Bar" , request . getFirstHeader ( "Foo" ) . getValue ( ) , "Custom header value" ) ;
response . setStatusCode ( 200 ) ;
latch . countDown ( ) ;
} catch ( Exception e ) {
System . out . println ( "Exception handling request: " + e . toString ( ) ) ;
e . printStackTrace ( ) ;
latch . countDown ( ) ;
}
setupServer ( "*" , ( request , response , _ ) - > {
try {
assertEquals ( path , request . getRequestLine ( ) . getUri ( ) , "Request path matches" ) ;
assertTrue ( request . containsHeader ( "Foo" ) , "Custom header included" ) ;
assertEquals ( "Bar" , request . getFirstHeader ( "Foo" ) . getValue ( ) , "Custom header value" ) ;
response . setStatusCode ( 200 ) ;
latch . countDown ( ) ;
} catch ( Exception e ) {
System . out . println ( "Exception handling request: " + e ) ;
e . printStackTrace ( ) ;
latch . countDown ( ) ;
}
} ) ;
@ -269,28 +262,23 @@ public class TbRestApiCallNodeTest extends AbstractRuleNodeUpgradeTest {
public void deleteRequestWithBody ( ) throws IOException , InterruptedException {
final CountDownLatch latch = new CountDownLatch ( 1 ) ;
final String path = "/path/to/delete" ;
setupServer ( "*" , new HttpRequestHandler ( ) {
@Override
public void handle ( HttpRequest request , HttpResponse response , HttpContext context )
throws HttpException , IOException {
try {
assertEquals ( path , request . getRequestLine ( ) . getUri ( ) , "Request path matches" ) ;
assertTrue ( request . containsHeader ( "Content-Type" ) , "Content-Type included" ) ;
assertEquals ( "application/json" ,
request . getFirstHeader ( "Content-Type" ) . getValue ( ) , "Content-Type value" ) ;
assertTrue ( request . containsHeader ( "Content-Length" ) , "Content-Length included" ) ;
assertEquals ( "2" ,
request . getFirstHeader ( "Content-Length" ) . getValue ( ) , "Content-Length value" ) ;
assertTrue ( request . containsHeader ( "Foo" ) , "Custom header included" ) ;
assertEquals ( "Bar" , request . getFirstHeader ( "Foo" ) . getValue ( ) , "Custom header value" ) ;
response . setStatusCode ( 200 ) ;
latch . countDown ( ) ;
} catch ( Exception e ) {
System . out . println ( "Exception handling request: " + e . toString ( ) ) ;
e . printStackTrace ( ) ;
latch . countDown ( ) ;
}
setupServer ( "*" , ( request , response , _ ) - > {
try {
assertEquals ( path , request . getRequestLine ( ) . getUri ( ) , "Request path matches" ) ;
assertTrue ( request . containsHeader ( "Content-Type" ) , "Content-Type included" ) ;
assertEquals ( "application/json" ,
request . getFirstHeader ( "Content-Type" ) . getValue ( ) , "Content-Type value" ) ;
assertTrue ( request . containsHeader ( "Content-Length" ) , "Content-Length included" ) ;
assertEquals ( "2" ,
request . getFirstHeader ( "Content-Length" ) . getValue ( ) , "Content-Length value" ) ;
assertTrue ( request . containsHeader ( "Foo" ) , "Custom header included" ) ;
assertEquals ( "Bar" , request . getFirstHeader ( "Foo" ) . getValue ( ) , "Custom header value" ) ;
response . setStatusCode ( 200 ) ;
latch . countDown ( ) ;
} catch ( Exception e ) {
System . out . println ( "Exception handling request: " + e ) ;
e . printStackTrace ( ) ;
latch . countDown ( ) ;
}
} ) ;
@ -323,6 +311,143 @@ public class TbRestApiCallNodeTest extends AbstractRuleNodeUpgradeTest {
assertEquals ( TbMsg . EMPTY_JSON_OBJECT , dataCaptor . getValue ( ) ) ;
}
@Test
public void postRequestWithBodyTemplate ( ) throws IOException , InterruptedException {
final CountDownLatch latch = new CountDownLatch ( 1 ) ;
final String path = "/api/token" ;
final String [ ] capturedBody = new String [ 1 ] ;
setupServerWithBodyCapture ( capturedBody , latch ) ;
TbRestApiCallNodeConfiguration config = new TbRestApiCallNodeConfiguration ( ) . defaultConfiguration ( ) ;
config . setRequestMethod ( "POST" ) ;
config . setRequestBodyTemplate ( "{\"grant_type\":\"client_credentials\",\"client_id\":\"${clientId}\",\"value\":\"$[token]\"}" ) ;
config . setRestEndpointUrlPattern ( String . format ( "http://localhost:%d%s" , server . getLocalPort ( ) , path ) ) ;
initWithConfig ( config ) ;
metaData . putValue ( "clientId" , "my-client-123" ) ;
TbMsg msg = TbMsg . newMsg ( )
. type ( TbMsgType . POST_TELEMETRY_REQUEST )
. originator ( originator )
. copyMetaData ( metaData )
. dataType ( TbMsgDataType . JSON )
. data ( "{\"token\":\"abc-xyz\"}" )
. ruleChainId ( ruleChainId )
. ruleNodeId ( ruleNodeId )
. build ( ) ;
restNode . onMsg ( ctx , msg ) ;
assertTrue ( latch . await ( 10 , TimeUnit . SECONDS ) , "Server handled request" ) ;
assertEquals ( "{\"grant_type\":\"client_credentials\",\"client_id\":\"my-client-123\",\"value\":\"abc-xyz\"}" , capturedBody [ 0 ] ) ;
}
@Test
public void postRequestWithBodyTemplateAndParseToPlainText ( ) throws IOException , InterruptedException {
final CountDownLatch latch = new CountDownLatch ( 1 ) ;
final String path = "/api/text" ;
final String [ ] capturedBody = new String [ 1 ] ;
setupServerWithBodyCapture ( capturedBody , latch ) ;
TbRestApiCallNodeConfiguration config = new TbRestApiCallNodeConfiguration ( ) . defaultConfiguration ( ) ;
config . setRequestMethod ( "POST" ) ;
config . setParseToPlainText ( true ) ;
config . setRequestBodyTemplate ( "Hello ${name}, your token is $[token]!" ) ;
config . setRestEndpointUrlPattern ( String . format ( "http://localhost:%d%s" , server . getLocalPort ( ) , path ) ) ;
initWithConfig ( config ) ;
metaData . putValue ( "name" , "World" ) ;
TbMsg msg = TbMsg . newMsg ( )
. type ( TbMsgType . POST_TELEMETRY_REQUEST )
. originator ( originator )
. copyMetaData ( metaData )
. dataType ( TbMsgDataType . JSON )
. data ( "{\"token\":\"abc-xyz\"}" )
. ruleChainId ( ruleChainId )
. ruleNodeId ( ruleNodeId )
. build ( ) ;
restNode . onMsg ( ctx , msg ) ;
assertTrue ( latch . await ( 10 , TimeUnit . SECONDS ) , "Server handled request" ) ;
assertEquals ( "Hello World, your token is abc-xyz!" , capturedBody [ 0 ] ) ;
}
@Test
public void postRequestWithBodyTemplateEscapesJsonSpecialChars ( ) throws IOException , InterruptedException {
final CountDownLatch latch = new CountDownLatch ( 1 ) ;
final String path = "/api/token" ;
final String [ ] capturedBody = new String [ 1 ] ;
setupServerWithBodyCapture ( capturedBody , latch ) ;
TbRestApiCallNodeConfiguration config = new TbRestApiCallNodeConfiguration ( ) . defaultConfiguration ( ) ;
config . setRequestMethod ( "POST" ) ;
config . setRequestBodyTemplate ( "{\"name\":\"${userName}\",\"desc\":\"$[description]\"}" ) ;
config . setRestEndpointUrlPattern ( String . format ( "http://localhost:%d%s" , server . getLocalPort ( ) , path ) ) ;
initWithConfig ( config ) ;
metaData . putValue ( "userName" , "John \"Doe\"" ) ;
TbMsg msg = TbMsg . newMsg ( )
. type ( TbMsgType . POST_TELEMETRY_REQUEST )
. originator ( originator )
. copyMetaData ( metaData )
. dataType ( TbMsgDataType . JSON )
. data ( "{\"description\":\"line1\\nline2\"}" )
. ruleChainId ( ruleChainId )
. ruleNodeId ( ruleNodeId )
. build ( ) ;
restNode . onMsg ( ctx , msg ) ;
assertTrue ( latch . await ( 10 , TimeUnit . SECONDS ) , "Server handled request" ) ;
assertEquals ( "{\"name\":\"John \\\"Doe\\\"\",\"desc\":\"line1\\nline2\"}" , capturedBody [ 0 ] ) ;
}
@Test
public void postRequestWithEmptyBodyTemplateUsesMessageData ( ) throws IOException , InterruptedException {
final CountDownLatch latch = new CountDownLatch ( 1 ) ;
final String path = "/api/data" ;
final String [ ] capturedBody = new String [ 1 ] ;
setupServerWithBodyCapture ( capturedBody , latch ) ;
TbRestApiCallNodeConfiguration config = new TbRestApiCallNodeConfiguration ( ) . defaultConfiguration ( ) ;
config . setRequestMethod ( "POST" ) ;
// requestBodyTemplate is null by default — should use msg.getData()
config . setRestEndpointUrlPattern ( String . format ( "http://localhost:%d%s" , server . getLocalPort ( ) , path ) ) ;
initWithConfig ( config ) ;
TbMsg msg = TbMsg . newMsg ( )
. type ( TbMsgType . POST_TELEMETRY_REQUEST )
. originator ( originator )
. copyMetaData ( metaData )
. dataType ( TbMsgDataType . JSON )
. data ( "{\"temperature\":25}" )
. ruleChainId ( ruleChainId )
. ruleNodeId ( ruleNodeId )
. build ( ) ;
restNode . onMsg ( ctx , msg ) ;
assertTrue ( latch . await ( 10 , TimeUnit . SECONDS ) , "Server handled request" ) ;
ArgumentCaptor < TbMsg > msgCaptor = ArgumentCaptor . forClass ( TbMsg . class ) ;
ArgumentCaptor < TbMsgMetaData > metadataCaptor = ArgumentCaptor . forClass ( TbMsgMetaData . class ) ;
ArgumentCaptor < String > dataCaptor = ArgumentCaptor . forClass ( String . class ) ;
verify ( ctx , timeout ( 10_000 ) ) . transformMsg ( msgCaptor . capture ( ) , metadataCaptor . capture ( ) , dataCaptor . capture ( ) ) ;
assertEquals ( "{\"temperature\":25}" , capturedBody [ 0 ] ) ;
}
private void setupServerWithBodyCapture ( String [ ] capturedBody , CountDownLatch latch ) throws IOException {
setupServer ( "*" , ( request , response , _ ) - > {
try {
if ( request instanceof org . apache . http . HttpEntityEnclosingRequest entityRequest ) {
InputStream is = entityRequest . getEntity ( ) . getContent ( ) ;
capturedBody [ 0 ] = new String ( is . readAllBytes ( ) , StandardCharsets . UTF_8 ) ;
}
response . setStatusCode ( 200 ) ;
latch . countDown ( ) ;
} catch ( Exception e ) {
e . printStackTrace ( ) ;
latch . countDown ( ) ;
}
} ) ;
}
private static Stream < Arguments > givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig ( ) {
return Stream . of (
Arguments . of ( 0 ,
Andrii Landiak
6 months ago