Add unittest toggle for remote and local tests

.github/workflows/acceptance_tests.yaml

@@ -111,10 +111,15 @@ jobs:
                 sleep 5
             fi
           done
-      - name: Run Acceptance Tests
+      - name: Run Acceptance Tests (Remote)
         run: |
           cd python-sdk
           export INDEXIFY_URL=http://localhost:8900
           poetry run python tests/test_graph_behaviours.py
           poetry run python tests/test_graph_update.py
           poetry run python tests/test_graph_validation.py
+      - name: Run Acceptance Tests (Local)
+        run: |
+          cd python-sdk
+          export TEST_LOCAL_RUN=True
+          poetry run python tests/test_graph_behaviours.py

python-sdk/tests/test_graph_behaviours.py

@@ -1,3 +1,4 @@
+import os
 import unittest
 from pathlib import Path
 from typing import List, Union
@@ -60,7 +61,10 @@ def simple_function_ctx_b(x: ComplexObject) -> int:
     return val + 1
 
 
-class SimpleFunctionCtxC(IndexifyFunction):
+class SimpleFunctionCtxC(
+
+
+):
     name = "SimpleFunctionCtxC"
 
     def __init__(self):
@@ -147,6 +151,9 @@ def create_pipeline_graph_with_map():
     return graph
 
 
+
+
+
 def create_pipeline_graph_with_map_reduce():
     graph = Graph(name="test_map_reduce", description="test", start_node=generate_seq)
     graph.add_edge(generate_seq, square)
@@ -183,38 +190,70 @@ def create_simple_pipeline():
     return p
 
 
+def remote_or_local_graph(g: Graph, is_local_run) -> Graph:
+    if is_local_run:
+        return g
+    else:
+        return RemoteGraph.deploy(g)
+
+
+def remote_or_local_pipeline(p: Pipeline, is_local_run) -> Graph:
+    if is_local_run:
+        return p
+    else:
+        return RemotePipeline.deploy(p)
+
+
 class TestGraphBehaviors(unittest.TestCase):
+    is_local_run = None
+
+    def setUp(self):
+        print("setup {}", os.getenv('TEST_LOCAL_RUN'))
+        self.is_local_run = True if os.getenv('TEST_LOCAL_RUN') == "True" else False
+
     def test_simple_function(self):
         graph = Graph(
             name="test_simple_function", description="test", start_node=simple_function
         )
-        graph = RemoteGraph.deploy(graph)
+
+        graph = remote_or_local_graph(graph, self.is_local_run)
         invocation_id = graph.run(block_until_done=True, x=MyObject(x="a"))
         output = graph.output(invocation_id, "simple_function")
         self.assertEqual(output, [MyObject(x="ab")])
 
     def test_simple_function_with_json_encoding(self):
         graph = Graph(
-            name="test_simple_function_with_json_encoding", description="test", start_node=simple_function_with_json_encoder
+            name="test_simple_function_with_json_encoding",
+            description="test",
+            start_node=simple_function_with_json_encoder,
         )
-        graph = RemoteGraph.deploy(graph)
+        graph = remote_or_local_graph(graph, self.is_local_run)
         invocation_id = graph.run(block_until_done=True, x=MyObject(x="a"))
-        output = graph.output(invocation_id,
-                              "simple_function_with_json_encoder")
-        self.assertEqual(output, [MyObject(x='ab')])
+        output = graph.output(invocation_id, "simple_function_with_json_encoder")
+        self.assertEqual(output, [MyObject(x="ab")])
 
     def test_simple_function_with_invalid_encoding(self):
         graph = Graph(
-            name="test_simple_function_with_invalid_encoding", description="test", start_node=simple_function_with_invalid_encoder
+            name="test_simple_function_with_invalid_encoding",
+            description="test",
+            start_node=simple_function_with_invalid_encoder,
         )
-        graph = RemoteGraph.deploy(graph)
-        invocation_id = graph.run(block_until_done=True, x=MyObject(x="a"))
-        output = graph.output(invocation_id, "simple_function_with_invalid_encoder")
-        self.assertEqual(output, [])
+        graph = remote_or_local_graph(graph, self.is_local_run)
+
+        if self.is_local_run:
+            with self.assertRaises(ValueError) as ve:
+                invocation_id = graph.run(block_until_done=True, x=MyObject(x="a"))
+                output = graph.output(invocation_id, "simple_function_with_invalid_encoder")
+                self.assertEqual(output, [])
+            self.assertEqual("Unknown serializer type: invalid", str(ve.exception))
+        else:
+            invocation_id = graph.run(block_until_done=True, x=MyObject(x="a"))
+            output = graph.output(invocation_id, "simple_function_with_invalid_encoder")
+            self.assertEqual(output, [])
 
     def test_map_operation(self):
         graph = create_pipeline_graph_with_map()
