@Test
  public void dataReaderCanStillOnlyReadAfterFailover() throws Exception {
    // Connect to the server that will fail
    ClientVM client = createAndInitializeClientAndCache("dataRead");

    // Client should be able to read and not write.
    client.invoke(() -> {
      Region<String, String> region = ClusterStartupRule.getClientCache().getRegion(regionName);
      // Assert that the client can get
      assertThat(region.get(server_k1)).isEqualTo(server_v1);
      assertThat(region.get(server_k2, null)).isEqualTo(server_v2);
      assertThat(region.getAll(serverData.keySet())).containsAllEntriesOf(serverData);
      assertThat(region.getAll(serverData.keySet(), null)).containsAllEntriesOf(serverData);
      // Assert that the client cannot put
      assertThatThrownBy(() -> region.put(client_k1, client_v1))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      assertThatThrownBy(() -> region.put(client_k2, client_v2, null))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      assertThatThrownBy(() -> region.putIfAbsent(client_k3, client_v3))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      assertThatThrownBy(() -> region.putAll(clientData45))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      assertThatThrownBy(() -> region.putAll(clientData67, null))
          .hasCauseInstanceOf(NotAuthorizedException.class);
    });

    // Initialize client cache and region. Get the port of the primary connected server.
    MemberVM server_to_fail = determinePrimaryServer(client);

    // Bring down primary server
    server_to_fail.stop(true);

    // Confirm failover
    MemberVM secondaryServer = (server1.getPort() == server_to_fail.getPort()) ? server2 : server1;
    await().until(() -> getPrimaryServerPort(client) == secondaryServer.getPort());

    // Confirm permissions: client should still only be able to read and not write.
    client.invoke(() -> {
      Region<String, String> region = ClusterStartupRule.getClientCache().getRegion(regionName);
      // Assert that the client can get
      assertThat(region.get(server_k1)).isEqualTo(server_v1);
      assertThat(region.get(server_k2, null)).isEqualTo(server_v2);
      assertThat(region.getAll(serverData.keySet())).containsAllEntriesOf(serverData);
      assertThat(region.getAll(serverData.keySet(), null)).containsAllEntriesOf(serverData);
      // Assert that the client cannot put
      assertThatThrownBy(() -> region.put(client_k1, client_v1))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      assertThatThrownBy(() -> region.put(client_k2, client_v2, null))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      assertThatThrownBy(() -> region.putIfAbsent(client_k3, client_v3))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      assertThatThrownBy(() -> region.putAll(clientData45))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      assertThatThrownBy(() -> region.putAll(clientData67, null))
          .hasCauseInstanceOf(NotAuthorizedException.class);
    });

    // Confirm that no puts went through
    secondaryServer.invoke(() -> {
      assertThat(ClusterStartupRule.getCache().getRegion(regionName))
          .containsOnlyKeys(server_k1, server_k2).containsAllEntriesOf(serverData);
    });
  }
This test class reproduces the tests present in
{@link ClientDataAuthorizationUsingLegacySecurityDUnitTest} and confirms that permissions are
maintained over failover.
geode-core/src/upgradeTest/java/org/apache/geode/security/ClientDataAuthorizationUsingLegacySecurityWithFailoverDUnitTest.java