code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.destination.dest_state_lifecycle_manager;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  import io.airbyte.commons.json.Jsons;
  import io.airbyte.protocol.models.AirbyteMessage;
  import io.airbyte.protocol.models.AirbyteMessage.Type;
  import io.airbyte.protocol.models.AirbyteStateMessage;
  import io.airbyte.protocol.models.AirbyteStateMessage.AirbyteStateType;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  
  class DestSingleStateLifecycleManagerTest {
  
    private static final AirbyteMessage MESSAGE1 = new AirbyteMessage()
        .withType(Type.STATE)
        .withState(new AirbyteStateMessage().withType(AirbyteStateType.GLOBAL).withData(Jsons.jsonNode("a")));
    private static final AirbyteMessage MESSAGE2 = new AirbyteMessage()
        .withType(Type.STATE)
        .withState(new AirbyteStateMessage().withType(AirbyteStateType.GLOBAL).withData(Jsons.jsonNode("b")));
  
    private DestSingleStateLifecycleManager mgr;
  
    @BeforeEach
    void setup() {
      mgr = new DestSingleStateLifecycleManager();
    }
  
    /**
     * Demonstrates expected lifecycle of a state object for documentation purposes. Subsequent test get
     * into the details.
     */
    @Test
    void testBasicLifeCycle() {
      // starts with no state.
      assertTrue(mgr.listPending().isEmpty());
      assertTrue(mgr.listFlushed().isEmpty());
      assertTrue(mgr.listCommitted().isEmpty());
  
      mgr.addState(MESSAGE1);
      // new state supersedes previous ones. we should only see MESSAGE2 from here on out.
      mgr.addState(MESSAGE2);
  
      // after adding a state, it is in pending only.
      assertEquals(MESSAGE2, mgr.listPending().poll());
      assertTrue(mgr.listFlushed().isEmpty());
      assertTrue(mgr.listCommitted().isEmpty());
  
      mgr.markPendingAsFlushed();
  
      // after flushing the state it is in flushed only.
      assertTrue(mgr.listPending().isEmpty());
      assertEquals(MESSAGE2, mgr.listFlushed().poll());
      assertTrue(mgr.listCommitted().isEmpty());
  
      // after committing the state it is in committed only.
      mgr.markFlushedAsCommitted();
  
      assertTrue(mgr.listPending().isEmpty());
      assertTrue(mgr.listFlushed().isEmpty());
      assertEquals(MESSAGE2, mgr.listCommitted().poll());
    }
  
    @Test
    void testPending() {
      mgr.addState(MESSAGE1);
      mgr.addState(MESSAGE2);
  
      // verify the LAST message is returned.
      assertEquals(MESSAGE2, mgr.listPending().poll());
      assertTrue(mgr.listFlushed().isEmpty());
      assertTrue(mgr.listCommitted().isEmpty());
    }
  
    @Test
    void testFlushed() {
      mgr.addState(MESSAGE1);
      mgr.addState(MESSAGE2);
      mgr.markPendingAsFlushed();
  
      assertTrue(mgr.listPending().isEmpty());
      assertEquals(MESSAGE2, mgr.listFlushed().poll());
      assertTrue(mgr.listCommitted().isEmpty());
  
      // verify that multiple calls to markPendingAsFlushed overwrite old states
      mgr.addState(MESSAGE1);
      mgr.markPendingAsFlushed();
      mgr.markPendingAsFlushed();
  
      assertTrue(mgr.listPending().isEmpty());
      assertEquals(MESSAGE1, mgr.listFlushed().poll());
      assertTrue(mgr.listCommitted().isEmpty());
    }
  
    @Test
    void testCommitted() {
      mgr.addState(MESSAGE1);
      mgr.addState(MESSAGE2);
      mgr.markPendingAsFlushed();
      mgr.markFlushedAsCommitted();
  
      assertTrue(mgr.listPending().isEmpty());
      assertTrue(mgr.listFlushed().isEmpty());
      assertEquals(MESSAGE2, mgr.listCommitted().poll());
  
      // verify that multiple calls to markFlushedAsCommitted overwrite old states
      mgr.addState(MESSAGE1);
      mgr.markPendingAsFlushed();
      mgr.markFlushedAsCommitted();
      mgr.markFlushedAsCommitted();
  
      assertTrue(mgr.listPending().isEmpty());
      assertTrue(mgr.listFlushed().isEmpty());
      assertEquals(MESSAGE1, mgr.listCommitted().poll());
    }
  
  }
  