# Copyright DataStax, Inc. # # Licensed 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. import sys import datetime import six import time from collections import namedtuple from packaging.version import Version from cassandra.datastax.graph.fluent import DseGraph from cassandra.graph import VertexProperty, GraphProtocol from cassandra.util import Point, Polygon, LineString from gremlin_python.process.graph_traversal import GraphTraversal, GraphTraversalSource from gremlin_python.process.traversal import P from gremlin_python.structure.graph import Edge as TravEdge from gremlin_python.structure.graph import Vertex as TravVertex, VertexProperty as TravVertexProperty from tests.util import wait_until_not_raised from tests.integration import DSE_VERSION from tests.integration.advanced.graph import ( GraphUnitTestCase, ClassicGraphSchema, CoreGraphSchema, VertexLabel) from tests.integration import requiredse try: import unittest2 as unittest except ImportError: import unittest # noqa import ipaddress def check_equality_base(testcase, original, read_value): if isinstance(original, float): testcase.assertAlmostEqual(original, read_value, delta=.01) elif isinstance(original, ipaddress.IPv4Address): testcase.assertAlmostEqual(original, ipaddress.IPv4Address(read_value)) elif isinstance(original, ipaddress.IPv6Address): testcase.assertAlmostEqual(original, ipaddress.IPv6Address(read_value)) else: testcase.assertEqual(original, read_value) def create_traversal_profiles(cluster, graph_name): ep_graphson2 = DseGraph().create_execution_profile( graph_name, graph_protocol=GraphProtocol.GRAPHSON_2_0) ep_graphson3 = DseGraph().create_execution_profile( graph_name, graph_protocol=GraphProtocol.GRAPHSON_3_0) cluster.add_execution_profile('traversal_graphson2', ep_graphson2) cluster.add_execution_profile('traversal_graphson3', ep_graphson3) return ep_graphson2, ep_graphson3 class _AbstractTraversalTest(GraphUnitTestCase): def setUp(self): super(_AbstractTraversalTest, self).setUp() self.ep_graphson2, self.ep_graphson3 = create_traversal_profiles(self.cluster, self.graph_name) def _test_basic_query(self, schema, graphson): """ Test to validate that basic graph queries works Creates a simple classic tinkerpot graph, and attempts to preform a basic query using Tinkerpop's GLV with both explicit and implicit execution ensuring that each one is correct. See reference graph here http://www.tinkerpop.com/docs/3.0.0.M1/ @since 1.0.0 @jira_ticket PYTHON-641 @expected_result graph should generate and all vertices and edge results should be @test_category dse graph """ g = self.fetch_traversal_source(graphson) self.execute_graph(schema.fixtures.classic(), graphson) traversal = g.V().has('name', 'marko').out('knows').values('name') results_list = self.execute_traversal(traversal, graphson) self.assertEqual(len(results_list), 2) self.assertIn('vadas', results_list) self.assertIn('josh', results_list) def _test_classic_graph(self, schema, graphson): """ Test to validate that basic graph generation, and vertex and edges are surfaced correctly Creates a simple classic tinkerpot graph, and iterates over the the vertices and edges using Tinkerpop's GLV with both explicit and implicit execution ensuring that each one iscorrect. See reference graph here http://www.tinkerpop.com/docs/3.0.0.M1/ @since 1.0.0 @jira_ticket PYTHON-641 @expected_result graph should generate and all vertices and edge results should be @test_category dse graph """ self.execute_graph(schema.fixtures.classic(), graphson) ep = self.get_execution_profile(graphson) g = self.fetch_traversal_source(graphson) traversal = g.V() vert_list = self.execute_traversal(traversal, graphson) for vertex in vert_list: schema.ensure_properties(self.session, vertex, execution_profile=ep) self._validate_classic_vertex(g, vertex) traversal = g.E() edge_list = self.execute_traversal(traversal, graphson) for edge in edge_list: schema.ensure_properties(self.session, edge, execution_profile=ep) self._validate_classic_edge(g, edge) def _test_graph_classic_path(self, schema, graphson): """ Test to validate that the path version of the result type is generated correctly. It also tests basic path results as that is not covered elsewhere @since 1.0.0 @jira_ticket PYTHON-641 @expected_result path object should be unpacked correctly including all nested edges and vertices @test_category dse graph """ self.execute_graph(schema.fixtures.classic(), graphson) g = self.fetch_traversal_source(graphson) traversal = g.V().hasLabel('person').has('name', 'marko').as_('a').outE('knows').inV().as_('c', 'd').outE('created').as_('e', 'f', 'g').inV().path() path_list = self.execute_traversal(traversal, graphson) self.assertEqual(len(path_list), 2) for path in path_list: self._validate_path_result_type(g, path) def _test_range_query(self, schema, graphson): """ Test to validate range queries are handled correctly. Creates a very large line graph script and executes it. Then proceeds to to a range limited query against it, and ensure that the results are formated correctly and that the result set is properly sized. @since 1.0.0 @jira_ticket PYTHON-641 @expected_result result set should be properly formated and properly sized @test_category dse graph """ self.execute_graph(schema.fixtures.line(150), graphson) ep = self.get_execution_profile(graphson) g = self.fetch_traversal_source(graphson) traversal = g.E().range(0, 10) edges = self.execute_traversal(traversal, graphson) self.assertEqual(len(edges), 10) for edge in edges: schema.ensure_properties(self.session, edge, execution_profile=ep) self._validate_line_edge(g, edge) def _test_result_types(self, schema, graphson): """ Test to validate that the edge and vertex version of results are constructed correctly. @since 1.0.0 @jira_ticket PYTHON-641 @expected_result edge/vertex result types should be unpacked correctly. @test_category dse graph """ self.execute_graph(schema.fixtures.line(150), graphson) g = self.fetch_traversal_source(graphson) traversal = g.V() vertices = self.execute_traversal(traversal, graphson) for vertex in vertices: self._validate_type(g, vertex) def _test_large_result_set(self, schema, graphson): """ Test to validate that large result sets return correctly. Creates a very large graph. Ensures that large result sets are handled appropriately. @since 1.0.0 @jira_ticket PYTHON-641 @expected_result when limits of result sets are hit errors should be surfaced appropriately @test_category dse graph """ self.execute_graph(schema.fixtures.large(), graphson) g = self.fetch_traversal_source(graphson) traversal = g.V() vertices = self.execute_traversal(traversal, graphson) for vertex in vertices: self._validate_generic_vertex_result_type(g, vertex) def _test_vertex_meta_properties(self, schema, graphson): """ Test verifying vertex property properties @since 1.0.0 @jira_ticket PYTHON-641 @test_category dse graph """ if schema is not ClassicGraphSchema: raise unittest.SkipTest('skipped because multiple properties are only supported with classic graphs') s = self.session s.execute_graph("schema.propertyKey('k0').Text().ifNotExists().create();") s.execute_graph("schema.propertyKey('k1').Text().ifNotExists().create();") s.execute_graph("schema.propertyKey('key').Text().properties('k0', 'k1').ifNotExists().create();") s.execute_graph("schema.vertexLabel('MLP').properties('key').ifNotExists().create();") s.execute_graph("schema.config().option('graph.allow_scan').set('true');") v = s.execute_graph('''v = graph.addVertex('MLP') v.property('key', 'meta_prop', 'k0', 'v0', 'k1', 'v1') v''')[0] g = self.fetch_traversal_source(graphson) traversal = g.V() # This should contain key, and value where value is a property # This should be a vertex property and should contain sub properties results = self.execute_traversal(traversal, graphson) self._validate_meta_property(g, results[0]) def _test_vertex_multiple_properties(self, schema, graphson): """ Test verifying vertex property form for various Cardinality All key types are encoded as a list, regardless of cardinality Single cardinality properties have only one value -- the last one added Default is single (this is config dependent) @since 1.0.0 @jira_ticket PYTHON-641 @test_category dse graph """ if schema is not ClassicGraphSchema: raise unittest.SkipTest('skipped because multiple properties are only supported with classic graphs') s = self.session s.execute_graph('''Schema schema = graph.schema(); schema.propertyKey('mult_key').Text().multiple().ifNotExists().create(); schema.propertyKey('single_key').Text().single().ifNotExists().create(); schema.vertexLabel('MPW1').properties('mult_key').ifNotExists().create(); schema.vertexLabel('MPW2').properties('mult_key').ifNotExists().create(); schema.vertexLabel('SW1').properties('single_key').ifNotExists().create();''') mpw1v = s.execute_graph('''v = graph.addVertex('MPW1') v.property('mult_key', 'value') v''')[0] mpw2v = s.execute_graph('''g.addV('MPW2').property('mult_key', 'value0').property('mult_key', 'value1')''')[0] g = self.fetch_traversal_source(graphson) traversal = g.V(mpw1v.id).properties() vertex_props = self.execute_traversal(traversal, graphson) self.assertEqual(len(vertex_props), 1) self.assertEqual(self.fetch_key_from_prop(vertex_props[0]), "mult_key") self.assertEqual(vertex_props[0].value, "value") # multiple_with_two_values #v = s.execute_graph('''g.addV(label, 'MPW2', 'mult_key', 'value0', 'mult_key', 'value1')''')[0] traversal = g.V(mpw2v.id).properties() vertex_props = self.execute_traversal(traversal, graphson) self.assertEqual(len(vertex_props), 2) self.assertEqual(self.fetch_key_from_prop(vertex_props[0]), 'mult_key') self.assertEqual(self.fetch_key_from_prop(vertex_props[1]), 'mult_key') self.assertEqual(vertex_props[0].value, 'value0') self.assertEqual(vertex_props[1].value, 'value1') # single_with_one_value v = s.execute_graph('''v = graph.addVertex('SW1') v.property('single_key', 'value') v''')[0] traversal = g.V(v.id).properties() vertex_props = self.execute_traversal(traversal, graphson) self.assertEqual(len(vertex_props), 1) self.assertEqual(self.fetch_key_from_prop(vertex_props[0]), "single_key") self.assertEqual(vertex_props[0].value, "value") def should_parse_meta_properties(self): g = self.fetch_traversal_source() g.addV("meta_v").property("meta_prop", "hello", "sub_prop", "hi", "sub_prop2", "hi2") def _test_all_graph_types_with_schema(self, schema, graphson): """ Exhaustively goes through each type that is supported by dse_graph. creates a vertex for each type using a dse-tinkerpop traversal, It then attempts to fetch it from the server and compares it to what was inserted Prime the graph with the correct schema first @since 1.0.0 @jira_ticket PYTHON-641 @expected_result inserted objects are equivalent to those retrieved @test_category dse graph """ self._write_and_read_data_types(schema, graphson) def _test_all_graph_types_without_schema(self, schema, graphson): """ Exhaustively goes through each type that is supported by dse_graph. creates a vertex for each type using a dse-tinkerpop traversal, It then attempts to fetch it from the server and compares it to what was inserted Do not prime the graph with the correct schema first @since 1.0.0 @jira_ticket PYTHON-641 @expected_result inserted objects are equivalent to those retrieved @test_category dse graph """ if schema is not ClassicGraphSchema: raise unittest.SkipTest('schema-less is only for classic graphs') self._write_and_read_data_types(schema, graphson, use_schema=False) def _test_dsl(self, schema, graphson): """ The test creates a SocialTraversal and a SocialTraversalSource as part of a DSL. Then calls it's method and checks the results to verify we have the expected results @since @since 1.1.0a1 @jira_ticket PYTHON-790 @expected_result only the vertex corresponding to marko is in the result @test_category dse graph """ class SocialTraversal(GraphTraversal): def knows(self, person_name): return self.out("knows").hasLabel("person").has("name", person_name).in_() class SocialTraversalSource(GraphTraversalSource): def __init__(self, *args, **kwargs): super(SocialTraversalSource, self).__init__(*args, **kwargs) self.graph_traversal = SocialTraversal def people(self, *names): return self.get_graph_traversal().V().has("name", P.within(*names)) self.execute_graph(schema.fixtures.classic(), graphson) if schema is CoreGraphSchema: self.execute_graph(""" schema.edgeLabel('knows').from('person').to('person').materializedView('person__knows__person_by_in_name'). ifNotExists().partitionBy('in_name').clusterBy('out_name', Asc).create() """, graphson) time.sleep(1) # give some time to the MV to be populated g = self.fetch_traversal_source(graphson, traversal_class=SocialTraversalSource) traversal = g.people("marko", "albert").knows("vadas") results = self.execute_traversal(traversal, graphson) self.assertEqual(len(results), 1) only_vertex = results[0] schema.ensure_properties(self.session, only_vertex, execution_profile=self.get_execution_profile(graphson)) self._validate_classic_vertex(g, only_vertex) def _test_bulked_results(self, schema, graphson): """ Send a query expecting a bulked result and the driver "undoes" the bulk and returns the expected list @since 1.1.0a1 @jira_ticket PYTHON-771 @expected_result the expanded list @test_category dse graph """ self.execute_graph(schema.fixtures.classic(), graphson) g = self.fetch_traversal_source(graphson) barrier_traversal = g.E().label().barrier() results = self.execute_traversal(barrier_traversal, graphson) self.assertEqual(sorted(["created", "created", "created", "created", "knows", "knows"]), sorted(results)) def _test_udt_with_classes(self, schema, graphson): class Address(object): def __init__(self, address, city, state): self.address = address self.city = city self.state = state def __eq__(self, other): return self.address == other.address and self.city == other.city and self.state == other.state class AddressWithTags(object): def __init__(self, address, city, state, tags): self.address = address self.city = city self.state = state self.tags = tags def __eq__(self, other): return (self.address == other.address and self.city == other.city and self.state == other.state and self.tags == other.tags) class ComplexAddress(object): def __init__(self, address, address_tags, city, state, props): self.address = address self.address_tags = address_tags self.city = city self.state = state self.props = props def __eq__(self, other): return (self.address == other.address and self.address_tags == other.address_tags and self.city == other.city and self.state == other.state and self.props == other.props) class ComplexAddressWithOwners(object): def __init__(self, address, address_tags, city, state, props, owners): self.address = address self.address_tags = address_tags self.city = city self.state = state self.props = props self.owners = owners def __eq__(self, other): return (self.address == other.address and self.address_tags == other.address_tags and self.city == other.city and self.state == other.state and self.props == other.props and self.owners == other.owners) self.__test_udt(schema, graphson, Address, AddressWithTags, ComplexAddress, ComplexAddressWithOwners) def _test_udt_with_namedtuples(self, schema, graphson): AddressTuple = namedtuple('Address', ('address', 'city', 'state')) AddressWithTagsTuple = namedtuple('AddressWithTags', ('address', 'city', 'state', 'tags')) ComplexAddressTuple = namedtuple('ComplexAddress', ('address', 'address_tags', 'city', 'state', 'props')) ComplexAddressWithOwnersTuple = namedtuple('ComplexAddressWithOwners', ('address', 'address_tags', 'city', 'state', 'props', 'owners')) self.__test_udt(schema, graphson, AddressTuple, AddressWithTagsTuple, ComplexAddressTuple, ComplexAddressWithOwnersTuple) def _write_and_read_data_types(self, schema, graphson, use_schema=True): g = self.fetch_traversal_source(graphson) ep = self.get_execution_profile(graphson) for data in six.itervalues(schema.fixtures.datatypes()): typ, value, deserializer = data vertex_label = VertexLabel([typ]) property_name = next(six.iterkeys(vertex_label.non_pk_properties)) if use_schema or schema is CoreGraphSchema: schema.create_vertex_label(self.session, vertex_label, execution_profile=ep) write_traversal = g.addV(str(vertex_label.label)).property('pkid', vertex_label.id).\ property(property_name, value) self.execute_traversal(write_traversal, graphson) read_traversal = g.V().hasLabel(str(vertex_label.label)).has(property_name).properties() results = self.execute_traversal(read_traversal, graphson) for result in results: if result.label == 'pkid': continue self._check_equality(g, value, result.value) def __test_udt(self, schema, graphson, address_class, address_with_tags_class, complex_address_class, complex_address_with_owners_class): if schema is not CoreGraphSchema or DSE_VERSION < Version('6.8'): raise unittest.SkipTest("Graph UDT is only supported with DSE 6.8+ and Core graphs.") ep = self.get_execution_profile(graphson) Address = address_class AddressWithTags = address_with_tags_class ComplexAddress = complex_address_class ComplexAddressWithOwners = complex_address_with_owners_class # setup udt self.session.execute_graph(""" schema.type('address').property('address', Text).property('city', Text).property('state', Text).create(); schema.type('addressTags').property('address', Text).property('city', Text).property('state', Text). property('tags', setOf(Text)).create(); schema.type('complexAddress').property('address', Text).property('address_tags', frozen(typeOf('addressTags'))). property('city', Text).property('state', Text).property('props', mapOf(Text, Int)).create(); schema.type('complexAddressWithOwners').property('address', Text). property('address_tags', frozen(typeOf('addressTags'))). property('city', Text).property('state', Text).property('props', mapOf(Text, Int)). property('owners', frozen(listOf(tupleOf(Text, Int)))).create(); """, execution_profile=ep) # wait max 10 seconds to get the UDT discovered. wait_until_not_raised( lambda: self.session.cluster.register_user_type(self.graph_name, 'address', Address), 1, 10) wait_until_not_raised( lambda: self.session.cluster.register_user_type(self.graph_name, 'addressTags', AddressWithTags), 1, 10) wait_until_not_raised( lambda: self.session.cluster.register_user_type(self.graph_name, 'complexAddress', ComplexAddress), 1, 10) wait_until_not_raised( lambda: self.session.cluster.register_user_type(self.graph_name, 'complexAddressWithOwners', ComplexAddressWithOwners), 1, 10) data = { "udt1": ["typeOf('address')", Address('1440 Rd Smith', 'Quebec', 'QC')], "udt2": ["tupleOf(typeOf('address'), Text)", (Address('1440 Rd Smith', 'Quebec', 'QC'), 'hello')], "udt3": ["tupleOf(frozen(typeOf('address')), Text)", (Address('1440 Rd Smith', 'Quebec', 'QC'), 'hello')], "udt4": ["tupleOf(tupleOf(Int, typeOf('address')), Text)", ((42, Address('1440 Rd Smith', 'Quebec', 'QC')), 'hello')], "udt5": ["tupleOf(tupleOf(Int, typeOf('addressTags')), Text)", ((42, AddressWithTags('1440 Rd Smith', 'Quebec', 'QC', {'t1', 't2'})), 'hello')], "udt6": ["tupleOf(tupleOf(Int, typeOf('complexAddress')), Text)", ((42, ComplexAddress('1440 Rd Smith', AddressWithTags('1440 Rd Smith', 'Quebec', 'QC', {'t1', 't2'}), 'Quebec', 'QC', {'p1': 42, 'p2': 33})), 'hello')], "udt7": ["tupleOf(tupleOf(Int, frozen(typeOf('complexAddressWithOwners'))), Text)", ((42, ComplexAddressWithOwners( '1440 Rd Smith', AddressWithTags('1440 CRd Smith', 'Quebec', 'QC', {'t1', 't2'}), 'Quebec', 'QC', {'p1': 42, 'p2': 33}, [('Mike', 43), ('Gina', 39)]) ), 'hello')] } g = self.fetch_traversal_source(graphson) for typ, value in six.itervalues(data): vertex_label = VertexLabel([typ]) property_name = next(six.iterkeys(vertex_label.non_pk_properties)) schema.create_vertex_label(self.session, vertex_label, execution_profile=ep) write_traversal = g.addV(str(vertex_label.label)).property('pkid', vertex_label.id). \ property(property_name, value) self.execute_traversal(write_traversal, graphson) #vertex = list(schema.add_vertex(self.session, vertex_label, property_name, value, execution_profile=ep))[0] #vertex_properties = list(schema.get_vertex_properties( # self.session, vertex, execution_profile=ep)) read_traversal = g.V().hasLabel(str(vertex_label.label)).has(property_name).properties() vertex_properties = self.execute_traversal(read_traversal, graphson) self.assertEqual(len(vertex_properties), 2) # include pkid for vp in vertex_properties: if vp.label == 'pkid': continue self.assertIsInstance(vp, (VertexProperty, TravVertexProperty)) self.assertEqual(vp.label, property_name) self.assertEqual(vp.value, value) @staticmethod def fetch_edge_props(g, edge): edge_props = g.E(edge.id).properties().toList() return edge_props @staticmethod def fetch_vertex_props(g, vertex): vertex_props = g.V(vertex.id).properties().toList() return vertex_props def _check_equality(self, g, original, read_value): return check_equality_base(self, original, read_value) def _validate_prop(key, value, unittest): if key == 'index': return if any(key.startswith(t) for t in ('int', 'short')): typ = int elif any(key.startswith(t) for t in ('long',)): if sys.version_info >= (3, 0): typ = int else: typ = long elif any(key.startswith(t) for t in ('float', 'double')): typ = float elif any(key.startswith(t) for t in ('polygon',)): typ = Polygon elif any(key.startswith(t) for t in ('point',)): typ = Point elif any(key.startswith(t) for t in ('Linestring',)): typ = LineString elif any(key.startswith(t) for t in ('neg',)): typ = six.string_types elif any(key.startswith(t) for t in ('date',)): typ = datetime.date elif any(key.startswith(t) for t in ('time',)): typ = datetime.time else: unittest.fail("Received unexpected type: %s" % key) @requiredse class BaseImplicitExecutionTest(GraphUnitTestCase): """ This test class will execute all tests of the AbstractTraversalTestClass using implicit execution This all traversal will be run directly using toList() """ def setUp(self): super(BaseImplicitExecutionTest, self).setUp() if DSE_VERSION: self.ep = DseGraph().create_execution_profile(self.graph_name) self.cluster.add_execution_profile(self.graph_name, self.ep) @staticmethod def fetch_key_from_prop(property): return property.key def fetch_traversal_source(self, graphson, **kwargs): ep = self.get_execution_profile(graphson, traversal=True) return DseGraph().traversal_source(self.session, self.graph_name, execution_profile=ep, **kwargs) def execute_traversal(self, traversal, graphson=None): return traversal.toList() def _validate_classic_vertex(self, g, vertex): # Checks the properties on a classic vertex for correctness vertex_props = self.fetch_vertex_props(g, vertex) vertex_prop_keys = [vp.key for vp in vertex_props] self.assertEqual(len(vertex_prop_keys), 2) self.assertIn('name', vertex_prop_keys) self.assertTrue('lang' in vertex_prop_keys or 'age' in vertex_prop_keys) def _validate_generic_vertex_result_type(self, g, vertex): # Checks a vertex object for it's generic properties properties = self.fetch_vertex_props(g, vertex) for attr in ('id', 'label'): self.assertIsNotNone(getattr(vertex, attr)) self.assertTrue(len(properties) > 2) def _validate_classic_edge_properties(self, g, edge): # Checks the properties on a classic edge for correctness edge_props = self.fetch_edge_props(g, edge) edge_prop_keys = [ep.key for ep in edge_props] self.assertEqual(len(edge_prop_keys), 1) self.assertIn('weight', edge_prop_keys) def _validate_classic_edge(self, g, edge): self._validate_generic_edge_result_type(edge) self._validate_classic_edge_properties(g, edge) def _validate_line_edge(self, g, edge): self._validate_generic_edge_result_type(edge) edge_props = self.fetch_edge_props(g, edge) edge_prop_keys = [ep.key for ep in edge_props] self.assertEqual(len(edge_prop_keys), 1) self.assertIn('distance', edge_prop_keys) def _validate_generic_edge_result_type(self, edge): self.assertIsInstance(edge, TravEdge) for attr in ('outV', 'inV', 'label', 'id'): self.assertIsNotNone(getattr(edge, attr)) def _validate_path_result_type(self, g, objects_path): for obj in objects_path: if isinstance(obj, TravEdge): self._validate_classic_edge(g, obj) elif isinstance(obj, TravVertex): self._validate_classic_vertex(g, obj) else: self.fail("Invalid object found in path " + str(obj.type)) def _validate_meta_property(self, g, vertex): meta_props = g.V(vertex.id).properties().toList() self.assertEqual(len(meta_props), 1) meta_prop = meta_props[0] self.assertEqual(meta_prop.value, "meta_prop") self.assertEqual(meta_prop.key, "key") nested_props = g.V(vertex.id).properties().properties().toList() self.assertEqual(len(nested_props), 2) for nested_prop in nested_props: self.assertTrue(nested_prop.key in ['k0', 'k1']) self.assertTrue(nested_prop.value in ['v0', 'v1']) def _validate_type(self, g, vertex): props = self.fetch_vertex_props(g, vertex) for prop in props: value = prop.value key = prop.key _validate_prop(key, value, self) class BaseExplicitExecutionTest(GraphUnitTestCase): def fetch_traversal_source(self, graphson, **kwargs): ep = self.get_execution_profile(graphson, traversal=True) return DseGraph().traversal_source(self.session, self.graph_name, execution_profile=ep, **kwargs) def execute_traversal(self, traversal, graphson): ep = self.get_execution_profile(graphson, traversal=True) ep = self.session.get_execution_profile(ep) context = None if graphson == GraphProtocol.GRAPHSON_3_0: context = { 'cluster': self.cluster, 'graph_name': ep.graph_options.graph_name.decode('utf-8') if ep.graph_options.graph_name else None } query = DseGraph.query_from_traversal(traversal, graphson, context=context) # Use an ep that is configured with the correct row factory, and bytecode-json language flat set result_set = self.execute_graph(query, graphson, traversal=True) return list(result_set)