code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.workers.internal;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.ArgumentMatchers.anyString;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.verify;
  import static org.mockito.Mockito.verifyNoInteractions;
  import static org.mockito.Mockito.verifyNoMoreInteractions;
  import static org.mockito.Mockito.when;
  
  import io.airbyte.commons.json.Jsons;
  import io.airbyte.commons.logging.MdcScope.Builder;
  import io.airbyte.protocol.models.AirbyteLogMessage;
  import io.airbyte.protocol.models.AirbyteMessage;
  import java.io.BufferedReader;
  import java.io.ByteArrayInputStream;
  import java.io.InputStream;
  import java.io.InputStreamReader;
  import java.nio.charset.StandardCharsets;
  import java.util.Collections;
  import java.util.stream.Collectors;
  import java.util.stream.Stream;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Disabled;
  import org.junit.jupiter.api.Test;
  import org.slf4j.Logger;
  
  class DefaultAirbyteStreamFactoryTest {
  
    private static final String STREAM_NAME = "user_preferences";
    private static final String FIELD_NAME = "favorite_color";
  
    private AirbyteProtocolPredicate protocolPredicate;
    private Logger logger;
  
    @BeforeEach
    void setup() {
      protocolPredicate = mock(AirbyteProtocolPredicate.class);
      when(protocolPredicate.test(any())).thenReturn(true);
      logger = mock(Logger.class);
    }
  
    @Test
    void testValid() {
      final AirbyteMessage record1 = AirbyteMessageUtils.createRecordMessage(STREAM_NAME, FIELD_NAME, "green");
  
      final Stream<AirbyteMessage> messageStream = stringToMessageStream(Jsons.serialize(record1));
      final Stream<AirbyteMessage> expectedStream = Stream.of(record1);
  
      assertEquals(expectedStream.collect(Collectors.toList()), messageStream.collect(Collectors.toList()));
      verifyNoInteractions(logger);
    }
  
    @Test
    void testLoggingLine() {
      final String invalidRecord = "invalid line";
  
      final Stream<AirbyteMessage> messageStream = stringToMessageStream(invalidRecord);
  
      assertEquals(Collections.emptyList(), messageStream.collect(Collectors.toList()));
      verify(logger).info(anyString());
      verifyNoMoreInteractions(logger);
    }
  
    @Test
    void testLoggingLevel() {
      final AirbyteMessage logMessage = AirbyteMessageUtils.createLogMessage(AirbyteLogMessage.Level.WARN, "warning");
  
      final Stream<AirbyteMessage> messageStream = stringToMessageStream(Jsons.serialize(logMessage));
  
      assertEquals(Collections.emptyList(), messageStream.collect(Collectors.toList()));
      verify(logger).warn("warning");
      verifyNoMoreInteractions(logger);
    }
  
    @Test
    void testFailValidation() {
      final String invalidRecord = "{ \"fish\": \"tuna\"}";
  
      when(protocolPredicate.test(Jsons.deserialize(invalidRecord))).thenReturn(false);
  
      final Stream<AirbyteMessage> messageStream = stringToMessageStream(invalidRecord);
  
      assertEquals(Collections.emptyList(), messageStream.collect(Collectors.toList()));
      verify(logger).error(anyString(), anyString());
      verifyNoMoreInteractions(logger);
    }
  
    @Test
    void testFailDeserialization() {
      final String invalidRecord = "{ \"type\": \"abc\"}";
  
      when(protocolPredicate.test(Jsons.deserialize(invalidRecord))).thenReturn(true);
  
      final Stream<AirbyteMessage> messageStream = stringToMessageStream(invalidRecord);
  
      assertEquals(Collections.emptyList(), messageStream.collect(Collectors.toList()));
      verify(logger).error(anyString(), anyString());
      verifyNoMoreInteractions(logger);
    }
  
    @Test
    @Disabled
    void testMissingNewLineBetweenValidRecords() {
      final AirbyteMessage record1 = AirbyteMessageUtils.createRecordMessage(STREAM_NAME, FIELD_NAME, "green");
      final AirbyteMessage record2 = AirbyteMessageUtils.createRecordMessage(STREAM_NAME, FIELD_NAME, "yellow");
  
      final String inputString = Jsons.serialize(record1) + Jsons.serialize(record2);
  
      final Stream<AirbyteMessage> messageStream = stringToMessageStream(inputString);
  
      assertEquals(Collections.emptyList(), messageStream.collect(Collectors.toList()));
      verify(logger).error(anyString(), anyString());
      verifyNoMoreInteractions(logger);
    }
  
    private Stream<AirbyteMessage> stringToMessageStream(final String inputString) {
      final InputStream inputStream = new ByteArrayInputStream(inputString.getBytes(StandardCharsets.UTF_8));
      final BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
      return new DefaultAirbyteStreamFactory(protocolPredicate, logger, new Builder()).create(bufferedReader);
    }
  
  }
  