feat(05-01): template list/detail/deploy API + RBAC + integration tests
- Create shared/api/templates.py with templates_router
- GET /api/portal/templates: list active templates (any authenticated user)
- GET /api/portal/templates/{id}: get template detail (any authenticated user)
- POST /api/portal/templates/{id}/deploy: create Agent snapshot (tenant_admin only)
- customer_operator returns 403 on deploy (RBAC enforced)
- Export templates_router from shared/api/__init__.py
- Mount templates_router in gateway/main.py (Phase 5 section)
- 11 integration tests pass (list, detail, deploy, RBAC, 404, snapshot independence)
This commit is contained in:
@@ -46,6 +46,7 @@ from shared.api import (
|
||||
invitations_router,
|
||||
llm_keys_router,
|
||||
portal_router,
|
||||
templates_router,
|
||||
usage_router,
|
||||
webhook_router,
|
||||
)
|
||||
@@ -140,6 +141,11 @@ app.include_router(webhook_router)
|
||||
# ---------------------------------------------------------------------------
|
||||
app.include_router(invitations_router)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Register Phase 5 Employee Design routers
|
||||
# ---------------------------------------------------------------------------
|
||||
app.include_router(templates_router)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Routes
|
||||
|
||||
@@ -9,6 +9,7 @@ from shared.api.channels import channels_router
|
||||
from shared.api.invitations import invitations_router
|
||||
from shared.api.llm_keys import llm_keys_router
|
||||
from shared.api.portal import portal_router
|
||||
from shared.api.templates import templates_router
|
||||
from shared.api.usage import usage_router
|
||||
|
||||
__all__ = [
|
||||
@@ -19,4 +20,5 @@ __all__ = [
|
||||
"llm_keys_router",
|
||||
"usage_router",
|
||||
"invitations_router",
|
||||
"templates_router",
|
||||
]
|
||||
|
||||
186
packages/shared/shared/api/templates.py
Normal file
186
packages/shared/shared/api/templates.py
Normal file
@@ -0,0 +1,186 @@
|
||||
"""
|
||||
FastAPI template API router — agent template gallery and deploy endpoints.
|
||||
|
||||
Templates are global (not tenant-scoped): any authenticated portal user can
|
||||
browse them. Only tenant admins can deploy a template (creates an Agent snapshot).
|
||||
|
||||
Mounted at /api/portal in gateway/main.py alongside other portal routers.
|
||||
|
||||
Endpoints:
|
||||
GET /api/portal/templates — list active templates (all authenticated users)
|
||||
GET /api/portal/templates/{id} — get template detail (all authenticated users)
|
||||
POST /api/portal/templates/{id}/deploy — deploy template as agent (tenant admin only)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from shared.api.portal import AgentResponse
|
||||
from shared.api.rbac import PortalCaller, get_portal_caller, require_tenant_admin
|
||||
from shared.db import get_session
|
||||
from shared.models.tenant import Agent, AgentTemplate
|
||||
from shared.rls import current_tenant_id
|
||||
|
||||
templates_router = APIRouter(prefix="/api/portal", tags=["templates"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pydantic schemas
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TemplateResponse(BaseModel):
|
||||
id: str
|
||||
name: str
|
||||
role: str
|
||||
description: str
|
||||
category: str
|
||||
persona: str
|
||||
system_prompt: str
|
||||
model_preference: str
|
||||
tool_assignments: list[Any]
|
||||
escalation_rules: list[Any]
|
||||
is_active: bool
|
||||
sort_order: int
|
||||
created_at: datetime
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
@classmethod
|
||||
def from_orm(cls, tmpl: AgentTemplate) -> "TemplateResponse":
|
||||
return cls(
|
||||
id=str(tmpl.id),
|
||||
name=tmpl.name,
|
||||
role=tmpl.role,
|
||||
description=tmpl.description,
|
||||
category=tmpl.category,
|
||||
persona=tmpl.persona,
|
||||
system_prompt=tmpl.system_prompt,
|
||||
model_preference=tmpl.model_preference,
|
||||
tool_assignments=tmpl.tool_assignments,
|
||||
escalation_rules=tmpl.escalation_rules,
|
||||
is_active=tmpl.is_active,
|
||||
sort_order=tmpl.sort_order,
|
||||
created_at=tmpl.created_at,
|
||||
)
|
||||
|
||||
|
||||
class TemplateDeployRequest(BaseModel):
|
||||
tenant_id: uuid.UUID
|
||||
|
||||
|
||||
class TemplateDeployResponse(BaseModel):
|
||||
agent: AgentResponse
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Endpoints
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@templates_router.get("/templates", response_model=list[TemplateResponse])
|
||||
async def list_templates(
|
||||
caller: PortalCaller = Depends(get_portal_caller),
|
||||
session: AsyncSession = Depends(get_session),
|
||||
) -> list[TemplateResponse]:
|
||||
"""
|
||||
List all active agent templates.
|
||||
|
||||
Available to all authenticated portal users (any role).
|
||||
Templates are global — not tenant-scoped, no RLS needed.
|
||||
Returns templates ordered by sort_order ascending, then name.
|
||||
"""
|
||||
result = await session.execute(
|
||||
select(AgentTemplate)
|
||||
.where(AgentTemplate.is_active == True) # noqa: E712
|
||||
.order_by(AgentTemplate.sort_order.asc(), AgentTemplate.name.asc())
|
||||
)
|
||||
templates = result.scalars().all()
|
||||
return [TemplateResponse.from_orm(t) for t in templates]
|
||||
|
||||
|
||||
@templates_router.get("/templates/{template_id}", response_model=TemplateResponse)
|
||||
async def get_template(
|
||||
template_id: uuid.UUID,
|
||||
caller: PortalCaller = Depends(get_portal_caller),
|
||||
session: AsyncSession = Depends(get_session),
|
||||
) -> TemplateResponse:
|
||||
"""
|
||||
Get a single active agent template by ID.
|
||||
|
||||
Returns 404 if the template does not exist or is inactive.
|
||||
Available to all authenticated portal users (any role).
|
||||
"""
|
||||
result = await session.execute(
|
||||
select(AgentTemplate).where(
|
||||
AgentTemplate.id == template_id,
|
||||
AgentTemplate.is_active == True, # noqa: E712
|
||||
)
|
||||
)
|
||||
tmpl = result.scalar_one_or_none()
|
||||
if tmpl is None:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found")
|
||||
return TemplateResponse.from_orm(tmpl)
|
||||
|
||||
|
||||
@templates_router.post(
|
||||
"/templates/{template_id}/deploy",
|
||||
response_model=TemplateDeployResponse,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
)
|
||||
async def deploy_template(
|
||||
template_id: uuid.UUID,
|
||||
body: TemplateDeployRequest,
|
||||
caller: PortalCaller = Depends(get_portal_caller),
|
||||
session: AsyncSession = Depends(get_session),
|
||||
) -> TemplateDeployResponse:
|
||||
"""
|
||||
Deploy a template as an independent Agent snapshot for a tenant.
|
||||
|
||||
Guard: tenant admin only (customer_operator gets 403).
|
||||
The deployed Agent is a point-in-time snapshot — subsequent changes to the
|
||||
template do not affect already-deployed agents.
|
||||
|
||||
Returns 404 if template not found. Returns 403 if caller lacks admin access
|
||||
to the specified tenant.
|
||||
"""
|
||||
# RBAC: tenant admin check (reuses require_tenant_admin from rbac.py)
|
||||
await require_tenant_admin(body.tenant_id, caller, session)
|
||||
|
||||
# Fetch template (404 if not found or inactive)
|
||||
result = await session.execute(
|
||||
select(AgentTemplate).where(AgentTemplate.id == template_id)
|
||||
)
|
||||
tmpl = result.scalar_one_or_none()
|
||||
if tmpl is None:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found")
|
||||
|
||||
# Create Agent snapshot scoped to the target tenant
|
||||
token = current_tenant_id.set(body.tenant_id)
|
||||
try:
|
||||
agent = Agent(
|
||||
tenant_id=body.tenant_id,
|
||||
name=tmpl.name,
|
||||
role=tmpl.role,
|
||||
persona=tmpl.persona,
|
||||
system_prompt=tmpl.system_prompt,
|
||||
model_preference=tmpl.model_preference,
|
||||
tool_assignments=tmpl.tool_assignments,
|
||||
escalation_rules=tmpl.escalation_rules,
|
||||
is_active=True,
|
||||
)
|
||||
session.add(agent)
|
||||
await session.commit()
|
||||
await session.refresh(agent)
|
||||
finally:
|
||||
current_tenant_id.reset(token)
|
||||
|
||||
return TemplateDeployResponse(agent=AgentResponse.from_orm(agent))
|
||||
449
tests/integration/test_templates.py
Normal file
449
tests/integration/test_templates.py
Normal file
@@ -0,0 +1,449 @@
|
||||
"""
|
||||
Integration tests for agent template API endpoints.
|
||||
|
||||
Covers:
|
||||
- GET /api/portal/templates returns 7+ seeded templates
|
||||
- GET /api/portal/templates/{id} returns correct template detail
|
||||
- POST /api/portal/templates/{id}/deploy creates agent snapshot with is_active=True
|
||||
- POST deploy as customer_operator returns 403
|
||||
- POST deploy with invalid UUID returns 404
|
||||
- Deployed agent fields match template (snapshot verification)
|
||||
|
||||
Test infrastructure follows the same pattern as test_portal_rbac.py:
|
||||
- Session override via app.dependency_overrides
|
||||
- X-Portal-User-Id / X-Portal-User-Role / X-Portal-Tenant-Id header injection
|
||||
- db_session fixture from tests/conftest.py (Alembic migrations applied)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from fastapi import FastAPI
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from shared.api.portal import portal_router
|
||||
from shared.api.templates import templates_router
|
||||
from shared.db import get_session
|
||||
from shared.models.auth import PortalUser, UserTenantRole
|
||||
from shared.models.tenant import Tenant
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# App factory
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def make_app(session: AsyncSession) -> FastAPI:
|
||||
"""Build a minimal FastAPI test app with portal + templates routers."""
|
||||
app = FastAPI()
|
||||
app.include_router(portal_router)
|
||||
app.include_router(templates_router)
|
||||
|
||||
async def override_get_session(): # type: ignore[return]
|
||||
yield session
|
||||
|
||||
app.dependency_overrides[get_session] = override_get_session
|
||||
return app
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RBAC header helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def platform_admin_headers(user_id: uuid.UUID) -> dict[str, str]:
|
||||
return {
|
||||
"X-Portal-User-Id": str(user_id),
|
||||
"X-Portal-User-Role": "platform_admin",
|
||||
}
|
||||
|
||||
|
||||
def customer_admin_headers(user_id: uuid.UUID, tenant_id: uuid.UUID) -> dict[str, str]:
|
||||
return {
|
||||
"X-Portal-User-Id": str(user_id),
|
||||
"X-Portal-User-Role": "customer_admin",
|
||||
"X-Portal-Tenant-Id": str(tenant_id),
|
||||
}
|
||||
|
||||
|
||||
def customer_operator_headers(user_id: uuid.UUID, tenant_id: uuid.UUID) -> dict[str, str]:
|
||||
return {
|
||||
"X-Portal-User-Id": str(user_id),
|
||||
"X-Portal-User-Role": "customer_operator",
|
||||
"X-Portal-Tenant-Id": str(tenant_id),
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# DB setup helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _create_tenant(session: AsyncSession, name: str | None = None) -> Tenant:
|
||||
suffix = uuid.uuid4().hex[:8]
|
||||
tenant = Tenant(
|
||||
id=uuid.uuid4(),
|
||||
name=name or f"Template Test Tenant {suffix}",
|
||||
slug=f"tmpl-test-{suffix}",
|
||||
settings={},
|
||||
)
|
||||
session.add(tenant)
|
||||
await session.flush()
|
||||
return tenant
|
||||
|
||||
|
||||
async def _create_user(session: AsyncSession, role: str = "customer_admin") -> PortalUser:
|
||||
import bcrypt
|
||||
|
||||
suffix = uuid.uuid4().hex[:8]
|
||||
hashed = bcrypt.hashpw(b"testpassword123", bcrypt.gensalt()).decode()
|
||||
user = PortalUser(
|
||||
id=uuid.uuid4(),
|
||||
email=f"tmpl-test-{suffix}@example.com",
|
||||
hashed_password=hashed,
|
||||
name=f"Template Test User {suffix}",
|
||||
role=role,
|
||||
)
|
||||
session.add(user)
|
||||
await session.flush()
|
||||
return user
|
||||
|
||||
|
||||
async def _grant_membership(
|
||||
session: AsyncSession,
|
||||
user: PortalUser,
|
||||
tenant: Tenant,
|
||||
role: str,
|
||||
) -> UserTenantRole:
|
||||
membership = UserTenantRole(
|
||||
id=uuid.uuid4(),
|
||||
user_id=user.id,
|
||||
tenant_id=tenant.id,
|
||||
role=role,
|
||||
)
|
||||
session.add(membership)
|
||||
await session.flush()
|
||||
return membership
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def template_setup(db_session: AsyncSession) -> dict[str, Any]:
|
||||
"""
|
||||
Set up tenant, users, and memberships for template tests.
|
||||
|
||||
Returns:
|
||||
- tenant: the primary test tenant
|
||||
- platform_admin: platform_admin user
|
||||
- customer_admin: customer_admin user with membership in tenant
|
||||
- operator: customer_operator user with membership in tenant
|
||||
"""
|
||||
tenant = await _create_tenant(db_session)
|
||||
platform_admin = await _create_user(db_session, role="platform_admin")
|
||||
customer_admin = await _create_user(db_session, role="customer_admin")
|
||||
operator = await _create_user(db_session, role="customer_operator")
|
||||
|
||||
await _grant_membership(db_session, customer_admin, tenant, "customer_admin")
|
||||
await _grant_membership(db_session, operator, tenant, "customer_operator")
|
||||
|
||||
await db_session.commit()
|
||||
|
||||
return {
|
||||
"tenant": tenant,
|
||||
"platform_admin": platform_admin,
|
||||
"customer_admin": customer_admin,
|
||||
"operator": operator,
|
||||
}
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def templates_client(db_session: AsyncSession) -> AsyncClient:
|
||||
"""HTTP client with templates router mounted."""
|
||||
app = make_app(db_session)
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: GET /api/portal/templates
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_templates(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
) -> None:
|
||||
"""GET /api/portal/templates returns 200 with 7+ active seeded templates."""
|
||||
admin = template_setup["platform_admin"]
|
||||
headers = platform_admin_headers(admin.id)
|
||||
|
||||
response = await templates_client.get("/api/portal/templates", headers=headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert isinstance(data, list)
|
||||
assert len(data) >= 7, f"Expected at least 7 seeded templates, got {len(data)}"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_templates_returns_expected_fields(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
) -> None:
|
||||
"""Template list items include all required response fields."""
|
||||
admin = template_setup["platform_admin"]
|
||||
headers = platform_admin_headers(admin.id)
|
||||
|
||||
response = await templates_client.get("/api/portal/templates", headers=headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
templates = response.json()
|
||||
assert len(templates) > 0
|
||||
|
||||
# Check first template has all required fields
|
||||
first = templates[0]
|
||||
for field in ("id", "name", "role", "description", "category", "persona",
|
||||
"system_prompt", "model_preference", "tool_assignments",
|
||||
"escalation_rules", "is_active", "sort_order", "created_at"):
|
||||
assert field in first, f"Missing field: {field}"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_templates_customer_admin_can_browse(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
) -> None:
|
||||
"""customer_admin can browse templates (any authenticated user can)."""
|
||||
admin = template_setup["customer_admin"]
|
||||
tenant = template_setup["tenant"]
|
||||
headers = customer_admin_headers(admin.id, tenant.id)
|
||||
|
||||
response = await templates_client.get("/api/portal/templates", headers=headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert len(response.json()) >= 7
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_templates_operator_can_browse(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
) -> None:
|
||||
"""customer_operator can also browse templates."""
|
||||
operator = template_setup["operator"]
|
||||
tenant = template_setup["tenant"]
|
||||
headers = customer_operator_headers(operator.id, tenant.id)
|
||||
|
||||
response = await templates_client.get("/api/portal/templates", headers=headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert len(response.json()) >= 7
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: GET /api/portal/templates/{id}
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_template_detail(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
) -> None:
|
||||
"""GET single template returns correct fields including AI transparency clause."""
|
||||
admin = template_setup["platform_admin"]
|
||||
headers = platform_admin_headers(admin.id)
|
||||
|
||||
# First get the list to find a real template ID
|
||||
list_resp = await templates_client.get("/api/portal/templates", headers=headers)
|
||||
assert list_resp.status_code == 200
|
||||
templates = list_resp.json()
|
||||
assert len(templates) > 0
|
||||
|
||||
template_id = templates[0]["id"]
|
||||
response = await templates_client.get(f"/api/portal/templates/{template_id}", headers=headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["id"] == template_id
|
||||
assert data["name"] == templates[0]["name"]
|
||||
assert data["role"] == templates[0]["role"]
|
||||
assert data["is_active"] is True
|
||||
# System prompt should contain AI transparency clause
|
||||
assert "When directly asked if you are an AI" in data["system_prompt"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_template_not_found(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
) -> None:
|
||||
"""GET with non-existent UUID returns 404."""
|
||||
admin = template_setup["platform_admin"]
|
||||
headers = platform_admin_headers(admin.id)
|
||||
nonexistent_id = str(uuid.uuid4())
|
||||
|
||||
response = await templates_client.get(f"/api/portal/templates/{nonexistent_id}", headers=headers)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: POST /api/portal/templates/{id}/deploy
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_deploy_template(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
db_session: AsyncSession,
|
||||
) -> None:
|
||||
"""POST deploy creates an independent Agent snapshot with is_active=True."""
|
||||
admin = template_setup["customer_admin"]
|
||||
tenant = template_setup["tenant"]
|
||||
headers = customer_admin_headers(admin.id, tenant.id)
|
||||
|
||||
# Get a template to deploy
|
||||
list_resp = await templates_client.get("/api/portal/templates", headers=headers)
|
||||
assert list_resp.status_code == 200
|
||||
templates = list_resp.json()
|
||||
template = templates[0]
|
||||
template_id = template["id"]
|
||||
|
||||
response = await templates_client.post(
|
||||
f"/api/portal/templates/{template_id}/deploy",
|
||||
json={"tenant_id": str(tenant.id)},
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
agent = data["agent"]
|
||||
|
||||
# Agent snapshot should be active
|
||||
assert agent["is_active"] is True
|
||||
# Agent fields should match template
|
||||
assert agent["name"] == template["name"]
|
||||
assert agent["role"] == template["role"]
|
||||
assert agent["persona"] == template["persona"]
|
||||
assert agent["system_prompt"] == template["system_prompt"]
|
||||
assert agent["model_preference"] == template["model_preference"]
|
||||
assert agent["tool_assignments"] == template["tool_assignments"]
|
||||
# Agent should be scoped to the tenant
|
||||
assert agent["tenant_id"] == str(tenant.id)
|
||||
# Agent should have an id
|
||||
assert agent["id"] is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_deploy_template_platform_admin(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
) -> None:
|
||||
"""platform_admin can deploy templates to any tenant."""
|
||||
admin = template_setup["platform_admin"]
|
||||
tenant = template_setup["tenant"]
|
||||
headers = platform_admin_headers(admin.id)
|
||||
|
||||
list_resp = await templates_client.get("/api/portal/templates", headers=headers)
|
||||
template = list_resp.json()[0]
|
||||
template_id = template["id"]
|
||||
|
||||
response = await templates_client.post(
|
||||
f"/api/portal/templates/{template_id}/deploy",
|
||||
json={"tenant_id": str(tenant.id)},
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.json()["agent"]["is_active"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_deploy_template_rbac_operator_forbidden(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
) -> None:
|
||||
"""POST deploy as customer_operator returns 403."""
|
||||
operator = template_setup["operator"]
|
||||
tenant = template_setup["tenant"]
|
||||
headers = customer_operator_headers(operator.id, tenant.id)
|
||||
|
||||
# Get a template ID
|
||||
admin = template_setup["platform_admin"]
|
||||
list_resp = await templates_client.get(
|
||||
"/api/portal/templates",
|
||||
headers=platform_admin_headers(admin.id),
|
||||
)
|
||||
template_id = list_resp.json()[0]["id"]
|
||||
|
||||
response = await templates_client.post(
|
||||
f"/api/portal/templates/{template_id}/deploy",
|
||||
json={"tenant_id": str(tenant.id)},
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_deploy_template_not_found(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
) -> None:
|
||||
"""POST deploy with invalid/nonexistent template UUID returns 404."""
|
||||
admin = template_setup["customer_admin"]
|
||||
tenant = template_setup["tenant"]
|
||||
headers = customer_admin_headers(admin.id, tenant.id)
|
||||
nonexistent_id = str(uuid.uuid4())
|
||||
|
||||
response = await templates_client.post(
|
||||
f"/api/portal/templates/{nonexistent_id}/deploy",
|
||||
json={"tenant_id": str(tenant.id)},
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_deploy_creates_independent_snapshot(
|
||||
templates_client: AsyncClient,
|
||||
template_setup: dict[str, Any],
|
||||
) -> None:
|
||||
"""Deploying the same template twice creates two independent agents."""
|
||||
admin = template_setup["customer_admin"]
|
||||
tenant = template_setup["tenant"]
|
||||
headers = customer_admin_headers(admin.id, tenant.id)
|
||||
|
||||
list_resp = await templates_client.get("/api/portal/templates", headers=headers)
|
||||
template_id = list_resp.json()[0]["id"]
|
||||
|
||||
resp1 = await templates_client.post(
|
||||
f"/api/portal/templates/{template_id}/deploy",
|
||||
json={"tenant_id": str(tenant.id)},
|
||||
headers=headers,
|
||||
)
|
||||
resp2 = await templates_client.post(
|
||||
f"/api/portal/templates/{template_id}/deploy",
|
||||
json={"tenant_id": str(tenant.id)},
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
assert resp1.status_code == 201
|
||||
assert resp2.status_code == 201
|
||||
# Both agents should have distinct IDs
|
||||
agent1_id = resp1.json()["agent"]["id"]
|
||||
agent2_id = resp2.json()["agent"]["id"]
|
||||
assert agent1_id != agent2_id
|
||||
Reference in New Issue
Block a user