<commit_before>package org.gbif.crawler.pipelines;

import org.gbif.api.model.pipelines.PipelineProcess;
import org.gbif.api.model.pipelines.PipelineStep;
import org.gbif.api.model.pipelines.StepType;
import org.gbif.api.service.registry.DatasetService;
import org.gbif.crawler.constants.PipelinesNodePaths;
import org.gbif.crawler.constants.PipelinesNodePaths.Fn;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Arrays;
import java.util.Collections;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import com.google.common.base.Charsets;
import com.google.common.collect.Sets;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.test.TestingServer;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;

import static java.time.format.DateTimeFormatter.ISO_LOCAL_DATE_TIME;

@RunWith(MockitoJUnitRunner.class)
public class PipelinesRunningProcessServiceImplTest {

  private static final String MESSAGE = "info";

  private static final BiConsumer<Set<PipelineProcess>, Set<String>> ASSERT_FN = (s, ids) -> {
    Consumer<PipelineStep> checkFn = step -> {
      Assert.assertTrue(Arrays.asList(StepType.values()).contains(step.getType()));
      Assert.assertNotNull(step.getStarted());
      Assert.assertNotNull(step.getFinished());
      Assert.assertEquals(PipelineStep.Status.COMPLETED, step.getState());
      Assert.assertEquals(MESSAGE, step.getMessage());
    };

    s.forEach(status -> {
      Assert.assertNotNull(status);
      Assert.assertEquals(6, status.getSteps().size());
      Assert.assertTrue(ids.contains(status.getDatasetKey() + "_" + status.getAttempt()));
      status.getSteps().forEach(step -> {

        if (step.getType() == StepType.DWCA_TO_VERBATIM
            || step.getType() == StepType.XML_TO_VERBATIM
            || step.getType() == StepType.ABCD_TO_VERBATIM
            || step.getType() == StepType.VERBATIM_TO_INTERPRETED) {
          checkFn.accept(step);
        }
        if (step.getType() == StepType.HDFS_VIEW) {
          Assert.assertTrue(Arrays.asList(StepType.values()).contains(step.getType()));
          Assert.assertNotNull(step.getStarted());
          Assert.assertNull(step.getFinished());
          Assert.assertEquals(PipelineStep.Status.FAILED, step.getState());
          Assert.assertEquals(MESSAGE, step.getMessage());
        }
        if (step.getType() == StepType.INTERPRETED_TO_INDEX) {
          Assert.assertTrue(Arrays.asList(StepType.values()).contains(step.getType()));
          Assert.assertNotNull(step.getStarted());
          Assert.assertNull(step.getFinished());
          Assert.assertEquals(PipelineStep.Status.RUNNING, step.getState());
          Assert.assertNull(MESSAGE, step.getMessage());
        }
      });
    });
  };


  private CuratorFramework curator;
  private PathChildrenCache pathChildrenCache;
  private TestingServer server;
  private PipelinesRunningProcessServiceImpl service;

  @Before
  public void setup() throws Exception {
    server = new TestingServer();
    curator =
        CuratorFrameworkFactory.builder()
            .connectString(server.getConnectString())
            .namespace("crawler")
            .retryPolicy(new RetryOneTime(1))
            .build();
    curator.start();
    service =<commit_msg>cleanup<commit_after>package org.gbif.crawler.pipelines;

import org.gbif.api.model.pipelines.PipelineProcess;
import org.gbif.api.model.pipelines.PipelineStep;
import org.gbif.api.model.pipelines.StepType;
import org.gbif.api.service.registry.DatasetService;
import org.gbif.crawler.constants.PipelinesNodePaths;
import org.gbif.crawler.constants.PipelinesNodePaths.Fn;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Arrays;
import java.util.Collections;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import com.google.common.base.Charsets;
import com.google.common.collect.Sets;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.test.TestingServer;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;

import static java.time.format.DateTimeFormatter.ISO_LOCAL_DATE_TIME;

@RunWith(MockitoJUnitRunner.class)
public class PipelinesRunningProcessServiceImplTest {

  private static final String MESSAGE = "info";

  private static final BiConsumer<Set<PipelineProcess>, Set<String>> ASSERT_FN = (s, ids) -> {
    Consumer<PipelineStep> checkFn = step -> {
      Assert.assertTrue(Arrays.asList(StepType.values()).contains(step.getType()));
      Assert.assertNotNull(step.getStarted());
      Assert.assertNotNull(step.getFinished());
      Assert.assertEquals(PipelineStep.Status.COMPLETED, step.getState());
      Assert.assertEquals(MESSAGE, step.getMessage());
    };

    s.forEach(status -> {
      Assert.assertNotNull(status);
      Assert.assertEquals(6, status.getSteps().size());
      Assert.assertTrue(ids.contains(status.getDatasetKey() + "_" + status.getAttempt()));
      status.getSteps().forEach(step -> {

        if (step.getType() == StepType.DWCA_TO_VERBATIM
            || step.getType() == StepType.XML_TO_VERBATIM
            || step.getType() == StepType.ABCD_TO_VERBATIM
            || step.getType() == StepType.VERBATIM_TO_INTERPRETED) {
          checkFn.accept(step);
        }
        if (step.getType() == StepType.HDFS_VIEW) {
          Assert.assertTrue(Arrays.asList(StepType.values()).contains(step.getType()));
          Assert.assertNotNull(step.getStarted());
          Assert.assertNull(step.getFinished());
          Assert.assertEquals(PipelineStep.Status.FAILED, step.getState());
          Assert.assertEquals(MESSAGE, step.getMessage());
        }
        if (step.getType() == StepType.INTERPRETED_TO_INDEX) {
          Assert.assertTrue(Arrays.asList(StepType.values()).contains(step.getType()));
          Assert.assertNotNull(step.getStarted());
          Assert.assertNull(step.getFinished());
          Assert.assertEquals(PipelineStep.Status.RUNNING, step.getState());
          Assert.assertNull(MESSAGE, step.getMessage());
        }
      });
    });
  };


  private CuratorFramework curator;
  private TestingServer server;
  private PipelinesRunningProcessServiceImpl service;

  @Before
  public void setup() throws Exception {
    server = new TestingServer();
    curator =
        CuratorFrameworkFactory.builder()
            .connectString(server.getConnectString())
            .namespace("crawler")
            .retryPolicy(new RetryOneTime(1))
            .build();
    curator.start();
    service =
{'fraction_non_alphanumeric': 0.1252853498052907, 'fraction_numerical': 0.000537129045253122, 'mean_word_length': 3.945551128818061, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 19, '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': '3322089', 'n_tokens_mistral': 2191, 'n_tokens_neox': 2218, 'n_words': 317}