feat: cli users should have a way to give a pipeline code in the command to upload a pipeline (HEXA-1216) (#242)

* Support code

* Test

* Minor

* Last test

* Fix test

Author: yolanfery

openhexa/cli/cli.py

@@ -322,6 +322,14 @@ def _handle_enter_pipeline_code():
         dir_okay=True,
     ),
 )
+@click.option(
+    "--code",
+    "-c",
+    type=str,
+    help="Code of the pipeline",
+    prompt="Code of the pipeline",
+    prompt_required=False,
+)
 @click.option(
     "--name",
     "-n",
@@ -352,6 +360,7 @@ def _handle_enter_pipeline_code():
 @click.option("--yes", is_flag=True, help="Skip confirmation")
 def pipelines_push(
     path: str,
+    code: str = None,
     name: str = None,
     description: str = None,
     link: str = None,
@@ -368,6 +377,8 @@ def pipelines_push(
             "activate a workspace.",
             err=True,
         )
+    if yes and not code:
+        _terminate("❌ You must provide a pipeline code (using -c or --code) when using the --yes flag.", err=True)
 
     ensure_is_pipeline_dir(path)
 
@@ -387,7 +398,11 @@ def pipelines_push(
         if settings.debug:
             click.echo(workspace_pipelines)
 
-        if yes:
+        if code:
+            selected_pipeline = get_pipeline_from_code(code) or _terminate(
+                f"❌ Pipeline with code '{code}' not found.", err=True
+            )
+        elif yes:
             selected_pipeline = workspace_pipelines[0] if workspace_pipelines else None
         else:
             selected_pipeline = select_pipeline(workspace_pipelines, number_of_pages, pipeline)

tests/test_cli.py

@@ -185,6 +185,110 @@ def test_push_pipeline(
                 result.output,
             )
 
+    @patch("openhexa.cli.api.graphql")
+    @patch.dict(os.environ, {"HEXA_API_URL": "https://www.bluesquarehub.com/", "HEXA_WORKSPACE": "workspace"})
+    def test_push_pipeline_with_yes_flag_without_code(self, mock_graphql):
+        """Test pushing a pipeline with the --yes flag without providing a --code."""
+        with self.runner.isolated_filesystem() as tmp:
+            with open(Path(tmp) / python_file_name, "w") as f:
+                f.write(python_code)
+            mock_graphql.return_value = setup_graphql_response()
+
+            result = self.runner.invoke(
+                pipelines_push,
+                [tmp, "--yes"],
+            )
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn(
+                "❌ You must provide a pipeline code (using -c or --code) when using the --yes flag.", result.output
+            )
+
+    @patch("openhexa.cli.api.graphql")
+    @patch("openhexa.cli.cli.get_pipeline")
+    @patch("openhexa.cli.cli.get_pipelines_pages")
+    @patch("openhexa.cli.cli.get_pipeline_from_code")
+    @patch.dict(os.environ, {"HEXA_API_URL": "https://www.bluesquarehub.com/", "HEXA_WORKSPACE": "workspace"})
+    def test_push_pipeline_with_non_existing_code(
+        self, mock_get_pipeline_from_code, mock_get_pipelines_pages, mock_get_pipeline, mock_graphql
+    ):
+        """Test pushing a pipeline with a non-existing --code flag."""
+        with self.runner.isolated_filesystem() as tmp:
+            with open(Path(tmp) / python_file_name, "w") as f:
+                f.write(python_code)
+            mock_graphql.return_value = setup_graphql_response()
+            mock_pipeline = MagicMock(spec=Pipeline)
+            mock_pipeline.name = pipeline_name
+            mock_get_pipeline.return_value = mock_pipeline
+            mock_get_pipelines_pages.return_value = {
+                "items": [
+                    {"name": "Pipeline1", "code": "code1"},
+                    {"name": "Pipeline2", "code": "code2"},
+                ],
+                "totalPages": 2,
+            }
+            mock_get_pipeline_from_code.return_value = None  # Simulate non-existing pipeline code
+
+            result = self.runner.invoke(
+                pipelines_push,
+                [tmp, "--code", "non_existing_code"],
+            )
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn("❌ Pipeline with code 'non_existing_code' not found.", result.output)
+
+    @patch("openhexa.cli.api.graphql")
+    @patch("openhexa.cli.cli.get_pipeline")
+    @patch("openhexa.cli.cli.get_pipelines_pages")
+    @patch("openhexa.cli.cli.upload_pipeline")
+    @patch.dict(os.environ, {"HEXA_API_URL": "https://www.bluesquarehub.com/", "HEXA_WORKSPACE": "workspace"})
+    def test_push_pipeline_with_code_flag(
+        self, mock_upload_pipeline, mock_get_pipelines_pages, mock_get_pipeline, mock_graphql
+    ):
+        """Test pushing a pipeline using the --code flag."""
+        code = "code1"
+        with self.runner.isolated_filesystem() as tmp:
+            with open(Path(tmp) / python_file_name, "w") as f:
+                f.write(python_code)
+            mock_graphql.return_value = {
+                "pipelineByCode": {
+                    "code": code,
+                    "currentVersion": {"zipfile": ""},
+                },
+                "pipelines": {"items": []},
+            }
+            mock_pipeline = MagicMock(spec=Pipeline)
+            mock_pipeline.name = pipeline_name
+            mock_get_pipeline.return_value = mock_pipeline
+            mock_get_pipelines_pages.return_value = {
+                "items": [
+                    {"name": "Pipeline1", "code": "code1"},
+                    {"name": "Pipeline2", "code": "code2"},
+                ],
+                "totalPages": 2,
+            }
+            mock_upload_pipeline.return_value = {
+                "versionName": version,
+                "pipeline": {
+                    "id": pipeline_id,
+                    "permissions": {"createTemplateVersion": True},
+                    "template": template,
+                },
+                "id": pipeline_version_id,
+            }
+
+            result = self.runner.invoke(
+                pipelines_push,
+                [tmp, "--name", version, "--code", code],
+                input="Y\nn\n",
+            )
+            self.assertEqual(result.exit_code, 0)
+            self.assertNotIn("Which pipeline do you want to update?", result.output)
+            self.assertTrue(mock_upload_pipeline.called)
+            self.assertEqual(mock_upload_pipeline.call_args[0][0], code)
+            self.assertIn(
+                f"✅ New version '{version}' created! ",
+                result.output,
+            )
+
     @patch("openhexa.cli.cli.click.prompt")
     def test_select_pipeline(self, mock_prompt):
         workspace_pipelines = [