package com.squareup.cascading2.function;

import cascading.flow.hadoop.HadoopFlowConnector;
import cascading.flow.hadoop.HadoopFlowProcess;
import cascading.operation.Function;
import cascading.operation.FunctionCall;
import cascading.pipe.Each;
import cascading.pipe.Pipe;
import cascading.scheme.hadoop.TextLine;
import cascading.tap.hadoop.Hfs;
import cascading.tuple.Fields;
import cascading.tuple.Tuple;
import cascading.tuple.TupleEntry;
import cascading.tuple.TupleEntryCollector;
import cascading.tuple.TupleEntryIterator;
import com.squareup.cascading2.generated.Example;
import com.squareup.cascading2.scheme.ProtobufScheme;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import junit.framework.TestCase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class ExpandRepeatedProtoTest extends TestCase {
    private static final Example.Partnership P1 = Example.Partnership.newBuilder()
      .setLeader(Example.Person.newBuilder().setName("John").setEmail("john@"))
      .setFollower(Example.Person.newBuilder().setName("Paul").setEmail("paul@"))
      .addSilent(Example.Person.newBuilder().setName("George").setEmail("george@"))
      .addSilent(Example.Person.newBuilder().setName("Ringo").setEmail("ringo@"))
      .build();

  private static final Example.Person.Builder BRYAN = Example.Person
      .newBuilder()
      .setName("bryan")
      .setId(1)
      .setEmail("bryan@mail.com")
      .setPosition(Example.Person.Position.CEO);
  private static final Example.Person.Builder LUCAS =
      Example.Person.newBuilder().setName("lucas").setId(2);
  private static final Example.Person.Builder TOM =
      Example.Person.newBuilder().setName("tom").setId(3);
  private static final Example.Person.Builder DICK =
      Example.Person.newBuilder().setName("dick").setId(3);
  private static final Example.Person.Builder HARRY =
      Example.Person.newBuilder().setName("harry").setId(3);


  public void testOnlyRepeated() throws Exception {
    ExpandRepeatedProto silent = new ExpandRepeatedProto(Example.Partnership.class, "silent");
    List<Tuple> results = TestExpandProto.operateFunction(silent, new TupleEntry(new Fields("value"), new Tuple(
        Example.Partnership
            .newBuilder()
            .setLeader(BRYAN)
            .setFollower(LUCAS)
            .addSilent(TOM)
            .addSilent(DICK)
            .addSilent(HARRY)
            .build())));

      List<Tuple> expected = new ArrayList<Tuple>();
      expected.add(new Tuple(TOM.build()));
      expected.add(new Tuple(DICK.build()));
      expected.add(new Tuple(HARRY.build()));

      assertEquals(expected, results);
  }

  public void testRepeated() throws Exception {
    ExpandRepeatedProto silent = new ExpandRepeatedProto(Example.Partnership.class, "silent");
    List<Tuple> results = TestExpandProto.operateFunction(silent, new TupleEntry(new Fields("value"), new Tuple(
        Example.Partnership
            .newBuilder()
            .setLeader(BRYAN)
            .setFollower(LUCAS)
            .addSilent(TOM)
            .addSilent(DICK)
            .addSilent(HARRY)
            .build())));

    List<Tuple> expected = new ArrayList<Tuple>();
    expected.add(new Tuple(BRYAN.build(), TOM.build()));
    expected.add(new Tuple(BRYAN.build(), DICK.build()));
    expected.add(new Tuple(BRYAN.build(), HARRY.build()));

    assertEquals(expected, results);
  }

   public void testConstructorErrorCases() throws Exception {
    try {
      new ExpandRepeatedProto<Example.Person>(Example.Person.class, "1");
      fail("should throw exception with non-found field");
    } catch(IllegalArgumentException e) {
      // ok
    }

    try {
      new ExpandRepeatedProto<Example.Partnership>(Example.Partnership.class, "leader");
      fail("should throw exception with non-repeated field");
    } catch(IllegalArgumentException e) {
      // ok
    }
  }
}

