| /* |
| * 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.camel.component.salesforce; |
| |
| import java.time.ZonedDateTime; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.Map; |
| import java.util.Objects; |
| |
| import com.fasterxml.jackson.annotation.JsonProperty; |
| import org.apache.camel.AsyncProcessor; |
| import org.apache.camel.CamelContext; |
| import org.apache.camel.Exchange; |
| import org.apache.camel.component.salesforce.api.dto.PlatformEvent; |
| import org.apache.camel.component.salesforce.internal.streaming.SubscriptionHelper; |
| import org.apache.camel.spi.ClassResolver; |
| import org.cometd.bayeux.Message; |
| import org.cometd.bayeux.client.ClientSessionChannel; |
| import org.cometd.common.HashMapMessage; |
| import org.junit.Before; |
| import org.junit.Test; |
| import org.junit.runner.RunWith; |
| import org.mockito.Mock; |
| import org.mockito.junit.MockitoJUnitRunner; |
| |
| import static org.mockito.ArgumentMatchers.any; |
| import static org.mockito.ArgumentMatchers.eq; |
| import static org.mockito.ArgumentMatchers.same; |
| import static org.mockito.Mockito.mock; |
| import static org.mockito.Mockito.never; |
| import static org.mockito.Mockito.verify; |
| import static org.mockito.Mockito.verifyNoMoreInteractions; |
| import static org.mockito.Mockito.when; |
| |
| @RunWith(MockitoJUnitRunner.class) |
| public class SalesforceConsumerTest { |
| |
| public static class AccountUpdates { |
| @JsonProperty("Id") |
| String id; |
| |
| @JsonProperty("Name") |
| String name; |
| |
| @JsonProperty("Phone") |
| String phone; |
| |
| @Override |
| public boolean equals(Object obj) { |
| if (!(obj instanceof AccountUpdates)) { |
| return false; |
| } |
| |
| final AccountUpdates other = (AccountUpdates)obj; |
| |
| return Objects.equals(id, other.id) && Objects.equals(name, other.name) && Objects.equals(phone, other.phone); |
| } |
| |
| @Override |
| public int hashCode() { |
| return Objects.hash(id, name, phone); |
| } |
| |
| } |
| |
| static final SubscriptionHelper NOT_USED = null; |
| |
| SalesforceEndpointConfig configuration = new SalesforceEndpointConfig(); |
| SalesforceEndpoint endpoint = mock(SalesforceEndpoint.class); |
| Exchange exchange = mock(Exchange.class); |
| org.apache.camel.Message in = mock(org.apache.camel.Message.class); |
| AsyncProcessor processor = mock(AsyncProcessor.class); |
| Message pushTopicMessage; |
| |
| @Mock |
| private Message mockChangeEvent; |
| @Mock |
| private Map<String, Object> mockChangeEventPayload; |
| @Mock |
| private Map<String, Object> mockChangeEventData; |
| @Mock |
| private Map<String, Object> mockChangeEventMap; |
| |
| @Before |
| public void setupMocks() { |
| when(endpoint.getConfiguration()).thenReturn(configuration); |
| when(endpoint.createExchange()).thenReturn(exchange); |
| when(exchange.getIn()).thenReturn(in); |
| final SalesforceComponent component = mock(SalesforceComponent.class); |
| when(endpoint.getComponent()).thenReturn(component); |
| final CamelContext camelContext = mock(CamelContext.class); |
| when(component.getCamelContext()).thenReturn(camelContext); |
| final ClassResolver classResolver = mock(ClassResolver.class); |
| when(camelContext.getClassResolver()).thenReturn(classResolver); |
| when(classResolver.resolveClass(AccountUpdates.class.getName())).thenReturn((Class)AccountUpdates.class); |
| |
| pushTopicMessage = createPushTopicMessage(); |
| |
| setupMockChangeEvent(); |
| } |
| |
| private void setupMockChangeEvent() { |
| final Map<String, Object> changeEventHeader = new HashMap<>(); |
| changeEventHeader.put("changeType", "CREATE"); |
| changeEventHeader.put("changeOrigin", "com/salesforce/api/rest/45.0"); |
| changeEventHeader.put("transactionKey", "000bc577-90c7-0d33-cebb-971bb50d75b8"); |
| changeEventHeader.put("sequenceNumber", 1L); |
| changeEventHeader.put("isTransactionEnd", Boolean.TRUE); |
| changeEventHeader.put("commitTimestamp", 1558949299000L); |
| changeEventHeader.put("commitUser", "0052p000009cl8uBBB"); |
| changeEventHeader.put("commitNumber", 10585193272713L); |
| changeEventHeader.put("entityName", "Account"); |
| changeEventHeader.put("recordIds", new Object[] {"0012p00002HHpNlAAL"}); |
| |
| when(mockChangeEventPayload.get("ChangeEventHeader")).thenReturn(changeEventHeader); |
| |
| when(mockChangeEventMap.get("replayId")).thenReturn(4L); |
| |
| when(mockChangeEventData.get("schema")).thenReturn("30H2pgzuWcF844p26Ityvg"); |
| when(mockChangeEventData.get("payload")).thenReturn(mockChangeEventPayload); |
| when(mockChangeEventData.get("event")).thenReturn(mockChangeEventMap); |
| |
| when(mockChangeEvent.getDataAsMap()).thenReturn(mockChangeEventData); |
| when(mockChangeEvent.getChannel()).thenReturn("/data/AccountChangeEvent"); |
| } |
| |
| @Test |
| public void shouldProcessMappedPayloadPushTopicMessages() throws Exception { |
| when(endpoint.getTopicName()).thenReturn("AccountUpdates"); |
| configuration.setSObjectClass(AccountUpdates.class.getName()); |
| |
| final SalesforceConsumer consumer = new SalesforceConsumer(endpoint, processor, NOT_USED); |
| |
| consumer.processMessage(mock(ClientSessionChannel.class), pushTopicMessage); |
| |
| final AccountUpdates accountUpdates = new AccountUpdates(); |
| accountUpdates.phone = "(415) 555-1212"; |
| accountUpdates.id = "001D000000KneakIAB"; |
| accountUpdates.name = "Blackbeard"; |
| |
| verify(in).setBody(accountUpdates); |
| verify(in).setHeader("CamelSalesforceEventType", "created"); |
| verify(in).setHeader("CamelSalesforceCreatedDate", "2016-09-16T19:45:27.454Z"); |
| verify(in).setHeader("CamelSalesforceReplayId", 1L); |
| verify(in).setHeader("CamelSalesforceTopicName", "AccountUpdates"); |
| verify(in).setHeader("CamelSalesforceChannel", "/topic/AccountUpdates"); |
| verify(in).setHeader("CamelSalesforceClientId", "lxdl9o32njygi1gj47kgfaga4k"); |
| |
| verify(processor).process(same(exchange), any()); |
| } |
| |
| @Test |
| public void shouldProcessPlatformEvents() throws Exception { |
| when(endpoint.getTopicName()).thenReturn("/event/TestEvent__e"); |
| |
| final Message message = new HashMapMessage(); |
| final Map<String, Object> data = new HashMap<>(); |
| data.put("schema", "30H2pgzuWcF844p26Ityvg"); |
| |
| final Map<String, Object> payload = new HashMap<>(); |
| payload.put("Test_Field__c", "abc"); |
| payload.put("CreatedById", "00541000002WYFpAAO"); |
| payload.put("CreatedDate", "2018-07-06T12:41:04Z"); |
| data.put("payload", payload); |
| data.put("event", Collections.singletonMap("replayId", 4L)); |
| message.put("data", data); |
| message.put("channel", "/event/TestEvent__e"); |
| |
| final SalesforceConsumer consumer = new SalesforceConsumer(endpoint, processor, NOT_USED); |
| |
| consumer.processMessage(mock(ClientSessionChannel.class), message); |
| |
| final ZonedDateTime created = ZonedDateTime.parse("2018-07-06T12:41:04Z"); |
| final PlatformEvent event = new PlatformEvent(created, "00541000002WYFpAAO"); |
| event.set("Test_Field__c", "abc"); |
| verify(in).setBody(event); |
| verify(in).setHeader("CamelSalesforceCreatedDate", created); |
| verify(in).setHeader("CamelSalesforceReplayId", 4L); |
| verify(in).setHeader("CamelSalesforceChannel", "/event/TestEvent__e"); |
| verify(in).setHeader("CamelSalesforceEventType", "TestEvent__e"); |
| verify(in).setHeader("CamelSalesforcePlatformEventSchema", "30H2pgzuWcF844p26Ityvg"); |
| |
| verify(processor).process(same(exchange), any()); |
| |
| verifyNoMoreInteractions(in, processor); |
| } |
| |
| @Test |
| public void shouldProcessPushTopicMessages() throws Exception { |
| when(endpoint.getTopicName()).thenReturn("AccountUpdates"); |
| |
| final SalesforceConsumer consumer = new SalesforceConsumer(endpoint, processor, NOT_USED); |
| |
| consumer.processMessage(mock(ClientSessionChannel.class), pushTopicMessage); |
| |
| @SuppressWarnings("unchecked") |
| final Object sobject = ((Map<String, Object>)pushTopicMessage.get("data")).get("sobject"); |
| verify(in).setBody(sobject); |
| verify(in).setHeader("CamelSalesforceEventType", "created"); |
| verify(in).setHeader("CamelSalesforceCreatedDate", "2016-09-16T19:45:27.454Z"); |
| verify(in).setHeader("CamelSalesforceReplayId", 1L); |
| verify(in).setHeader("CamelSalesforceTopicName", "AccountUpdates"); |
| verify(in).setHeader("CamelSalesforceChannel", "/topic/AccountUpdates"); |
| verify(in).setHeader("CamelSalesforceClientId", "lxdl9o32njygi1gj47kgfaga4k"); |
| |
| verify(processor).process(same(exchange), any()); |
| } |
| |
| @Test |
| public void shouldProcessRawPayloadPushTopicMessages() throws Exception { |
| when(endpoint.getTopicName()).thenReturn("AccountUpdates"); |
| configuration.setRawPayload(true); |
| |
| final SalesforceConsumer consumer = new SalesforceConsumer(endpoint, processor, NOT_USED); |
| |
| consumer.processMessage(mock(ClientSessionChannel.class), pushTopicMessage); |
| |
| verify(in).setBody("{\"Phone\":\"(415) 555-1212\",\"Id\":\"001D000000KneakIAB\",\"Name\":\"Blackbeard\"}"); |
| verify(in).setHeader("CamelSalesforceEventType", "created"); |
| verify(in).setHeader("CamelSalesforceCreatedDate", "2016-09-16T19:45:27.454Z"); |
| verify(in).setHeader("CamelSalesforceReplayId", 1L); |
| verify(in).setHeader("CamelSalesforceTopicName", "AccountUpdates"); |
| verify(in).setHeader("CamelSalesforceChannel", "/topic/AccountUpdates"); |
| verify(in).setHeader("CamelSalesforceClientId", "lxdl9o32njygi1gj47kgfaga4k"); |
| |
| verify(processor).process(same(exchange), any()); |
| } |
| |
| @Test |
| public void shouldProcessRawPlatformEvents() throws Exception { |
| when(endpoint.getTopicName()).thenReturn("/event/TestEvent__e"); |
| configuration.setRawPayload(true); |
| |
| final Message message = new HashMapMessage(); |
| final Map<String, Object> data = new HashMap<>(); |
| data.put("schema", "30H2pgzuWcF844p26Ityvg"); |
| |
| final Map<String, Object> payload = new HashMap<>(); |
| payload.put("Test_Field__c", "abc"); |
| payload.put("CreatedById", "00541000002WYFpAAO"); |
| payload.put("CreatedDate", "2018-07-06T12:41:04Z"); |
| data.put("payload", payload); |
| data.put("event", Collections.singletonMap("replayId", 4L)); |
| message.put("data", data); |
| message.put("channel", "/event/TestEvent__e"); |
| |
| final SalesforceConsumer consumer = new SalesforceConsumer(endpoint, processor, NOT_USED); |
| |
| consumer.processMessage(mock(ClientSessionChannel.class), message); |
| |
| verify(in).setBody(message); |
| verify(in).setHeader("CamelSalesforceCreatedDate", ZonedDateTime.parse("2018-07-06T12:41:04Z")); |
| verify(in).setHeader("CamelSalesforceReplayId", 4L); |
| verify(in).setHeader("CamelSalesforceChannel", "/event/TestEvent__e"); |
| verify(in).setHeader("CamelSalesforceEventType", "TestEvent__e"); |
| verify(in).setHeader("CamelSalesforcePlatformEventSchema", "30H2pgzuWcF844p26Ityvg"); |
| |
| verify(processor).process(same(exchange), any()); |
| |
| verifyNoMoreInteractions(in, processor); |
| } |
| |
| @Test |
| public void shouldProcessChangeEvents() throws Exception { |
| when(endpoint.getTopicName()).thenReturn("/data/AccountChangeEvent"); |
| |
| final SalesforceConsumer consumer = new SalesforceConsumer(endpoint, processor, NOT_USED); |
| |
| consumer.processMessage(mock(ClientSessionChannel.class), mockChangeEvent); |
| |
| verify(in).setBody(mockChangeEventPayload); |
| verify(in).setHeader("CamelSalesforceChannel", "/data/AccountChangeEvent"); |
| verify(in).setHeader("CamelSalesforceReplayId", 4L); |
| verify(in).setHeader("CamelSalesforceChangeEventSchema", "30H2pgzuWcF844p26Ityvg"); |
| verify(in).setHeader("CamelSalesforceEventType", "AccountChangeEvent"); |
| verify(in).setHeader("CamelSalesforceChangeType", "CREATE"); |
| verify(in).setHeader("CamelSalesforceChangeOrigin", "com/salesforce/api/rest/45.0"); |
| verify(in).setHeader("CamelSalesforceTransactionKey", "000bc577-90c7-0d33-cebb-971bb50d75b8"); |
| verify(in).setHeader("CamelSalesforceSequenceNumber", 1L); |
| verify(in).setHeader("CamelSalesforceIsTransactionEnd", Boolean.TRUE); |
| verify(in).setHeader("CamelSalesforceCommitTimestamp", 1558949299000L); |
| verify(in).setHeader("CamelSalesforceCommitUser", "0052p000009cl8uBBB"); |
| verify(in).setHeader("CamelSalesforceCommitNumber", 10585193272713L); |
| verify(in).setHeader("CamelSalesforceEntityName", "Account"); |
| verify(in).setHeader("CamelSalesforceRecordIds", new Object[] {"0012p00002HHpNlAAL"}); |
| |
| verify(mockChangeEventPayload).remove("ChangeEventHeader"); |
| |
| verify(processor).process(same(exchange), any()); |
| |
| verifyNoMoreInteractions(in, processor); |
| } |
| |
| @Test |
| public void processNoReplayIdChangeEventsShouldNotSetReplayIdHeader() throws Exception { |
| when(endpoint.getTopicName()).thenReturn("/data/AccountChangeEvent"); |
| when(mockChangeEventMap.get("replayId")).thenReturn(null); |
| |
| final SalesforceConsumer consumer = new SalesforceConsumer(endpoint, processor, NOT_USED); |
| |
| consumer.processMessage(mock(ClientSessionChannel.class), mockChangeEvent); |
| |
| verify(in, never()).setHeader(eq("CamelSalesforceReplayId"), any()); |
| } |
| |
| @Test |
| public void processRawPayloadChangeEventsShouldSetInputMessageAsBody() throws Exception { |
| when(endpoint.getTopicName()).thenReturn("/data/AccountChangeEvent"); |
| configuration.setRawPayload(true); |
| |
| final SalesforceConsumer consumer = new SalesforceConsumer(endpoint, processor, NOT_USED); |
| |
| consumer.processMessage(mock(ClientSessionChannel.class), mockChangeEvent); |
| |
| verify(in).setBody(mockChangeEvent); |
| } |
| |
| static Message createPushTopicMessage() { |
| final Message pushTopicMessage = new HashMapMessage(); |
| pushTopicMessage.put("clientId", "lxdl9o32njygi1gj47kgfaga4k"); |
| |
| final Map<String, Object> data = new HashMap<>(); |
| pushTopicMessage.put("data", data); |
| |
| final Map<String, Object> event = new HashMap<>(); |
| data.put("event", event); |
| |
| event.put("createdDate", "2016-09-16T19:45:27.454Z"); |
| event.put("replayId", 1L); |
| event.put("type", "created"); |
| |
| final Map<String, Object> sobject = new HashMap<>(); |
| data.put("sobject", sobject); |
| |
| sobject.put("Phone", "(415) 555-1212"); |
| sobject.put("Id", "001D000000KneakIAB"); |
| sobject.put("Name", "Blackbeard"); |
| |
| pushTopicMessage.put("channel", "/topic/AccountUpdates"); |
| return pushTopicMessage; |
| } |
| } |
