PGETaskInstance implements WorkflowTaskInstance { protected void runIngestCrawler(ProductCrawler crawler) throws Exception { if (crawler instanceof StdProductCrawler){ this.processOutput(); } List<File> crawlDirs = new LinkedList<File>(); for (OutputDir outputDir : pgeConfig.getOuputDirs()) { crawlDirs.add(new File(outputDir.getPath())); } updateStatus(CRAWLING.getWorkflowStatusName()); boolean attemptIngestAll = Boolean.parseBoolean(pgeMetadata .getMetadata(ATTEMPT_INGEST_ALL)); for (File crawlDir : crawlDirs) { logger.info("Crawling for products in [" + crawlDir + "]"); crawler.crawl(crawlDir); if (!attemptIngestAll) { verifyIngests(crawler); } } if (attemptIngestAll) { verifyIngests(crawler); } } protected  PGETaskInstance(); @Override void run(Metadata metadata, WorkflowTaskConfiguration config); String getWorkflowInstId(); void setWorkflowInstId(String workflowInstId); void setWmClient(WorkflowManagerClient wmClient); @Override void finalize();  }
@Test public void testRunIngestCrawler() throws Exception { PGETaskInstance pgeTask = createTestInstance(); pgeTask.pgeConfig.addOuputDirAndExpressions(new OutputDir("/tmp/dir1", true)); pgeTask.pgeConfig.addOuputDirAndExpressions(new OutputDir("/tmp/dir2", true)); pgeTask.pgeMetadata.replaceMetadata(ATTEMPT_INGEST_ALL, Boolean.toString(true)); pgeTask.setWorkflowInstId("WorkflowInstanceId"); pgeTask.setWmClient(createMock(WorkflowManagerClient.class)); expect(pgeTask.getWorkflowManagerClient() .updateWorkflowInstanceStatus(pgeTask.getWorkflowInstId(), CRAWLING.getWorkflowStatusName()) ).andReturn(true); replay(pgeTask.getWorkflowManagerClient()); AutoDetectProductCrawler pc = createMock(AutoDetectProductCrawler.class); pc.crawl(new File("/tmp/dir1")); pc.crawl(new File("/tmp/dir2")); expect(pc.getIngestStatus()).andReturn(Collections.<IngestStatus>emptyList()); replay(pc); pgeTask.runIngestCrawler(pc); verify(pgeTask.getWorkflowManagerClient()); verify(pc); pgeTask.setWmClient(createMock(WorkflowManagerClient.class)); expect(pgeTask.getWorkflowManagerClient().updateWorkflowInstanceStatus(pgeTask.getWorkflowInstId(), CRAWLING.getWorkflowStatusName())).andReturn(false); replay(pgeTask.getWorkflowManagerClient()); pc = createMock(AutoDetectProductCrawler.class); replay(pc); try { pgeTask.runIngestCrawler(pc); fail("Should have thrown"); } catch (Exception e) { } verify(pgeTask.getWorkflowManagerClient()); verify(pc); pgeTask.setWmClient(createMock(WorkflowManagerClient.class)); expect(pgeTask.getWorkflowManagerClient().updateWorkflowInstanceStatus(pgeTask.getWorkflowInstId(), CRAWLING.getWorkflowStatusName())).andReturn(true); replay(pgeTask.getWorkflowManagerClient()); pc = createMock(AutoDetectProductCrawler.class); pc.crawl(new File("/tmp/dir1")); pc.crawl(new File("/tmp/dir2")); IngestStatus failedIngestStatus = new IngestStatus() { @Override public String getMessage() { return "Ingest Failure"; } @Override public File getProduct() { return new File("/tmp/dir1"); } @Override public Result getResult() { return Result.FAILURE; } }; expect(pc.getIngestStatus()).andReturn( Lists.newArrayList(failedIngestStatus)); replay(pc); try { pgeTask.runIngestCrawler(pc); fail("Should have thrown"); } catch (Exception e) { } verify(pgeTask.getWorkflowManagerClient()); verify(pc); }