tests/itests-salesforce/src/test/java/org/apache/camel/kafkaconnector/salesforce/sink/CamelSinkSalesforceITCase.java

/*
 * 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.kafkaconnector.salesforce.sink;

import java.io.IOException;
import java.util.concurrent.ExecutionException;

import org.apache.camel.kafkaconnector.common.AbstractKafkaTest;
import org.apache.camel.kafkaconnector.common.ConnectorPropertyFactory;
import org.apache.camel.kafkaconnector.common.clients.kafka.KafkaClient;
import org.apache.camel.kafkaconnector.salesforce.clients.SalesforceCliContainer;
import org.apache.camel.kafkaconnector.salesforce.clients.SfdxCommand;
import org.apache.camel.test.infra.common.TestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Container.ExecResult;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

import static org.apache.camel.kafkaconnector.salesforce.clients.SalesforceCliContainer.verifyCommand;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;


/* This test is disabled by default because requires setup on Salesforce end.

Creating API keys:
https://help.salesforce.com/articleView?id=connected_app_create_api_integration.htm

You need to set the following system properties to run this test:
-Dit.test.salesforce.enable=true to enable the test
-Dit.test.salesforce.client.id=<client ID>
-Dit.test.salesforce.client.secret=<client secret>
-Dit.test.salesforce.password=<user password>
-Dit.test.salesforce.username=<user name>
-Dit.test.salesforce.sfdx.path=/path/to/sfdx

The it.test.salesforce.sfdx.path property should point to the directory containing the sfdx
CLI client configuration. This can be generated using the following steps:

1. Run the Salesforce CLI container:
docker run --rm --name salesforce-cli -it -v /path/to/sfdx:/root/.sfdx salesforce/salesforcedx

2. Within the container, use the following command to login:
sfdx force:auth:device:login -s -d -i <client ID>

3. Provide the client secret when request and execute the steps requested by the CLI.

4. Verify that you are logged in correctly using the following command
sfdx force:auth:list

It should present an output like:

#### authenticated orgs
ALIAS  USERNAME              ORG ID              INSTANCE URL                 OAUTH METHOD
─────  ────────────────────  ──────────────────  ───────────────────────────  ────────────
       angusyoung@gmail.com  SOME NUMERIC ID     https://eu31.salesforce.com  web


Note: after leaving the container you might need to adjust the permissions of the directory
containing the sfdx configuration files (/path/to/sfdx).
*/
@Testcontainers
@EnabledIfSystemProperty(named = "it.test.salesforce.enable", matches = "true")
public class CamelSinkSalesforceITCase extends AbstractKafkaTest {
    private static final Logger LOG = LoggerFactory.getLogger(CamelSinkSalesforceITCase.class);

    @Container
    public final SalesforceCliContainer container = new SalesforceCliContainer();

    private final String clientId = System.getProperty("it.test.salesforce.client.id");
    private final String clientSecret = System.getProperty("it.test.salesforce.client.secret");
    private final String password = System.getProperty("it.test.salesforce.password");
    private final String userName = System.getProperty("it.test.salesforce.username");

    private String accountName;
    private boolean recordCreated;
    private String topicName;

    @Override
    protected String[] getConnectorsInTest() {
        return new String[] {"camel-salesforce-kafka-connector"};
    }

    @BeforeEach
    public void setUp() {
        accountName = "TestSinkAccount" + TestUtils.randomWithRange(1, 100);
        topicName = getTopicForTest(this);
    }

    @AfterEach
    public void tearDown() throws IOException, InterruptedException {
        SfdxCommand sfdxCommand = SfdxCommand.forceDataRecordDelete()
                .withArgument("-u", userName)
                .withArgument("--sobjecttype", "Account")
                .withArgument("--where", String.format("Name=%s", accountName));

        LOG.debug("Deleting the test account {}", accountName);
        ExecResult result = container.execCommand(sfdxCommand);
        if (!verifyCommand(sfdxCommand, result)) {
            fail("Unable to delete the test account on Salesforce");
        }

        accountName = null;
    }


    private boolean waitForRecordCreation() {
        SfdxCommand sfdxCommand = SfdxCommand.forceDataRecordGet()
                .withArgument("-u", userName)
                .withArgument("--sobjecttype", "Account")
                .withArgument("--where", String.format("Name=%s", accountName));

        LOG.debug("Check if the test account {} was created on Salesforce", accountName);
        try {
            ExecResult result = container.execCommand(sfdxCommand);

            if (verifyCommand(sfdxCommand, result)) {
                recordCreated = true;
                return true;
            }

        } catch (IOException e) {
            LOG.warn("I/O exception while checking if the record was created: {}", e.getMessage(), e);

            return false;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();

            LOG.warn("The thread was interrupted while waiting for the record creation");
            return false;
        }

        return false;
    }


    private void runTest(ConnectorPropertyFactory connectorPropertyFactory) throws ExecutionException, InterruptedException {
        connectorPropertyFactory.log();
        getKafkaConnectService().initializeConnectorBlocking(connectorPropertyFactory, 1);

        KafkaClient<String, String> kafkaClient = new KafkaClient<>(getKafkaService().getBootstrapServers());

        // Ideally we should use the DTOs, but they cause the source check to fail
        String data = String.format("{\"attributes\":{\"referenceId\":null,\"type\":\"Account\",\"url\":null},"
                + "\"Description\":\"%s\",\"Name\":\"%s\"}", "Created during sink test", accountName);

        LOG.info("Sending new account {}", data);

        kafkaClient.produce(topicName, data);
    }

    @Test
    @Timeout(180)
    public void testBasicProduce() throws ExecutionException, InterruptedException {
        ConnectorPropertyFactory factory = CamelSalesforcePropertyFactory.basic()
                .withKafkaTopic(topicName)
                .withUserName(userName)
                .withPassword(password)
                .withClientId(clientId)
                .withClientSecret(clientSecret)
                .withRawPayload(true)
                .withPackages("org.apache.camel.salesforce.dto")
                .withSObjectName("Account")
                .withOperationName("createSObject");

        runTest(factory);

        TestUtils.waitFor(this::waitForRecordCreation);
        assertTrue(recordCreated, "The record was not created");
    }
}