hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(cli): refactor PaaS auth flow and fix container deploy payload

Browse Source 
Extract _exchange_iam_token method for cleaner auth caching logic.
Fix deploy_env to send full container payload on PUT (PaaS requires it).
Add e2e test scaffold.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
Zoo Queen
2026-02-20 19:09:05 -08:00
co-authored by Hanzo Dev
parent 7d77ee615f
commit e67acae9db
3 changed files with 329 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-cli/hanzo_cli/paas/client.py
+62 -34
View File
@@ -65,33 +65,8 @@ class PaaSClient:
# -- bootstrap -----------------------------------------------------------
@classmethod
def from_auth(cls) -> PaaSClient:
"""Build a client, exchanging the IAM token for a PaaS session.
1. Check for a cached PaaS session
2. If none, exchange the IAM token via POST /v1/auth/login
3. Cache the PaaS session for reuse
"""
base_url = os.getenv("HANZO_PAAS_URL", DEFAULT_PAAS_URL).rstrip("/")
# Try cached PaaS session first
session = _load_session()
if session and session.get("at"):
inst = cls(
base_url=base_url,
access_token=session["at"],
refresh_token=session.get("rt"),
)
# Quick health check — if 401, re-login
try:
resp = inst.http.get("/v1/cluster/setup-status")
if resp.status_code != 401:
return inst
except Exception:
pass
inst.close()
# Need the IAM token
def _exchange_iam_token(cls, base_url: str) -> PaaSClient:
"""Exchange the stored IAM token for a fresh PaaS session."""
token_data = get_token_info()
if not token_data or not token_data.get("access_token"):
click.echo(
@@ -102,7 +77,6 @@ class PaaSClient:
iam_token = token_data["access_token"]
# Exchange IAM token for PaaS session
with httpx.Client(base_url=base_url, timeout=30.0) as tmp:
resp = tmp.post(
"/v1/auth/login",
@@ -134,6 +108,35 @@ class PaaSClient:
_save_session({"at": at, "rt": rt, "login_time": int(time.time())})
return cls(base_url=base_url, access_token=at, refresh_token=rt)
@classmethod
def from_auth(cls) -> PaaSClient:
"""Build a client, exchanging the IAM token for a PaaS session.
1. Check for a cached PaaS session (validate with authenticated endpoint)
2. If expired/invalid, exchange the IAM token via POST /v1/auth/login
3. Cache the PaaS session for reuse
"""
base_url = os.getenv("HANZO_PAAS_URL", DEFAULT_PAAS_URL).rstrip("/")
# Try cached PaaS session first
session = _load_session()
if session and session.get("at"):
inst = cls(
base_url=base_url,
access_token=session["at"],
refresh_token=session.get("rt"),
)
# Validate with an authenticated endpoint
try:
resp = inst.http.get("/v1/org")
if resp.status_code != 401:
return inst
except Exception:
pass
inst.close()
return cls._exchange_iam_token(base_url)
@property
def http(self) -> httpx.Client:
if self._http is None:
@@ -159,7 +162,15 @@ class PaaSClient:
# -- helpers -------------------------------------------------------------
def _ok(self, resp: httpx.Response) -> Any:
def _reauth(self) -> None:
"""Re-exchange the IAM token for a fresh PaaS session."""
self.close()
fresh = self._exchange_iam_token(self._base_url)
self._at = fresh._at
self._rt = fresh._rt
# _http will be lazily rebuilt on next access
def _ok(self, resp: httpx.Response, *, _retried: bool = False) -> Any:
# PaaS may return refreshed tokens in headers
new_at = resp.headers.get("Access-Token")
new_rt = resp.headers.get("Refresh-Token")
@@ -173,6 +184,19 @@ class PaaSClient:
# Rebuild client with new token
self.close()
# Auto-retry on 401: re-exchange IAM token and replay the request
if resp.status_code == 401 and not _retried:
self._reauth()
# Replay the original request using just the path
req = resp.request
path = req.url.raw_path.decode("ascii") # e.g. /v1/org
retry_resp = self.http.request(
method=req.method,
url=path,
content=req.content if req.content else None,
)
return self._ok(retry_resp, _retried=True)
if resp.status_code >= 400:
try:
err = resp.json()
@@ -273,11 +297,15 @@ class PaaSClient:
def redeploy_container(
self, org_id: str, project_id: str, env_id: str, container_id: str,
) -> dict[str, Any]:
url = (
f"{self._container_base(org_id, project_id, env_id)}"
f"/{container_id}/redeploy"
)
return self._ok(self.http.post(url))
"""Trigger a redeploy by re-PUTting the container config.
The PaaS API doesn't expose a dedicated /redeploy endpoint.
Updating the container with its current config triggers a rolling restart.
"""
# Fetch current config, then PUT it back to trigger redeployment
container = self.get_container(org_id, project_id, env_id, container_id)
url = f"{self._container_base(org_id, project_id, env_id)}/{container_id}"
return self._ok(self.http.put(url, json=container))
def get_container_pods(
self, org_id: str, project_id: str, env_id: str, container_id: str,
pkg/hanzo-cli/hanzo_cli/paas/commands.py
+4 -1
View File
@@ -612,8 +612,11 @@ def deploy_env(
existing[nv["name"]] = nv["value"]
merged = [{"name": k, "value": v} for k, v in existing.items()]
# PaaS requires the full container payload on PUT — merge vars into it
container["variables"] = merged
client.update_container(
org_id, project_id, env_id, cid, {"variables": merged},
org_id, project_id, env_id, cid, container,
)
console.print(f"[green]Set {len(new_vars)} variable(s) on '{name}'.[/green]")
finally:
pkg/hanzo-cli/tests/test_e2e.py
+263
View File
@@ -0,0 +1,263 @@
"""End-to-end tests for hanzo CLI against live services.
Requires:
- hanzo-cli installed (pip install -e pkg/hanzo-cli)
- Valid credentials stored at ~/.hanzo/auth/token.json (run hanzo login first)
- Live services: hanzo.id, platform.hanzo.ai
Run:
pytest pkg/hanzo-cli/tests/test_e2e.py -v -s
"""
from __future__ import annotations
import json
import os
import subprocess
import time
import pytest
HANZO = "hanzo"
TIMEOUT = 30
def run(args: list[str], check: bool = True) -> subprocess.CompletedProcess:
"""Run a hanzo CLI command and return the result."""
result = subprocess.run(
[HANZO] + args,
capture_output=True,
text=True,
timeout=TIMEOUT,
)
if check and result.returncode != 0:
pytest.fail(
f"hanzo {' '.join(args)} failed (rc={result.returncode}):\n"
f"stdout: {result.stdout}\n"
f"stderr: {result.stderr}"
)
return result
# =========================================================================
# Auth
# =========================================================================
class TestAuth:
def test_version(self):
r = run(["--version"])
assert "0.1.0" in r.stdout
def test_help(self):
r = run(["--help"])
assert "login" in r.stdout
assert "paas" in r.stdout
assert "iam" in r.stdout
assert "kms" in r.stdout
def test_whoami(self):
r = run(["whoami"])
assert "[email protected]" in r.stdout or "z" in r.stdout
assert "hanzo.id" in r.stdout
def test_whoami_shows_org(self):
r = run(["whoami"])
assert "hanzo" in r.stdout
# =========================================================================
# IAM
# =========================================================================
class TestIAM:
def test_iam_help(self):
r = run(["iam", "--help"])
assert "users" in r.stdout
assert "set-password" in r.stdout
def test_iam_users(self):
"""List users — should return at least user z."""
r = run(["iam", "users"], check=False)
# May fail if bearer token doesn't have admin access,
# but the command should at least not crash
assert r.returncode == 0 or "Not authenticated" not in r.stderr
def test_iam_orgs(self):
r = run(["iam", "orgs"], check=False)
assert r.returncode == 0 or "Not authenticated" not in r.stderr
# =========================================================================
# PaaS — Read Operations
# =========================================================================
class TestPaaSRead:
def test_paas_help(self):
r = run(["paas", "--help"])
assert "deploy" in r.stdout
assert "orgs" in r.stdout
assert "use" in r.stdout
def test_paas_orgs(self):
r = run(["paas", "orgs"])
assert "Hanzo" in r.stdout
assert "Organizations" in r.stdout
def test_paas_projects(self):
r = run(["paas", "projects", "--org", "698cda6739f65183b3009313"])
assert "Platform" in r.stdout
def test_paas_envs(self):
r = run([
"paas", "envs",
"--org", "698cda6739f65183b3009313",
"--project", "698cda6739f65183b3009318",
])
assert "production" in r.stdout
def test_paas_context_set_and_show(self):
run([
"paas", "use",
"--org", "698cda6739f65183b3009313",
"--project", "698cda6739f65183b3009318",
"--env", "698cda6739f65183b300931c",
])
r = run(["paas", "context"])
assert "698cda6739f65183b3009313" in r.stdout
assert "698cda6739f65183b3009318" in r.stdout
assert "698cda6739f65183b300931c" in r.stdout
def test_paas_deploy_list(self):
"""List containers — may be empty, but should not error."""
r = run(["paas", "deploy", "list"])
assert r.returncode == 0
# =========================================================================
# PaaS — Deploy Lifecycle (create → status → logs → redeploy → delete)
# =========================================================================
class TestPaaSDeployLifecycle:
"""Full container lifecycle test using a lightweight Docker image."""
CONTAINER_NAME = "e2e-test-nginx"
@classmethod
def setup_class(cls):
"""Ensure context is set and clean up any leftover test containers."""
run([
"paas", "use",
"--org", "698cda6739f65183b3009313",
"--project", "698cda6739f65183b3009318",
"--env", "698cda6739f65183b300931c",
])
# Clean up leftover from previous failed runs
r = run(["paas", "deploy", "delete", cls.CONTAINER_NAME, "-y"], check=False)
@classmethod
def teardown_class(cls):
"""Clean up test container."""
run(["paas", "deploy", "delete", cls.CONTAINER_NAME, "-y"], check=False)
def test_01_create(self):
r = run([
"paas", "deploy", "create", self.CONTAINER_NAME,
"--image", "nginx:latest",
"--port", "80",
])
assert "Created container" in r.stdout
assert self.CONTAINER_NAME in r.stdout
def test_02_list_contains_container(self):
r = run(["paas", "deploy", "list"])
assert self.CONTAINER_NAME in r.stdout
def test_03_status_shows_pod(self):
"""Wait for the pod to appear and check status."""
for attempt in range(6):
r = run(["paas", "deploy", "status", self.CONTAINER_NAME])
if "Running" in r.stdout or "Pending" in r.stdout:
break
time.sleep(5)
assert self.CONTAINER_NAME in r.stdout
assert "deployment" in r.stdout
def test_04_wait_for_running(self):
"""Wait up to 60s for the pod to be Running."""
for attempt in range(12):
r = run(["paas", "deploy", "status", self.CONTAINER_NAME])
if "Running" in r.stdout:
return
time.sleep(5)
pytest.fail(f"Pod not Running after 60s:\n{r.stdout}")
def test_05_logs(self):
"""Container logs should contain nginx startup messages."""
r = run(["paas", "deploy", "logs", self.CONTAINER_NAME])
assert "nginx" in r.stdout.lower() or "worker process" in r.stdout
def test_06_env_show(self):
"""Show env vars — should work even if empty."""
r = run(["paas", "deploy", "env", self.CONTAINER_NAME])
assert r.returncode == 0
def test_07_env_set(self):
"""Set an env var on the container."""
r = run([
"paas", "deploy", "env", self.CONTAINER_NAME,
"TEST_VAR=hello_e2e",
])
assert "Set 1 variable" in r.stdout
def test_08_env_verify(self):
"""Verify the env var was set."""
r = run(["paas", "deploy", "env", self.CONTAINER_NAME])
assert "TEST_VAR" in r.stdout
def test_09_redeploy(self):
"""Trigger a redeploy."""
r = run(["paas", "deploy", "redeploy", self.CONTAINER_NAME])
assert "Redeployment triggered" in r.stdout
def test_10_delete(self):
"""Delete the container."""
r = run(["paas", "deploy", "delete", self.CONTAINER_NAME, "-y"])
assert "Deleted" in r.stdout
def test_11_verify_deleted(self):
"""Verify the container no longer exists."""
r = run(["paas", "deploy", "list"])
assert self.CONTAINER_NAME not in r.stdout
# =========================================================================
# PaaS — Error Handling
# =========================================================================
class TestPaaSErrors:
def test_missing_context(self):
"""Commands without required context should fail gracefully."""
run(["paas", "use", "--clear"])
r = run(["paas", "deploy", "list"], check=False)
assert r.returncode != 0
assert "required" in r.stdout.lower() or "required" in r.stderr.lower()
def test_nonexistent_container(self):
run([
"paas", "use",
"--org", "698cda6739f65183b3009313",
"--project", "698cda6739f65183b3009318",
"--env", "698cda6739f65183b300931c",
])
r = run(["paas", "deploy", "status", "nonexistent-container-xyz"], check=False)
assert r.returncode != 0
assert "not found" in r.stdout.lower() or "not found" in r.stderr.lower()
def test_create_missing_image_and_repo(self):
r = run(["paas", "deploy", "create", "bad-deploy"], check=False)
assert r.returncode != 0