Move run_standalone to VCRIntegrationTest

Signed-off-by: Yves Bastide <yves@botify.com>

Author: ybastide

tests/integration/__init__.py

@@ -2,11 +2,20 @@
 import os
 
 import boto.swf
+from click.testing import CliRunner
+from sure import expect
 from vcr import VCR
 
 import simpleflow.command  # NOQA
 from tests.utils import IntegrationTestCase
 
+from simpleflow.utils import json_dumps
+
+if False:
+    from typing import List, Union
+    from click.testing import Result
+
+
 # Default SWF parameters
 os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
 os.environ["SWF_DOMAIN"] = "TestDomain"
@@ -70,3 +79,37 @@ def get_events(self, run_id):
             events.extend(response['events'])
             next_page = response.get('nextPageToken')
         return events
+
+    def invoke(self, command, arguments):
+        # type: (str, Union(str, List[str])) -> Result
+        if not hasattr(self, "runner"):
+            self.runner = CliRunner()
+        if isinstance(arguments, str):
+            arguments = arguments.split(" ")
+        print('simpleflow {} {}'.format(command, ' '.join(arguments)))
+        return self.runner.invoke(command, arguments, catch_exceptions=False)
+
+    def run_standalone(self, workflow_name, *args, **kwargs):
+        input = json_dumps(dict(args=args, kwargs=kwargs))
+        result = self.invoke(
+            simpleflow.command.cli,
+            [
+                "standalone",
+                "--workflow-id",
+                str(self.workflow_id),
+                "--input",
+                input,
+                "--nb-deciders",
+                "2",
+                "--nb-workers",
+                "2",
+                workflow_name,
+            ],
+        )
+        expect(result.exit_code).to.equal(0)
+        lines = result.output.split("\n")
+        start_line = [line for line in lines if line.startswith(self.workflow_id)][0]
+        _, run_id = start_line.split(" ", 1)
+
+        events = self.get_events(run_id)
+        return events

tests/integration/test_markers.py

@@ -1,50 +1,9 @@
-import simpleflow.command
-from click.testing import CliRunner
 from sure import expect
 
-from simpleflow.utils import json_dumps
 from tests.integration import VCRIntegrationTest, vcr
 
-if False:
-    from typing import List, Union
-    from click.testing import Result
-
 
 class TestMarkers(VCRIntegrationTest):
-    def invoke(self, command, arguments):
-        # type: (str, Union(str, List[str])) -> Result
-        if not hasattr(self, "runner"):
-            self.runner = CliRunner()
-        if isinstance(arguments, str):
-            arguments = arguments.split(" ")
-        print('simpleflow {} {}'.format(command, ' '.join(arguments)))
-        return self.runner.invoke(command, arguments, catch_exceptions=False)
-
-    def run_standalone(self, workflow_name, *args, **kwargs):
-        input = json_dumps(dict(args=args, kwargs=kwargs))
-        result = self.invoke(
-            simpleflow.command.cli,
-            [
-                "standalone",
-                "--workflow-id",
-                str(self.workflow_id),
-                "--input",
-                input,
-                "--nb-deciders",
-                "2",
-                "--nb-workers",
-                "2",
-                workflow_name,
-            ],
-        )
-        expect(result.exit_code).to.equal(0)
-        lines = result.output.split("\n")
-        start_line = [line for line in lines if line.startswith(self.workflow_id)][0]
-        _, run_id = start_line.split(" ", 1)
-
-        events = self.get_events(run_id)
-        return events
-
     @vcr.use_cassette
     def test_without_replays(self):
         events = self.run_standalone('tests.integration.workflow.MarkerWorkflow', False)