test(live): add sandboxed live test suite #58

serhii-workato · 2026-01-28T16:52:39Z

@hovu96 What do you think - should we parse WORKATO_HOST more explicitly to classify the environment? Right now I decide to add as an example checking substrings (trial, preview, sandbox). We also have URLs like https://app-.devenv.awstf.workato.com which won’t match. Should we allow a regex/whitelist for those, or require WORKATO_LIVE_SANDBOX=1 for any non‑standard host/define type of test running?

Context: How should we define/standardize environment identification for live test runs?

-Original file line number
+Diff line change
@@ Expand Up / @@ -47,6 +47,9 @@ test-unit: @@
     test-integration:
     	uv run pytest tests/integration/ -v
+    test-live:
+    	uv run pytest -m live -vv -s
     test-client:
     	uv run pytest src/workato_platform_cli/client/workato_api/test/ -v
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up @@
     make test      # Run tests and check code style
     ```
+    ### Live Tests (Sandbox Only)
+    Live tests hit real Workato APIs and require a sandbox or trial workspace.
+    ```bash
+    WORKATO_LIVE_SANDBOX=1 make test-live
+    ```
+    Required environment variables:
+    - `WORKATO_HOST`
+    - `WORKATO_API_TOKEN`
+    Optional environment variables and flags are documented in `tests/live/README.md`.
     ### Testing with Test PyPI
     When testing pre-release versions from Test PyPI, you need to use both Test PyPI and regular PyPI to resolve dependencies:
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -197,7 +197,19 @@ ignore_missing_imports = true @@
     # Pytest configuration
     [tool.pytest.ini_options]
     minversion = "7.0"
-    addopts = "-ra -q"
+    # Default pytest flags:
+    # -ra: show extra summary for skipped/failed/etc
+    # -q: quiet
+    # -m "not live": never run real sandbox tests unless explicitly requested
+    addopts = '-ra -q -m "not live"'
+    # Markers used across the suite
+    markers = [
+        "live: hits real sandbox endpoints (skipped by default; requires env vars like WORKATO_HOST/WORKATO_API_TOKEN)",
+        "stubbed: uses mocked/stubbed HTTP (default tier for integration workflow tests)",
+    ]
     testpaths = ["tests"]
     python_files = ["test_*.py", "*_test.py"]
     python_classes = ["Test*"]
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -50,11 +50,20 @@ def mock_workato_client() -> Mock: @@
     @pytest.fixture(autouse=True)
-    def isolate_tests(monkeypatch: pytest.MonkeyPatch, temp_config_dir: Path) -> None:
+    def isolate_tests(
+        request: pytest.FixtureRequest,
+        monkeypatch: pytest.MonkeyPatch,
+        temp_config_dir: Path,
+    ) -> None:
         """Isolate tests by using temporary directories and env vars."""
         # Prevent tests from accessing real config files
         monkeypatch.setenv("WORKATO_CONFIG_DIR", str(temp_config_dir))
         monkeypatch.setenv("WORKATO_DISABLE_UPDATE_CHECK", "1")
+        monkeypatch.setattr("pathlib.Path.home", lambda: temp_config_dir)
+        # If this is a live test, do NOT force test mode and do NOT wipe creds
+        if request.node.get_closest_marker("live"):
+            return
         # Ensure we don't make real API calls
         monkeypatch.setenv("WORKATO_TEST_MODE", "1")
@@ Expand Down @@

-Original file line number
+Diff line change
@@ -0,0 +1,26 @@
+    # Live Tests (Sandbox Only)
+    These tests hit real Workato APIs. Run only against sandbox/trial environments.
+    ## Required Environment Variables
+    - `WORKATO_HOST`
+    - `WORKATO_API_TOKEN`
+    - `WORKATO_LIVE_SANDBOX=1` (explicit confirmation)
+    ## Optional Environment Variables
+    - `WORKATO_TEST_PROJECT_ID` or `WORKATO_TEST_PROJECT_NAME`
+    - `WORKATO_TEST_PROFILE` (default: `live-test`)
+    - `WORKATO_LIVE_ALLOW_PUSH=1` (enable `workato push`)
+    - `WORKATO_TEST_RECIPE_ID` (enable recipe start/stop)
+    - `WORKATO_LIVE_ALLOW_RECIPE_CONTROL=1` (enable recipe start/stop)
+    - `WORKATO_TEST_CONNECTION_ID` (enable OAuth URL and pick-list tests)
+    - `WORKATO_TEST_PICKLIST_NAME` (enable pick-list test)
+    ## Run
+    ```bash
+    WORKATO_LIVE_SANDBOX=1 make test-live
+    ```
+    ## Safety Notes
+    - Destructive actions are opt-in.
+    - Project creation test skips if the token lacks permissions.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

test(live): add sandboxed live test suite #58

Uh oh!

Diff view

Diff view

There are no files selected for viewing

serhii-workato Jan 28, 2026 •

edited

Loading

Uh oh!

Uh oh!

-Original file line number
+Diff line change
@@ -0,0 +1,29 @@
+    import contextlib
+    from collections.abc import AsyncGenerator
+    from typing import Any
+    import pytest
+    import pytest_asyncio
+    from workato_platform_cli import Workato
+    @pytest_asyncio.fixture(autouse=True)
+    async def close_workato_clients(
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> AsyncGenerator[None, None]:
+        clients: list[Workato] = []
+        original_init = Workato.__init__
+        def tracking_init(self: Workato, *args: Any, **kwargs: Any) -> None:
+            original_init(self, *args, **kwargs)
+            clients.append(self)
+        monkeypatch.setattr(Workato, "__init__", tracking_init)
+        yield
+        for client in clients:
+            with contextlib.suppress(Exception):
+                await client.close()

-Original file line number
+Diff line change
@@ -0,0 +1,170 @@
+    import json
+    import os
+    from pathlib import Path
+    from typing import cast
+    import certifi
+    import pytest
+    from asyncclick.testing import CliRunner
+    from workato_platform_cli import Workato
+    from workato_platform_cli.cli import cli
+    from workato_platform_cli.cli.commands.projects.project_manager import ProjectManager
+    from workato_platform_cli.client.workato_api.configuration import Configuration
+    REQUIRED_BASE = ["WORKATO_HOST", "WORKATO_API_TOKEN"]
+    _PROJECT_ID_CACHE: str | None = None
+    def ensure_sandbox() -> None:
+        host = os.getenv("WORKATO_HOST", "").lower()
+        if os.getenv("WORKATO_LIVE_SANDBOX", "").lower() in {"1", "true", "yes"}:
+            return
+        if any(token in host for token in ("trial", "preview", "sandbox")):
+            return
+        pytest.skip(
+            "Set WORKATO_LIVE_SANDBOX=1 to confirm tests are running against a sandbox."
+        )
+    def require_live_env(extra: list[str] | None = None) -> None:
+        ensure_sandbox()
+        required = REQUIRED_BASE + (extra or [])
+        missing = [key for key in required if not os.getenv(key)]
+        if missing:
+            pytest.skip(f"Missing env vars for live tests: {', '.join(missing)}")
+    def allow_live_action(flag: str) -> bool:
+        return os.getenv(flag, "").lower() in {"1", "true", "yes"}
+    def force_keyring_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
+        import workato_platform_cli.cli.utils.config.profiles as profiles
+        def _raise_no_keyring() -> None:
+            raise Exception("No keyring backend")
+        monkeypatch.setattr(profiles.keyring, "get_keyring", _raise_no_keyring)
+    def prepare_live_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+        monkeypatch.setenv("WORKATO_DISABLE_UPDATE_CHECK", "1")
+        monkeypatch.setattr(Path, "home", lambda: tmp_path)
+        force_keyring_fallback(monkeypatch)
+    def _build_api_client() -> Workato:
+        api_host = os.environ["WORKATO_HOST"]
+        api_token = os.environ["WORKATO_API_TOKEN"]
+        api_config = Configuration(
+            access_token=api_token,
+            host=api_host,
+            ssl_ca_cert=certifi.where(),
+        )
+        return Workato(configuration=api_config)
+    async def resolve_project_id() -> str:
+        global _PROJECT_ID_CACHE
+        if _PROJECT_ID_CACHE:
+            return _PROJECT_ID_CACHE
+        project_id = os.getenv("WORKATO_TEST_PROJECT_ID")
+        if project_id:
+            _PROJECT_ID_CACHE = project_id
+            return project_id
+        project_name = os.getenv("WORKATO_TEST_PROJECT_NAME")
+        async with _build_api_client() as workato_api_client:
+            project_manager = ProjectManager(workato_api_client=workato_api_client)
+            projects = await project_manager.get_all_projects()
+        if not projects:
+            pytest.skip("No projects found in workspace.")
+        if project_name:
+            for project in projects:
+                if project.name == project_name:
+                    _PROJECT_ID_CACHE = str(project.id)
+                    return str(project.id)
+            pytest.skip(f"No project found with name '{project_name}'.")
+        _PROJECT_ID_CACHE = str(projects[0].id)
+        return _PROJECT_ID_CACHE
+    async def delete_project(project_id: int) -> None:
+        async with _build_api_client() as workato_api_client:
+            await workato_api_client.projects_api.delete_project(project_id=project_id)
+    async def run_init(cli_runner: CliRunner, project_id: str) -> dict[str, object]:
+        api_host = os.environ["WORKATO_HOST"]
+        profile_name = os.getenv("WORKATO_TEST_PROFILE", "live-test")
+        result = await cli_runner.invoke(
+            cli,
+            [
+                "init",
+                "--non-interactive",
+                "--profile",
+                profile_name,
+                "--project-id",
+                str(project_id),
+                "--output-mode",
+                "json",
+                "--region",
+                "custom",
+                "--api-url",
+                api_host,
+            ],
+        )
+        assert result.exit_code == 0, result.output
+        lines = [line for line in result.output.splitlines() if line.strip()]
+        data = cast(dict[str, object], json.loads(lines[-1]))
+        assert data.get("status") == "success"
+        return data
+    async def run_init_create_project(
+        cli_runner: CliRunner,
+        project_name: str,
+    ) -> dict[str, object]:
+        api_host = os.environ["WORKATO_HOST"]
+        profile_name = os.getenv("WORKATO_TEST_PROFILE", "live-test")
+        result = await cli_runner.invoke(
+            cli,
+            [
+                "init",
+                "--non-interactive",
+                "--profile",
+                profile_name,
+                "--project-name",
+                project_name,
+                "--output-mode",
+                "json",
+                "--region",
+                "custom",
+                "--api-url",
+                api_host,
+            ],
+        )
+        if result.exit_code != 0:
+            if "UNAUTHORIZED" in result.output or "Authentication failed" in result.output:
+                pytest.skip(
+                    "Project creation/pull not authorized for this token. "
+                    "Check sandbox permissions or token scope."
+                )
+            assert result.exit_code == 0, result.output
+        lines = [line for line in result.output.splitlines() if line.strip()]
+        data = cast(dict[str, object], json.loads(lines[-1]))
+        assert data.get("status") == "success"
+        return data

-Original file line number
+Diff line change
@@ -0,0 +1,58 @@
+    import os
+    from pathlib import Path
+    import pytest
+    from asyncclick.testing import CliRunner
+    from tests.live.helpers import prepare_live_env, require_live_env
+    from workato_platform_cli.cli import cli
+    @pytest.mark.live
+    @pytest.mark.asyncio
+    async def test_connections_oauth_flow_live(
+        cli_runner: CliRunner,
+        monkeypatch: pytest.MonkeyPatch,
+        tmp_path: Path,
+    ) -> None:
+        require_live_env(["WORKATO_TEST_CONNECTION_ID"])
+        prepare_live_env(monkeypatch, tmp_path)
+        connection_id = os.environ["WORKATO_TEST_CONNECTION_ID"]
+        list_result = await cli_runner.invoke(cli, ["connections", "list"])
+        assert list_result.exit_code == 0, list_result.output
+        oauth_result = await cli_runner.invoke(
+            cli, ["connections", "get-oauth-url", "--id", str(connection_id)]
+        )
+        assert oauth_result.exit_code == 0, oauth_result.output
+    @pytest.mark.live
+    @pytest.mark.asyncio
+    async def test_connection_picklist_live(
+        cli_runner: CliRunner,
+        monkeypatch: pytest.MonkeyPatch,
+        tmp_path: Path,
+    ) -> None:
+        require_live_env(["WORKATO_TEST_CONNECTION_ID", "WORKATO_TEST_PICKLIST_NAME"])
+        prepare_live_env(monkeypatch, tmp_path)
+        connection_id = os.environ["WORKATO_TEST_CONNECTION_ID"]
+        picklist_name = os.environ["WORKATO_TEST_PICKLIST_NAME"]
+        result = await cli_runner.invoke(
+            cli,
+            [
+                "connections",
+                "pick-list",
+                "--id",
+                str(connection_id),
+                "--pick-list-name",
+                picklist_name,
+            ],
+        )
+        assert result.exit_code == 0, result.output

test(live): add sandboxed live test suite #58

Are you sure you want to change the base?

Uh oh!

test(live): add sandboxed live test suite #58

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

serhii-workato Jan 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

serhii-workato Jan 28, 2026 •

edited

Loading