| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| # |
| |
| from gremlin_python.process.anonymous_traversal import traversal |
| from gremlin_python.process.graph_traversal import __ |
| from gremlin_python.driver import serializer |
| from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection |
| from radish import before, after, world |
| |
| outV = __.outV |
| label = __.label |
| inV = __.inV |
| project = __.project |
| tail = __.tail |
| |
| |
| @before.all |
| def prepare_static_traversal_source(features, marker): |
| # as the various traversal sources for testing do not change their data, there is no need to re-create remotes |
| # and client side lookup data over and over. it can be created once for all tests and be reused. |
| cache = {} |
| for graph_name in (("modern", "gmodern"), ("classic", "gclassic"), ("crew", "gcrew"), ("grateful", "ggrateful"), ("sink", "gsink")): |
| cache[graph_name[0]] = {} |
| remote = __create_remote(graph_name[1]) |
| cache[graph_name[0]]["remote_conn"] = __create_remote(graph_name[1]) |
| cache[graph_name[0]]["lookup_v"] = __create_lookup_v(remote) |
| cache[graph_name[0]]["lookup_e"] = __create_lookup_e(remote) |
| |
| # store the cache on the global context so that remotes can be shutdown cleanly at the end of the tests |
| world.cache = cache |
| |
| # iterate each feature and apply the cached remotes/lookups to each scenario context so that they are |
| # accessible to the feature steps for test logic |
| for feature in features: |
| for scenario in feature.all_scenarios: |
| scenario.context.remote_conn = {} |
| scenario.context.lookup_v = {} |
| scenario.context.lookup_e = {} |
| |
| for graph_name in ("modern", "classic", "crew", "grateful", "sink"): |
| scenario.context.remote_conn[graph_name] = cache[graph_name]["remote_conn"] |
| scenario.context.lookup_v[graph_name] = cache[graph_name]["lookup_v"] |
| scenario.context.lookup_e[graph_name] = cache[graph_name]["lookup_e"] |
| |
| # setup the "empty" lookups as needed |
| scenario.context.lookup_v["empty"] = {} |
| scenario.context.lookup_e["empty"] = {} |
| |
| |
| @before.each_scenario |
| def prepare_traversal_source(scenario): |
| # some tests create data - create a fresh remote to the empty graph and clear that graph prior to each test |
| remote = __create_remote("ggraph") |
| scenario.context.remote_conn["empty"] = remote |
| g = traversal().withRemote(remote) |
| g.V().drop().iterate() |
| |
| |
| @after.each_scenario |
| def close_traversal_source(scenario): |
| scenario.context.remote_conn["empty"].close() |
| |
| |
| @after.all |
| def close_static_traversal_source(features, marker): |
| for key, value in world.cache.iteritems(): |
| value["remote_conn"].close() |
| |
| |
| def __create_remote(server_graph_name): |
| if not("serializer" in world.config.user_data): |
| raise ValueError('test configuration requires setting of --user-data="serializer={mime-type}"') |
| |
| if world.config.user_data["serializer"] == "application/vnd.gremlin-v3.0+json": |
| s = serializer.GraphSONSerializersV3d0() |
| elif world.config.user_data["serializer"] == "application/vnd.graphbinary-v1.0": |
| s = serializer.GraphBinarySerializersV1() |
| else: |
| raise ValueError('serializer not found - ' + world.config.user_data["serializer"]) |
| |
| return DriverRemoteConnection('ws://localhost:45940/gremlin', server_graph_name, message_serializer=s) |
| |
| |
| def __create_lookup_v(remote): |
| g = traversal().withRemote(remote) |
| |
| # hold a map of name/vertex for use in asserting results |
| return g.V().group().by('name').by(tail()).next() |
| |
| |
| def __create_lookup_e(remote): |
| g = traversal().withRemote(remote) |
| |
| # hold a map of the "name"/edge for use in asserting results - "name" in this context is in the form of |
| # outgoingV-label->incomingV |
| return g.E().group(). \ |
| by(lambda: ("it.outVertex().value('name') + '-' + it.label() + '->' + it.inVertex().value('name')", "gremlin-groovy")). \ |
| by(tail()).next() |