public void checkHeadMethods(final Storage storage, final String inputLocation, final String outputLocation, final Class outputGraphParserClass, final Class outputMemoryParserClass) throws Exception {
        // TEST INPUT GRAPH
        assertFalse(storage.exists(outputLocation));
        if (inputLocation.endsWith(".json") && storage.exists(inputLocation)) { // gryo is not text readable
            assertEquals(6, IteratorUtils.count(storage.head(inputLocation)));
            for (int i = 0; i < 7; i++) {
                assertEquals(i, IteratorUtils.count(storage.head(inputLocation, i)));
            }
            assertEquals(6, IteratorUtils.count(storage.head(inputLocation, 10)));
        }

        ////////////////////

        final ComputerResult result = graphProvider.getGraphComputer(graph).program(PeerPressureVertexProgram.build().create(graph)).mapReduce(ClusterCountMapReduce.build().memoryKey("clusterCount").create()).submit().get();

        // TEST OUTPUT GRAPH
        assertTrue(storage.exists(outputLocation));
        assertTrue(storage.exists(Constants.getGraphLocation(outputLocation)));
        assertEquals(6, result.graph().traversal().V().count().next().longValue());
        assertEquals(0, result.graph().traversal().E().count().next().longValue());
        assertEquals(6, result.graph().traversal().V().values("name").count().next().longValue());
        assertEquals(6, result.graph().traversal().V().values(PeerPressureVertexProgram.CLUSTER).count().next().longValue());
        assertEquals(2, result.graph().traversal().V().values(PeerPressureVertexProgram.CLUSTER).dedup().count().next().longValue());
        assertEquals(6, IteratorUtils.count(storage.head(Constants.getGraphLocation(outputLocation), outputGraphParserClass)));
        for (int i = 0; i < 7; i++) {
            assertEquals(i, IteratorUtils.count(storage.head(Constants.getGraphLocation(outputLocation), outputGraphParserClass, i)));
        }
        assertEquals(6, IteratorUtils.count(storage.head(Constants.getGraphLocation(outputLocation), outputGraphParserClass, 346)));
        /////
        // TEST MEMORY PERSISTENCE
        assertEquals(2, (int) result.memory().get("clusterCount"));
        assertTrue(storage.exists(Constants.getMemoryLocation(outputLocation, "clusterCount")));
        assertEquals(1, IteratorUtils.count(storage.head(outputLocation, "clusterCount", outputMemoryParserClass)));
        assertEquals(2, storage.head(outputLocation, "clusterCount", outputMemoryParserClass).next().getValue());
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/AbstractStorageCheck.java