@Test
public void testAssignAndUnassignExecutor() throws Exception {

  final String host = "localhost";
  final int port = 12345;
  final Executor executor = this.executorDao.addExecutor(host, port);
  final ExecutableFlow flow = TestUtils.createTestExecutableFlow("exectest1", "exec1");
  this.executionFlowDao.uploadExecutableFlow(flow);
  this.assignExecutor.assignExecutor(executor.getId(), flow.getExecutionId());
  final Executor fetchExecutor = this.executorDao
      .fetchExecutorByExecutionId(flow.getExecutionId());
  assertThat(fetchExecutor).isEqualTo(executor);
  this.assignExecutor.unassignExecutor(flow.getExecutionId());
  assertThat(this.executorDao.fetchExecutorByExecutionId(flow.getExecutionId())).isNull();
}