<commit_before>package org.apache.flume.lifecycle;

import org.apache.flume.Context;
import org.apache.flume.CounterGroup;
import org.apache.flume.LogicalNode;
import org.apache.flume.lifecycle.LifecycleSupervisor.SupervisorPolicy;
import org.apache.flume.sink.NullSink;
import org.apache.flume.source.SequenceGeneratorSource;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

public class TestLifecycleSupervisor {

  private LifecycleSupervisor supervisor;

  @Before
  public void setUp() {
    supervisor = new LifecycleSupervisor();
  }

  @Test
  public void testLifecycle() throws LifecycleException, InterruptedException {
    Context context = new Context();

    supervisor.start(context);
    supervisor.stop(context);
  }

  @Test
  public void testSupervise() throws LifecycleException, InterruptedException {
    Context context = new Context();

    supervisor.start(context);

    /* Attempt to supervise a known-to-fail config. */
    /*
     * LogicalNode node = new LogicalNode(); SupervisorPolicy policy = new
     * SupervisorPolicy.OnceOnlyPolicy(); supervisor.supervise(node, policy,
     * LifecycleState.START);
     */

    LogicalNode node = new LogicalNode();
    node.setName("node1");
    node.setSource(new SequenceGeneratorSource());
    node.setSink(new NullSink());

    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();
    supervisor.supervise(node, policy, LifecycleState.START);

    Thread.sleep(10000);

    node = new LogicalNode();
    node.setName("node2");
    node.setSource(new SequenceGeneratorSource());
    node.setSink(new NullSink());

    policy = new SupervisorPolicy.OnceOnlyPolicy();
    supervisor.supervise(node, policy, LifecycleState.START);

    Thread.sleep(5000);

    supervisor.stop(context);
  }

  @Test
  public void testSuperviseBroken() throws LifecycleException,
      InterruptedException {
    Context context = new Context();

    supervisor.start(context);

    /* Attempt to supervise a known-to-fail config. */
    LogicalNode node = new LogicalNode();
    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();
    supervisor.supervise(node, policy, LifecycleState.START);

    Thread.sleep(5000);

    supervisor.stop(context);
  }

  @Test
  public void testSuperviseSupervisor() throws LifecycleException,
      InterruptedException {

    Context context = new Context();

    supervisor.start(context);

    LifecycleSupervisor supervisor2 = new LifecycleSupervisor();

    /* Attempt to supervise a known-to-fail config. */
    LogicalNode node = new LogicalNode();
    node.setName("node1");
    node.setSource(new SequenceGeneratorSource());
    node.setSink(new NullSink());

    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();
    supervisor2.supervise(node, policy, LifecycleState.START);

    supervisor.supervise(supervisor2,
        new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);

    Thread.sleep(10000);

    supervisor.stop(context);
  }

  @Test
  public void testUnsuperviseServce() throws LifecycleException,
      InterruptedException {
    Context context = new Context();

    supervisor.start(context);

    LifecycleAware service = new CountingLifecycleAware();
    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();

    supervisor.supervise(service, policy, LifecycleState.START);
    supervisor.unsupervise(service);

    service.stop(context);

    supervisor.stop(context);
  }

  @Test
  public void testStopServce() throws LifecycleException, InterruptedException {
    Context context = new Context();

    supervisor.start(context);

    CountingLifecycleAware service = new CountingLifecycleAware();
    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();

    Assert.assertEquals(Long.valueOf(0), service.counterGroup.get("start"));
    Assert.assertEquals(Long.valueOf(0), service.counterGroup.get("stop"));

    supervisor.supervise(service, policy, LifecycleState.START);

    Thread.sleep(3200);

    Assert.assertEquals(Long.valueOf(1), service.counterGroup.get("start"));
    Assert.assertEquals(Long.valueOf(0), service.counterGroup.get("stop"));

    supervisor.setDesiredState(service, LifecycleState.STOP);

    Thread.sleep(3200);

    Assert.assertEquals(Long.valueOf(1), service.counterGroup.get("start"));
    Assert.assertEquals(Long.valueOf(1), service.counterGroup.get("stop"));

    supervisor.stop(context);
  }

  public static class CountingLifecycleAware implements LifecycleAware {

    public CounterGroup counterGroup;
    private LifecycleState lifecycleState;

    public CountingLifecycleAware() {
      lifecycleState = LifecycleState.IDLE;
      counterGroup = new CounterGroup();
    }

    @Override
    public void start(Context context) throws LifecycleException,
        InterruptedException {

      counterGroup.incrementAndGet("start");

      lifecycleState = LifecycleState.START;
    }

    @Override
    public void stop(Context context) throws LifecycleException,
        InterruptedException {

      counterGroup.incrementAndGet("stop");

      lifecycleState = LifecycleState.STOP;
    }

    @Override
    public LifecycleState getLifecycleState() {
      return lifecycleState;
    }

  }

}<commit_msg>- LifecycleSupervisor test no longer depends on LogicalNode and friends (self-contained).<commit_after>package org.apache.flume.lifecycle;

