fix: Remove slash from api_url

.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.

.gitignore

@@ -127,3 +127,6 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# IDE
+.idea/

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,45 @@
 """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 patch, MagicMock
 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():
+    """Helper method to 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()):
+    """Helper method to 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 +86,60 @@ 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 = setup_graphql_response()
 
-            mock_graphql.return_value = {
-                "pipelineByCode": {"currentVersion": {"zipfile": base64.b64encode(zip_buffer.getvalue()).decode()}}
-            }
             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")