fix: unify deploy provider environment handling

tests/cli/test_cli_clean.py

@@ -0,0 +1,141 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from typing import Any
+
+from click.testing import CliRunner
+
+from veadk.cli.cli_clean import clean
+
+
+class _FakeVeFaaS:
+    calls: list[dict[str, Any]] = []
+
+    def __init__(self, **kwargs: Any) -> None:
+        self.calls.append({"init": kwargs, "deleted": []})
+
+    def find_app_id_by_name(self, name: str) -> str | None:
+        self.calls[-1].setdefault("lookups", []).append(name)
+        if len(self.calls[-1]["lookups"]) == 1:
+            return "app-123"
+        return None
+
+    def delete(self, app_id: str | None) -> None:
+        self.calls[-1]["deleted"].append(app_id)
+
+
+def test_clean_defaults_to_volcengine_environment(
+    monkeypatch,
+) -> None:
+    monkeypatch.delenv("AGENTKIT_CLOUD_PROVIDER", raising=False)
+    monkeypatch.delenv("CLOUD_PROVIDER", raising=False)
+    monkeypatch.setenv("VOLCENGINE_ACCESS_KEY", "volc-ak")
+    monkeypatch.setenv("VOLCENGINE_SECRET_KEY", "volc-sk")
+    monkeypatch.setenv("VOLCENGINE_SESSION_TOKEN", "volc-token")
+    monkeypatch.setenv("REGION", "cn-shanghai")
+    monkeypatch.setattr("veadk.integrations.ve_faas.ve_faas.VeFaaS", _FakeVeFaaS)
+    _FakeVeFaaS.calls.clear()
+
+    result = CliRunner().invoke(
+        clean,
+        ["--vefaas-app-name", "studio-app"],
+        input="y\n",
+    )
+
+    assert result.exit_code == 0, result.output
+    assert _FakeVeFaaS.calls == [
+        {
+            "init": {
+                "access_key": "volc-ak",
+                "secret_key": "volc-sk",
+                "session_token": "volc-token",
+                "region": "cn-shanghai",
+                "provider": "volcengine",
+            },
+            "deleted": ["app-123"],
+            "lookups": ["studio-app", "studio-app"],
+        }
+    ]
+
+
+def test_clean_uses_byteplus_environment(
+    monkeypatch,
+) -> None:
+    monkeypatch.setenv("CLOUD_PROVIDER", "byteplus")
+    monkeypatch.delenv("AGENTKIT_CLOUD_PROVIDER", raising=False)
+    monkeypatch.setenv("BYTEPLUS_ACCESS_KEY", "byteplus-ak")
+    monkeypatch.setenv("BYTEPLUS_SECRET_KEY", "byteplus-sk")
+    monkeypatch.setenv("BYTEPLUS_SESSION_TOKEN", "byteplus-token")
+    monkeypatch.setenv("BYTEPLUS_REGION", "ap-southeast-1")
+    monkeypatch.setenv("VOLCENGINE_ACCESS_KEY", "volc-ak")
+    monkeypatch.setenv("VOLCENGINE_SECRET_KEY", "volc-sk")
+    monkeypatch.setattr("veadk.integrations.ve_faas.ve_faas.VeFaaS", _FakeVeFaaS)
+    _FakeVeFaaS.calls.clear()
+
+    result = CliRunner().invoke(
+        clean,
+        ["--vefaas-app-name", "studio-app"],
+        input="y\n",
+    )
+
+    assert result.exit_code == 0, result.output
+    assert _FakeVeFaaS.calls[0]["init"] == {
+        "access_key": "byteplus-ak",
+        "secret_key": "byteplus-sk",
+        "session_token": "byteplus-token",
+        "region": "ap-southeast-1",
+        "provider": "byteplus",
+    }
+    assert _FakeVeFaaS.calls[0]["deleted"] == ["app-123"]
+
+
+def test_clean_explicit_byteplus_options_override_environment(
+    monkeypatch,
+) -> None:
+    monkeypatch.setenv("CLOUD_PROVIDER", "volcengine")
+    monkeypatch.setenv("BYTEPLUS_ACCESS_KEY", "env-ak")
+    monkeypatch.setenv("BYTEPLUS_SECRET_KEY", "env-sk")
+    monkeypatch.setenv("BYTEPLUS_SESSION_TOKEN", "env-token")
+    monkeypatch.setattr("veadk.integrations.ve_faas.ve_faas.VeFaaS", _FakeVeFaaS)
+    _FakeVeFaaS.calls.clear()
+
+    result = CliRunner().invoke(
+        clean,
+        [
+            "--provider",
+            "byteplus",
+            "--region",
+            "ap-southeast-1",
+            "--byteplus-access-key",
+            "cli-ak",
+            "--byteplus-secret-key",
+            "cli-sk",
+            "--byteplus-session-token",
+            "cli-token",
+            "--vefaas-app-name",
+            "studio-app",
+        ],
+        input="y\n",
+    )
+
+    assert result.exit_code == 0, result.output
+    assert _FakeVeFaaS.calls[0]["init"] == {
+        "access_key": "cli-ak",
+        "secret_key": "cli-sk",
+        "session_token": "cli-token",
+        "region": "ap-southeast-1",
+        "provider": "byteplus",
+    }