import org.apache.flume.Context;
import org.apache.flume.CounterGroup;
import org.apache.flume.lifecycle.LifecycleSupervisor.SupervisorPolicy;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

public class TestLifecycleSupervisor {

  private LifecycleSupervisor supervisor;

  @Before
  public void setUp() {
    supervisor = new LifecycleSupervisor();
  }

  @Test
  public void testLifecycle() throws LifecycleException, InterruptedException {
    Context context = new Context();

    supervisor.start(context);
    supervisor.stop(context);
  }

  @Test
  public void testSupervise() throws LifecycleException, InterruptedException {
    Context context = new Context();

    supervisor.start(context);

    /* Attempt to supervise a known-to-fail config. */
    /*
     * LogicalNode node = new LogicalNode(); SupervisorPolicy policy = new
     * SupervisorPolicy.OnceOnlyPolicy(); supervisor.supervise(node, policy,
     * LifecycleState.START);
     */

    CountingLifecycleAware node = new CountingLifecycleAware();

    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();
    supervisor.supervise(node, policy, LifecycleState.START);

    Thread.sleep(10000);

    node = new CountingLifecycleAware();

    policy = new SupervisorPolicy.OnceOnlyPolicy();
    supervisor.supervise(node, policy, LifecycleState.START);

    Thread.sleep(5000);

    supervisor.stop(context);
  }

  @Test
  public void testSuperviseBroken() throws LifecycleException,
      InterruptedException {
    Context context = new Context();

    supervisor.start(context);

    /* Attempt to supervise a known-to-fail config. */
    LifecycleAware node = new LifecycleAware() {

      @Override
      public void stop(Context context) {
      }

      @Override
      public void start(Context context) {
        throw new NullPointerException("Boom!");
      }

      @Override
      public LifecycleState getLifecycleState() {
        return LifecycleState.IDLE;
      }
    };

    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();
    supervisor.supervise(node, policy, LifecycleState.START);

    Thread.sleep(5000);

    supervisor.stop(context);
  }

  @Test
  public void testSuperviseSupervisor() throws LifecycleException,
      InterruptedException {

    Context context = new Context();

    supervisor.start(context);

    LifecycleSupervisor supervisor2 = new LifecycleSupervisor();

    CountingLifecycleAware node = new CountingLifecycleAware();

    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();
    supervisor2.supervise(node, policy, LifecycleState.START);

    supervisor.supervise(supervisor2,
        new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);

    Thread.sleep(10000);

    supervisor.stop(context);
  }

  @Test
  public void testUnsuperviseServce() throws LifecycleException,
      InterruptedException {
    Context context = new Context();

    supervisor.start(context);

    LifecycleAware service = new CountingLifecycleAware();
    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();

    supervisor.supervise(service, policy, LifecycleState.START);
    supervisor.unsupervise(service);

    service.stop(context);

    supervisor.stop(context);
  }

  @Test
  public void testStopServce() throws LifecycleException, InterruptedException {
    Context context = new Context();

    supervisor.start(context);

    CountingLifecycleAware service = new CountingLifecycleAware();
    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();

    Assert.assertEquals(Long.valueOf(0), service.counterGroup.get("start"));
    Assert.assertEquals(Long.valueOf(0), service.counterGroup.get("stop"));

    supervisor.supervise(service, policy, LifecycleState.START);

    Thread.sleep(3200);

    Assert.assertEquals(Long.valueOf(1), service.counterGroup.get("start"));
    Assert.assertEquals(Long.valueOf(0), service.counterGroup.get("stop"));

    supervisor.setDesiredState(service, LifecycleState.STOP);

    Thread.sleep(3200);

    Assert.assertEquals(Long.valueOf(1), service.counterGroup.get("start"));
    Assert.assertEquals(Long.valueOf(1), service.counterGroup.get("stop"));

    supervisor.stop(context);
  }

  public static class CountingLifecycleAware implements LifecycleAware {

    public CounterGroup counterGroup;
    private LifecycleState lifecycleState;

    public CountingLifecycleAware() {
      lifecycleState = LifecycleState.IDLE;
      counterGroup = new CounterGroup();
    }

    @Override
    public void start(Context context) {

      counterGroup.incrementAndGet("start");

      lifecycleState = LifecycleState.START;
    }

    @Override
    public void stop(Context context) {

      counterGroup.incrementAndGet("stop");

      lifecycleState = LifecycleState.STOP;
    }

    @Override
    public LifecycleState getLifecycleState() {
      return lifecycleState;
    }

  }

}
{'fraction_non_alphanumeric': 0.10067567567567567, 'fraction_numerical': 0.0070463320463320465, 'mean_word_length': 4.064027370478984, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '3637354', 'n_tokens_mistral': 3070, 'n_tokens_neox': 2845, 'n_words': 588}