content: package co.airy.core.sources.twilio;
  
  import co.airy.avro.communication.DeliveryState;
  import co.airy.avro.communication.Message;
  import co.airy.core.sources.twilio.services.Api;
  import co.airy.kafka.schema.Topic;
  import co.airy.kafka.schema.application.ApplicationCommunicationChannels;
  import co.airy.kafka.schema.application.ApplicationCommunicationMessages;
  import co.airy.kafka.test.KafkaTestHelper;
  import co.airy.kafka.test.junit.SharedKafkaTestResource;
  import co.airy.spring.core.AirySpringBootApplication;
  import co.airy.spring.test.WebTestHelper;
  import com.fasterxml.jackson.databind.JsonNode;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import org.apache.kafka.clients.producer.ProducerRecord;
  import org.junit.jupiter.api.AfterAll;
  import org.junit.jupiter.api.BeforeAll;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.extension.ExtendWith;
  import org.junit.jupiter.api.extension.RegisterExtension;
  import org.mockito.ArgumentCaptor;
  import org.mockito.InjectMocks;
  import org.mockito.MockitoAnnotations;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
  import org.springframework.boot.test.context.SpringBootTest;
  import org.springframework.boot.test.mock.mockito.MockBean;
  import org.springframework.test.context.TestPropertySource;
  import org.springframework.test.context.junit.jupiter.SpringExtension;
  
  import java.time.Instant;
  import java.util.UUID;
  import java.util.concurrent.TimeUnit;
  
  import static co.airy.test.Timing.retryOnException;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.mockito.Mockito.doNothing;
  import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
  
  @SpringBootTest(classes = AirySpringBootApplication.class)
  @TestPropertySource(value = "classpath:test.properties")
  @AutoConfigureMockMvc
  @ExtendWith(SpringExtension.class)
  class CreateConversationTest {
  
      @RegisterExtension
      public static final SharedKafkaTestResource sharedKafkaTestResource = new SharedKafkaTestResource();
      private static KafkaTestHelper kafkaTestHelper;
  
      private static final Topic applicationCommunicationChannels = new ApplicationCommunicationChannels();
      private static final Topic applicationCommunicationMessages = new ApplicationCommunicationMessages();
  
      @MockBean
      Api api;
  
      @Autowired
      @InjectMocks
      private Connector worker;
  
      @Autowired
      private WebTestHelper webTestHelper;
  
      @Autowired
      private Stores stores;
  
      @BeforeAll
      static void beforeAll() throws Exception {
          kafkaTestHelper = new KafkaTestHelper(sharedKafkaTestResource,
                  applicationCommunicationChannels,
                  applicationCommunicationMessages
          );
  
          kafkaTestHelper.beforeAll();
      }
  
      @AfterAll
      static void afterAll() throws Exception {
          kafkaTestHelper.afterAll();
      }
  
      @BeforeEach
      void beforeEach() throws InterruptedException {
          MockitoAnnotations.openMocks(this);
          webTestHelper.waitUntilHealthy();
      }
  
      @Test
      void canCreateConversation() throws Exception {
          final String conversationId = UUID.randomUUID().toString();
          final String messageId = UUID.randomUUID().toString();
          final String sourceRecipientId = "+4912345678";
          final String sourceChannelId = "+4987654321";
          final String payload = "{\"Body\":\"Hello World\"}";
  
          ArgumentCaptor<String> payloadCaptor = ArgumentCaptor.forClass(String.class);
          ArgumentCaptor<String> fromCaptor = ArgumentCaptor.forClass(String.class);
          ArgumentCaptor<String> toCaptor = ArgumentCaptor.forClass(String.class);
  
          doNothing().when(api).sendMessage(fromCaptor.capture(), toCaptor.capture(), payloadCaptor.capture());
  
          // Test that phone number input gets cleaned up
          final String channelPayload = "{\"phone_number\":\"+49 8765 4321 \",\"name\":\"my channel\"}";
          final String response = webTestHelper.post("/channels.twilio.sms.connect", channelPayload)
                  .andExpect(status().isOk())
                  .andReturn().getResponse().getContentAsString();
          final JsonNode jsonNode = new ObjectMapper().readTree(response);
          final String channelId = jsonNode.get("id").textValue();
  
          TimeUnit.SECONDS.sleep(5);
  
          kafkaTestHelper.produceRecord(new ProducerRecord<>(applicationCommunicationMessages.name(), messageId,
                  Message.newBuilder()
                          .setId(messageId)
                          .setSenderId("sender-id")
                          .setSentAt(Instant.now().toEpochMilli())
                          .setSourceRecipientId(sourceRecipientId)
                          .setIsFromContact(false)
                          .setDeliveryState(DeliveryState.PENDING)
                          .setConversationId(conversationId)
                          .setChannelId(channelId)
                          .setSource("twilio.sms")
                          .setContent(payload)
                          .build())
          );
  
          retryOnException(() -> {
              assertEquals(payload, payloadCaptor.getValue());
              assertEquals(sourceRecipientId, toCaptor.getValue());
              assertEquals(sourceChannelId, fromCaptor.getValue());
          }, "Twilio API was not called");
      }
  }
  