package com.squareup.cascading2.function;

import cascading.flow.hadoop.HadoopFlowProcess;
import cascading.operation.Function;
import cascading.operation.FunctionCall;
import cascading.tuple.Fields;
import cascading.tuple.Tuple;
import cascading.tuple.TupleEntry;
import cascading.tuple.TupleEntryCollector;
import com.squareup.cascading2.generated.Example;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicReference;
import junit.framework.TestCase;

public class ExtractProtoTest extends TestCase {
  private static final Example.Partnership P1 = Example.Partnership.newBuilder()
      .setFollower(Example.Person.newBuilder()
          .setName("Andy")
          .setEmail("andy@")
      )
      .setLeader(Example.Person.newBuilder()
          .setName("Bryan")
          .setEmail("bryan@")
      )
      .build();

  private static final Example.Partnership P2 = Example.Partnership.newBuilder()
      .setLeader(Example.Person.newBuilder().setName("Bryan").setEmail("bryan@"))
      .build();

  public void testValidatesNonExistingField() throws Exception {
    try {
      new ExtractProto(Example.Partnership.class, "follower.has_mind_powers");
      fail("Expected an exception!");
    } catch (IllegalArgumentException e) {
      // yay!
    }
  }

  public void testThrowsOnRepeatedField() throws Exception {
    try {
      new ExtractProto(Example.Partnership.class, "silent.name");
      fail("Expected an exception!");
    } catch (IllegalArgumentException e) {
      // yay!
    }
    try {
      new ExtractProto(Example.Partnership.class, "silent");
      fail("Expected an exception!");
    } catch (IllegalArgumentException e) {
      // yay!
    }
  }

  public void testNullFirstObject() throws Exception {
    assertEquals(new Tuple(null, null, null, null),
        exec(new ExtractProto(Example.Partnership.class, "follower.name", "follower.email", "leader.name", "leader.email"), new Tuple((Object)null)));
  }

  public void testAllPresent() throws Exception {
    assertEquals(new Tuple("Andy", "andy@", "Bryan", "bryan@"),
        exec(new ExtractProto(Example.Partnership.class, "follower.name", "follower.email", "leader.name", "leader.email"), new Tuple(P1)));
  }

  public void testSomePresent() throws Exception {
    assertEquals(new Tuple(null, null, "Bryan", "bryan@"),
        exec(new ExtractProto(Example.Partnership.class, "follower.name", "follower.email", "leader.name", "leader.email"), new Tuple(P2)));
  }

  private static Tuple exec(Function f, final Tuple input) {
    final AtomicReference<Tuple> output = new AtomicReference<Tuple>();

    f.operate(new HadoopFlowProcess(), new FunctionCall() {
      @Override public TupleEntry getArguments() {
        return new TupleEntry(new Fields("blah"), input);
      }

      @Override public Fields getDeclaredFields() {
        return null;
      }

      @Override public TupleEntryCollector getOutputCollector() {
        return new TupleEntryCollector() {
          @Override protected void collect(TupleEntry tupleEntry) throws IOException {
            output.set(tupleEntry.getTuple());
          }
        };
      }

      @Override public Object getContext() {
        return null;
      }

      @Override public void setContext(Object o) {
      }

      @Override public Fields getArgumentFields() {
        return null;
      }
    });
    return output.get();
  }
}

@@ -1015,16 +1015,268 @@
 ild();%0A%0A
+  private static final Example.Partnership P3 = Example.Partnership.newBuilder()%0A      .setLeader(Example.Person.newBuilder()%0A          .setName(%22Jack%22)%0A          .setEmail(%22jack@%22)%0A          .setPosition(Example.Person.Position.CEO))%0A      .build();%0A%0A
   public
@@ -2721,16 +2721,264 @@
 );%0A  %7D%0A%0A
+  public void testEnum() throws Exception %7B%0A    assertEquals(new Tuple(%22Jack%22, %22jack@%22, Example.Person.Position.CEO),%0A        exec(new ExtractProto(Example.Partnership.class, %22leader.name%22, %22leader.email%22, %22leader.position%22), new Tuple(P3)));%0A  %7D%0A%0A
   privat

Add test for ExtractProto with enum.