<commit_before>package com.palominolabs.benchpress;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.inject.AbstractModule;
import com.google.inject.Inject;
import com.google.inject.Stage;
import com.ning.http.client.AsyncCompletionHandler;
import com.ning.http.client.AsyncHttpClient;
import com.ning.http.client.Response;
import com.palominolabs.benchpress.config.ZookeeperConfig;
import com.palominolabs.benchpress.controller.ControllerResourceModule;
import com.palominolabs.benchpress.controller.JobFarmer;
import com.palominolabs.benchpress.controller.zookeeper.ZKServer;
import com.palominolabs.benchpress.controller.zookeeper.ZKServerConfig;
import com.palominolabs.benchpress.controller.zookeeper.ZKServerModule;
import com.palominolabs.benchpress.curator.InstanceSerializerModule;
import com.palominolabs.benchpress.http.server.DefaultJerseyServletModule;
import com.palominolabs.benchpress.ipc.Ipc;
import com.palominolabs.benchpress.ipc.IpcHttpClientModule;
import com.palominolabs.benchpress.ipc.IpcJsonModule;
import com.palominolabs.benchpress.job.json.Job;
import com.palominolabs.benchpress.job.json.Task;
import com.palominolabs.benchpress.job.registry.JobRegistryModule;
import com.palominolabs.benchpress.job.task.TaskFactoryFactoryRegistryModule;
import com.palominolabs.benchpress.job.task.TaskPartitionerRegistryModule;
import com.palominolabs.benchpress.task.reporting.TaskProgressClientModule;
import com.palominolabs.benchpress.task.simplehttp.SimpleHttpTaskModule;
import com.palominolabs.benchpress.worker.WorkerAdvertiser;
import com.palominolabs.benchpress.worker.WorkerControlFactory;
import com.palominolabs.benchpress.worker.WorkerMetadata;
import com.palominolabs.benchpress.worker.http.WorkerResourceModule;
import com.palominolabs.benchpress.zookeeper.CuratorModule;
import com.palominolabs.config.ConfigModuleBuilder;
import com.palominolabs.http.server.HttpServer;
import com.palominolabs.http.server.HttpServerConfig;
import com.palominolabs.http.server.HttpServerFactory;
import org.apache.commons.configuration.MapConfiguration;
import org.apache.commons.io.FileUtils;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceInstance;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.bridge.SLF4JBridgeHandler;

import javax.ws.rs.core.MediaType;
import java.io.File;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.LogManager;

import static com.google.inject.Guice.createInjector;
import static com.mongodb.util.MyAsserts.assertFalse;
import static javax.ws.rs.core.Response.Status.ACCEPTED;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

public class SingleVmIntegrationTest {

    @Inject
    WorkerAdvertiser workerAdvertiser;
    @Inject
    HttpServerFactory httpServerFactory;
    @Inject
    ZKServer zkServer;
    @Inject
    ZKServerConfig zkServerConfig;
    @Inject
    CuratorModule.CuratorLifecycleHook curatorLifecycleHook;
    @Inject
    @Ipc
    AsyncHttpClient asyncHttpClient;
    @Inject
    ServiceDiscovery<WorkerMetadata> serviceDiscovery;
    @Inject
    ZookeeperConfig zookeeperConfig;
    @Inject
    WorkerControlFactory workerControlFactory;
    @Inject
    JobFarmer jobFarmer;
    @Ipc
    @Inject
    ObjectWriter objectWriter;
    @Ipc
    @Inject
    ObjectReader objectReader;
    @Inject
    SimpleHttpResource simpleHttpResource;

    ExecutorService executorService;
    HttpServer httpServer;
    private HttpServerConfig httpServerConfig;

    @BeforeClass
    public static void setUpClass() {
        LogManager.getLogManager().reset();
        SLF4JBridgeHandler.install();
    }

