package water;

import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import water.api.UnlockTask;
import water.fvec.Frame;
import water.util.Log;

public class UnlockTest extends TestUtil {
  @BeforeClass
  public static void stall() { stall_till_cloudsize(3); }

  static class DJob extends Job {
    DJob(Key a) { super(a,"DJob",1); }
  }

  @Test
  public void run() {
    // Put chunks into KV store
    Frame f = new TestUtil().parse_test_file("smalldata/junit/syn_2659x1049.csv.gz");
    // Create two lockable frames in KV store
    Frame fr1 = new Frame(Key.make(), f.names(), f.vecs());
    Frame fr2 = new Frame(Key.make(), f.names(), f.vecs());
    DJob dj = new DJob(Key.make());
    // Lock the frames against writes
    fr1.delete_and_lock(dj._key);
    fr2.delete_and_lock(dj._key);
    int i = 0;
    try {
      // try to delete the write-locked frames -> will throw an exception
      fr1.delete();
      fr2.delete(); // won't be reached
    } catch (Throwable t) {
      Log.info("Correctly caught exception: " + t.getClass()); //either AssertionError if local or DistributedException if remote
      i++;
    } finally {
      // second attempt: will unlock and delete properly
      new UnlockTask().doAllNodes(); // without this line, there will be a leak (and assertion won't be shown)
      fr1.delete();
      fr2.delete();
      f.delete();
      Log.info("Able to delete after unlocking.");
    }
    Assert.assertTrue(i == 1);
  }
}

@@ -286,94 +286,16 @@
  %7D%0A%0A
-  static class DJob extends Job %7B%0A    DJob(Key a) %7B super(a,%22DJob%22,1); %7D%0A  %7D%0A%0A
   @Test%0A
   pu
@@ -290,16 +290,16 @@
   @Test%0A
+
   public
@@ -600,44 +600,8 @@
 ));%0A
-    DJob dj = new DJob(Key.make());%0A
     
@@ -654,31 +654,28 @@
 te_and_lock(
-dj._key
+null
 );%0A    fr2.d
@@ -693,15 +693,12 @@
 ock(
-dj._key
+null
 );%0A 
@@ -913,24 +913,37 @@
 tly 
-caught exception
+unable to delete (was locked)
 : %22 

Clean up UnlockTest, no need for dummy job.