<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.flume.source;


import static org.junit.Assert.*;

import java.io.*;
import java.nio.charset.Charset;
import java.util.*;
import java.util.regex.Pattern;

import org.apache.commons.io.FileUtils;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
import org.apache.flume.lifecycle.LifecycleException;
import org.junit.*;

import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import com.google.common.io.Files;

public class TestExecSource {

  private AbstractSource source;
  private Channel channel = new MemoryChannel();

  private Context context = new Context();

  private ChannelSelector rcs = new ReplicatingChannelSelector();


  @Before
  public void setUp() {
    context.put("keep-alive", "1");
    context.put("capacity", "1000");
    context.put("transactionCapacity", "1000");
    Configurables.configure(channel, context);
    rcs.setChannels(Lists.newArrayList(channel));

    source = new ExecSource();
    source.setChannelProcessor(new ChannelProcessor(rcs));
  }

  @After
  public void tearDown() {
    source.stop();
  }

  @Test
  public void testProcess() throws InterruptedException, LifecycleException,
  EventDeliveryException, IOException {

    context.put("command", "cat /etc/passwd");
    context.put("keep-alive", "1");
    context.put("capacity", "1000");
    context.put("transactionCapacity", "1000");
    Configurables.configure(source, context);

    source.start();
    Transaction transaction = channel.getTransaction();

    transaction.begin();
    Event event;

    FileOutputStream outputStream = new FileOutputStream(
        "/tmp/flume-execsource." + Thread.currentThread().getId());

    while ((event = channel.take()) != null) {
      outputStream.write(event.getBody());
      outputStream.write('\n');
    }

    outputStream.close();
    transaction.commit();
    transaction.close();

    File file1 = new File("/tmp/flume-execsource."
        + Thread.currentThread().getId());
    File file2 = new File("/etc/passwd");
    Assert.assertEquals(FileUtils.checksumCRC32(file1),
        FileUtils.checksumCRC32(file2));
    FileUtils.forceDelete(file1);
  }

  @Test
  public void testShellCommandSimple() throws InterruptedException, LifecycleException,
  EventDeliveryException, IOException {
    runTestShellCmdHelper("/bin/sh -c", "seq 5"
            , new String[]{"1","2","3","4","5" } );
  }

  @Test
  public void testShellCommandBackTicks() throws InterruptedException, LifecycleException,
  EventDeliveryException, IOException {
    // command with backticks
    runTestShellCmdHelper("/bin/sh -c", "echo `seq 5`" , new String[]{"1 2 3 4 5" } );
    runTestShellCmdHelper("/bin/sh -c", "echo $(seq 5)" , new String[]{"1 2 3 4 5" } );
  }

  @Test
  public void testShellCommandComplex() throws InterruptedException, LifecycleException,
  EventDeliveryException, IOException {
    // command with wildcards & pipes
    String[] expected = {"1234", "abcd", "ijk", "xyz", "zzz"};

    // pipes
    runTestShellCmdHelper("/bin/sh -c", "echo zzz 1234 xyz abcd ijk | xargs -n1 echo | sort -f"
            ,  expected );
  }

  @Test
  public void testShellCommandScript() throws InterruptedException, LifecycleException,
  EventDeliveryException, IOException {
    // mini script
    runTestShellCmdHelper("/bin/sh -c", "for i in {1..5}; do echo $i;done"
            , new String[]{"1","2","3","4","5" } );
    // shell arithmetic
    runTestShellCmdHelper("/bin/sh -c", "if ((2+2>3)); then  echo good; else echo not good; fi" , new String[]{"good"} );
  }

  @Test
  public void testShellCommandEmbeddingAndEscaping() throws InterruptedException, LifecycleException,
    EventDeliveryException, IOException {
      System.out.println( "######### PWD = " + new java.io.File( "." ).getCanonicalPath() );
    // mini script
      BufferedReader reader = new BufferedReader(new FileReader("src/test/resources/test_command.txt") );
      try {
        String command1 = reader.readLine();
        Assert.assertNotNull(command1);
        String[] output1 = new String[] {"'1'", "\"2\"", "\\3", "\\4"};
        runTestShellCmdHelper("/bin/sh -c", command1 , output1);
        String command2 = reader.readLine();
        Assert.assertNotNull(command2);
        String[] output2 = new String[]{"1","2","3","4","5" };
        runTestShellCmdHelper("/bin/sh -c", command2 , output2);
        String command3 = reader.readLine();
        Assert.assertNotNull(command3);
        String[] output3 = new String[]{"2","3","4","5","6" };
        runTestShellCmdHelper("/bin/sh -c", command3 , output3);
      } finally {
        reader.close();
      }
    }


