server/replication/SchemaReplicationTest.java

	SchemaReplicationTest.java revision ea1068c292e9b341af6d6b563cd8988a96be20a9
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster/*
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * CDDL HEADER START
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster *
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * The contents of this file are subject to the terms of the
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * Common Development and Distribution License, Version 1.0 only
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * (the "License").  You may not use this file except in compliance
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * with the License.
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster *
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * You can obtain a copy of the license at legal-notices/CDDLv1_0.txt
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * or http://forgerock.org/license/CDDLv1.0.html.
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * See the License for the specific language governing permissions
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * and limitations under the License.
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster *
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * When distributing Covered Code, include this CDDL HEADER in each
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * file and include the License file at legal-notices/CDDLv1_0.txt.
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * If applicable, add the following below this CDDL HEADER, with the
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * fields enclosed by brackets "[]" replaced with your own identifying
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * information:
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster *      Portions Copyright [yyyy] [name of copyright owner]
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster *
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * CDDL HEADER END
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster *
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster *
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster *      Copyright 2008-2010 Sun Microsystems, Inc.
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster *      Portions Copyright 2012-2015 ForgeRock AS.
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster */
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterpackage org.opends.server.replication;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport java.io.File;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport java.io.FileInputStream;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport java.util.ArrayList;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport java.util.List;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.assertj.core.api.Assertions;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.forgerock.i18n.LocalizableMessage;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.forgerock.i18n.slf4j.LocalizedLogger;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.forgerock.opendj.ldap.ModificationType;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.TestCaseUtils;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.admin.std.server.SynchronizationProviderCfg;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.api.SynchronizationProvider;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.core.DirectoryServer;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.core.ModifyOperation;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.protocols.internal.InternalClientConnection;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.replication.common.CSNGenerator;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.replication.plugin.EntryHistorical;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.replication.protocol.ModifyMsg;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.replication.protocol.ReplicationMsg;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.replication.service.ReplicationBroker;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.opends.server.types.*;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.forgerock.opendj.ldap.ResultCode;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.testng.annotations.BeforeClass;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport org.testng.annotations.Test;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterimport static org.testng.Assert.*;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster/**
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster * Test for the schema replication.
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster */
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster@SuppressWarnings("javadoc")
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Fosterpublic class SchemaReplicationTest extends ReplicationTestCase
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster{
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster  private List<Modification> rcvdMods;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster  private int replServerPort;
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster
a688bcbb4bcff5398fdd29b86f83450257dc0df4Allan Foster  /**
   * Set up the environment for performing the tests in this Class.
   */
  @Override
  @BeforeClass
  public void setUp() throws Exception
  {
    super.setUp();

    // This test suite depends on having the schema available.
    replServerPort = TestCaseUtils.findFreePort();

    // Create an internal connection
    connection = InternalClientConnection.getRootConnection();

    // Change log
    String replServerLdif =
      "dn: " + "cn=Replication Server, " + SYNCHRO_PLUGIN_DN + "\n"
        + "objectClass: top\n"
        + "objectClass: ds-cfg-replication-server\n"
        + "cn: Replication Server\n"
        + "ds-cfg-replication-port: " + replServerPort + "\n"
        + "ds-cfg-replication-db-directory: SchemaReplicationTest\n"
        + "ds-cfg-replication-db-implementation: " + replicationDbImplementation + "\n"
        + "ds-cfg-replication-server-id: 105\n";

    // suffix synchronized
    String testName = "schemaReplicationTest";
    String domainLdif =
      "dn: cn=" + testName + ", cn=domains, " + SYNCHRO_PLUGIN_DN + "\n"
        + "objectClass: top\n"
        + "objectClass: ds-cfg-replication-domain\n"
        + "cn: " + testName + "\n"
        + "ds-cfg-base-dn: cn=schema\n"
        + "ds-cfg-replication-server: localhost:" + replServerPort + "\n"
        + "ds-cfg-server-id: 1\n";

    configureReplication(replServerLdif, domainLdif);
  }

  /**
   * Checks that changes done to the schema are pushed to the replicationServer
   * clients.
   */
  @Test
  public void pushSchemaChange() throws Exception
  {
    logger.error(LocalizableMessage.raw("Starting replication test : pushSchemaChange "));

    cleanUpReplicationServersDB();

    final DN baseDN = DN.valueOf("cn=schema");

    ReplicationBroker broker =
      openReplicationSession(baseDN, 2, 100, replServerPort, 5000);

    try
    {
      // Modify the schema
      Attribute attr = Attributes.create("attributetypes",
          "( 2.5.44.77.33 NAME 'dummy' )");
      List<Modification> mods = new ArrayList<Modification>();
      Modification mod = new Modification(ModificationType.ADD, attr);
      mods.add(mod);
      ModifyOperation modOp = connection.processModify(baseDN, mods);
      assertEquals(modOp.getResultCode(), ResultCode.SUCCESS,
          "The original operation failed");

      // See if the client has received the msg
      ReplicationMsg msg = broker.receive();
      Assertions.assertThat(msg).isInstanceOf(ModifyMsg.class);
      ModifyMsg modMsg = (ModifyMsg) msg;

      Operation receivedOp = modMsg.createOperation(connection);
      assertEquals(modMsg.getDN(), baseDN, "The received message is not for cn=schema");
      Assertions.assertThat(receivedOp).isInstanceOf(ModifyOperation.class);
      ModifyOperation receivedModifyOperation = (ModifyOperation) receivedOp;

      this.rcvdMods = new ArrayList<Modification>();
      for (RawModification m : receivedModifyOperation.getRawModifications())
      {
        this.rcvdMods.add(m.toModification());
      }

      assertTrue(this.rcvdMods.contains(mod),
                 "The received mod does not contain the original change");

      /*
       * Now cleanup the schema for the next test
       */
      mod = new Modification(ModificationType.DELETE, attr);
      mods.clear();
      mods.add(mod);
      modOp = connection.processModify(baseDN, mods);
      assertEquals(modOp.getResultCode(), ResultCode.SUCCESS,
          "The original operation failed");

      // See if the client has received the msg
      msg = broker.receive();
      Assertions.assertThat(msg).isInstanceOf(ModifyMsg.class);
    }
    finally
    {
      broker.stop();
    }
  }

  /**
   * Checks that changes to the schema pushed to the replicationServer
   * are received and correctly replayed by replication plugin.
   */
  @Test(enabled=true,dependsOnMethods = { "pushSchemaChange" })
  public void replaySchemaChange() throws Exception
  {
    logger.error(LocalizableMessage.raw("Starting replication test : replaySchemaChange "));

    cleanUpReplicationServersDB();

    final DN baseDN = DN.valueOf("cn=schema");

    ReplicationBroker broker =
      openReplicationSession(baseDN, 2, 100, replServerPort, 5000);

    try
    {
      CSNGenerator gen = new CSNGenerator( 2, 0);

      ModifyMsg modMsg = new ModifyMsg(gen.newCSN(), baseDN, rcvdMods,
          EntryHistorical.getEntryUUID(DirectoryServer.getEntry(baseDN)));
      broker.publish(modMsg);

      boolean found = checkEntryHasAttribute(baseDN, "attributetypes",
        "( 2.5.44.77.33 NAME 'dummy' )",
        10000, true);
      assertTrue(found, "The modification has not been correctly replayed.");
    }
    finally
    {
      broker.stop();
    }
  }

  /**
   * Checks that changes done to the schema files are pushed to the
   * ReplicationServers and that the ServerState is updated in the schema
   * file.
   */
  @Test(enabled=true, dependsOnMethods = { "replaySchemaChange" })
  public void pushSchemaFilesChange() throws Exception
  {
    logger.error(LocalizableMessage.raw("Starting replication test : pushSchemaFilesChange "));

    cleanUpReplicationServersDB();

    final DN baseDN = DN.valueOf("cn=schema");

    ReplicationBroker broker =
      openReplicationSession(baseDN, 3, 100, replServerPort, 5000);

    try
    {
      // create a schema change Notification
      Attribute attr = Attributes.create("attributetypes",
        "( 2.5.44.76.35 NAME 'push' )");
      List<Modification> mods = new ArrayList<Modification>();
      Modification mod = new Modification(ModificationType.ADD, attr);
      mods.add(mod);

      for (SynchronizationProvider<SynchronizationProviderCfg> provider : DirectoryServer.
        getSynchronizationProviders())
      {
        provider.processSchemaChange(mods);
      }

      // receive the message on the broker side.
      ReplicationMsg msg = broker.receive();
      Assertions.assertThat(msg).isInstanceOf(ModifyMsg.class);
      ModifyMsg modMsg = (ModifyMsg) msg;

      Operation receivedOp = modMsg.createOperation(connection);
      assertEquals(modMsg.getDN(), baseDN, "The received message is not for cn=schema");
      Assertions.assertThat(receivedOp).isInstanceOf(ModifyOperation.class);
      ModifyOperation receivedModifyOperation = (ModifyOperation) receivedOp;

      this.rcvdMods = new ArrayList<Modification>();
      for (RawModification m : receivedModifyOperation.getRawModifications())
      {
        this.rcvdMods.add(m.toModification());
      }

      assertTrue(this.rcvdMods.contains(mod),
        "The received mod does not contain the original change");

      // check that the schema files were updated with the new ServerState.
      // by checking that the CSN of msg we just received has been
      // added to the user schema file.

      // build the string to find in the schema file
      String stateStr = modMsg.getCSN().toString();

      // open the schema file
      String buildRoot = System.getProperty(TestCaseUtils.PROPERTY_BUILD_ROOT);
      String buildDir = System.getProperty(TestCaseUtils.PROPERTY_BUILD_DIR,
              buildRoot + File.separator + "build");
      String path = buildDir + File.separator +
        "unit-tests" + File.separator + "package-instance" + File.separator +
        "config" + File.separator + "schema" + File.separator +
        "99-user.ldif";

      // it is necessary to loop on this check because the state is not
      // written immediately but only every so often.
      int count = 0;
      while (true)
      {
        File file = new File(path);
        FileInputStream input = new FileInputStream(file);
        byte[] bytes = new byte[input.available()];
        input.read(bytes);
        String fileStr = new String(bytes);
        if (fileStr.indexOf(stateStr) != -1)
        {
          break;
        }
        assertTrue(count++ <= 50, "The Schema persistentState (CSN:" + stateStr
            + ") has not been saved to " + path + " : " + fileStr);
        TestCaseUtils.sleep(100);
      }
    } finally
    {
      broker.stop();
    }
    logger.error(LocalizableMessage.raw("Ending replication test : pushSchemaFilesChange "));
  }
}