    @Before
    public void setUp() throws Exception {

        final Map<String, Object> configMap = new HashMap<String, Object>();

        createInjector(Stage.PRODUCTION, new AbstractModule() {
            @Override
            protected void configure() {
                install(new ConfigModuleBuilder().addConfiguration(new MapConfiguration(configMap))
                    .build());

                // basic zookeeper
                install(new ZKServerModule());
                install(new CuratorModule());

                install(new DefaultJerseyServletModule());

                // controller
                install(new InstanceSerializerModule());
                install(new IpcJsonModule());
                install(new TaskPartitionerRegistryModule());
                install(new ControllerResourceModule());

                // worker
                install(new JobRegistryModule());
                install(new TaskProgressClientModule());
                install(new IpcHttpClientModule());
                install(new TaskFactoryFactoryRegistryModule());
                install(new WorkerResourceModule());

                // custom task
                install(new SimpleHttpTaskModule());
                bind(SimpleHttpResource.class);
            }
        }).injectMembers(this);

        FileUtils.deleteDirectory(new File(zkServerConfig.getTmpDir()));

        executorService = Executors.newCachedThreadPool();
        executorService.submit(zkServer);

        httpServerConfig = new HttpServerConfig();
        httpServer = httpServerFactory.getHttpServer(this.httpServerConfig);
        httpServer.start();

        curatorLifecycleHook.start();

        jobFarmer.setListenAddress(httpServer.getHttpListenHost());
        jobFarmer.setListenPort(httpServer.getHttpListenPort());
    }

    @After
    public void tearDown() throws Exception {
        httpServer.stop();

        executorService.shutdownNow();

        serviceDiscovery.close();
    }

    @Test
    public void testWorkerLockDeAdvertises() throws Exception {
        WorkerMetadata workerMetadata = advertiseWorker();

        assertTrue(lock(workerMetadata));

        assertNoWorkersAdvertised();
    }

