fix: Remove slash from api_url (#219)

* fix: Remove slash from api_url

* fix: Remove slash from api_url

* fix: Remove slash from api_url

---------

Co-authored-by: YolanFery <yolanfery@MacBook-Pro-de-Yolan.local>

Author: yolanfery

.github/pull_request_template.md

@@ -0,0 +1,16 @@
+A short, meaningful PR description. Explain the goal of the PR and the ideas behind it.
+
+## Changes
+
+Please list / describe the changes in the codebase for the reviewer(s).
+
+- Change 1
+- Change 2
+
+## How/what to test
+
+A few words about what needs to be tested, along with specific testing instructions if needed.
+
+## Screenshots / screencast
+
+If relevant, please add some screenshots or a short video.

openhexa/cli/settings.py

@@ -53,15 +53,13 @@ def api_url(self):
         """Return the API URL from the settings file or environment variables."""
         url_from_env = os.getenv("HEXA_API_URL") or os.getenv("HEXA_SERVER_URL")
         if url_from_env is None:
-            return self._file_config["openhexa"]["url"]
+            url_from_env = self._file_config["openhexa"]["url"]
+        return url_from_env.rstrip("/")
 
     @property
     def public_api_url(self):
         """Return the public API URL from the settings file or environment variables."""
-        url_from_env = os.getenv("HEXA_API_URL") or os.getenv("HEXA_SERVER_URL")
-        if url_from_env is not None:
-            return url_from_env
-        return self._file_config["openhexa"]["url"].replace("api", "app")
+        return self.api_url.replace("api", "app")
 
     @property
     def current_workspace(self):

tests/test_cli.py

@@ -1,17 +1,41 @@
 """CLI test module."""
 
 import base64
+import os
 from io import BytesIO
 from pathlib import Path
 from tempfile import mkdtemp
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 from zipfile import ZipFile
 
 import pytest
 from click.testing import CliRunner
 
-from openhexa.cli.cli import pipelines_download, pipelines_run, workspaces_add
+from openhexa.cli.cli import pipelines_download, pipelines_push, pipelines_run, workspaces_add
+from openhexa.sdk.pipelines import Pipeline
+
+python_file_name = "pipeline.py"
+python_code = "print('pipeline.py file')"
+version = "v1.0"
+pipeline_name = "MyPipeline"
+
+
+def create_zip_with_pipeline():
+    """Create a zip file containing the pipeline.py file."""
+    zip_buffer = BytesIO()
+    fake_zipfile = ZipFile(zip_buffer, "w")
+    fake_zipfile.writestr(python_file_name, python_code)
+    fake_zipfile.close()
+    return zip_buffer
+
+
+def setup_graphql_response(zip_buffer=create_zip_with_pipeline()):
+    """Set up the mock GraphQL response pipelines."""
+    return {
+        "pipelineByCode": {"currentVersion": {"zipfile": base64.b64encode(zip_buffer.getvalue()).decode()}},
+        "pipelines": {"items": []},  # (empty workspace initially)
+    }
 
 
 @pytest.mark.usefixtures("settings")
@@ -58,52 +82,65 @@ def test_download_pipeline_no_pipeline(self, mock_graphql):
     def test_download_pipeline_overwrite(self, mock_graphql):
         """Test the download pipeline command with an existing director with and without content."""
         with self.runner.isolated_filesystem() as tmp:
-            # Create a zipfile with a pipeline.py file in a buffer
-            zip_buffer = BytesIO()
-            fake_zipfile = ZipFile(zip_buffer, "w")
-            fake_zipfile.writestr("pipeline.py", "print('pipeline.py file')")
-            fake_zipfile.close()
-
-            # Known Pipeline & non empty directory
-            with open(tmp + "/pipeline.py", "w") as f:
+            # Known Pipeline & non-empty directory
+            with open(Path(tmp) / python_file_name, "w") as f:
                 f.write("<content>")
-            mock_graphql.return_value = {
-                "pipelineByCode": {"currentVersion": {"zipfile": base64.b64encode(zip_buffer.getvalue()).decode()}}
-            }
+
+            mock_graphql.return_value = setup_graphql_response()
+
             result = self.runner.invoke(pipelines_download, ["test_pipeline", tmp], input="N\n")
             self.assertIn("Overwrite the files?", result.output)
             self.assertIn(f"{tmp} is not empty", result.output)
-            self.assertEqual(open(tmp + "/pipeline.py").read(), "<content>")
+            self.assertEqual(open(Path(tmp) / python_file_name).read(), "<content>")
 
             # Overwrite the files in the directory
-            mock_graphql.return_value = {
-                "pipelineByCode": {"currentVersion": {"zipfile": base64.b64encode(zip_buffer.getvalue()).decode()}}
-            }
+            mock_graphql.return_value = setup_graphql_response()
+
             result = self.runner.invoke(pipelines_download, ["test_pipeline", tmp], input="y\n")
             self.assertIn("Overwrite the files?", result.output)
-            self.assertEqual(open(tmp + "/pipeline.py").read(), "print('pipeline.py file')")
+            self.assertEqual(open(Path(tmp) / python_file_name).read(), python_code)
 
     @patch("openhexa.cli.api.graphql")
     def test_download_pipeline(self, mock_graphql):
         """Test the download pipeline command."""
         with self.runner.isolated_filesystem() as tmp:
             # Create a zipfile with a pipeline.py file in a buffer
-            zip_buffer = BytesIO()
-            fake_zipfile = ZipFile(zip_buffer, "w")
-            fake_zipfile.writestr("pipeline.py", "print('pipeline.py file')")
-            fake_zipfile.close()
-
-            mock_graphql.return_value = {
-                "pipelineByCode": {"currentVersion": {"zipfile": base64.b64encode(zip_buffer.getvalue()).decode()}}
-            }
+            mock_graphql.return_value = setup_graphql_response()
+
             result = self.runner.invoke(pipelines_download, ["test_pipeline", tmp])
             self.assertEqual(result.exit_code, 0)
-            self.assertTrue((Path(tmp) / "pipeline.py").exists())
-            self.assertEqual(open(tmp + "/pipeline.py").read(), "print('pipeline.py file')")
+            path = Path(tmp) / python_file_name
+            self.assertTrue(path.exists())
+            self.assertEqual(open(path).read(), python_code)
+
+    @patch("openhexa.cli.api.graphql")
+    @patch("openhexa.cli.cli.get_pipeline")
+    @patch("openhexa.cli.cli.upload_pipeline")
+    @patch.dict(os.environ, {"HEXA_API_URL": "https://www.bluesquarehub.com/", "HEXA_WORKSPACE": "workspace"})
+    def test_push_pipeline(self, mock_upload_pipeline, mock_get_pipeline, mock_graphql):
+        """Test pushing a pipeline."""
+        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.code = pipeline_name
+            mock_get_pipeline.return_value = mock_pipeline
+
+            result = self.runner.invoke(pipelines_push, [tmp, "--name", version])
+            self.assertEqual(result.exit_code, 0)
+            self.assertIn(
+                (
+                    f"✅ New version '{version}' created! "
+                    f"You can view the pipeline in OpenHEXA on https://www.bluesquarehub.com/workspaces/workspace/pipelines/{pipeline_name}"
+                ),
+                result.output,
+            )
+            self.assertTrue(mock_upload_pipeline.called)
 
     @patch("openhexa.cli.api.graphql")
     def test_workspaces_add_not_found(self, mock_graphql):
-        """Test the add workspace command when the workspae doesn't exist on the current server."""
+        """Test the add workspace command when the workspace doesn't exist on the current server."""
         with self.runner.isolated_filesystem():
             mock_graphql.return_value = {"workspace": None}
             result = self.runner.invoke(workspaces_add, ["test_workspace"], input="random_token \n")