### Question:
Client { public JobExecutionResult run(final PackagedProgram prog, int parallelism, boolean wait) throws ProgramInvocationException { Thread.currentThread().setContextClassLoader(prog.getUserCodeClassLoader()); if (prog.isUsingProgramEntryPoint()) { return run(prog.getPlanWithJars(), parallelism, wait); } else if (prog.isUsingInteractiveMode()) { ContextEnvironment env = new ContextEnvironment(this, prog.getAllLibraries(), prog.getUserCodeClassLoader()); if (parallelism > 0) { env.setDegreeOfParallelism(parallelism); } env.setAsContext(); if (wait) { prog.invokeInteractiveModeForExecution(); } else { Thread backGroundRunner = new Thread("Program Runner") { public void run() { try { prog.invokeInteractiveModeForExecution(); } catch (Throwable t) { LOG.error("The program execution failed.", t); } }; }; backGroundRunner.start(); } return null; } else { throw new RuntimeException(); } } Client(InetSocketAddress jobManagerAddress, Configuration config);  Client(Configuration config); void setPrintStatusDuringExecution(boolean print); String getJobManagerAddress(); int getJobManagerPort(); String getOptimizedPlanAsJson(PackagedProgram prog, int parallelism); OptimizedPlan getOptimizedPlan(PackagedProgram prog, int parallelism); OptimizedPlan getOptimizedPlan(Plan p, int parallelism); OptimizedPlan getOptimizedPlan(JobWithJars prog, int parallelism); JobGraph getJobGraph(PackagedProgram prog, OptimizedPlan optPlan); JobExecutionResult run(final PackagedProgram prog, int parallelism, boolean wait); JobExecutionResult run(PackagedProgram prog, OptimizedPlan optimizedPlan, boolean wait); JobExecutionResult run(JobWithJars prog, int parallelism, boolean wait); JobExecutionResult run(OptimizedPlan compiledPlan, List<File> libraries, boolean wait); JobExecutionResult run(JobGraph jobGraph, boolean wait);  }### Answer:
@Test public void shouldSubmitToJobClient() throws ProgramInvocationException, IOException { when(jobSubmissionResultMock.getReturnCode()).thenReturn(ReturnCode.SUCCESS); Client out = new Client(configMock); out.run(program.getPlanWithJars(), -1, false); program.deleteExtractedLibraries(); verify(this.compilerMock, times(1)).compile(planMock); verify(this.generatorMock, times(1)).compileJobGraph(optimizedPlanMock); verify(this.jobClientMock, times(1)).submitJob(); }
@Test(expected=ProgramInvocationException.class) public void shouldThrowException() throws Exception { when(jobSubmissionResultMock.getReturnCode()).thenReturn(ReturnCode.ERROR); Client out = new Client(configMock); out.run(program.getPlanWithJars(), -1, false); program.deleteExtractedLibraries(); verify(this.jobClientMock).submitJob(); }