<commit_before>package org.I0Itec.zkclient;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

public class ZkClientTest {

    private static final Logger LOG = Logger.getLogger(ZkClientTest.class);

    @Before
    public void setUp() {
        LOG.info("------------ BEFORE -------------");
    }

    @After
    public void tearDown() {
        LOG.info("------------ AFTER -------------");
    }

    @Test(expected = IOException.class, timeout = 5000)
    public void testUnableToConnect() throws Exception {
        LOG.info("--- testUnableToConnect");
        // we are using port 4711 to avoid conflicts with the zk server that is
        // started by the Spring context
        new ZkClient("localhost:4711", 1000);
    }

    @Test
    public void testWriteAndRead() throws Exception {
        LOG.info("--- testWriteAndRead");
        ZkServer zkServer = ZkTestUtil.startZkServer("ZkClientTest-testWriteAndRead", 4711);
        ZkClient client = new ZkClient("localhost:4711", 5000);
        String data = "something";
        String path = "/a";
        client.createPersistent(path, data);
        String data2 = client.readData(path);
        Assert.assertEquals(data, data2);
        client.delete(path);

        client.close();
        zkServer.shutdown();
        zkServer.join();
    }

    @Test
    public void testDelete() throws Exception {
        LOG.info("--- testDelete");
        ZkServer zkServer = ZkTestUtil.startZkServer("ZkClientTest-testDelete", 4711);
        ZkClient client = new ZkClient("localhost:4711", 5000);
        String path = "/a";
        assertFalse(client.delete(path));
        client.createPersistent(path, null);
        assertTrue(client.delete(path));
        assertFalse(client.delete(path));

        client.close();
        zkServer.shutdown();
        zkServer.join();
    }

    @Test
    public void testDeleteRecursive() throws Exception {
        LOG.info("--- testDeleteRecursive");
        ZkServer zkServer = ZkTestUtil.startZkServer("ZkClientTest-testDeleteRecursive", 4711);
        ZkClient client = new ZkClient("localhost:4711", 5000);

        // should be able to call this on a not existing directory
        client.deleteRecursive("/doesNotExist");

        client.close();
        zkServer.shutdown();
        zkServer.join();
    }

    @Test(timeout = 15000)
    public void testRetryUntilConnected() throws Exception {
        LOG.info("--- testRetryUntilConnected");
        final ZkServer zkServer = ZkTestUtil.startZkServer("ZkClientTest-testRetryUntilConnected", 4711);
        final ZkConnection connection = new ZkConnection("localhost:4711");
        final ZkClient client = new ZkClient(connection);

        zkServer.shutdown();
        zkServer.join();

        // start server in 250ms
        new DeferredZookeeperStarter(zkServer, 250).start();

        // this should work as soon as the connection is reestablished, if it
        // fails it throws a ConnectionLossException
        client.retryUntilConnected(new Callable<Object>() {

            @Override
            public Object call() throws Exception {
                connection.exists("/a", false);
                return null;
            }
        });

        client.close();
        zkServer.shutdown();
        zkServer.join();
    }

    @Test(timeout = 15000)
    public void testWaitUntilConnected() throws Exception {
        LOG.info("--- testWaitUntilConnected");
        ZkServer zkServer = ZkTestUtil.startZkServer("ZkClientTest-testWaitUntilConnected", 4711);
        ZkClient client = new ZkClient("localhost:4711", 5000);

        zkServer.shutdown();
        zkServer.join();

        // the client state should change to KeeperState.Disconnected
        assertTrue(client.waitForKeeperState(KeeperState.Disconnected, 1, TimeUnit.SECONDS));

        // connection should not be possible and timeout after 100ms
        assertFalse(client.waitUntilConnected(100, TimeUnit.MILLISECONDS));

        client.close();
        zkServer.shutdown();
        zkServer.join();
    }

    @Test
    public void testWaitUntilExists() throws InterruptedException, IOException, KeeperException {
        LOG.info("--- testWaitUntilExists");
        ZkServer zkServer = ZkTestUtil.startZkServer("ZkClientTest-testWaitUntilExists", 4711);
        final ZkClient client = new ZkClient("localhost:4711", 5000);

        // create /gaga node asynchronously
        new Thread() {
            @Override
            public void run() {
                try {
                    Thread.sleep(100);
                    client.createPersistent("/gaga");
                } catch (Exception e) {
                    // ignore
                }
            }
        }.start();

        // wait until this was created
        assertTrue(client.waitUntilExists("/gaga", TimeUnit.SECONDS, 5));
        assertTrue(client.exists("/gaga"));

        // waiting for /neverCreated should timeout
        assertFalse(client.waitUntilExists("/neverCreated", TimeUnit.MILLISECONDS, 100));

        client.close();
        zkServer.shutdown();
        zkServer.join();
    }

