  @Test
  public void testDeleteState() {

    Stream stream = mock(Stream.class);

    Context ctx = new StreamContext(stream, "test-context-id");

    List<String> deleteList = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
      deleteList.add(RandomString.make(70));
    }

    FutureByteString okResponse = new FutureByteString("test-correlation-id");
    TpStateDeleteResponse deleteReponse = TpStateDeleteResponse.newBuilder().setStatus(TpStateDeleteResponse.Status.OK)
        .addAllAddresses(deleteList).build();
    okResponse.setResult(deleteReponse.toByteString());

    FutureByteString emptyResponse = new FutureByteString("test-correlation-id");
    TpStateDeleteResponse emptyDeleteReponse = TpStateDeleteResponse.newBuilder()
        .setStatus(TpStateDeleteResponse.Status.OK).addAllAddresses(new ArrayList<String>()).build();
    emptyResponse.setResult(emptyDeleteReponse.toByteString());

    FutureByteString errResponse = new FutureByteString("test-correlation-id");
    errResponse.setResult(TpStateDeleteResponse.newBuilder().setStatus(TpStateDeleteResponse.Status.AUTHORIZATION_ERROR)
        .build().toByteString());

    FutureByteString garbageResponse = new FutureByteString("test-correlation-id");
    garbageResponse.setResult(ByteString.copyFrom("something-garbage-like".getBytes()));

    try {
      when(stream.send(any(), any())).thenReturn(okResponse);
      Collection<String> addressesDeleted = ctx.deleteState(deleteList);
      // Compare the in and out addresses, without trusting that either are ordered
      Set<String> inSet = new HashSet<>(deleteList);
      Set<String> outSet = new HashSet<>(addressesDeleted);
      if (!inSet.equals(outSet)) {
        fail("Addresses requested to be deleted do not match those which were actually deleted");
      }
    } catch (InternalError exc) {
      exc.printStackTrace();
      fail("Happy path should not generate an InternalError");
    } catch (InvalidTransactionException exc) {
      exc.printStackTrace();
      fail("Happy path should not generate an InvalidTransactionException");
    }

    try {
      when(stream.send(any(), any())).thenReturn(emptyResponse);
      Collection<String> addressesDeleted = ctx.deleteState(deleteList);
      assertTrue("addressesDeleted should be empty", addressesDeleted.isEmpty());
    } catch (InternalError exc) {
      fail("Returning an empty list of deleted addresses should not result in an InternalError");
    } catch (InvalidTransactionException exc) {
      exc.printStackTrace();
      fail("Empty path should not generate an InvalidTransactionException");
    }

    try {
      when(stream.send(any(), any())).thenReturn(garbageResponse);
      Collection<String> addressesDeleted = ctx.deleteState(deleteList);
      fail("An error should have been thrown since we responded with something wrong!");
    } catch (InternalError ie) {
      // Expected
    } catch (InvalidTransactionException exc) {
      exc.printStackTrace();
      fail(" should not generate an InvalidTransactionException");
    }

    try {
      when(stream.send(any(), any())).thenReturn(errResponse);
      Collection<String> addressesDeleted = ctx.deleteState(deleteList);
      fail("An error should have been thrown since we responded with something wrong!");
    } catch (InternalError ie) {
      ie.printStackTrace();
      fail(" should not generate an InternalError");
    } catch (InvalidTransactionException exc) {
      // Expected
    }
  }