You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by up...@apache.org on 2016/01/28 19:13:29 UTC
[27/70] [partial] incubator-geode git commit: WAN and CQ code drop
under the Pivotal SGA
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/cli/commands/DurableClientCommandsDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/cli/commands/DurableClientCommandsDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/cli/commands/DurableClientCommandsDUnitTest.java
new file mode 100644
index 0000000..353d7fa
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/cli/commands/DurableClientCommandsDUnitTest.java
@@ -0,0 +1,404 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.management.internal.cli.commands;
+
+import java.util.Properties;
+import com.gemstone.gemfire.cache.AttributesFactory;
+import com.gemstone.gemfire.cache.CacheException;
+import com.gemstone.gemfire.cache.DataPolicy;
+import com.gemstone.gemfire.cache.PartitionAttributesFactory;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.RegionAttributes;
+import com.gemstone.gemfire.cache.Scope;
+import com.gemstone.gemfire.cache.client.ClientCache;
+import com.gemstone.gemfire.cache.client.ClientCacheFactory;
+import com.gemstone.gemfire.cache.query.CqAttributesFactory;
+import com.gemstone.gemfire.cache.query.CqException;
+import com.gemstone.gemfire.cache.query.CqExistsException;
+import com.gemstone.gemfire.cache.query.QueryService;
+import com.gemstone.gemfire.cache.query.RegionNotFoundException;
+import com.gemstone.gemfire.cache.query.data.Portfolio;
+import com.gemstone.gemfire.cache.server.CacheServer;
+import com.gemstone.gemfire.cache30.CacheSerializableRunnable;
+import com.gemstone.gemfire.distributed.internal.DistributionConfig;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+import com.gemstone.gemfire.internal.cache.DistributedRegion;
+import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
+import com.gemstone.gemfire.internal.cache.PartitionedRegion;
+import com.gemstone.gemfire.internal.cache.tier.sockets.CacheServerTestUtil;
+import com.gemstone.gemfire.management.cli.Result.Status;
+import com.gemstone.gemfire.management.internal.cli.i18n.CliStrings;
+import com.gemstone.gemfire.management.internal.cli.result.CommandResult;
+import com.gemstone.gemfire.management.internal.cli.util.CommandStringBuilder;
+
+import dunit.Host;
+import dunit.SerializableCallable;
+import dunit.VM;
+
+
+
+public class DurableClientCommandsDUnitTest extends CliCommandTestBase {
+
+ private static final long serialVersionUID = 1L;
+ final String regionName = "stocks";
+ final String cq1 = "cq1";
+ final String cq2 = "cq2";
+ final String cq3 = "cq3";
+ final String clientName = "dc1";
+
+ public DurableClientCommandsDUnitTest(String name) {
+ super(name);
+ }
+
+ public void testListDurableClientCqs() throws Exception {
+ setupSystem();
+ setupCqs();
+
+ CommandStringBuilder csb = new CommandStringBuilder(CliStrings.LIST_DURABLE_CQS);
+ csb.addOption(CliStrings.LIST_DURABLE_CQS__DURABLECLIENTID, clientName);
+ String commandString = csb.toString();
+ writeToLog("Command String :\n ", commandString);
+ CommandResult commandResult = executeCommand(commandString);
+ String resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.OK.equals(commandResult.getStatus()));
+ assertTrue(resultAsString.contains(cq1));
+ assertTrue(resultAsString.contains(cq2));
+ assertTrue(resultAsString.contains(cq3));
+
+ closeCq(cq1);
+ closeCq(cq2);
+ closeCq(cq3);
+
+ csb = new CommandStringBuilder(CliStrings.LIST_DURABLE_CQS);
+ csb.addOption(CliStrings.LIST_DURABLE_CQS__DURABLECLIENTID, clientName);
+ commandString = csb.toString();
+ writeToLog("Command String :\n ", commandString);
+ commandResult = executeCommand(commandString);
+ resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.ERROR.equals(commandResult.getStatus()));
+ String errorMessage = CliStrings.format(CliStrings.LIST_DURABLE_CQS__NO__CQS__FOR__CLIENT, clientName);
+ assertTrue(resultAsString.contains(errorMessage));
+ }
+
+ public void testCloseDurableClients() throws Exception {
+ setupSystem();
+ setupCqs();
+ closeDurableClient();
+
+ CommandStringBuilder csb = new CommandStringBuilder(CliStrings.CLOSE_DURABLE_CLIENTS);
+ csb.addOption(CliStrings.CLOSE_DURABLE_CLIENTS__CLIENT__ID, clientName);
+ String commandString = csb.toString();
+ writeToLog("Command String : ", commandString);
+ CommandResult commandResult = executeCommand(commandString);
+ String resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.OK.equals(commandResult.getStatus()));
+
+ //Execute again to see the error condition
+ writeToLog("Command String : ", commandString);
+ commandResult = executeCommand(commandString);
+ resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.ERROR.equals(commandResult.getStatus()));
+ String errorMessage = CliStrings.format(CliStrings.NO_CLIENT_FOUND_WITH_CLIENT_ID, clientName);
+ assertTrue(resultAsString.contains(errorMessage));
+ }
+
+
+ public void testCloseDurableCQ() throws Exception{
+ setupSystem();
+ setupCqs();
+
+ closeDurableClient();
+
+ CommandStringBuilder csb = new CommandStringBuilder(CliStrings.CLOSE_DURABLE_CQS);
+ csb.addOption(CliStrings.CLOSE_DURABLE_CQS__DURABLE__CLIENT__ID, clientName);
+ csb.addOption(CliStrings.CLOSE_DURABLE_CQS__NAME, cq1);
+ String commandString = csb.toString();
+ writeToLog("Command String : ", commandString);
+ CommandResult commandResult = executeCommand(commandString);
+ String resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.OK.equals(commandResult.getStatus()));
+
+ csb = new CommandStringBuilder(CliStrings.CLOSE_DURABLE_CQS);
+ csb.addOption(CliStrings.CLOSE_DURABLE_CQS__DURABLE__CLIENT__ID, clientName);
+ csb.addOption(CliStrings.CLOSE_DURABLE_CQS__NAME, cq1);
+ commandString = csb.toString();
+ writeToLog("Command String : ", commandString);
+ commandResult = executeCommand(commandString);
+ resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result : ", resultAsString);
+ assertTrue(Status.ERROR.equals(commandResult.getStatus()));
+
+ }
+
+ public void testCountSubscriptionQueueSize() throws Exception {
+ setupSystem();
+ setupCqs();
+ doPuts(regionName, Host.getHost(0).getVM(1));
+
+ CommandStringBuilder csb = new CommandStringBuilder(CliStrings.COUNT_DURABLE_CQ_EVENTS);
+ csb.addOption(CliStrings.COUNT_DURABLE_CQ_EVENTS__DURABLE__CLIENT__ID, clientName);
+ String commandString = csb.toString();
+ writeToLog("Command String : ", commandString);
+ CommandResult commandResult = executeCommand(commandString);
+ String resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.OK.equals(commandResult.getStatus()));
+ assertTrue(resultAsString.contains("4"));
+
+ csb = new CommandStringBuilder(CliStrings.COUNT_DURABLE_CQ_EVENTS);
+ csb.addOption(CliStrings.COUNT_DURABLE_CQ_EVENTS__DURABLE__CLIENT__ID, clientName);
+ csb.addOption(CliStrings.COUNT_DURABLE_CQ_EVENTS__DURABLE__CQ__NAME, cq3);
+ commandString = csb.toString();
+ writeToLog("Command String : ", commandString);
+ commandResult = executeCommand(commandString);
+ resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.OK.equals(commandResult.getStatus()));
+
+ //CLOSE all the cqs
+ closeCq(cq1);
+ closeCq(cq2);
+ closeCq(cq3);
+
+ //Run the commands again
+ csb = new CommandStringBuilder(CliStrings.COUNT_DURABLE_CQ_EVENTS);
+ csb.addOption(CliStrings.COUNT_DURABLE_CQ_EVENTS__DURABLE__CLIENT__ID, clientName);
+ csb.addOption(CliStrings.COUNT_DURABLE_CQ_EVENTS__DURABLE__CQ__NAME, cq1);
+ commandString = csb.toString();
+ writeToLog("Command String : ", commandString);
+ commandResult = executeCommand(commandString);
+ resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.ERROR.equals(commandResult.getStatus()));
+ String errorMessage = CliStrings.format(CliStrings.COUNT_DURABLE_CQ_EVENTS__DURABLE_CQ_NOT_FOUND, clientName, cq1);
+ assertTrue(resultAsString.contains(errorMessage));
+
+ csb = new CommandStringBuilder(CliStrings.COUNT_DURABLE_CQ_EVENTS);
+ csb.addOption(CliStrings.COUNT_DURABLE_CQ_EVENTS__DURABLE__CLIENT__ID, clientName);
+ commandString = csb.toString();
+ writeToLog("Command String : ", commandString);
+ commandResult = executeCommand(commandString);
+ resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.OK.equals(commandResult.getStatus()));
+
+ //Disconnect the client
+ closeDurableClient();
+
+ //Close the client
+ csb = new CommandStringBuilder(CliStrings.CLOSE_DURABLE_CLIENTS);
+ csb.addOption(CliStrings.CLOSE_DURABLE_CLIENTS__CLIENT__ID, clientName);
+ commandString = csb.toString();
+ writeToLog("Command String : ", commandString);
+ commandResult = executeCommand(commandString);
+ resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.OK.equals(commandResult.getStatus()));
+
+ csb = new CommandStringBuilder(CliStrings.COUNT_DURABLE_CQ_EVENTS);
+ csb.addOption(CliStrings.COUNT_DURABLE_CQ_EVENTS__DURABLE__CLIENT__ID, clientName);
+ commandString = csb.toString();
+ writeToLog("Command String : ", commandString);
+ commandResult = executeCommand(commandString);
+ resultAsString = commandResultToString(commandResult);
+ writeToLog("Command Result :\n", resultAsString);
+ assertTrue(Status.ERROR.equals(commandResult.getStatus()));
+ assertTrue(resultAsString.contains(CliStrings.format(CliStrings.NO_CLIENT_FOUND_WITH_CLIENT_ID, clientName)));
+ }
+
+ private void writeToLog(String text, String resultAsString) {
+ getLogWriter().info(testName + "\n");
+ getLogWriter().info(resultAsString);
+ }
+
+ private void setupSystem() throws Exception {
+ disconnectAllFromDS();
+ final int[] port = AvailablePortHelper.getRandomAvailableTCPPorts(2);
+ createDefaultSetup(getServerProperties());
+
+ final VM manager = Host.getHost(0).getVM(0);
+ final VM server1 = Host.getHost(0).getVM(1);
+ final VM client1 = Host.getHost(0).getVM(2);
+
+ startCacheServer(server1, port[0], false, regionName);
+ startDurableClient(client1, server1, port[0], clientName, "300");
+ }
+
+ /**
+ * Close the cq from the client-side
+ * @param cqName , Name of the cq which is to be close.
+ */
+ private void closeCq(final String cqName) {
+ final VM vm2 = Host.getHost(0).getVM(2);
+ vm2.invoke(new SerializableCallable() {
+ public Object call() {
+ QueryService qs = getCache().getQueryService();
+ CqAttributesFactory cqAf = new CqAttributesFactory();
+ try {
+ qs.getCq(cqName).close();
+
+ }
+ catch (CqException e) {
+ e.printStackTrace();
+ return false;
+ }
+ return true;
+ }
+ });
+ }
+
+ private void setupCqs() {
+ final VM vm2 = Host.getHost(0).getVM(2);
+ vm2.invoke(new SerializableCallable() {
+ public Object call() {
+ QueryService qs = getCache().getQueryService();
+ CqAttributesFactory cqAf = new CqAttributesFactory();
+ try {
+ qs.newCq(cq1, "select * from /" + regionName, cqAf.create(), true).execute();
+ qs.newCq(cq2, "select * from /" + regionName + " where id = 1", cqAf.create(), true).execute();
+ qs.newCq(cq3, "select * from /" + regionName + " where id > 2", cqAf.create(), true).execute();
+ }
+ catch (CqException e) {
+ e.printStackTrace();
+ return false;
+ }
+ catch (CqExistsException e) {
+ e.printStackTrace();
+
+ return false;
+ }
+ catch(RegionNotFoundException e) {
+ e.printStackTrace();
+
+ return false;
+ }
+ return true;
+ }
+ });
+ }
+
+ private void startCacheServer(VM server, final int port,
+ final boolean createPR, final String regionName) throws Exception {
+
+ server.invoke(new SerializableCallable() {
+ public Object call() throws Exception {
+ getSystem(getServerProperties());
+
+ GemFireCacheImpl cache = (GemFireCacheImpl)getCache();
+ AttributesFactory factory = new AttributesFactory();
+ if (createPR) {
+ PartitionAttributesFactory paf = new PartitionAttributesFactory();
+ paf.setRedundantCopies(1);
+ paf.setTotalNumBuckets(11);
+ factory.setPartitionAttributes(paf.create());
+ } else {
+ factory.setScope(Scope.DISTRIBUTED_ACK);
+ factory.setDataPolicy(DataPolicy.REPLICATE);
+ }
+ Region region = createRootRegion(regionName, factory.create());
+ if (createPR) {
+ assertTrue(region instanceof PartitionedRegion);
+ } else {
+ assertTrue(region instanceof DistributedRegion);
+ }
+ CacheServer cacheServer = getCache().addCacheServer();
+ cacheServer.setPort(port);
+ cacheServer.start();
+
+ return null;
+ }
+ });
+ }
+
+ private void startDurableClient(VM client, final VM server, final int port,
+ final String durableClientId, final String durableClientTimeout) {
+ client.invoke(new CacheSerializableRunnable("Start client") {
+ public void run2() throws CacheException {
+ Properties props = getClientProps(durableClientId, durableClientTimeout);
+ getSystem(props);
+
+ final ClientCacheFactory ccf = new ClientCacheFactory(props);
+ ccf.addPoolServer(getServerHostName(server.getHost()), port);
+ ccf.setPoolSubscriptionEnabled(true);
+
+ ClientCache cache = (ClientCache)getClientCache(ccf);
+ }
+ });
+ }
+
+ /* Does few puts on the region on the server
+ *
+ */
+
+ private void doPuts(final String regionName, VM server) {
+ server.invoke(new SerializableCallable() {
+ public Object call() throws Exception {
+ GemFireCacheImpl cache = (GemFireCacheImpl)getCache();
+ Region region = cache.getRegion(regionName);
+ Portfolio p1 = new Portfolio();
+ p1.ID = 1;
+ p1.names = new String [] {"AAPL", "VMW"};
+
+ Portfolio p2 = new Portfolio();
+ p2.ID = 2;
+ p2.names = new String [] {"EMC", "IBM"};
+
+ Portfolio p3 = new Portfolio();
+ p3.ID = 5;
+ p3.names = new String [] {"DOW","TON"};
+
+ Portfolio p4 = new Portfolio();
+ p4.ID = 5;
+ p4.names = new String [] {"ABC", "EBAY"};
+
+ region.put("p1", p1);
+ region.put("p2", p2);
+ region.put("p3", p3);
+ region.put("p4", p4);
+ return null;
+ }
+ });
+ }
+
+ //Closes the durable-client from the client side.
+ private void closeDurableClient() {
+ final VM client = Host.getHost(0).getVM(2);
+ client.invoke(new CacheSerializableRunnable("Stop client") {
+ public void run2() throws CacheException {
+ ClientCacheFactory.getAnyInstance().close(true);
+ }
+ });
+ }
+
+ protected Properties getClientProps(String durableClientId, String durableClientTimeout) {
+ Properties p = new Properties();
+ p.setProperty(DistributionConfig.MCAST_PORT_NAME, "0");
+ p.setProperty(DistributionConfig.LOCATORS_NAME, "");
+ p.setProperty(DistributionConfig.DURABLE_CLIENT_ID_NAME, durableClientId);
+ p.setProperty(DistributionConfig.DURABLE_CLIENT_TIMEOUT_NAME, String.valueOf(durableClientTimeout));
+ return p;
+ }
+
+ protected Properties getServerProperties() {
+ Properties p = new Properties();
+ p.setProperty(DistributionConfig.LOCATORS_NAME, "localhost["+getDUnitLocatorPort()+"]");
+ return p;
+ }
+
+ public void tearDown2() throws Exception {
+ super.tearDown2();
+ Host.getHost(0).getVM(0).invoke(CacheServerTestUtil.class, "closeCache");
+ Host.getHost(0).getVM(1).invoke(CacheServerTestUtil.class, "closeCache");
+ Host.getHost(0).getVM(2).invoke(CacheServerTestUtil.class, "closeCache");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestCQDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestCQDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestCQDUnitTest.java
new file mode 100644
index 0000000..d855b16
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestCQDUnitTest.java
@@ -0,0 +1,139 @@
+/*
+ * =========================================================================
+ * Copyright (c) 2002-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * more patents listed at http://www.pivotal.io/patents.
+ * ========================================================================
+ */
+package com.gemstone.gemfire.management.internal.pulse;
+
+import com.gemstone.gemfire.cache.query.cq.dunit.CqQueryDUnitTest;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+import com.gemstone.gemfire.management.DistributedSystemMXBean;
+import com.gemstone.gemfire.management.ManagementService;
+import com.gemstone.gemfire.management.ManagementTestBase;
+import dunit.VM;
+
+/**
+ * This is for testing continuous query.
+ * @author ajayp
+ *
+ */
+
+public class TestCQDUnitTest extends ManagementTestBase {
+
+ private static final long serialVersionUID = 1L;
+ private static final String queryName = "testClientWithFeederAndCQ_0";
+ private static final String queryName2 = "testClientWithFeederAndCQ_3";
+
+ protected CqQueryDUnitTest cqDUnitTest = new CqQueryDUnitTest(
+ "CqDataDUnitTest");
+
+ public TestCQDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+ super.setUp();
+
+ }
+
+ public void tearDown2() throws Exception {
+ super.tearDown2();
+ }
+
+ public static long getNumOfCQ() {
+
+ final WaitCriterion waitCriteria = new WaitCriterion() {
+ @Override
+ public boolean done() {
+ final ManagementService service = getManagementService();
+ final DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
+ if (bean != null) {
+ if(bean.getActiveCQCount() > 0){
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public String description() {
+ return "wait for getNumOfCQ to complete and get results";
+ }
+ };
+ waitForCriterion(waitCriteria, 2 * 60 * 1000, 3000, true);
+ final DistributedSystemMXBean bean = getManagementService().getDistributedSystemMXBean();
+ assertNotNull(bean);
+ return bean.getActiveCQCount();
+ }
+
+ public void testNumOfCQ() throws Exception {
+ initManagement(false);
+ getLogWriter().info("started testNumOfCQ");
+
+ VM server = managedNodeList.get(1);
+ VM client = managedNodeList.get(2);
+
+ final String host0 = getServerHostName(server.getHost());
+
+ int serverPort = AvailablePortHelper.getRandomAvailableTCPPort();
+ cqDUnitTest.createServer(server, serverPort);
+
+ final int port = server.invokeInt(CqQueryDUnitTest.class,
+ "getCacheServerPort");
+
+ cqDUnitTest.createClient(client, port, host0);
+ cqDUnitTest.createCQ(client, queryName, cqDUnitTest.cqs[0]);
+ cqDUnitTest.executeCQ(client, queryName, false, null);
+
+ cqDUnitTest.createCQ(client, queryName2, cqDUnitTest.cqs[3]);
+ cqDUnitTest.executeCQ(client, queryName2, false, null);
+
+ final int size = 1000;
+ cqDUnitTest.createValues(client, cqDUnitTest.regions[0], size);
+ cqDUnitTest.waitForCreated(client, queryName, CqQueryDUnitTest.KEY + size);
+ cqDUnitTest.waitForCreated(client, queryName2, CqQueryDUnitTest.KEY + size );
+
+
+ cqDUnitTest.validateCQ(client, queryName,
+ /* resultSize: */CqQueryDUnitTest.noTest,
+ /* creates: */size,
+ /* updates: */0,
+ /* deletes; */0,
+ /* queryInserts: */size,
+ /* queryUpdates: */0,
+ /* queryDeletes: */0,
+ /* totalEvents: */size);
+
+ cqDUnitTest.validateCQ(client, queryName2,
+ /* resultSize: */CqQueryDUnitTest.noTest,
+ /* creates: */size,
+ /* updates: */0,
+ /* deletes; */0,
+ /* queryInserts: */size,
+ /* queryUpdates: */0,
+ /* queryDeletes: */0,
+ /* totalEvents: */size);
+
+ long numOfCQ = ((Number) managingNode.invoke(TestCQDUnitTest.class,
+ "getNumOfCQ")).intValue();
+
+ getLogWriter().info("testNumOfCQ numOfCQ= " + numOfCQ);
+
+ cqDUnitTest.closeClient(client);
+ cqDUnitTest.closeServer(server);
+
+ assertTrue(numOfCQ > 0 ? true : false);
+ }
+
+ public void verifyStatistics() {
+
+ }
+
+ public void invokeOperations() {
+
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestClientsDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestClientsDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestClientsDUnitTest.java
new file mode 100644
index 0000000..cb473df
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestClientsDUnitTest.java
@@ -0,0 +1,101 @@
+/*
+ * =========================================================================
+ * Copyright (c) 2002-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * more patents listed at http://www.pivotal.io/patents.
+ * ========================================================================
+ */
+package com.gemstone.gemfire.management.internal.pulse;
+
+import com.gemstone.gemfire.cache.query.cq.dunit.CqQueryDUnitTest;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+import com.gemstone.gemfire.management.DistributedSystemMXBean;
+import com.gemstone.gemfire.management.ManagementService;
+import com.gemstone.gemfire.management.ManagementTestBase;
+
+import dunit.VM;
+
+/**
+ * This is for testing Number of clients and can be extended for relevant test
+ * addition
+ *
+ * @author ajayp
+ *
+ */
+
+public class TestClientsDUnitTest extends ManagementTestBase {
+
+ private static final long serialVersionUID = 1L;
+
+ public TestClientsDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+ super.setUp();
+
+ }
+
+ public void tearDown2() throws Exception {
+ super.tearDown2();
+ }
+
+ public static Integer getNumOfClients() {
+ final WaitCriterion waitCriteria = new WaitCriterion() {
+ @Override
+ public boolean done() {
+ final ManagementService service = getManagementService();
+ final DistributedSystemMXBean bean = service
+ .getDistributedSystemMXBean();
+ if (bean != null) {
+ if (bean.getNumClients() > 0) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public String description() {
+ return "wait for getNumOfClients bean to complete and get results";
+ }
+ };
+
+ waitForCriterion(waitCriteria, 2 * 60 * 1000, 3000, true);
+ final DistributedSystemMXBean bean = getManagementService()
+ .getDistributedSystemMXBean();
+ assertNotNull(bean);
+ return Integer.valueOf(bean.getNumClients());
+ }
+
+ protected CqQueryDUnitTest cqDUnitTest = new CqQueryDUnitTest(
+ "CqDataDUnitTest");
+
+ public void testNumOfClients() throws Exception {
+ initManagement(false);
+ VM server = managedNodeList.get(1);
+ VM client = managedNodeList.get(2);
+ int serverPort = AvailablePortHelper.getRandomAvailableTCPPort();
+ cqDUnitTest.createServer(server, serverPort);
+ final int port = server.invokeInt(CqQueryDUnitTest.class,
+ "getCacheServerPort");
+ final String host0 = getServerHostName(server.getHost());
+ cqDUnitTest.createClient(client, port, host0);
+ Integer numOfClients = (Integer) managingNode.invoke(
+ TestClientsDUnitTest.class, "getNumOfClients");
+ getLogWriter().info("testNumOfClients numOfClients = " + numOfClients);
+ cqDUnitTest.closeClient(client);
+ cqDUnitTest.closeServer(server);
+ assertEquals(1, numOfClients.intValue());
+ }
+
+ public void verifyStatistics() {
+
+ }
+
+ public void invokeOperations() {
+
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestServerDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestServerDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestServerDUnitTest.java
new file mode 100644
index 0000000..6b8d21b
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/management/internal/pulse/TestServerDUnitTest.java
@@ -0,0 +1,91 @@
+/*
+ * =========================================================================
+ * Copyright (c) 2002-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * more patents listed at http://www.pivotal.io/patents.
+ * ========================================================================
+ */
+package com.gemstone.gemfire.management.internal.pulse;
+
+import com.gemstone.gemfire.cache.query.cq.dunit.CqQueryDUnitTest;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+import com.gemstone.gemfire.management.DistributedSystemMXBean;
+import com.gemstone.gemfire.management.ManagementService;
+import com.gemstone.gemfire.management.ManagementTestBase;
+import dunit.VM;
+
+/**
+ * This is for testing server count details from MBean
+ * @author ajayp
+ *
+ */
+
+public class TestServerDUnitTest extends ManagementTestBase {
+
+ private static final long serialVersionUID = 1L;
+
+ public TestServerDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+ super.setUp();
+
+ }
+
+ public void tearDown2() throws Exception {
+ super.tearDown2();
+ }
+
+ public static int getNumOfServersFromMBean() {
+
+ final WaitCriterion waitCriteria = new WaitCriterion() {
+ @Override
+ public boolean done() {
+ final ManagementService service = getManagementService();
+ final DistributedSystemMXBean bean = service.getDistributedSystemMXBean();
+ if(bean !=null){
+ if (bean.listCacheServers().length > 0) {
+ return true;
+ }
+ }
+ return false;
+ }
+ @Override
+ public String description() {
+ return "wait for getDistributedSystemMXBean to complete and get results";
+ }
+ };
+
+ waitForCriterion(waitCriteria, 2 * 60 * 1000, 3000, true);
+ final DistributedSystemMXBean bean = getManagementService().getDistributedSystemMXBean();
+ assertNotNull(bean);
+ return bean.listCacheServers().length;
+
+ }
+
+ protected CqQueryDUnitTest cqDUnitTest = new CqQueryDUnitTest(
+ "CqDataDUnitTest");
+
+ public void testNumOfServersDUnitTest() throws Exception {
+ initManagement(false);
+ VM server = managedNodeList.get(1);
+ int serverPort = AvailablePortHelper.getRandomAvailableTCPPort();
+ cqDUnitTest.createServer(server, serverPort);
+ int serverCount = ((Number) managingNode.invoke(TestServerDUnitTest.class,
+ "getNumOfServersFromMBean")).intValue();
+ getLogWriter().info("TestServerDUnitTest serverCount =" + serverCount);
+ cqDUnitTest.closeServer(server);
+ assertEquals(1, serverCount);
+ }
+
+ public void verifyStatistics() {
+
+ }
+
+ public void invokeOperations() {
+
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientAuthorizationTwoDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientAuthorizationTwoDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientAuthorizationTwoDUnitTest.java
new file mode 100644
index 0000000..4596381
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientAuthorizationTwoDUnitTest.java
@@ -0,0 +1,237 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.security;
+
+import com.gemstone.gemfire.cache.operations.OperationContext.OperationCode;
+
+import dunit.Host;
+
+/**
+ * Tests for authorization from client to server. This tests for authorization
+ * of all operations with both valid and invalid credentials/modules with
+ * pre-operation callbacks. It also checks for authorization in case of
+ * failover.
+ *
+ * This is the second part of the test which had become long enough to
+ * occasionally go beyond the 10min limit.
+ *
+ * @author sumedh
+ * @since 5.5
+ */
+public class ClientAuthorizationTwoDUnitTest extends
+ ClientAuthorizationTestBase {
+
+
+ /** constructor */
+ public ClientAuthorizationTwoDUnitTest(String name) {
+ super(name);
+ }
+
+ @Override
+ public void setUp() throws Exception {
+
+ super.setUp();
+ final Host host = Host.getHost(0);
+ server1 = host.getVM(0);
+ server2 = host.getVM(1);
+ client1 = host.getVM(2);
+ client2 = host.getVM(3);
+
+ server1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { serverExpectedExceptions });
+ server2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { serverExpectedExceptions });
+ client1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { clientExpectedExceptions });
+ client2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { clientExpectedExceptions });
+ SecurityTestUtil.registerExpectedExceptions(clientExpectedExceptions);
+ }
+
+ // Region: Tests
+
+ public void testAllOpsWithFailover2() {
+ addExpectedException("Read timed out");
+ addExpectedException("Connection reset");
+ addExpectedException("SocketTimeoutException");
+ addExpectedException("ServerConnectivityException");
+ addExpectedException("Socket Closed");
+
+ OperationWithAction[] allOps = {
+ // Register interest in all keys using list
+ new OperationWithAction(OperationCode.REGISTER_INTEREST, 3,
+ OpFlags.USE_LIST | OpFlags.CHECK_NOTAUTHZ, 4),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST, 1,
+ OpFlags.USE_LIST, 4),
+ // UPDATE and test with GET
+ new OperationWithAction(OperationCode.PUT, 2),
+ new OperationWithAction(OperationCode.GET, 1, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 4),
+
+ // Unregister interest in all keys using list
+ new OperationWithAction(OperationCode.UNREGISTER_INTEREST, 1,
+ OpFlags.USE_OLDCONN | OpFlags.USE_LIST, 4),
+ // UPDATE and test with GET for no updates
+ new OperationWithAction(OperationCode.PUT, 2, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 4),
+ new OperationWithAction(OperationCode.GET, 1, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 4),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Register interest in all keys using regular expression
+ new OperationWithAction(OperationCode.REGISTER_INTEREST, 3,
+ OpFlags.USE_REGEX | OpFlags.CHECK_NOTAUTHZ, 4),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST, 2,
+ OpFlags.USE_REGEX, 4),
+ // UPDATE and test with GET
+ new OperationWithAction(OperationCode.PUT),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 4),
+
+ // Unregister interest in all keys using regular expression
+ new OperationWithAction(OperationCode.UNREGISTER_INTEREST, 2,
+ OpFlags.USE_OLDCONN | OpFlags.USE_REGEX, 4),
+ // UPDATE and test with GET for no updates
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 4),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 4),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Register interest in all keys using ALL_KEYS
+ new OperationWithAction(OperationCode.REGISTER_INTEREST, 3,
+ OpFlags.USE_ALL_KEYS | OpFlags.CHECK_NOTAUTHZ, 4),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST, 2,
+ OpFlags.USE_ALL_KEYS, 4),
+ // UPDATE and test with GET
+ new OperationWithAction(OperationCode.PUT),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 4),
+
+ // Unregister interest in all keys using ALL_KEYS
+ new OperationWithAction(OperationCode.UNREGISTER_INTEREST, 2,
+ OpFlags.USE_OLDCONN | OpFlags.USE_ALL_KEYS, 4),
+ // UPDATE and test with GET for no updates
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 4),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 4),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Register CQ
+ new OperationWithAction(OperationCode.PUT, 2, OpFlags.USE_NEWVAL, 4),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 3,
+ OpFlags.CHECK_NOTAUTHZ, 4),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 1,
+ OpFlags.USE_NEWVAL, 4),
+ // UPDATE and test with GET
+ new OperationWithAction(OperationCode.PUT, 2, OpFlags.USE_OLDCONN, 4),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 1,
+ OpFlags.USE_OLDCONN | OpFlags.LOCAL_OP, 4),
+
+ // Stop CQ
+ new OperationWithAction(OperationCode.STOP_CQ, 3, OpFlags.USE_OLDCONN
+ | OpFlags.CHECK_EXCEPTION, 4),
+ new OperationWithAction(OperationCode.STOP_CQ, 1, OpFlags.USE_OLDCONN,
+ 4),
+ // UPDATE and test with GET for no updates
+ new OperationWithAction(OperationCode.PUT, 2, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 4),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 1,
+ OpFlags.USE_OLDCONN | OpFlags.CHECK_FAIL | OpFlags.LOCAL_OP, 4),
+
+ // Restart the CQ
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 3, OpFlags.USE_NEWVAL
+ | OpFlags.CHECK_NOTAUTHZ, 4),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 1,
+ OpFlags.USE_OLDCONN | OpFlags.USE_NEWVAL, 4),
+ // UPDATE and test with GET
+ new OperationWithAction(OperationCode.PUT, 2, OpFlags.USE_OLDCONN, 4),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 1,
+ OpFlags.USE_OLDCONN | OpFlags.LOCAL_OP, 4),
+
+ // Close CQ
+ new OperationWithAction(OperationCode.CLOSE_CQ, 3, OpFlags.USE_OLDCONN,
+ 4),
+ new OperationWithAction(OperationCode.CLOSE_CQ, 1, OpFlags.USE_OLDCONN,
+ 4),
+ // UPDATE and test with GET for no updates
+ new OperationWithAction(OperationCode.PUT, 2, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 4),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 1,
+ OpFlags.USE_OLDCONN | OpFlags.CHECK_FAIL | OpFlags.LOCAL_OP, 4),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Do REGION_CLEAR and check with GET
+ new OperationWithAction(OperationCode.REGION_CLEAR, 3,
+ OpFlags.CHECK_NOTAUTHZ, 1),
+ new OperationWithAction(OperationCode.REGION_CLEAR, 1, OpFlags.NONE, 1),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_FAIL, 8),
+ // Repopulate the region
+ new OperationWithAction(OperationCode.PUT),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Do REGION_CREATE and check with CREATE/GET
+ new OperationWithAction(OperationCode.REGION_CREATE, 3,
+ OpFlags.ENABLE_DRF | OpFlags.CHECK_NOTAUTHZ, 1),
+ new OperationWithAction(OperationCode.REGION_CREATE, 1,
+ OpFlags.ENABLE_DRF, 1),
+ new OperationWithAction(OperationCode.PUT, 3, OpFlags.USE_OLDCONN
+ | OpFlags.USE_SUBREGION | OpFlags.CHECK_NOTAUTHZ, 4),
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_SUBREGION, 4),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.CHECK_NOKEY
+ | OpFlags.USE_SUBREGION, 4),
+
+ // Do REGION_DESTROY of the sub-region and check with GET
+ new OperationWithAction(OperationCode.REGION_DESTROY, 3,
+ OpFlags.USE_OLDCONN | OpFlags.USE_SUBREGION
+ | OpFlags.NO_CREATE_SUBREGION | OpFlags.CHECK_NOTAUTHZ, 1),
+ new OperationWithAction(OperationCode.REGION_DESTROY, 1,
+ OpFlags.USE_OLDCONN | OpFlags.USE_SUBREGION
+ | OpFlags.NO_CREATE_SUBREGION, 1),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.USE_SUBREGION | OpFlags.CHECK_EXCEPTION, 4),
+
+ // Do REGION_DESTROY of the region and check with GET
+ new OperationWithAction(OperationCode.REGION_DESTROY, 3,
+ OpFlags.CHECK_NOTAUTHZ, 1),
+ new OperationWithAction(OperationCode.REGION_DESTROY, 1, OpFlags.NONE,
+ 1),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_EXCEPTION, 4),
+
+ // Skip failover for region destroy since it shall fail
+ // without restarting the server
+ OperationWithAction.OPBLOCK_NO_FAILOVER };
+
+ runOpsWithFailover(allOps, "testAllOpsWithFailover2");
+ }
+
+ // End Region: Tests
+
+ @Override
+ public void tearDown2() throws Exception {
+
+ super.tearDown2();
+ // close the clients first
+ client1.invoke(SecurityTestUtil.class, "closeCache");
+ client2.invoke(SecurityTestUtil.class, "closeCache");
+ SecurityTestUtil.closeCache();
+ // then close the servers
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+ server2.invoke(SecurityTestUtil.class, "closeCache");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientAuthzObjectModDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientAuthzObjectModDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientAuthzObjectModDUnitTest.java
new file mode 100644
index 0000000..c3adb3f
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientAuthzObjectModDUnitTest.java
@@ -0,0 +1,408 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.security;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+
+import security.CredentialGenerator;
+import security.DummyAuthzCredentialGenerator;
+import security.DummyCredentialGenerator;
+import templates.security.UserPasswordAuthInit;
+
+import com.gemstone.gemfire.DataSerializable;
+import com.gemstone.gemfire.Instantiator;
+import com.gemstone.gemfire.cache.operations.OperationContext.OperationCode;
+import com.gemstone.gemfire.distributed.internal.DistributionConfig;
+import com.gemstone.gemfire.internal.AvailablePort;
+import com.gemstone.gemfire.internal.security.ObjectWithAuthz;
+
+import dunit.DistributedTestCase;
+import dunit.Host;
+import dunit.SerializableRunnable;
+
+/**
+ * Tests for authorization callback that modify objects and callbacks from
+ * client to server.
+ *
+ * The scheme of these tests is thus: A user name terminating in integer 'i' can
+ * get values terminating with 'i', '2*i', '3*i' and so on. So 'gemfire1' can
+ * get 'value1', 'value2', ...; 'gemfire2' can get 'value2', 'value4', ... and
+ * so on. On the server side this is done by adding the index 'i' to the object
+ * in the pre-processing phase, and checked by comparing against the user name
+ * index during the post-processing phase.
+ *
+ * This enables testing of object and callback modification both in
+ * pre-processing and post-processing phases.
+ *
+ * @author sumedh
+ * @since 5.5
+ */
+public class ClientAuthzObjectModDUnitTest extends ClientAuthorizationTestBase {
+
+
+ /** constructor */
+ public ClientAuthzObjectModDUnitTest(String name) {
+ super(name);
+ }
+
+ private static final String preAccessor = "com.gemstone.gemfire.internal."
+ + "security.FilterPreAuthorization.create";
+
+ private static final String postAccessor = "com.gemstone.gemfire.internal."
+ + "security.FilterPostAuthorization.create";
+
+ private static class TestPostCredentialGenerator implements
+ TestCredentialGenerator {
+
+ public TestPostCredentialGenerator() {
+ }
+
+ public Properties getAllowedCredentials(OperationCode[] opCodes,
+ String[] regionNames, int[] keyIndices, int num) {
+
+ int userIndex = 1;
+ byte role = DummyAuthzCredentialGenerator.getRequiredRole(opCodes);
+ if (role == DummyAuthzCredentialGenerator.READER_ROLE) {
+ userIndex = keyIndices[0] + 1;
+ }
+ Properties props = new Properties();
+ props.setProperty(UserPasswordAuthInit.USER_NAME, "user" + userIndex);
+ props.setProperty(UserPasswordAuthInit.PASSWORD, "user" + userIndex);
+ return props;
+ }
+
+ public Properties getDisallowedCredentials(OperationCode[] opCodes,
+ String[] regionNames, int[] keyIndices, int num) {
+
+ int userIndex = 0;
+ for (int index = 0; index < keyIndices.length; ++index) {
+ if (keyIndices[index] != index) {
+ userIndex = index + 1;
+ break;
+ }
+ }
+ Properties props = new Properties();
+ props.setProperty(UserPasswordAuthInit.USER_NAME, "gemfire" + userIndex);
+ props.setProperty(UserPasswordAuthInit.PASSWORD, "gemfire" + userIndex);
+ return props;
+ }
+
+ public CredentialGenerator getCredentialGenerator() {
+
+ return null;
+ }
+ }
+
+ public void setUp() throws Exception {
+
+ super.setUp();
+ final Host host = Host.getHost(0);
+ server1 = host.getVM(0);
+ server2 = host.getVM(1);
+ client1 = host.getVM(2);
+ client2 = host.getVM(3);
+
+ server1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { serverExpectedExceptions });
+ server2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { serverExpectedExceptions });
+ client2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { clientExpectedExceptions });
+ SecurityTestUtil.registerExpectedExceptions(clientExpectedExceptions);
+
+ //required by FilterPreAuthorization and FilterPostAuthorization. Normally,
+ //this would be automatically registered in the static initializer, but with dunit
+ //a previous test may have already loaded these classes. We clear the instantiators
+ //between each test.
+ SerializableRunnable registerInstantiator = new SerializableRunnable() {
+ public void run() {
+ Instantiator.register(new MyInstantiator(), false);
+ }
+ };
+ server1.invoke(registerInstantiator);
+ server2.invoke(registerInstantiator);
+ }
+
+ public void tearDown2() throws Exception {
+ super.tearDown2();
+ DistributedTestCase.cleanupAllVms();
+ }
+
+ // Region: Utility and static functions invoked by the tests
+
+ private static Properties buildProperties(String authenticator,
+ Properties extraProps, String preAccessor, String postAccessor) {
+
+ Properties authProps = new Properties();
+ if (authenticator != null) {
+ authProps.setProperty(
+ DistributionConfig.SECURITY_CLIENT_AUTHENTICATOR_NAME, authenticator);
+ }
+ if (preAccessor != null) {
+ authProps.setProperty(DistributionConfig.SECURITY_CLIENT_ACCESSOR_NAME,
+ preAccessor);
+ }
+ if (postAccessor != null) {
+ authProps.setProperty(
+ DistributionConfig.SECURITY_CLIENT_ACCESSOR_PP_NAME, postAccessor);
+ }
+ if (extraProps != null) {
+ authProps.putAll(extraProps);
+ }
+ return authProps;
+ }
+
+ public static Integer createCacheServer(Integer mcastPort,
+ Properties authProps) {
+
+ if (mcastPort == null) {
+ mcastPort = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ }
+ return SecurityTestUtil.createCacheServer(authProps, null, mcastPort, null,
+ null, Boolean.FALSE, new Integer(SecurityTestUtil.NO_EXCEPTION));
+ }
+
+ public static void createCacheServer(Integer loctorPort, Integer serverPort,
+ Properties authProps) {
+
+ if (loctorPort == null) {
+ loctorPort = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ }
+ SecurityTestUtil.createCacheServer(authProps, null, loctorPort, null,
+ serverPort, Boolean.FALSE, new Integer(SecurityTestUtil.NO_EXCEPTION));
+ }
+
+ // End Region: Utility and static functions invoked by the tests
+
+ // Region: Tests
+
+ public void testAllOpsObjectModWithFailover() {
+
+ OperationWithAction[] allOps = {
+ // Perform CREATE and verify with GET
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.NONE, 8),
+ // For second client connect with valid credentials for key2, key4,
+ // key6, key8 and check that other keys are not accessible
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.CHECK_NOKEY,
+ new int[] { 1, 3, 5, 7 }),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.CHECK_NOKEY
+ | OpFlags.USE_OLDCONN | OpFlags.CHECK_NOTAUTHZ, new int[] { 0, 2,
+ 4, 6 }),
+ // For third client check that key3, key6 are accessible but others are
+ // not
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.CHECK_NOKEY,
+ new int[] { 2, 5 }),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.CHECK_NOKEY
+ | OpFlags.USE_OLDCONN | OpFlags.CHECK_NOTAUTHZ, new int[] { 0, 1,
+ 3, 4, 6, 7 }),
+
+ // OPBLOCK_END indicates end of an operation block that needs to
+ // be executed on each server when doing failover
+ OperationWithAction.OPBLOCK_END,
+
+ // Perform UPDATE and verify with GET
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 8),
+ // For second client check that key2, key4, key6, key8 are accessible
+ // but others are not
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, new int[] { 1, 3, 5, 7 }),
+ new OperationWithAction(OperationCode.GET, 2,
+ OpFlags.USE_OLDCONN | OpFlags.USE_NEWVAL | OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_NOTAUTHZ, new int[] { 0, 2, 4, 6 }),
+ // For third client check that key3, key6 are accessible but others are
+ // not
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, new int[] { 2, 5 }),
+ new OperationWithAction(OperationCode.GET, 3,
+ OpFlags.USE_OLDCONN | OpFlags.USE_NEWVAL | OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_NOTAUTHZ, new int[] { 0, 1, 3, 4, 6, 7 }),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Perform UPDATE and verify with GET_ALL
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 8),
+ // For second client check that key2, key4, key6, key8 are accessible
+ // but others are not; getAll test in doOp uses a combination of local
+ // entries and remote fetches
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL | OpFlags.USE_ALL_KEYS,
+ new int[] { 1, 3, 5, 7 }),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL | OpFlags.USE_ALL_KEYS | OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_FAIL, new int[] { 0, 2, 4, 6 }),
+ // For third client check that key3, key6 are accessible but others are
+ // not
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL | OpFlags.USE_ALL_KEYS, new int[] { 2, 5 }),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL | OpFlags.USE_ALL_KEYS | OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_FAIL, new int[] { 0, 1, 3, 4, 6, 7 }),
+
+ // locally destroy the keys to also test create after failover
+ new OperationWithAction(OperationCode.DESTROY, 1, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 8),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Perform PUTALL and verify with GET
+ new OperationWithAction(OperationCode.PUTALL, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 8),
+ // For second client check that key2, key4, key6, key8 are accessible
+ // but others are not
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, new int[] { 1, 3, 5, 7 }),
+ new OperationWithAction(OperationCode.GET, 2,
+ OpFlags.USE_OLDCONN | OpFlags.USE_NEWVAL | OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_NOTAUTHZ, new int[] { 0, 2, 4, 6 }),
+ // For third client check that key3, key6 are accessible but others are
+ // not
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, new int[] { 2, 5 }),
+ new OperationWithAction(OperationCode.GET, 3,
+ OpFlags.USE_OLDCONN | OpFlags.USE_NEWVAL | OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_NOTAUTHZ, new int[] { 0, 1, 3, 4, 6, 7 }),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Test UPDATE and verify with a QUERY
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN, 8),
+ // For second client check that key2, key4, key6, key8 are accessible
+ // but others are not
+ new OperationWithAction(OperationCode.QUERY, 2, OpFlags.USE_OLDCONN,
+ new int[] { 1, 3, 5, 7 }),
+ new OperationWithAction(OperationCode.QUERY, 2, OpFlags.USE_OLDCONN
+ | OpFlags.CHECK_FAIL, new int[] { 0, 2, 4, 6 }),
+ // For third client check that key3, key6 are accessible but others are
+ // not
+ new OperationWithAction(OperationCode.QUERY, 3, OpFlags.USE_OLDCONN,
+ new int[] { 2, 5 }),
+ new OperationWithAction(OperationCode.QUERY, 3, OpFlags.USE_OLDCONN
+ | OpFlags.CHECK_FAIL, new int[] { 0, 1, 3, 4, 6, 7 }),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Test UPDATE and verify with a EXECUTE_CQ initial results
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 8),
+ // For second client check that key2, key4, key6, key8 are accessible
+ // but others are not
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 2,
+ OpFlags.USE_OLDCONN | OpFlags.USE_NEWVAL, new int[] { 1, 3, 5, 7 }),
+ new OperationWithAction(OperationCode.CLOSE_CQ, 2, OpFlags.USE_OLDCONN,
+ 1),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 2,
+ OpFlags.USE_OLDCONN | OpFlags.USE_NEWVAL | OpFlags.CHECK_FAIL,
+ new int[] { 0, 2, 4, 6 }),
+ new OperationWithAction(OperationCode.CLOSE_CQ, 2, OpFlags.USE_OLDCONN,
+ 1),
+ // For third client check that key3, key6 are accessible but others are
+ // not
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 3,
+ OpFlags.USE_OLDCONN | OpFlags.USE_NEWVAL, new int[] { 2, 5 }),
+ new OperationWithAction(OperationCode.CLOSE_CQ, 3, OpFlags.USE_OLDCONN,
+ 1),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 3,
+ OpFlags.USE_OLDCONN | OpFlags.USE_NEWVAL | OpFlags.CHECK_FAIL,
+ new int[] { 0, 1, 3, 4, 6, 7 }),
+ new OperationWithAction(OperationCode.CLOSE_CQ, 3, OpFlags.USE_OLDCONN,
+ 1),
+
+ OperationWithAction.OPBLOCK_END };
+
+ TestPostCredentialGenerator tgen = new TestPostCredentialGenerator();
+
+ CredentialGenerator gen = new DummyCredentialGenerator();
+ gen.init();
+ Properties extraProps = gen.getSystemProperties();
+ Properties javaProps = gen.getJavaProperties();
+ String authInit = gen.getAuthInit();
+ String authenticator = gen.getAuthenticator();
+
+ getLogWriter().info(
+ "testPutsGetsObjectModWithFailover: Using authinit: " + authInit);
+ getLogWriter().info(
+ "testPutsGetsObjectModWithFailover: Using authenticator: "
+ + authenticator);
+ getLogWriter().info(
+ "testPutsGetsObjectModWithFailover: Using pre-operation accessor: "
+ + preAccessor);
+ getLogWriter().info(
+ "testPutsGetsObjectModWithFailover: Using post-operation accessor: "
+ + postAccessor);
+
+ // Start servers with all required properties
+ Properties serverProps = buildProperties(authenticator, extraProps,
+ preAccessor, postAccessor);
+ // Get ports for the servers
+ Integer port1 = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ Integer port2 = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+
+ // Perform all the ops on the clients
+ List opBlock = new ArrayList();
+ Random rnd = new Random();
+ for (int opNum = 0; opNum < allOps.length; ++opNum) {
+ // Start client with valid credentials as specified in
+ // OperationWithAction
+ OperationWithAction currentOp = allOps[opNum];
+ if (currentOp.equals(OperationWithAction.OPBLOCK_END)
+ || currentOp.equals(OperationWithAction.OPBLOCK_NO_FAILOVER)) {
+ // End of current operation block; execute all the operations
+ // on the servers with failover
+ if (opBlock.size() > 0) {
+ // Start the first server and execute the operation block
+ server1.invoke(ClientAuthorizationTestBase.class,
+ "createCacheServer", new Object[] {
+ SecurityTestUtil.getLocatorPort(), port1, serverProps,
+ javaProps });
+ server2.invoke(SecurityTestUtil.class, "closeCache");
+ executeOpBlock(opBlock, port1, port2, authInit, extraProps, null,
+ tgen, rnd);
+ if (!currentOp.equals(OperationWithAction.OPBLOCK_NO_FAILOVER)) {
+ // Failover to the second server and run the block again
+ server2.invoke(ClientAuthorizationTestBase.class,
+ "createCacheServer", new Object[] {
+ SecurityTestUtil.getLocatorPort(), port2, serverProps,
+ javaProps });
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+ executeOpBlock(opBlock, port1, port2, authInit, extraProps, null,
+ tgen, rnd);
+ }
+ opBlock.clear();
+ }
+ }
+ else {
+ currentOp.setOpNum(opNum);
+ opBlock.add(currentOp);
+ }
+ }
+ }
+
+ private static class MyInstantiator extends Instantiator {
+
+ public MyInstantiator(Class clazz, int classId) {
+ super(clazz, classId);
+ }
+ public MyInstantiator() {
+ this(ObjectWithAuthz.class, ObjectWithAuthz.CLASSID);
+ }
+
+ public DataSerializable newInstance() {
+ return new ObjectWithAuthz();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientCQPostAuthorizationDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientCQPostAuthorizationDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientCQPostAuthorizationDUnitTest.java
new file mode 100644
index 0000000..0155613
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientCQPostAuthorizationDUnitTest.java
@@ -0,0 +1,514 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.security;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+
+import security.AuthzCredentialGenerator;
+import security.CredentialGenerator;
+
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.operations.OperationContext.OperationCode;
+import com.gemstone.gemfire.cache.query.CqAttributes;
+import com.gemstone.gemfire.cache.query.CqAttributesFactory;
+import com.gemstone.gemfire.cache.query.CqException;
+import com.gemstone.gemfire.cache.query.CqListener;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.QueryService;
+import com.gemstone.gemfire.cache.query.SelectResults;
+import com.gemstone.gemfire.cache.query.cq.dunit.CqQueryTestListener;
+import com.gemstone.gemfire.cache.query.internal.cq.ClientCQImpl;
+import com.gemstone.gemfire.cache.query.internal.cq.CqService;
+import com.gemstone.gemfire.cache.query.internal.cq.InternalCqQuery;
+import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
+import com.gemstone.gemfire.internal.AvailablePort;
+import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
+import com.gemstone.gemfire.internal.logging.InternalLogWriter;
+
+import dunit.DistributedTestCase;
+import dunit.Host;
+import dunit.SerializableRunnable;
+/**
+ * This is for multiuser-authentication
+ *
+ * @author ashetkar
+ *
+ */
+public class ClientCQPostAuthorizationDUnitTest extends
+ ClientAuthorizationTestBase {
+
+// public static final String regionName = "ClientCQPostAuthorizationDUnitTest_region";
+
+ public static final Map<String, String> cqNameToQueryStrings = new HashMap<String, String>();
+
+ static {
+ cqNameToQueryStrings.put("CQ_0", "SELECT * FROM ");
+ cqNameToQueryStrings.put("CQ_1", "SELECT * FROM ");
+ }
+
+ public ClientCQPostAuthorizationDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+
+ super.setUp();
+ getSystem();
+ invokeInEveryVM(new SerializableRunnable("getSystem") {
+ public void run() {
+ getSystem();
+ }
+ });
+
+ final Host host = Host.getHost(0);
+ server1 = host.getVM(0);
+ server2 = host.getVM(1);
+ client1 = host.getVM(2);
+ client2 = host.getVM(3);
+
+ server1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { serverExpectedExceptions });
+ server2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { serverExpectedExceptions });
+ client2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { clientExpectedExceptions });
+ SecurityTestUtil.registerExpectedExceptions(clientExpectedExceptions);
+ }
+
+ public void tearDown2() throws Exception {
+ super.tearDown2();
+ client1.invoke(SecurityTestUtil.class, "closeCache");
+ client2.invoke(SecurityTestUtil.class, "closeCache");
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+ server2.invoke(SecurityTestUtil.class, "closeCache");
+ }
+
+ public void testAllowCQForAllMultiusers() throws Exception {
+ /*
+ * Start a server
+ * Start a client1 with two users with valid credentials and post-authz'ed for CQ
+ * Each user registers a unique CQ
+ * Client2 does some operations on the region which satisfies both the CQs
+ * Validate that listeners for both the CQs are invoked.
+ */
+ doStartUp(Integer.valueOf(2), Integer.valueOf(5), new Boolean[] {true,
+ true});
+ }
+
+ public void testDisallowCQForAllMultiusers() throws Exception {
+ /*
+ * Start a server
+ * Start a client1 with two users with valid credentials but not post-authz'ed for CQ
+ * Each user registers a unique CQ
+ * Client2 does some operations on the region which satisfies both the CQs
+ * Validate that listeners for none of the CQs are invoked.
+ */
+ doStartUp(Integer.valueOf(2), Integer.valueOf(5), new Boolean[] {false,
+ false});
+ }
+
+ public void testDisallowCQForSomeMultiusers() throws Exception {
+ /*
+ * Start a server
+ * Start a client1 with two users with valid credentials
+ * User1 is post-authz'ed for CQ but user2 is not.
+ * Each user registers a unique CQ
+ * Client2 does some operations on the region which satisfies both the CQs
+ * Validate that listener for User1's CQ is invoked but that for User2's CQ is not invoked.
+ */
+ doStartUp(Integer.valueOf(2), Integer.valueOf(5), new Boolean[] {true,
+ false});
+ }
+
+ public void testAllowCQForAllMultiusersWithFailover() throws Exception {
+ /*
+ * Start a server1
+ * Start a client1 with two users with valid credentials and post-authz'ed for CQ
+ * Each user registers a unique CQ
+ * Client2 does some operations on the region which satisfies both the CQs
+ * Validate that listeners for both the CQs are invoked.
+ * Start server2 and shutdown server1
+ * Client2 does some operations on the region which satisfies both the CQs
+ * Validate that listeners for both the CQs are get updates.
+ */
+ doStartUp(Integer.valueOf(2), Integer.valueOf(5), new Boolean[] {true,
+ true}, Boolean.TRUE);
+ }
+
+ public void doStartUp(Integer numOfUsers, Integer numOfPuts,
+ Boolean[] postAuthzAllowed) throws Exception {
+ doStartUp(numOfUsers, numOfPuts, postAuthzAllowed, Boolean.FALSE /* failover */);
+ }
+
+ public void doStartUp(Integer numOfUsers, Integer numOfPuts,
+ Boolean[] postAuthzAllowed, Boolean failover) throws Exception {
+ AuthzCredentialGenerator gen = this.getXmlAuthzGenerator();
+ CredentialGenerator cGen = gen.getCredentialGenerator();
+ Properties extraAuthProps = cGen.getSystemProperties();
+ Properties javaProps = cGen.getJavaProperties();
+ Properties extraAuthzProps = gen.getSystemProperties();
+ String authenticator = cGen.getAuthenticator();
+ String accessor = gen.getAuthorizationCallback();
+ String authInit = cGen.getAuthInit();
+ TestAuthzCredentialGenerator tgen = new TestAuthzCredentialGenerator(gen);
+
+ Properties serverProps = buildProperties(authenticator, accessor, true,
+ extraAuthProps, extraAuthzProps);
+
+ Properties opCredentials;
+ cGen = tgen.getCredentialGenerator();
+ Properties javaProps2 = null;
+ if (cGen != null) {
+ javaProps2 = cGen.getJavaProperties();
+ }
+
+ int[] indices = new int[numOfPuts];
+ for (int index = 0; index < numOfPuts; ++index) {
+ indices[index] = index;
+ }
+
+ Random rnd = new Random();
+ Properties[] authProps = new Properties[numOfUsers];
+ for (int i = 0; i < numOfUsers; i++) {
+ int rand = rnd.nextInt(100) + 1;
+ if (postAuthzAllowed[i]) {
+ opCredentials = tgen.getAllowedCredentials(new OperationCode[] {
+ OperationCode.EXECUTE_CQ, OperationCode.GET}, // For callback, GET should be allowed
+ new String[] {regionName}, indices, rand);
+// authProps[i] = gen.getAllowedCredentials(
+// new OperationCode[] {OperationCode.EXECUTE_CQ},
+// new String[] {regionName}, rnd.nextInt(100) + 1);
+ } else {
+ opCredentials = tgen.getDisallowedCredentials(new OperationCode[] {
+ OperationCode.GET}, // For callback, GET should be disallowed
+ new String[] {regionName}, indices, rand);
+// authProps[i] = gen.getDisallowedCredentials(
+// new OperationCode[] {OperationCode.EXECUTE_CQ},
+// new String[] {regionName}, rnd.nextInt(100) + 1);
+ }
+ authProps[i] = SecurityTestUtil.concatProperties(new Properties[] {
+ opCredentials, extraAuthProps, extraAuthzProps});
+ }
+
+ // Get ports for the servers
+ Integer port1 = Integer.valueOf(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ Integer port2 = Integer.valueOf(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ Integer locatorPort = Integer.valueOf(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ // Close down any running servers
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+ server2.invoke(SecurityTestUtil.class, "closeCache");
+
+ server1.invoke(ClientCQPostAuthorizationDUnitTest.class,
+ "createServerCache", new Object[] {serverProps, javaProps, locatorPort, port1});
+ client1.invoke(ClientCQPostAuthorizationDUnitTest.class,
+ "createClientCache", new Object[] {javaProps2, authInit, authProps,
+ new Integer[] {port1, port2}, numOfUsers, postAuthzAllowed});
+ client2.invoke(ClientCQPostAuthorizationDUnitTest.class,
+ "createClientCache", new Object[] {javaProps2, authInit, authProps,
+ new Integer[] {port1, port2}, numOfUsers, postAuthzAllowed});
+
+ client1.invoke(ClientCQPostAuthorizationDUnitTest.class, "createCQ",
+ new Object[] {numOfUsers});
+ client1.invoke(ClientCQPostAuthorizationDUnitTest.class, "executeCQ",
+ new Object[] {numOfUsers, new Boolean[] {false, false}, numOfPuts,
+ new String[numOfUsers], postAuthzAllowed});
+
+ client2.invoke(ClientCQPostAuthorizationDUnitTest.class, "doPuts",
+ new Object[] {numOfPuts, Boolean.TRUE/* put last key */});
+ if (!postAuthzAllowed[0]) {
+ // There is no point waiting as no user is authorized to receive cq events.
+ try {Thread.sleep(1000);} catch (InterruptedException ie) {}
+ } else {
+ client1.invoke(ClientCQPostAuthorizationDUnitTest.class,
+ "waitForLastKey", new Object[] {Integer.valueOf(0)});
+ if (postAuthzAllowed[1]) {
+ client1.invoke(ClientCQPostAuthorizationDUnitTest.class,
+ "waitForLastKey", new Object[] {Integer.valueOf(1)});
+ }
+ }
+ client1.invoke(ClientCQPostAuthorizationDUnitTest.class,
+ "checkCQListeners", new Object[] {numOfUsers, postAuthzAllowed,
+ numOfPuts + 1/* last key */, 0, !failover});
+ if (failover) {
+ server2.invoke(ClientCQPostAuthorizationDUnitTest.class,
+ "createServerCache", new Object[] {serverProps, javaProps, locatorPort, port2});
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+
+ // Allow time for client1 to register its CQs on server2
+ server2.invoke(ClientCQPostAuthorizationDUnitTest.class,
+ "allowCQsToRegister", new Object[] {Integer.valueOf(2)});
+
+ client2.invoke(ClientCQPostAuthorizationDUnitTest.class, "doPuts",
+ new Object[] {numOfPuts, Boolean.TRUE/* put last key */});
+ client1.invoke(ClientCQPostAuthorizationDUnitTest.class,
+ "waitForLastKeyUpdate", new Object[] {Integer.valueOf(0)});
+ client1.invoke(ClientCQPostAuthorizationDUnitTest.class,
+ "checkCQListeners", new Object[] {numOfUsers, postAuthzAllowed,
+ numOfPuts + 1/* last key */, numOfPuts + 1/* last key */,
+ Boolean.TRUE});
+ }
+ }
+
+ public static void createServerCache(Properties serverProps,
+ Properties javaProps, Integer serverPort) {
+ Integer locatorPort = Integer.valueOf(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ SecurityTestUtil.createCacheServer((Properties)serverProps, javaProps,
+ locatorPort, null, serverPort, Boolean.TRUE, Integer.valueOf(
+ SecurityTestUtil.NO_EXCEPTION));
+ }
+
+ public static void createServerCache(Properties serverProps,
+ Properties javaProps, Integer locatorPort, Integer serverPort) {
+ SecurityTestUtil.createCacheServer((Properties)serverProps, javaProps,
+ locatorPort, null, serverPort, Boolean.TRUE, Integer.valueOf(
+ SecurityTestUtil.NO_EXCEPTION));
+ }
+
+ public static void createClientCache(Properties javaProps, String authInit,
+ Properties[] authProps, Integer ports[], Integer numOfUsers,
+ Boolean[] postAuthzAllowed) {
+ SecurityTestUtil.createCacheClientForMultiUserMode(numOfUsers, authInit,
+ authProps, javaProps, ports, null, Boolean.FALSE,
+ SecurityTestUtil.NO_EXCEPTION);
+ }
+
+ public static void createCQ(Integer num) {
+ for (int i = 0; i < num; i++) {
+ QueryService cqService = SecurityTestUtil.proxyCaches[i].getQueryService();
+ String cqName = "CQ_" + i;
+ String queryStr = cqNameToQueryStrings.get(cqName)
+ + SecurityTestUtil.proxyCaches[i].getRegion(regionName).getFullPath();
+ // Create CQ Attributes.
+ CqAttributesFactory cqf = new CqAttributesFactory();
+ CqListener[] cqListeners = {new CqQueryTestListener(getLogWriter())};
+ ((CqQueryTestListener)cqListeners[0]).cqName = cqName;
+
+ cqf.initCqListeners(cqListeners);
+ CqAttributes cqa = cqf.create();
+
+ // Create CQ.
+ try {
+ CqQuery cq1 = cqService.newCq(cqName, queryStr, cqa);
+ assertTrue("newCq() state mismatch", cq1.getState().isStopped());
+ } catch (Exception ex) {
+ AssertionError err = new AssertionError("Failed to create CQ " + cqName
+ + " . ");
+ err.initCause(ex);
+ getLogWriter().info("CqService is :" + cqService, err);
+ throw err;
+ }
+ }
+ }
+
+ public static void executeCQ(Integer num, Boolean[] initialResults,
+ Integer expectedResultsSize, String[] expectedErr, Boolean[] postAuthzAllowed) {
+ InternalLogWriter logWriter = InternalDistributedSystem.getStaticInternalLogWriter();
+ for (int i = 0; i < num; i++) {
+ try {
+ if (expectedErr[i] != null) {
+ logWriter.info(
+ "<ExpectedException action=add>" + expectedErr[i]
+ + "</ExpectedException>");
+ }
+ CqQuery cq1 = null;
+ String cqName = "CQ_" + i;
+ String queryStr = cqNameToQueryStrings.get(cqName)
+ + SecurityTestUtil.proxyCaches[i].getRegion(regionName)
+ .getFullPath();
+ QueryService cqService = SecurityTestUtil.proxyCaches[i]
+ .getQueryService();
+
+ // Get CqQuery object.
+ try {
+ cq1 = cqService.getCq(cqName);
+ if (cq1 == null) {
+ getLogWriter().info(
+ "Failed to get CqQuery object for CQ name: " + cqName);
+ fail("Failed to get CQ " + cqName);
+ } else {
+ getLogWriter().info("Obtained CQ, CQ name: " + cq1.getName());
+ assertTrue("newCq() state mismatch", cq1.getState().isStopped());
+ }
+ } catch (Exception ex) {
+ getLogWriter().info("CqService is :" + cqService);
+ getLogWriter().error(ex);
+ AssertionError err = new AssertionError("Failed to execute CQ "
+ + cqName);
+ err.initCause(ex);
+ throw err;
+ }
+
+ if (initialResults[i]) {
+ SelectResults cqResults = null;
+
+ try {
+ cqResults = cq1.executeWithInitialResults();
+ } catch (CqException ce) {
+ if (ce.getCause() instanceof NotAuthorizedException && !postAuthzAllowed[i]) {
+ getLogWriter().info("Got expected exception for CQ " + cqName);
+ } else {
+ getLogWriter().info("CqService is: " + cqService);
+ ce.printStackTrace();
+ AssertionError err = new AssertionError("Failed to execute CQ "
+ + cqName);
+ err.initCause(ce);
+ throw err;
+ }
+ } catch (Exception ex) {
+ getLogWriter().info("CqService is: " + cqService);
+ ex.printStackTrace();
+ AssertionError err = new AssertionError("Failed to execute CQ "
+ + cqName);
+ err.initCause(ex);
+ throw err;
+ }
+ getLogWriter().info("initial result size = " + cqResults.size());
+ assertTrue("executeWithInitialResults() state mismatch", cq1
+ .getState().isRunning());
+ if (expectedResultsSize >= 0) {
+ assertEquals("unexpected results size", expectedResultsSize
+ .intValue(), cqResults.size());
+ }
+ } else {
+ try {
+ cq1.execute();
+ } catch (CqException ce) {
+ if (ce.getCause() instanceof NotAuthorizedException && !postAuthzAllowed[i]) {
+ getLogWriter().info("Got expected exception for CQ " + cqName);
+ } else {
+ getLogWriter().info("CqService is: " + cqService);
+ ce.printStackTrace();
+ AssertionError err = new AssertionError("Failed to execute CQ "
+ + cqName);
+ err.initCause(ce);
+ throw err;
+ }
+ } catch (Exception ex) {
+ AssertionError err = new AssertionError("Failed to execute CQ "
+ + cqName);
+ err.initCause(ex);
+ if (expectedErr == null) {
+ getLogWriter().info("CqService is: " + cqService, err);
+ }
+ throw err;
+ }
+ assertTrue("execute() state mismatch", cq1.getState().isRunning() == postAuthzAllowed[i]);
+ }
+ } finally {
+ if (expectedErr[i] != null) {
+ logWriter.info(
+ "<ExpectedException action=remove>" + expectedErr[i]
+ + "</ExpectedException>");
+ }
+ }
+ }
+ }
+
+ public static void doPuts(Integer num, Boolean putLastKey) {
+// Region region = GemFireCache.getInstance().getRegion(regionName);
+ Region region = SecurityTestUtil.proxyCaches[0].getRegion(regionName);
+ for (int i = 0; i < num; i++) {
+ region.put("CQ_key"+i, "CQ_value"+i);
+ }
+ if (putLastKey) {
+ region.put("LAST_KEY", "LAST_KEY");
+ }
+ }
+
+ public static void putLastKey() {
+ Region region = GemFireCacheImpl.getInstance().getRegion(regionName);
+ region.put("LAST_KEY", "LAST_KEY");
+ }
+
+ public static void waitForLastKey(Integer cqIndex) {
+ String cqName = "CQ_" + cqIndex;
+ QueryService qService = SecurityTestUtil.proxyCaches[cqIndex].getQueryService();
+ ClientCQImpl cqQuery = (ClientCQImpl)qService.getCq(cqName);
+ ((CqQueryTestListener)cqQuery.getCqListeners()[0])
+ .waitForCreated("LAST_KEY");
+// WaitCriterion wc = new WaitCriterion() {
+// public boolean done() {
+// Region region = GemFireCache.getInstance().getRegion(regionName);
+// Region.Entry entry = region.getEntry("LAST_KEY");
+// if (entry != null && entry.getValue() != null) {
+// return false;
+// } else if (entry.getValue() != null) {
+// return true;
+// }
+// return false;
+// }
+// public String description() {
+// return "Last key not received.";
+// }
+// };
+// DistributedTestCase.waitForCriterion(wc, 60 * 1000, 100, false);
+ }
+
+ public static void waitForLastKeyUpdate(Integer cqIndex) {
+ String cqName = "CQ_" + cqIndex;
+ QueryService qService = SecurityTestUtil.proxyCaches[cqIndex].getQueryService();
+ ClientCQImpl cqQuery = (ClientCQImpl)qService.getCq(cqName);
+ ((CqQueryTestListener)cqQuery.getCqListeners()[0])
+ .waitForUpdated("LAST_KEY");
+ }
+
+ public static void allowCQsToRegister(Integer number) {
+ final int num = number;
+ WaitCriterion wc = new WaitCriterion() {
+ public boolean done() {
+ CqService cqService = GemFireCacheImpl.getInstance().getCqService();
+ cqService.start();
+ Collection<? extends InternalCqQuery> cqs = cqService.getAllCqs();
+ if (cqs != null) {
+ return cqs.size() >= num;
+ } else {
+ return false;
+ }
+ }
+
+ public String description() {
+ return num + "Waited for " + num
+ + " CQs to be registered on this server.";
+ }
+ };
+ DistributedTestCase.waitForCriterion(wc, 60 * 1000, 100, false);
+ }
+
+ public static void checkCQListeners(Integer numOfUsers,
+ Boolean[] expectedListenerInvocation, Integer createEventsSize,
+ Integer updateEventsSize, Boolean closeCache) {
+ for (int i = 0; i < numOfUsers; i++) {
+ String cqName = "CQ_" + i;
+ QueryService qService = SecurityTestUtil.proxyCaches[i].getQueryService();
+ ClientCQImpl cqQuery = (ClientCQImpl)qService.getCq(cqName);
+ if (expectedListenerInvocation[i]) {
+ for (CqListener listener : cqQuery.getCqListeners()) {
+ assertEquals(createEventsSize.intValue(),
+ ((CqQueryTestListener)listener).getCreateEventCount());
+ assertEquals(updateEventsSize.intValue(),
+ ((CqQueryTestListener)listener).getUpdateEventCount());
+ }
+ } else {
+ for (CqListener listener : cqQuery.getCqListeners()) {
+ assertEquals(0, ((CqQueryTestListener)listener).getTotalEventCount());
+ }
+ }
+ if (closeCache) {
+ SecurityTestUtil.proxyCaches[i].close();
+ }
+ }
+ }
+}