Change test & more

3 files changed, 97 insertions, 51 deletions

diff --git a/redis/commands/graph/__init__.py b/redis/commands/graph/__init__.py
index 846105e..b27d699 100644
--- a/redis/commands/graph/__init__.py
+++ b/redis/commands/graph/__init__.py
@@ -3,6 +3,7 @@ from .commands import AsyncGraphCommands, GraphCommands
 from .edge import Edge  # noqa
 from .node import Node  # noqa
 from .path import Path  # noqa
+import redis
 
 DB_LABELS = "DB.LABELS"
 DB_RAELATIONSHIPTYPES = "DB.RELATIONSHIPTYPES"
@@ -18,6 +19,9 @@ class Graph(GraphCommands):
         """
         Create a new graph.
         """
+        # Set the module commands' callbacks
+        self.MODULE_CALLBACKS = {}
+
         self.NAME = name  # Graph key
         self.client = client
         self.execute_command = client.execute_command
@@ -161,6 +165,74 @@ class Graph(GraphCommands):
     def property_keys(self):
         return self.call_procedure(DB_PROPERTYKEYS, read_only=True).result_set
 
+    def pipeline(self, transaction=True, shard_hint=None):
+        """Creates a pipeline for the GRAPH module, that can be used for executing
+        GRAPH commands, as well as classic core commands.
+
+        Usage example: # TODO: add graph usage
+            r = redis.Redis()
+            pipe = r.graph().pipeline()
+            query = \"\"\"CREATE (p:person{name:'a',age:32, array:[0,1,2]})\"\"\"
+            pipe.query(query)
+
+            query = \"\"\"WITH [0,1,2] as x return x\"\"\"
+            pipe.query(query)
+            results = pipe.execute() # Execute the pipeline and return the results
+
+        """
+        if isinstance(self.client, redis.RedisCluster):
+            p = ClusterPipeline(
+                nodes_manager=self.client.nodes_manager,
+                commands_parser=self.client.commands_parser,
+                startup_nodes=self.client.nodes_manager.startup_nodes,
+                result_callbacks=self.client.result_callbacks,
+                cluster_response_callbacks=self.client.cluster_response_callbacks,
+                cluster_error_retry_attempts=self.client.cluster_error_retry_attempts,
+                read_from_replicas=self.client.read_from_replicas,
+                reinitialize_steps=self.client.reinitialize_steps,
+                lock=self.client._lock,
+            )
+
+        else:
+            p = Pipeline(
+                connection_pool=self.client.connection_pool,
+                response_callbacks=self.MODULE_CALLBACKS,
+                transaction=transaction,
+                shard_hint=shard_hint,
+            )
+
+        p._clear_schema = self._clear_schema
+        p._refresh_schema = self._refresh_schema
+        p._refresh_labels = self._refresh_labels
+        p._refresh_relations = self._refresh_relations
+        p._refresh_attributes = self._refresh_attributes
+        p.get_label = self.get_label
+        p.get_relation = self.get_relation
+        p.get_property = self.get_property
+        p.add_node = self.add_node
+        p.add_edge = self.add_edge
+        p._build_params_header = self._build_params_header
+        p.call_procedure = self.call_procedure
+        p.labels = self.labels
+        p.relationship_types = self.relationship_types
+        p.property_keys = self.property_keys
+
+
+        p.nodes = self.nodes
+        p.edges = self.edges
+        p.name = self.name
+
+
+        return p
+
+
+class ClusterPipeline(GraphCommands, redis.cluster.ClusterPipeline):
+    """Cluster pipeline for the module."""
+
+
+class Pipeline(GraphCommands, redis.client.Pipeline):
+    """Pipeline for the module."""
+
 
 class AsyncGraph(Graph, AsyncGraphCommands):
     """Async version for Graph"""
@@ -253,51 +325,3 @@ class AsyncGraph(Graph, AsyncGraphCommands):
         return (
             await self.call_procedure(DB_RAELATIONSHIPTYPES, read_only=True)
         ).result_set