    private void runTestShellCmdHelper(String shell, String command, String[] expectedOutput)
             throws InterruptedException, LifecycleException, EventDeliveryException, IOException {
      context.put("shell", shell);
      context.put("command", command);
      Configurables.configure(source, context);
      source.start();
      File outputFile = File.createTempFile("flumeExecSourceTest_", "");
      FileOutputStream outputStream = new FileOutputStream(outputFile);
      Transaction transaction = channel.getTransaction();
      transaction.begin();
      try {
        Event event;
        while ((event = channel.take()) != null) {
          outputStream.write(event.getBody());
          outputStream.write('\n');
        }
        outputStream.close();
        transaction.commit();
        List<String> output  = Files.readLines(outputFile, Charset.defaultCharset());
<commit_msg>FLUME-1943: ExecSource tests failing on Jenkins<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.flume.source;


import static org.junit.Assert.*;

import java.io.*;
import java.nio.charset.Charset;
import java.util.*;
import java.util.regex.Pattern;

import org.apache.commons.io.FileUtils;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
import org.apache.flume.lifecycle.LifecycleException;
import org.junit.*;

import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import com.google.common.io.Files;

public class TestExecSource {

  private AbstractSource source;
  private Channel channel = new MemoryChannel();

  private Context context = new Context();

  private ChannelSelector rcs = new ReplicatingChannelSelector();


  @Before
  public void setUp() {
    context.put("keep-alive", "1");
    context.put("capacity", "1000");
    context.put("transactionCapacity", "1000");
    Configurables.configure(channel, context);
    rcs.setChannels(Lists.newArrayList(channel));

    source = new ExecSource();
    source.setChannelProcessor(new ChannelProcessor(rcs));
  }

  @After
  public void tearDown() {
    source.stop();
  }

  @Test
  public void testProcess() throws InterruptedException, LifecycleException,
  EventDeliveryException, IOException {

    context.put("command", "cat /etc/passwd");
    context.put("keep-alive", "1");
    context.put("capacity", "1000");
    context.put("transactionCapacity", "1000");
    Configurables.configure(source, context);

    source.start();
    Transaction transaction = channel.getTransaction();

    transaction.begin();
    Event event;

    FileOutputStream outputStream = new FileOutputStream(
        "/tmp/flume-execsource." + Thread.currentThread().getId());

    while ((event = channel.take()) != null) {
      outputStream.write(event.getBody());
      outputStream.write('\n');
    }

    outputStream.close();
    transaction.commit();
    transaction.close();

    File file1 = new File("/tmp/flume-execsource."
        + Thread.currentThread().getId());
    File file2 = new File("/etc/passwd");
    Assert.assertEquals(FileUtils.checksumCRC32(file1),
        FileUtils.checksumCRC32(file2));
    FileUtils.forceDelete(file1);
  }

  @Test
  public void testShellCommandSimple() throws InterruptedException, LifecycleException,
  EventDeliveryException, IOException {
    runTestShellCmdHelper("/bin/bash -c", "seq 5"
            , new String[]{"1","2","3","4","5" } );
  }

  @Test
  public void testShellCommandBackTicks() throws InterruptedException, LifecycleException,
  EventDeliveryException, IOException {
    // command with backticks
    runTestShellCmdHelper("/bin/bash -c", "echo `seq 5`" , new String[]{"1 2 3 4 5" } );
    runTestShellCmdHelper("/bin/bash -c", "echo $(seq 5)" , new String[]{"1 2 3 4 5" } );
  }

  @Test
  public void testShellCommandComplex() throws InterruptedException, LifecycleException,
  EventDeliveryException, IOException {
    // command with wildcards & pipes
    String[] expected = {"1234", "abcd", "ijk", "xyz", "zzz"};

    // pipes
    runTestShellCmdHelper("/bin/bash -c", "echo zzz 1234 xyz abcd ijk | xargs -n1 echo | sort -f"
            ,  expected );
  }

  @Test
  public void testShellCommandScript() throws InterruptedException, LifecycleException,
  EventDeliveryException, IOException {
    // mini script
    runTestShellCmdHelper("/bin/bash -c", "for i in {1..5}; do echo $i;done"
            , new String[]{"1","2","3","4","5" } );
    // shell arithmetic
    runTestShellCmdHelper("/bin/bash -c", "if ((2+2>3)); then  echo good; else echo not good; fi" , new String[]{"good"} );
  }

  @Test
  public void testShellCommandEmbeddingAndEscaping() throws InterruptedException, LifecycleException,
    EventDeliveryException, IOException {
      System.out.println( "######### PWD = " + new java.io.File( "." ).getCanonicalPath() );
    // mini script
      BufferedReader reader = new BufferedReader(new FileReader("src/test/resources/test_command.txt") );
      try {
        String command1 = reader.readLine();
        Assert.assertNotNull(command1);
        String[] output1 = new String[] {"'1'", "\"2\"", "\\3", "\\4"};
        runTestShellCmdHelper("/bin/bash -c", command1 , output1);
        String command2 = reader.readLine();
        Assert.assertNotNull(command2);
        String[] output2 = new String[]{"1","2","3","4","5" };
        runTestShellCmdHelper("/bin/bash -c", command2 , output2);
        String command3 = reader.readLine();
        Assert.assertNotNull(command3);
        String[] output3 = new String[]{"2","3","4","5","6" };
        runTestShellCmdHelper("/bin/bash -c", command3 , output3);
      } finally {
        reader.close();
      }
    }


    private void runTestShellCmdHelper(String shell, String command, String[] expectedOutput)
             throws InterruptedException, LifecycleException, EventDeliveryException, IOException {
      context.put("shell", shell);
      context.put("command", command);
      Configurables.configure(source, context);
      source.start();
      File outputFile = File.createTempFile("flumeExecSourceTest_", "");
      FileOutputStream outputStream = new FileOutputStream(outputFile);
      Transaction transaction = channel.getTransaction();
      transaction.begin();
      try {
        Event event;
        while ((event = channel.take()) != null) {
          outputStream.write(event.getBody());
          outputStream.write('\n');
        }
        outputStream.close();
        transaction.commit();
        List<String> output  = Files.readLines(outputFile, Charset.defaultCharset());
        System.out.println("command : " + command);
        System.out.println("output : ");
        for( String line : output )
          System.out.println();
{'fraction_non_alphanumeric': 0.13229713952130764, 'fraction_numerical': 0.014448336252189142, 'mean_word_length': 3.650492025788938, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6817716', 'n_tokens_mistral': 3983, 'n_tokens_neox': 3818, 'n_words': 989}