    @Test
    public void testWorkerUnlockReAdvertises() throws Exception {
        WorkerMetadata workerMetadata = advertiseWorker();

        lock(workerMetadata);

        assertNoWorkersAdvertised();

        assertTrue(unlock(workerMetadata));
<commit_msg>Still can't reproduce Drew's test failures -- add more careful inter-test cleanup<commit_after>package com.palominolabs.benchpress;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.inject.AbstractModule;
import com.google.inject.Inject;
import com.google.inject.Stage;
import com.ning.http.client.AsyncCompletionHandler;
import com.ning.http.client.AsyncHttpClient;
import com.ning.http.client.Response;
import com.palominolabs.benchpress.config.ZookeeperConfig;
import com.palominolabs.benchpress.controller.ControllerResourceModule;
import com.palominolabs.benchpress.controller.JobFarmer;
import com.palominolabs.benchpress.controller.zookeeper.ZKServer;
import com.palominolabs.benchpress.controller.zookeeper.ZKServerConfig;
import com.palominolabs.benchpress.controller.zookeeper.ZKServerModule;
import com.palominolabs.benchpress.curator.InstanceSerializerModule;
import com.palominolabs.benchpress.http.server.DefaultJerseyServletModule;
import com.palominolabs.benchpress.ipc.Ipc;
import com.palominolabs.benchpress.ipc.IpcHttpClientModule;
import com.palominolabs.benchpress.ipc.IpcJsonModule;
import com.palominolabs.benchpress.job.json.Job;
import com.palominolabs.benchpress.job.json.Task;
import com.palominolabs.benchpress.job.registry.JobRegistryModule;
import com.palominolabs.benchpress.job.task.TaskFactoryFactoryRegistryModule;
import com.palominolabs.benchpress.job.task.TaskPartitionerRegistryModule;
import com.palominolabs.benchpress.task.reporting.TaskProgressClientModule;
import com.palominolabs.benchpress.task.simplehttp.SimpleHttpTaskModule;
import com.palominolabs.benchpress.worker.WorkerAdvertiser;
import com.palominolabs.benchpress.worker.WorkerControlFactory;
import com.palominolabs.benchpress.worker.WorkerMetadata;
import com.palominolabs.benchpress.worker.http.WorkerResourceModule;
import com.palominolabs.benchpress.zookeeper.CuratorModule;
import com.palominolabs.config.ConfigModuleBuilder;
import com.palominolabs.http.server.HttpServer;
import com.palominolabs.http.server.HttpServerConfig;
import com.palominolabs.http.server.HttpServerFactory;
import org.apache.commons.configuration.MapConfiguration;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceInstance;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.slf4j.bridge.SLF4JBridgeHandler;

import javax.ws.rs.core.MediaType;
import java.io.File;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.logging.LogManager;

import static com.google.inject.Guice.createInjector;
import static com.mongodb.util.MyAsserts.assertFalse;
import static javax.ws.rs.core.Response.Status.ACCEPTED;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

public class SingleVmIntegrationTest {

    @Inject
    WorkerAdvertiser workerAdvertiser;
    @Inject
    HttpServerFactory httpServerFactory;
    @Inject
    ZKServer zkServer;
    @Inject
    ZKServerConfig zkServerConfig;
    @Inject
    CuratorModule.CuratorLifecycleHook curatorLifecycleHook;
    @Inject
    @Ipc
    AsyncHttpClient asyncHttpClient;
    @Inject
    ServiceDiscovery<WorkerMetadata> serviceDiscovery;
    @Inject
    ZookeeperConfig zookeeperConfig;
    @Inject
    WorkerControlFactory workerControlFactory;
    @Inject
    JobFarmer jobFarmer;
    @Ipc
    @Inject
    ObjectWriter objectWriter;
    @Ipc
    @Inject
    ObjectReader objectReader;
    @Inject
    SimpleHttpResource simpleHttpResource;

    @Rule
    public TemporaryFolder temporaryFolder = new TemporaryFolder();

    ExecutorService executorService;
    HttpServer httpServer;
    private HttpServerConfig httpServerConfig;

    @BeforeClass
    public static void setUpClass() {
        LogManager.getLogManager().reset();
        SLF4JBridgeHandler.install();
    }

    @Before
    public void setUp() throws Exception {

        File zkTmpDir = temporaryFolder.newFolder();

        final Map<String, Object> configMap = new HashMap<String, Object>();
        configMap.put("benchpress.controller.zookeeper.embedded-server.tmp-dir", zkTmpDir.getAbsolutePath());

        createInjector(Stage.PRODUCTION, new AbstractModule() {
            @Override
            protected void configure() {
                install(new ConfigModuleBuilder().addConfiguration(new MapConfiguration(configMap))
                    .build());

                // basic zookeeper
                install(new ZKServerModule());
                install(new CuratorModule());

                install(new DefaultJerseyServletModule());

                // controller
                install(new InstanceSerializerModule());
                install(new IpcJsonModule());
                install(new TaskPartitionerRegistryModule());
                install(new ControllerResourceModule());

                // worker
                install(new JobRegistryModule());
                install(new TaskProgressClientModule());
                install(new IpcHttpClientModule());
                install(new TaskFactoryFactoryRegistryModule());
                install(new WorkerResourceModule());

                // custom task
                install(new SimpleHttpTaskModule());
                bind(SimpleHttpResource.class);
            }
        }).injectMembers(this);

        executorService = Executors.newCachedThreadPool();
        executorService.submit(zkServer);

        httpServerConfig = new HttpServerConfig();
        httpServer = httpServerFactory.getHttpServer(this.httpServerConfig);
        httpServer.start();

        curatorLifecycleHook.start();

        jobFarmer.setListenAddress(httpServer.getHttpListenHost());
        jobFarmer.setListenPort(httpServer.getHttpListenPort());
    }

    @After
    public void tearDown() throws Exception {
        httpServer.stop();

        executorService.shutdownNow();
        if (!executorService.awaitTermination(1, TimeUnit.SECONDS)) {
            throw new RuntimeException("Executor did not shut down");
        }

        serviceDiscovery.close();
    }

    @Test
    public void testWorkerLockDeAdvertises() throws Exception {
        WorkerMetadata workerMetadata = advertiseWorker();

        assertTrue(lock(workerMetadata));

        assertNoWorkersAdvertised();
    }

    @Test
    public void testWorkerUnlockReAdvertises() throws Exception {
        WorkerMetadata workerMetadata = advertiseWorker();

        lock(workerMetadata);

        assertNoWorkersAdvertised();

        assertTrue(unlock(workerMetadata));

{'fraction_non_alphanumeric': 0.08770769004558153, 'fraction_numerical': 0.0005146302014409646, 'mean_word_length': 3.8756272401433693, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 10}, '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': '3960981', 'n_tokens_mistral': 3833, 'n_tokens_neox': 3685, 'n_words': 613}