-        graph = RemoteGraph.deploy(graph)
+        graph = remote_or_local_graph(graph, self.is_local_run)
         invocation_id = graph.run(block_until_done=True, x=3)
         output_seq = graph.output(invocation_id, "generate_seq")
         self.assertEqual(sorted(output_seq), [0, 1, 2])
@@ -223,7 +262,7 @@ def test_map_operation(self):
 
     def test_map_reduce_operation(self):
         graph = create_pipeline_graph_with_map_reduce()
-        graph = RemoteGraph.deploy(graph)
+        graph = remote_or_local_graph(graph, self.is_local_run)
         invocation_id = graph.run(block_until_done=True, x=3)
         output_sum_sq = graph.output(invocation_id, "sum_of_squares")
         self.assertEqual(output_sum_sq, [Sum(val=5)])
@@ -232,7 +271,7 @@ def test_map_reduce_operation(self):
 
     def test_map_reduce_operation_with_json_encoding(self):
         graph = create_pipeline_graph_with_map_reduce_with_json_encoder()
-        graph = RemoteGraph.deploy(graph)
+        graph = remote_or_local_graph(graph, self.is_local_run)
         invocation_id = graph.run(block_until_done=True, x=3)
         output_square_sq_with_json_encoding = graph.output(invocation_id, "square_with_json_encoder")
         self.assertEqual(output_square_sq_with_json_encoding, [9])
@@ -241,7 +280,7 @@ def test_map_reduce_operation_with_json_encoding(self):
 
     def test_router_graph_behavior(self):
         graph = create_router_graph()
-        graph = RemoteGraph.deploy(graph)
+        graph = remote_or_local_graph(graph, self.is_local_run)
         invocation_id = graph.run(block_until_done=True, x=3)
 
         output_add_two = graph.output(invocation_id, "add_two")
@@ -260,7 +299,7 @@ def test_invoke_file(self):
         graph = Graph(
             name="test_handle_file", description="test", start_node=handle_file
         )
-        graph = RemoteGraph.deploy(graph)
+        graph = remote_or_local_graph(graph, self.is_local_run)
         import os
 
         data = Path(os.path.dirname(__file__) + "/test_file").read_text()
@@ -276,7 +315,7 @@ def test_invoke_file(self):
 
     def test_pipeline(self):
         p = create_simple_pipeline()
-        p = RemotePipeline.deploy(p)
+        p = remote_or_local_pipeline(p, self.is_local_run)
         p.run(x=3)
         invocation_id = p.run(block_until_done=True, x=3)
         output = p.output(invocation_id, "make_it_string")
@@ -300,7 +339,7 @@ def add_two(x: int) -> int:
         graph = Graph(name="test_ignore_none", description="test", start_node=gen_seq)
         graph.add_edge(gen_seq, ignore_none)
         graph.add_edge(ignore_none, add_two)
-        graph = RemoteGraph.deploy(graph)
+        graph = remote_or_local_graph(graph, self.is_local_run)
         invocation_id = graph.run(block_until_done=True, x=5)
         output = graph.output(invocation_id, "add_two")
         self.assertEqual(sorted(output), [2, 4, 6])
@@ -310,15 +349,16 @@ def test_graph_context(self):
             name="test_context", description="test", start_node=simple_function_ctx
         )
         graph.add_edge(simple_function_ctx, simple_function_ctx_b)
-        graph = RemoteGraph.deploy(graph)
+        graph = remote_or_local_graph(graph, self.is_local_run)
         invocation_id = graph.run(block_until_done=True, x=MyObject(x="a"))
         output2 = graph.output(invocation_id, "simple_function_ctx_b")
         self.assertEqual(output2[0], 11)
+
         graph1 = Graph(
             name="test_context", description="test", start_node=simple_function_ctx
         )
         graph1.add_edge(simple_function_ctx, SimpleFunctionCtxC)
-        graph1 = RemoteGraph.deploy(graph1)
+        graph1 = remote_or_local_graph(graph1, self.is_local_run)
         invocation_id = graph1.run(block_until_done=True, x=MyObject(x="a"))
         output2 = graph1.output(invocation_id, "SimpleFunctionCtxC")
         self.assertEqual(output2[0], 11)