package com.viadeo.avrodiff;

import java.io.File;

import org.apache.avro.Schema;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.generic.GenericData.Record;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumWriter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;

import com.viadeo.avrodiff.GenerateSample.TestSchema;

public class DiffTest {

	@Rule
	public TemporaryFolder tmpFolder = new TemporaryFolder();

    public void create(String dirname, String filename, Record[] records) throws  Exception {


        Schema schema = TestSchema.getSchema();

        File dir = new File(tmpFolder.getRoot(), dirname);

        if(!dir.exists() ){
            dir.mkdirs();
        }

        File file = new File(dir,filename);
        DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema);
        DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<GenericRecord>(datumWriter);
        dataFileWriter.create(schema, file);
        TestSchema.append(dataFileWriter,records);
        dataFileWriter.close();

    }


    @Before
    public void setUp() throws Exception {
        Record[] records = {TestSchema.record("2", 2), TestSchema.record("3", 3)};
        Record[] records2 = {TestSchema.record("4", 4), TestSchema.record("3", 3)};

        create("a", "input.avro", records);

        create("b", "input.avro", records2);
    }


	@Test
	public void test42() throws Exception {

		String outStr = tmpFolder.getRoot().getPath() + "/out-generic2";

        Path diffin 	= new Path( new File(tmpFolder.getRoot(), "a").toURI().toString());
        Path diffout 	= new Path( new File(tmpFolder.getRoot(), "b").toURI().toString());
        Path output 	= new Path(outStr);

        Configuration jobConf = new Configuration();
        DiffJob job = new DiffJob();
        Assert.assertTrue(job.internalRun(diffin, diffout, output, jobConf).waitForCompletion(true));

        System.out.println("----------------------------" + outStr);
	}
}

@@ -76,16 +76,60 @@
 Schema;%0A
+import org.apache.avro.file.DataFileReader;%0A
 import o
@@ -207,32 +207,83 @@
 ricData.Record;%0A
+import org.apache.avro.generic.GenericDatumReader;%0A
 import org.apach
@@ -355,32 +355,71 @@
 .GenericRecord;%0A
+import org.apache.avro.io.DatumReader;%0A
 import org.apach
@@ -1775,16 +1775,546 @@
     %7D%0A%0A%0A
+    public static void assertContainsOnly(String base, String dir, Record record) throws Exception %7B%0A        String path = base + dir + %22/part-r-00000.avro%22;%0A%0A        DatumReader%3CGenericRecord%3E datumReader = new GenericDatumReader%3CGenericRecord%3E();%0A        DataFileReader%3CGenericRecord%3E dataFileReader = new DataFileReader%3CGenericRecord%3E(new File(path), datumReader);%0A%0A        //Read only one Record%0A        GenericRecord user = dataFileReader.next();%0A%0A        Assert.assertEquals(%22should only contains %22, user, record);%0A%0A    %7D%0A%0A%0A
 %09@Test%0A%09
@@ -2849,65 +2849,254 @@
    S
-ystem.out.println(%22----------------------------%22 + outStr
+tring base = outStr + %22/%22;%0A        assertContainsOnly(base, %22type=kernel%22, TestSchema.record(%223%22, 3));%0A        assertContainsOnly(base, %22type=add%22, TestSchema.record(%224%22, 4));%0A        assertContainsOnly(base, %22type=del%22, TestSchema.record(%222%22, 2)
 );%0A%09

add functional tests