@Test
public void testWorkerStarted() throws Exception {

	new Context() {{
		// set the initial state with a (recovered) launched worker
		MesosWorkerStore.Worker worker1launched = MesosWorkerStore.Worker.newWorker(task1).launchWorker(slave1, slave1host);
		when(rmServices.workerStore.getFrameworkID()).thenReturn(Option.apply(framework1));
		when(rmServices.workerStore.recoverWorkers()).thenReturn(singletonList(worker1launched));
		startResourceManager();
		assertThat(resourceManager.workersInLaunch, hasEntry(extractResourceID(task1), worker1launched));
		// send registration message
		CompletableFuture<RegistrationResponse> successfulFuture =
			resourceManager.registerTaskExecutor(task1Executor.address, task1Executor.resourceID, slotReport, timeout);
		RegistrationResponse response = successfulFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
		assertTrue(response instanceof TaskExecutorRegistrationSuccess);
		// verify the internal state
		assertThat(resourceManager.workersInLaunch, hasEntry(extractResourceID(task1), worker1launched));
	}};
}