package water.parser;

import static org.junit.Assert.assertTrue;
import org.junit.Test;
import java.io.File;
import water.*;
import water.fvec.*;

public class ParseExceptionTest extends TestUtil {
  public ParseExceptionTest() { super(3); }

  @Test public void testParserRecoversFromException() {
    Throwable ex = null;
    Key fkey0=null,fkey1=null,fkey2=null,okey=null;
    try {
      okey = Key.make("junk.hex");

      fkey0 = NFSFileVec.make(new File("smalldata/parse_folder_test/prostate_0.csv"))._key;
      fkey1 = NFSFileVec.make(new File("smalldata/parse_folder_test/prostate_1.csv"))._key;
      fkey2 = NFSFileVec.make(new File("smalldata/parse_folder_test/prostate_2.csv"))._key;
      // Now "break" one of the files.  Globally
      new Break(fkey1).doAllNodes();

      water.parser.ParseDataset2.parse(okey, fkey0,fkey1,fkey2);

    } catch( Throwable e2 ) {
      ex = e2; // Record expected exception
    }
    try {
      // Cleanup is buggy, in the other JVMs run-on, and produce more output
      // keys even after the job is canceled.  Sleep till they hopefully
      // shutdown, then remove keys.
      try { Thread.sleep(100); } catch( InterruptedException ignore ) { }
      //Value v = DKV.get(fkey0);
      //if( v != null ) {
      //  NFSFileVec nfs = v.get();
      //  System.out.println(nfs.toString());
      //}
      assertTrue( "Parse should throw an NPE",ex!=null);
      assertTrue( "All input & output keys not removed", DKV.get(fkey0)==null );
      assertTrue( "All input & output keys not removed", DKV.get(fkey1)==null );
      assertTrue( "All input & output keys not removed", DKV.get(fkey2)==null );
      assertTrue( "All input & output keys not removed", DKV.get(okey )==null );
      // Try again, in the same test, same inputs & outputs but not broken.
      // Should recover completely.
      okey = Key.make("junk.hex");
      fkey0 = NFSFileVec.make(new File("smalldata/parse_folder_test/prostate_0.csv"))._key;
      fkey1 = NFSFileVec.make(new File("smalldata/parse_folder_test/prostate_1.csv"))._key;
      fkey2 = NFSFileVec.make(new File("smalldata/parse_folder_test/prostate_2.csv"))._key;
      Frame fr = water.parser.ParseDataset2.parse(okey, fkey0,fkey1,fkey2);
      fr.delete();
      
      assertTrue( "All input & output keys not removed", DKV.get(fkey0)==null );
      assertTrue( "All input & output keys not removed", DKV.get(fkey1)==null );
      assertTrue( "All input & output keys not removed", DKV.get(fkey2)==null );
      assertTrue( "All input & output keys not removed", DKV.get(okey )==null );
    } finally {
      Keyed.remove(fkey0);
      Keyed.remove(fkey1);
      Keyed.remove(fkey2);
    }
  }

  private static class Break extends MRTask<Break> {
    final Key _key;
    Break(Key key ) { _key = key; }
    @Override public void setupLocal() {
      Vec vec = DKV.get(_key).get();
      Chunk chk = vec.chunkForChunkIdx(0); // Load the chunk (which otherwise loads only lazily)
      chk.crushBytes(); // Illegal setup: Chunk _mem should never be null; will trigger NPE
      tryComplete();
    }
  }

}

@@ -458,32 +458,38 @@
 File(%22smalldata/
+junit/
 parse_folder_tes
@@ -476,37 +476,32 @@
 nit/parse_folder
-_test
 /prostate_0.csv%22
@@ -551,32 +551,38 @@
 File(%22smalldata/
+junit/
 parse_folder_tes
@@ -569,37 +569,32 @@
 nit/parse_folder
-_test
 /prostate_1.csv%22
@@ -644,32 +644,38 @@
 File(%22smalldata/
+junit/
 parse_folder_tes
@@ -662,37 +662,32 @@
 nit/parse_folder
-_test
 /prostate_2.csv%22
@@ -1921,32 +1921,38 @@
 File(%22smalldata/
+junit/
 parse_folder_tes
@@ -1939,37 +1939,32 @@
 nit/parse_folder
-_test
 /prostate_0.csv%22
@@ -2014,32 +2014,38 @@
 File(%22smalldata/
+junit/
 parse_folder_tes
@@ -2032,37 +2032,32 @@
 nit/parse_folder
-_test
 /prostate_1.csv%22
@@ -2115,16 +2115,22 @@
 alldata/
+junit/
 parse_fo
@@ -2137,13 +2137,8 @@
 lder
-_test
 /pro

Fix directory names on broken test.