BlockingStatusTest.java

package azkaban.test.execapp.event;

import org.junit.Assert;
import org.junit.Test;

import azkaban.execapp.event.BlockingStatus;
import azkaban.executor.Status;

public class BlockingStatusTest {

  public class WatchingThread extends Thread {
    private BlockingStatus status;
    private long diff = 0;

    public WatchingThread(BlockingStatus status) {
      this.status = status;
    }

    public void run() {
      long startTime = System.currentTimeMillis();
      status.blockOnFinishedStatus();
      diff = System.currentTimeMillis() - startTime;
    }

    public long getDiff() {
      return diff;
    }
  }

  @Test
  public void testFinishedBlock() {
    BlockingStatus status = new BlockingStatus(1, "test", Status.SKIPPED);

    WatchingThread thread = new WatchingThread(status);
    thread.start();
    try {
      thread.join();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("Diff " + thread.getDiff());
    Assert.assertTrue(thread.getDiff() < 100);
  }

  @Test
  public void testUnfinishedBlock() throws InterruptedException {
    BlockingStatus status = new BlockingStatus(1, "test", Status.QUEUED);

    WatchingThread thread = new WatchingThread(status);
    thread.start();

    synchronized (this) {
      wait(3000);
    }

    status.changeStatus(Status.SUCCEEDED);
    thread.join();

    System.out.println("Diff " + thread.getDiff());
    Assert.assertTrue(thread.getDiff() >= 3000 && thread.getDiff() < 3100);
  }

  @Test
  public void testUnfinishedBlockSeveralChanges() throws InterruptedException {
    BlockingStatus status = new BlockingStatus(1, "test", Status.QUEUED);

    WatchingThread thread = new WatchingThread(status);
    thread.start();

    synchronized (this) {
      wait(3000);
    }

    status.changeStatus(Status.PAUSED);

    synchronized (this) {
      wait(1000);
    }

    status.changeStatus(Status.FAILED);

    thread.join(1000);

    System.out.println("Diff " + thread.getDiff());
    Assert.assertTrue(thread.getDiff() >= 4000 && thread.getDiff() < 4100);
  }

  @Test
  public void testMultipleWatchers() throws InterruptedException {
    BlockingStatus status = new BlockingStatus(1, "test", Status.QUEUED);

    WatchingThread thread1 = new WatchingThread(status);
    thread1.start();

    synchronized (this) {
      wait(2000);
    }

    WatchingThread thread2 = new WatchingThread(status);
    thread2.start();

    synchronized (this) {
      wait(2000);
    }

    status.changeStatus(Status.FAILED);
    thread2.join(1000);
    thread1.join(1000);

    System.out.println("Diff thread 1 " + thread1.getDiff());
    System.out.println("Diff thread 2 " + thread2.getDiff());
    Assert.assertTrue(thread1.getDiff() >= 4000 && thread1.getDiff() < 4100);
    Assert.assertTrue(thread2.getDiff() >= 2000 && thread2.getDiff() < 2100);
  }
}