### Question:
PublisherEuropeanaDao { public DBCursor buildCursorForDocumentStatistics (final int batchSize, final DateTime dateFilter) { final BasicDBObject findQuery = new BasicDBObject(); final BasicDBObject retrievedFields = new BasicDBObject(); if (null != dateFilter) { findQuery.put("updatedAt", new BasicDBObject("$gt", dateFilter.toDate())); } final BasicDBList orList = new BasicDBList(); orList.add(new BasicDBObject("state", ProcessingState.ERROR.name())); orList.add(new BasicDBObject("state", ProcessingState.FAILED.name())); orList.add(new BasicDBObject("state", ProcessingState.SUCCESS.name())); findQuery.put("$or", orList); retrievedFields.put("sourceDocumentReferenceId", 1); retrievedFields.put("referenceOwner", 1); retrievedFields.put("processingJobSubTaskStats", 1); retrievedFields.put("urlSourceType", 1); retrievedFields.put("taskType", 1); retrievedFields.put("updatedAt", 1); retrievedFields.put("_id", 0); final DBCursor cursor = mongoDB.getCollection("SourceDocumentProcessingStatistics") .find(findQuery, retrievedFields) .sort(new BasicDBObject("updatedAt", 1)) .limit(batchSize); return cursor.addOption(Bytes.QUERYOPTION_NOTIMEOUT); } PublisherEuropeanaDao(MongoConfig mongoConfig); DB getMongoDB(); List<HarvesterRecord> retrieveRecords(final DBCursor cursor, final String publishingBatchId); @Deprecated long countNumberOfDocumentUpdatedBefore(final DateTime dateFilter); List<SourceDocumentReferenceMetaInfo> retrieveMetaInfo(final Collection<String> sourceDocumentReferenceIds); DBCursor buildCursorForDocumentStatistics(final int batchSize, final DateTime dateFilter);  }### Answer:
@Test public void test_buildCursor_withoutDataFilter() { final DBCursor cursor = europeanaDao.buildCursorForDocumentStatistics(100, null); assertEquals (orQuery, cursor.getQuery()); }
@Test public void test_buildCursor_withDataFilter() { final DBCursor cursor = europeanaDao.buildCursorForDocumentStatistics(100, publisherConfig.getStartTimestamp()); final BasicDBObject query = new BasicDBObject(); query.put ("updatedAt", new BasicDBObject("$gt", publisherConfig.getStartTimestamp().toDate())); query.put ("$or", orQuery.get("$or")); assertEquals(query, cursor.getQuery()); }
@Test public void test_extractDocuments_batchSizeOne () { final DBCursor cursor = europeanaDao.buildCursorForDocumentStatistics(1, null); checkDocuments(cursor, 1, null); }
@Test public void test_extractDocuments_batchSizeTwo () { final DBCursor cursor = europeanaDao.buildCursorForDocumentStatistics(2, null); checkDocuments(cursor, 2, null); }
@Test public void test_extractDocument_withDataFilter_batchSizeTwo() { final DBCursor cursor = europeanaDao.buildCursorForDocumentStatistics(2, publisherConfig.getStartTimestamp()); checkDocuments(cursor, 2, publisherConfig.getStartTimestamp()); }