code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.source.postgres;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  import com.fasterxml.jackson.databind.JsonNode;
  import io.airbyte.commons.json.Jsons;
  import io.airbyte.protocol.models.AirbyteStream;
  import io.airbyte.protocol.models.SyncMode;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import org.junit.jupiter.api.Test;
  
  class PostgresCdcCatalogHelperTest {
  
    @Test
    public void testRemoveIncrementalWithoutPk() {
      // no primary key
      final AirbyteStream noPrimaryKeyNoIncremental = new AirbyteStream()
          .withSourceDefinedPrimaryKey(Collections.emptyList())
          .withSupportedSyncModes(List.of(SyncMode.FULL_REFRESH));
      final AirbyteStream noPrimaryKeyWithIncremental = new AirbyteStream()
          .withSourceDefinedPrimaryKey(Collections.emptyList())
          .withSupportedSyncModes(List.of(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL));
  
      assertEquals(noPrimaryKeyNoIncremental,
          PostgresCdcCatalogHelper.removeIncrementalWithoutPk(Jsons.clone(noPrimaryKeyWithIncremental)));
      assertEquals(noPrimaryKeyNoIncremental,
          PostgresCdcCatalogHelper.removeIncrementalWithoutPk(Jsons.clone(noPrimaryKeyNoIncremental)));
  
      // with primary key
      final AirbyteStream withPrimaryKeyNoIncremental = new AirbyteStream()
          .withSourceDefinedPrimaryKey(List.of(List.of("id")))
          .withSupportedSyncModes(List.of(SyncMode.FULL_REFRESH));
      final AirbyteStream withPrimaryKeyWithIncremental = new AirbyteStream()
          .withSourceDefinedPrimaryKey(List.of(List.of("id")))
          .withSupportedSyncModes(List.of(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL));
  
      assertEquals(withPrimaryKeyWithIncremental,
          PostgresCdcCatalogHelper.removeIncrementalWithoutPk(Jsons.clone(withPrimaryKeyWithIncremental)));
      assertEquals(withPrimaryKeyNoIncremental,
          PostgresCdcCatalogHelper.removeIncrementalWithoutPk(Jsons.clone(withPrimaryKeyNoIncremental)));
    }
  
    @Test
    public void testSetIncrementalToSourceDefined() {
      final AirbyteStream withIncremental = new AirbyteStream()
          .withSourceDefinedCursor(false)
          .withSupportedSyncModes(List.of(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL));
      assertTrue(PostgresCdcCatalogHelper
          .setIncrementalToSourceDefined(withIncremental)
          .getSourceDefinedCursor());
  
      final AirbyteStream noIncremental = new AirbyteStream()
          .withSourceDefinedCursor(false)
          .withSupportedSyncModes(List.of(SyncMode.FULL_REFRESH));
      assertFalse(PostgresCdcCatalogHelper
          .setIncrementalToSourceDefined(noIncremental)
          .getSourceDefinedCursor());
    }
  
    @Test
    public void testAddCdcMetadataColumns() {
      final AirbyteStream before = new AirbyteStream()
          .withJsonSchema(Jsons.jsonNode(Map.of("properties", Jsons.deserialize("{}"))));
  
      final JsonNode properties = Jsons.jsonNode(Map.of(
          "_ab_cdc_lsn", Jsons.jsonNode(Map.of("type", "number")),
          "_ab_cdc_updated_at", Jsons.jsonNode(Map.of("type", "string")),
          "_ab_cdc_deleted_at", Jsons.jsonNode(Map.of("type", "string"))));
      final AirbyteStream after = new AirbyteStream()
          .withJsonSchema(Jsons.jsonNode(Map.of("properties", properties)));
  
      assertEquals(after, PostgresCdcCatalogHelper.addCdcMetadataColumns(before));
    }
  
  }
  