### Question:
ProvisioningService extends AbstractIdleService { public ClusterStatus getClusterStatus(ProgramRunId programRunId, ProgramOptions programOptions, Cluster cluster, String userId) throws Exception { Map<String, String> systemArgs = programOptions.getArguments().asMap(); String name = SystemArguments.getProfileProvisioner(systemArgs); Provisioner provisioner = provisionerInfo.get().provisioners.get(name); if (provisioner == null) { return ClusterStatus.NOT_EXISTS; } Map<String, String> properties = SystemArguments.getProfileProperties(systemArgs); ProvisionerContext context; try { DefaultSSHContext defaultSSHContext = null; if (!getRuntimeJobManager(programRunId, programOptions).isPresent()) { defaultSSHContext = new DefaultSSHContext( Networks.getAddress(cConf, Constants.NETWORK_PROXY_ADDRESS), null, null); } context = createContext(programRunId, userId, properties, defaultSSHContext, SystemArguments.getRuntimeMonitorType(cConf, programOptions), name); } catch (InvalidMacroException e) { runWithProgramLogging(programRunId, systemArgs, () -> LOG.error("Could not evaluate macros while checking cluster status.", e)); return ClusterStatus.NOT_EXISTS; } return Retries.callWithRetries(() -> provisioner.getClusterStatus(context, cluster), RetryStrategies.exponentialDelay(1, 5, TimeUnit.SECONDS), RetryableProvisionException.class::isInstance); } @Inject  ProvisioningService(CConfiguration cConf, ProvisionerProvider provisionerProvider,
                      ProvisionerConfigProvider provisionerConfigProvider,
                      ProvisionerNotifier provisionerNotifier, LocationFactory locationFactory,
                      SecureStore secureStore, ProgramStateWriter programStateWriter,
                      ProvisionerStore provisionerStore, TransactionRunner transactionRunner,
                      MetricsCollectionService metricsCollectionService); ClusterStatus getClusterStatus(ProgramRunId programRunId, ProgramOptions programOptions,
                                        Cluster cluster, String userId); Optional<ProvisioningTaskInfo> cancelProvisionTask(ProgramRunId programRunId); Runnable provision(ProvisionRequest provisionRequest, StructuredTableContext context); Optional<RuntimeJobManager> getRuntimeJobManager(ProgramRunId programRunId, ProgramOptions programOptions); Runnable deprovision(ProgramRunId programRunId, StructuredTableContext context); Collection<ProvisionerDetail> getProvisionerDetails(); @Nullable ProvisionerDetail getProvisionerDetail(String name); void validateProperties(String provisionerName, Map<String, String> properties);  }### Answer:
@Test public void testGetClusterStatus() throws Exception { TaskFields taskFields = createTaskInfo(new MockProvisioner.PropertyBuilder() .setFirstClusterStatus(ClusterStatus.RUNNING) .failRetryablyEveryN(2) .build()); Cluster cluster = new Cluster("test", ClusterStatus.NOT_EXISTS, Collections.emptyList(), Collections.emptyMap()); Assert.assertEquals(ClusterStatus.RUNNING, provisioningService.getClusterStatus(taskFields.programRunId, taskFields.programOptions, cluster, "cdap")); }
@Test(expected = Exception.class) public void testGetClusterStatusFailure() throws Exception { TaskFields taskFields = createTaskInfo(new MockProvisioner.PropertyBuilder() .setFirstClusterStatus(ClusterStatus.RUNNING) .failGet() .build()); Cluster cluster = new Cluster("test", ClusterStatus.NOT_EXISTS, Collections.emptyList(), Collections.emptyMap()); provisioningService.getClusterStatus(taskFields.programRunId, taskFields.programOptions, cluster, "cdap"); }