@Test
    public void shouldSuccessfullyEvaluateInterceptedTraversals() throws Exception {
        final Configuration configuration = getBaseConfiguration();
        configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, TestFiles.PATHS.get("tinkerpop-modern-v3.kryo"));
        configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
        configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
        configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, TestHelper.makeTestDataDirectory(SparkSingleIterationStrategyTest.class, UUID.randomUUID().toString()));
        configuration.setProperty(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER, SparkGraphComputer.class.getCanonicalName());
        configuration.setProperty(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true);
        ///
        Graph graph = GraphFactory.open(configuration);

        // removed ProductiveByStrategy for purpose of the testing semantics which were here prior to it
        GraphTraversalSource g = graph.traversal().withComputer().withoutStrategies(SparkSingleIterationStrategy.class, ProductiveByStrategy.class);
        assertFalse(g.getStrategies().getStrategy(SparkSingleIterationStrategy.class).isPresent());
        assertFalse(g.V().count().explain().toString().contains(SparkSingleIterationStrategy.class.getSimpleName()));
        assertTrue(g.getStrategies().getStrategy(SparkInterceptorStrategy.class).isPresent());
        assertTrue(g.V().count().explain().toString().contains(SparkInterceptorStrategy.class.getSimpleName()));
        /// SparkCountInterceptor matches
        test(SparkStarBarrierInterceptor.class, 6l, g.V().count());
        test(SparkStarBarrierInterceptor.class, 2l, g.V().hasLabel("software").count());
        test(SparkStarBarrierInterceptor.class, 2l, g.V().hasLabel("person").has("age", P.gt(30)).count());
        test(SparkStarBarrierInterceptor.class, 2l, g.V().hasLabel("person").has("age", P.gt(30)).values("name").count());
        test(SparkStarBarrierInterceptor.class, 2l, g.V().hasLabel("person").has("age", P.gt(30)).properties("name").count());
        test(SparkStarBarrierInterceptor.class, 4l, g.V().hasLabel("person").has("age", P.gt(30)).properties("name", "age").count());
        test(SparkStarBarrierInterceptor.class, 3l, g.V().hasLabel("person").has("age", P.gt(30)).out().count());
        test(SparkStarBarrierInterceptor.class, 0l, g.V().hasLabel("person").has("age", P.gt(30)).out("knows").count());
        test(SparkStarBarrierInterceptor.class, 3l, g.V().has(T.label, P.not(P.within("robot", "android")).and(P.within("person", "software"))).hasLabel("person").has("age", P.gt(30)).out("created").count());
        test(SparkStarBarrierInterceptor.class, 3l, g.V(1).out().count());
        test(SparkStarBarrierInterceptor.class, 2l, g.V(1).out("knows").count());
        test(SparkStarBarrierInterceptor.class, 3l, g.V(1).out("knows", "created").count());
        test(SparkStarBarrierInterceptor.class, 5l, g.V(1, 4).out("knows", "created").count());
        test(SparkStarBarrierInterceptor.class, 1l, g.V(2).in("knows").count());
        test(SparkStarBarrierInterceptor.class, 0l, g.V(6).has("name", "peter").in().count());
        test(SparkStarBarrierInterceptor.class, 6l, g.V().as("a").values("name").as("b").count());
        test(SparkStarBarrierInterceptor.class, 6l, g.V().as("a").count());
        test(SparkStarBarrierInterceptor.class, 1l, g.V().has("name", "marko").as("a").values("name").as("b").count());
        test(SparkStarBarrierInterceptor.class, 4l, g.V().has(T.label, P.not(P.within("robot", "android")).and(P.within("person", "software"))).hasLabel("person").has("age").out("created").count());
        test(SparkStarBarrierInterceptor.class, 123, g.V().has("age").values("age").sum());
        test(SparkStarBarrierInterceptor.class, 67, g.V().has("age").has("age", P.gt(30)).values("age").sum());
        test(SparkStarBarrierInterceptor.class, 27, g.V().hasLabel("person").values("age").min());
        test(SparkStarBarrierInterceptor.class, 35, g.V().hasLabel("person").values("age").max());
        test(SparkStarBarrierInterceptor.class, new HashMap<String, Long>() {{
            put("software", 2l);
            put("person", 4l);
        }}, g.V().<String>groupCount().by(T.label));
        test(SparkStarBarrierInterceptor.class, Collections.singletonMap("person", 2l), g.V().has("person", "age", P.lt(30)).<String>groupCount().by(T.label));
        test(SparkStarBarrierInterceptor.class, new HashMap<String, Long>() {{
            put("software", 2l);
            put("person", 4l);
        }}, g.V().<String, Long>group().by(T.label).by(__.count()));
        test(SparkStarBarrierInterceptor.class, 123l, g.V().hasLabel("person").values("age").fold(0l, Operator.sum));
        /// No interceptor matches
        test(2l, g.V().out().out().count());
        test(6l, g.E().count());
        test(2l, g.V().out().out().count());
        test(6l, g.V().out().values("name").count());
        test(2l, g.V().out("knows").values("name").count());
        test(3l, g.V().in().has("name", "marko").count());
        test(0l, g.V().repeat(__.dedup()).times(2).count());
        test(6l, g.V().dedup().count());
        test(4l, g.V().hasLabel("person").order().by("age").count());
        test(1l, g.V().count().count());
        test(2l, g.V().limit(2).count());
        test(3l, g.V().tail(3).count());
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkInterceptorStrategyTest.java