@@ -57,813 +57,220 @@
 ing.
-flow.hadoop.HadoopFlowConnector;%0Aimport cascading.flow.hadoop.HadoopFlowProcess;%0Aimport cascading.operation.Function;%0Aimport cascading.operation.FunctionCall;%0Aimport cascading.pipe.Each;%0Aimport cascading.pipe.Pipe;%0Aimport cascading.scheme.hadoop.TextLine;%0Aimport cascading.tap.hadoop.Hfs;%0Aimport cascading.tuple.Fields;%0Aimport cascading.tuple.Tuple;%0Aimport cascading.tuple.TupleEntry;%0Aimport cascading.tuple.TupleEntryCollector;%0Aimport cascading.tuple.TupleEntryIterator;%0Aimport com.squareup.cascading2.generated.Example;%0Aimport com.squareup.cascading2.scheme.ProtobufScheme;%0Aimport java.io.IOException;%0Aimport java.util.ArrayList;%0Aimport java.util.List;%0Aimport junit.framework.TestCase;%0Aimport org.apache.hadoop.conf.Configuration;%0Aimport org.apache.hadoop.fs.FileSystem;%0Aimport org.apache.hadoop.fs.Path
+tuple.Fields;%0Aimport cascading.tuple.Tuple;%0Aimport cascading.tuple.TupleEntry;%0Aimport com.squareup.cascading2.generated.Example;%0Aimport junit.framework.TestCase;%0A%0Aimport java.util.ArrayList;%0Aimport java.util.List
 ;%0A%0Ap
@@ -1001,24 +1001,16 @@
 e.Person
-.Builder
  LUCAS =
@@ -1069,16 +1069,24 @@
 setId(2)
+.build()
 ;%0A  priv
@@ -1116,24 +1116,16 @@
 e.Person
-.Builder
  TOM =%0A 
@@ -1172,32 +1172,40 @@
 (%22tom%22).setId(3)
+.build()
 ;%0A  private stat
@@ -1227,24 +1227,16 @@
 e.Person
-.Builder
  DICK =%0A
@@ -1293,16 +1293,24 @@
 setId(3)
+.build()
 ;%0A  priv
@@ -1340,24 +1340,16 @@
 e.Person
-.Builder
  HARRY =
@@ -1408,16 +1408,24 @@
 setId(3)
+.build()
 ;%0A%0A%0A  pu
@@ -1994,24 +1994,16 @@
 uple(TOM
-.build()
 ));%0A    
@@ -2027,32 +2027,24 @@
 w Tuple(DICK
-.build()
 ));%0A      ex
@@ -2065,32 +2065,24 @@
  Tuple(HARRY
-.build()
 ));%0A%0A      a
@@ -2122,762 +2122,8 @@
  %7D%0A%0A
-  public void testRepeated() throws Exception %7B%0A    ExpandRepeatedProto silent = new ExpandRepeatedProto(Example.Partnership.class, %22silent%22);%0A    List%3CTuple%3E results = TestExpandProto.operateFunction(silent, new TupleEntry(new Fields(%22value%22), new Tuple(%0A        Example.Partnership%0A            .newBuilder()%0A            .setLeader(BRYAN)%0A            .setFollower(LUCAS)%0A            .addSilent(TOM)%0A            .addSilent(DICK)%0A            .addSilent(HARRY)%0A            .build())));%0A%0A    List%3CTuple%3E expected = new ArrayList%3CTuple%3E();%0A    expected.add(new Tuple(BRYAN.build(), TOM.build()));%0A    expected.add(new Tuple(BRYAN.build(), DICK.build()));%0A    expected.add(new Tuple(BRYAN.build(), HARRY.build()));%0A%0A    assertEquals(expected, results);%0A  %7D%0A%0A 
   pu

delete redundant test case