tests/cli/test_cli_deploy.py

@@ -0,0 +1,111 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+from click.testing import CliRunner
+
+from veadk.cli.cli_deploy import deploy
+from veadk.config import veadk_environments
+
+
+def _write_agent_project(path: Path) -> None:
+    path.mkdir()
+    (path / "__init__.py").write_text("from . import agent\n", encoding="utf-8")
+    (path / "agent.py").write_text("root_agent = object()\n", encoding="utf-8")
+
+
+def test_deploy_reads_byteplus_provider_environment(
+    monkeypatch,
+    tmp_path: Path,
+) -> None:
+    project = tmp_path / "agent-proj"
+    _write_agent_project(project)
+    captured: dict[str, Any] = {}
+
+    def fake_cookiecutter(
+        template: str,
+        output_dir: str,
+        no_input: bool,
+        extra_context: dict[str, Any],
+    ) -> None:
+        captured["template"] = template
+        captured["no_input"] = no_input
+        captured["extra_context"] = extra_context
+        generated_agent_dir = (
+            Path(output_dir) / extra_context["local_dir_name"] / "src" / "agent_proj"
+        )
+        generated_agent_dir.mkdir(parents=True)
+
+    async def fake_main() -> None:
+        captured["main_env"] = {
+            "CLOUD_PROVIDER": os.environ.get("CLOUD_PROVIDER"),
+            "AGENTKIT_CLOUD_PROVIDER": os.environ.get("AGENTKIT_CLOUD_PROVIDER"),
+            "BYTEPLUS_REGION": os.environ.get("BYTEPLUS_REGION"),
+            "REGION": os.environ.get("REGION"),
+            "VOLCENGINE_ACCESS_KEY": os.environ.get("VOLCENGINE_ACCESS_KEY"),
+            "VOLCENGINE_SECRET_KEY": os.environ.get("VOLCENGINE_SECRET_KEY"),
+            "VOLCENGINE_SESSION_TOKEN": os.environ.get("VOLCENGINE_SESSION_TOKEN"),
+        }
+
+    monkeypatch.setenv("CLOUD_PROVIDER", "byteplus")
+    monkeypatch.delenv("AGENTKIT_CLOUD_PROVIDER", raising=False)
+    monkeypatch.delenv("REGION", raising=False)
+    monkeypatch.delenv("VOLCENGINE_ACCESS_KEY", raising=False)
+    monkeypatch.delenv("VOLCENGINE_SECRET_KEY", raising=False)
+    monkeypatch.delenv("VOLCENGINE_SESSION_TOKEN", raising=False)
+    monkeypatch.setenv("BYTEPLUS_REGION", "ap-southeast-1")
+    monkeypatch.setenv("BYTEPLUS_ACCESS_KEY", "byteplus-ak")
+    monkeypatch.setenv("BYTEPLUS_SECRET_KEY", "byteplus-sk")
+    monkeypatch.setenv("BYTEPLUS_SESSION_TOKEN", "byteplus-token")
+    for key in (
+        "CLOUD_PROVIDER",
+        "AGENTKIT_CLOUD_PROVIDER",
+        "BYTEPLUS_REGION",
+    ):
+        monkeypatch.delitem(veadk_environments, key, raising=False)
+    monkeypatch.setattr("veadk.utils.misc.formatted_timestamp", lambda: "20260824")
+    monkeypatch.setattr("cookiecutter.main.cookiecutter", fake_cookiecutter)
+    monkeypatch.setattr(
+        "veadk.utils.misc.load_module_from_file",
+        lambda **_kwargs: SimpleNamespace(main=fake_main),
+    )
+
+    result = CliRunner().invoke(
+        deploy,
+        [
+            "--vefaas-app-name",
+            "studio-app",
+            "--path",
+            str(project),
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert captured["extra_context"]["provider"] == "byteplus"
+    assert captured["extra_context"]["region"] == "ap-southeast-1"
+    assert captured["main_env"] == {
+        "CLOUD_PROVIDER": "byteplus",
+        "AGENTKIT_CLOUD_PROVIDER": "byteplus",
+        "BYTEPLUS_REGION": "ap-southeast-1",
+        "REGION": "ap-southeast-1",
+        "VOLCENGINE_ACCESS_KEY": "byteplus-ak",
+        "VOLCENGINE_SECRET_KEY": "byteplus-sk",
+        "VOLCENGINE_SESSION_TOKEN": "byteplus-token",
+    }

tests/cli/test_frontend_sandbox_options.py

@@ -34,14 +34,14 @@ def test_serve_provider_is_bootstrapped_before_command_modules_load(
 
     _bootstrap_serve_provider(["frontend"])
 
-    assert os.environ["AGENTKIT_CLOUD_PROVIDER"] == "volcengine"
-    assert os.environ["CLOUD_PROVIDER"] == "volcengine"
-
-    _bootstrap_serve_provider(["studio", "--provider", "byteplus"])
-
     assert os.environ["AGENTKIT_CLOUD_PROVIDER"] == "byteplus"
     assert os.environ["CLOUD_PROVIDER"] == "byteplus"
 
+    _bootstrap_serve_provider(["studio", "--provider", "volcengine"])
+
+    assert os.environ["AGENTKIT_CLOUD_PROVIDER"] == "volcengine"
+    assert os.environ["CLOUD_PROVIDER"] == "volcengine"
+
 
 @pytest.mark.parametrize("command", [frontend, studio])
 def test_sandbox_tool_options_are_shared_by_local_serve_commands(
@@ -162,7 +162,7 @@ def test_local_studio_mounts_snapshot_tools_into_sandbox_services() -> None:
 
 
 @pytest.mark.parametrize("command", [frontend, studio])
-def test_local_serve_commands_default_to_volcengine(
+def test_local_serve_commands_defer_provider_resolution_when_not_explicit(
     monkeypatch: pytest.MonkeyPatch,
     command: Command,
 ) -> None:
@@ -176,7 +176,7 @@ def test_local_serve_commands_default_to_volcengine(
     result = CliRunner().invoke(command)
 
     assert result.exit_code == 0, result.output
-    assert captured["provider"] == "volcengine"
+    assert captured["provider"] is None
 
 
 @pytest.mark.parametrize("command", [frontend, studio])

tests/cli/test_generated_agent_backend_codegen_extended.py

@@ -88,6 +88,12 @@
 }
 
 
+@pytest.fixture(autouse=True)
+def _default_to_volcengine_provider(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("CLOUD_PROVIDER", raising=False)
+    monkeypatch.delenv("AGENTKIT_CLOUD_PROVIDER", raising=False)
+
+
 def _file_map(project: GeneratedProject) -> dict[str, str]:
     return {file.path: file.content for file in project.files}
 

tests/cli/test_studio_deploy_permissions.py

@@ -14,6 +14,8 @@
 
 from __future__ import annotations
 
+import os
+
 import pytest
 from click.testing import CliRunner
 
@@ -357,6 +359,55 @@ def _precheck(*, specs, **_kwargs):
     assert "Pre-check only: no cloud resources were created." in result.output
 
 
+def test_cli_restores_provider_process_env_after_byteplus_deploy(
+    monkeypatch,
+) -> None:
+    for key in (
+        "CLOUD_PROVIDER",
+        "AGENTKIT_CLOUD_PROVIDER",
+        "BYTEPLUS_REGION",
+        "BYTEPLUS_ACCESS_KEY",
+        "BYTEPLUS_SECRET_KEY",
+        "BYTEPLUS_SESSION_TOKEN",
+        "IAM_ROLE",
+    ):
+        monkeypatch.delenv(key, raising=False)
+
+    def _precheck(*, specs, **_kwargs):
+        return [
+            permissions.PermissionResult(spec=spec, satisfied=True) for spec in specs
+        ]
+
+    monkeypatch.setattr(
+        permissions,
+        "run_studio_deploy_permission_precheck",
+        _precheck,
+    )
+
+    result = CliRunner().invoke(
+        studio,
+        [
+            "deploy",
+            "--vefaas-app-name",
+            "studio-test",
+            "--provider",
+            "byteplus",
+            "--byteplus-access-key",
+            "ak",
+            "--byteplus-secret-key",
+            "sk",
+            "--precheck-only",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert os.environ.get("CLOUD_PROVIDER") is None
+    assert os.environ.get("AGENTKIT_CLOUD_PROVIDER") is None
+    assert os.environ.get("BYTEPLUS_REGION") is None
+    assert os.environ.get("BYTEPLUS_ACCESS_KEY") is None
+    assert os.environ.get("BYTEPLUS_SECRET_KEY") is None
+
+
 def test_cli_precheck_only_rejects_overlong_site_title_before_iam(monkeypatch) -> None:
     precheck_called = False