    @Test
    public void testDataChanges() throws InterruptedException, IOException, KeeperException {
        ZkServer zkServer = ZkTestUtil.startZkServer("ZkClientTest-testWaitUntilExists", 4711);
        final ZkClient client = new ZkClient("localhost:4711", 5000);
        String path = "/a";
        final Holder<String> holder = new Holder<String>();

        IZkDataListener<String> listener = new IZkDataListener<String>() {

            public void handleDataDeleted(String dataPath) {
                holder.set(null);
            }

            public void handleDataChange(String dataPath, String data) {
                holder.set(data);
            }

            public void handleDataAdded(String dataPath, String data) {
                holder.set(data);
            }
        };
        client.subscribeDataChanges(path, listener);
        client.createPersistent(path, "aaa");
        // wait some time to make sure the event was triggered
        Thread.sleep(500);
        assertEquals("aaa", holder.get());

        client.close();
        zkServer.shutdown();
        zkServer.join();
    }
}<commit_msg>refactor the test so it can be used for in memory connection as well.<commit_after>package org.I0Itec.zkclient;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

public class ZkClientTest {

	private static final Logger LOG = Logger.getLogger(ZkClientTest.class);
	private ZkServer _zkServer;
	private ZkClient _client;
	private AtomicInteger _counter = new AtomicInteger();

	@Before
	public void setUp() throws InterruptedException, IOException,
			KeeperException {
		LOG.info("------------ BEFORE -------------");
		_zkServer = ZkTestUtil.startZkServer("ZkClientTest_"+_counter.addAndGet(1), 4711);
		_client = new ZkClient("localhost:4711", 5000);
	}

	@After
	public void tearDown() throws InterruptedException {
		LOG.info("------------ AFTER -------------");
		_client.close();
		_zkServer.shutdown();
		_zkServer.join();
	}

	@Test(expected = IOException.class, timeout = 5000)
	public void testUnableToConnect() throws Exception {
		LOG.info("--- testUnableToConnect");
		// trying to a non existing zk server
		new ZkClient("localhost:4799", 1000);
	}

	@Test
	public void testWriteAndRead() throws Exception {
		LOG.info("--- testWriteAndRead");

		String data = "something";
		String path = "/a";
		_client.createPersistent(path, data);
		String data2 = _client.readData(path);
		Assert.assertEquals(data, data2);
		_client.delete(path);

	}

	@Test
	public void testDelete() throws Exception {
		LOG.info("--- testDelete");
		String path = "/a";
		assertFalse(_client.delete(path));
		_client.createPersistent(path, null);
		assertTrue(_client.delete(path));
		assertFalse(_client.delete(path));
	}

	@Test
	public void testDeleteRecursive() throws Exception {
		LOG.info("--- testDeleteRecursive");
		ZkClient _client = new ZkClient("localhost:4711", 5000);
		// should be able to call this on a not existing directory
		_client.deleteRecursive("/doesNotExist");
	}

	@Test(timeout = 15000)
	public void testRetryUntilConnected() throws Exception {
		LOG.info("--- testRetryUntilConnected");
		final ZkConnection connection = new ZkConnection("localhost:4711");
		final ZkClient _client = new ZkClient(connection);
		_zkServer.shutdown();
		_zkServer.join();

		// start server in 250ms
		new DeferredZookeeperStarter(_zkServer, 250).start();

		// this should work as soon as the connection is reestablished, if it
		// fails it throws a ConnectionLossException
		_client.retryUntilConnected(new Callable<Object>() {

			@Override
			public Object call() throws Exception {
				connection.exists("/a", false);
				return null;
			}
		});
	}

	@Test(timeout = 15000)
	public void testWaitUntilConnected() throws Exception {
		LOG.info("--- testWaitUntilConnected");
		ZkClient _client = new ZkClient("localhost:4711", 5000);

		_zkServer.shutdown();
		_zkServer.join();

		// the _client state should change to KeeperState.Disconnected
		assertTrue(_client.waitForKeeperState(KeeperState.Disconnected, 1,
				TimeUnit.SECONDS));

		// connection should not be possible and timeout after 100ms
		assertFalse(_client.waitUntilConnected(100, TimeUnit.MILLISECONDS));
	}

	@Test
	public void testWaitUntilExists() throws InterruptedException, IOException,
			KeeperException {
		LOG.info("--- testWaitUntilExists");

		// create /gaga node asynchronously
		new Thread() {
			@Override
			public void run() {
				try {
					Thread.sleep(100);
					_client.createPersistent("/gaga");
				} catch (Exception e) {
					// ignore
				}
			}
		}.start();

		// wait until this was created
		assertTrue(_client.waitUntilExists("/gaga", TimeUnit.SECONDS, 5));
		assertTrue(_client.exists("/gaga"));

		// waiting for /neverCreated should timeout
		assertFalse(_client.waitUntilExists("/neverCreated",
				TimeUnit.MILLISECONDS, 100));
	}
}
{'fraction_non_alphanumeric': 0.12032159689492654, 'fraction_numerical': 0.01959153497828297, 'mean_word_length': 2.9309843806756266, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6641634', 'n_tokens_mistral': 3402, 'n_tokens_neox': 3042, 'n_words': 729}