diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..576d717 --- /dev/null +++ b/.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. \ No newline at end of file diff --git a/.gitignore b/.gitignore index b6e4761..3efd499 100644 --- a/.gitignore +++ b/.gitignore @@ -127,3 +127,6 @@ dmypy.json # Pyre type checker .pyre/ + +# IDE +.idea/ diff --git a/openhexa/cli/settings.py b/openhexa/cli/settings.py index 93bc770..e0d0bfa 100644 --- a/openhexa/cli/settings.py +++ b/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): diff --git a/tests/test_cli.py b/tests/test_cli.py index 0cedb6e..954f87a 100644 --- a/tests/test_cli.py +++ b/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("") - 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(), "") + self.assertEqual(open(Path(tmp) / python_file_name).read(), "") # 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")