code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.destination.cassandra;
  
  import com.fasterxml.jackson.databind.JsonNode;
  import io.airbyte.commons.json.Jsons;
  import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
  import io.airbyte.integrations.util.HostPortResolver;
  import java.util.Comparator;
  import java.util.List;
  import java.util.stream.Collectors;
  import org.junit.jupiter.api.BeforeAll;
  
  public class CassandraDestinationAcceptanceTest extends DestinationAcceptanceTest {
  
    private JsonNode configJson;
  
    private CassandraCqlProvider cassandraCqlProvider;
  
    private CassandraNameTransformer cassandraNameTransformer;
  
    private static CassandraContainerInitializr.ConfiguredCassandraContainer cassandraContainer;
  
    @BeforeAll
    static void initContainer() {
      cassandraContainer = CassandraContainerInitializr.initContainer();
    }
  
    @Override
    protected void setup(TestDestinationEnv testEnv) {
      configJson = TestDataFactory.createJsonConfig(
          cassandraContainer.getUsername(),
          cassandraContainer.getPassword(),
          HostPortResolver.resolveHost(cassandraContainer),
          HostPortResolver.resolvePort(cassandraContainer));
      var cassandraConfig = new CassandraConfig(configJson);
      cassandraCqlProvider = new CassandraCqlProvider(cassandraConfig);
      cassandraNameTransformer = new CassandraNameTransformer(cassandraConfig);
    }
  
    @Override
    protected void tearDown(TestDestinationEnv testEnv) {
      cassandraCqlProvider.retrieveMetadata().forEach(meta -> {
        var keyspace = meta.value1();
        meta.value2().forEach(table -> cassandraCqlProvider.truncate(keyspace, table));
      });
    }
  
    @Override
    protected String getImageName() {
      return "airbyte/destination-cassandra:dev";
    }
  
    @Override
    protected JsonNode getConfig() {
      return configJson;
    }
  
    @Override
    protected boolean implementsNamespaces() {
      return true;
    }
  
    @Override
    protected JsonNode getFailCheckConfig() {
      return TestDataFactory.createJsonConfig(
          "usr",
          "pw",
          "127.0.192.1",
          8080);
    }
  
    @Override
    protected List<JsonNode> retrieveRecords(TestDestinationEnv testEnv,
                                             String streamName,
                                             String namespace,
                                             JsonNode streamSchema) {
      var keyspace = cassandraNameTransformer.outputKeyspace(namespace);
      var table = cassandraNameTransformer.outputTable(streamName);
      return cassandraCqlProvider.select(keyspace, table).stream()
          .sorted(Comparator.comparing(CassandraRecord::getTimestamp))
          .map(CassandraRecord::getData)
          .map(Jsons::deserialize)
          .collect(Collectors.toList());
    }
  
  }
  