-
-    def pipeline(self, transaction=True, shard_hint=None):
-        """Creates a pipeline for the GRAPH module, that can be used for executing
-        GRAPH commands, as well as classic core commands.
-
-        Usage example: # TODO: add graph usage
-            r = redis.Redis()
-            pipe = r.graph().pipeline()
-            query = \"\"\"CREATE (p:person{name:'a',age:32, array:[0,1,2]})\"\"\"
-            pipe.query(query)
-
-            query = \"\"\"WITH [0,1,2] as x return x\"\"\"
-            pipe.query(query)
-            results = pipe.execute() # Execute the pipeline and return the results
-
-        """
-        if isinstance(self.client, redcumis.RedisCluster):
-            p = ClusterPipeline(
-                nodes_manager=self.client.nodes_manager,
-                commands_parser=self.client.commands_parser,
-                startup_nodes=self.client.nodes_manager.startup_nodes,
-                result_callbacks=self.client.result_callbacks,
-                cluster_response_callbacks=self.client.cluster_response_callbacks,
-                cluster_error_retry_attempts=self.client.cluster_error_retry_attempts,
-                read_from_replicas=self.client.read_from_replicas,
-                reinitialize_steps=self.client.reinitialize_steps,
-                lock=self.client._lock,
-            )
-
-        else:
-            p = Pipeline(
-                connection_pool=self.client.connection_pool,
-                response_callbacks=self.MODULE_CALLBACKS,
-                transaction=transaction,
-                shard_hint=shard_hint,
-            )
-
-        p._encode = self._encode
-        p._decode = self._decode
-        return p
-
-
-class ClusterPipeline(GRAPHCommands, redis.cluster.ClusterPipeline):
-    """Cluster pipeline for the module."""
-
-
-class Pipeline(GRAPHCommands, redis.client.Pipeline):
-    """Pipeline for the module."""
diff --git a/redis/commands/json/__init__.py b/redis/commands/json/__init__.py
index 7d55023..c45f1f7 100644
--- a/redis/commands/json/__init__.py
+++ b/redis/commands/json/__init__.py
@@ -95,9 +95,9 @@ class JSON(JSONCommands):
 
         r = redis.Redis()
         pipe = r.json().pipeline()
-        pipe.jsonset('foo', '.', {'hello!': 'world'})
-        pipe.jsonget('foo')
-        pipe.jsonget('notakey')
+        pipe.set('foo', '.', {'hello!': 'world'})
+        pipe.get('foo')
+        pipe.get('notakey')
         """
         if isinstance(self.client, redis.RedisCluster):
             p = ClusterPipeline(
diff --git a/tests/test_graph.py b/tests/test_graph.py
index d71df48..5256060 100644
--- a/tests/test_graph.py
+++ b/tests/test_graph.py
@@ -123,6 +123,28 @@ def test_path(client):
 
 
 @pytest.mark.redismod
+def test_path_in_pipeline(client):
+    node0 = Node(node_id=0, label="L1")
+    node1 = Node(node_id=1, label="L1")
+    edge01 = Edge(node0, "R1", node1, edge_id=0, properties={"value": 1})
+
+    pipe = client.graph().pipeline()
+    pipe.add_node(node0)
+    pipe.add_node(node1)
+    pipe.add_edge(edge01)
+    pipe.flush()
+
+    path01 = Path.new_empty_path().add_node(node0).add_edge(edge01).add_node(node1)
+    expected_results = [[path01]]
+
+    query = "MATCH p=(:L1)-[:R1]->(:L1) RETURN p ORDER BY p"
+    result = pipe.query(query)
+    print("pipe.execute(): ")
+    print(pipe.execute())
+    assert expected_results == result.result_set
+
+
+@pytest.mark.redismod
 def test_param(client):
     params = [1, 2.3, "str", True, False, None, [0, 1, 2]]
     query = "RETURN $param"