214 lines
7.8 KiB
Python
214 lines
7.8 KiB
Python
"""Tests for env management routes."""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.agent import Agent
|
|
from app.models.task import Task
|
|
from app.models.tenant import Tenant
|
|
from app.routes.env import inspect_env, update_env
|
|
from app.schemas.env import EnvInspectRequest, EnvUpdateRequest
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestInspectEnv:
|
|
"""Tests for the inspect_env endpoint."""
|
|
|
|
async def test_happy_path(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent: Agent
|
|
):
|
|
"""Valid request creates an ENV_INSPECT task."""
|
|
request = EnvInspectRequest(
|
|
tenant_id=test_tenant.id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
keys=["FRONTEND_URL", "BACKEND_URL"],
|
|
)
|
|
|
|
task = await inspect_env(agent_id=test_agent.id, request=request, db=db)
|
|
|
|
assert task.id is not None
|
|
assert task.tenant_id == test_tenant.id
|
|
assert task.agent_id == test_agent.id
|
|
assert task.type == "ENV_INSPECT"
|
|
assert task.status == "pending"
|
|
assert task.payload["path"] == "/opt/letsbe/env/chatwoot.env"
|
|
assert task.payload["keys"] == ["FRONTEND_URL", "BACKEND_URL"]
|
|
|
|
async def test_without_keys(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent: Agent
|
|
):
|
|
"""Request without keys omits keys from payload."""
|
|
request = EnvInspectRequest(
|
|
tenant_id=test_tenant.id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
keys=None,
|
|
)
|
|
|
|
task = await inspect_env(agent_id=test_agent.id, request=request, db=db)
|
|
|
|
assert task.type == "ENV_INSPECT"
|
|
assert task.payload["path"] == "/opt/letsbe/env/chatwoot.env"
|
|
assert "keys" not in task.payload
|
|
|
|
async def test_tenant_not_found(self, db: AsyncSession, test_agent: Agent):
|
|
"""Returns 404 when tenant doesn't exist."""
|
|
fake_tenant_id = uuid.uuid4()
|
|
request = EnvInspectRequest(
|
|
tenant_id=fake_tenant_id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await inspect_env(agent_id=test_agent.id, request=request, db=db)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert f"Tenant {fake_tenant_id} not found" in str(exc_info.value.detail)
|
|
|
|
async def test_agent_not_found(self, db: AsyncSession, test_tenant: Tenant):
|
|
"""Returns 404 when agent doesn't exist."""
|
|
fake_agent_id = uuid.uuid4()
|
|
request = EnvInspectRequest(
|
|
tenant_id=test_tenant.id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await inspect_env(agent_id=fake_agent_id, request=request, db=db)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert f"Agent {fake_agent_id} not found" in str(exc_info.value.detail)
|
|
|
|
async def test_task_persisted_to_database(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent: Agent
|
|
):
|
|
"""Verify the task is actually persisted and can be retrieved."""
|
|
request = EnvInspectRequest(
|
|
tenant_id=test_tenant.id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
)
|
|
|
|
task = await inspect_env(agent_id=test_agent.id, request=request, db=db)
|
|
|
|
# Query the task back from the database
|
|
result = await db.execute(select(Task).where(Task.id == task.id))
|
|
retrieved_task = result.scalar_one_or_none()
|
|
|
|
assert retrieved_task is not None
|
|
assert retrieved_task.type == "ENV_INSPECT"
|
|
assert retrieved_task.tenant_id == test_tenant.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUpdateEnv:
|
|
"""Tests for the update_env endpoint."""
|
|
|
|
async def test_happy_path(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent: Agent
|
|
):
|
|
"""Valid request creates an ENV_UPDATE task."""
|
|
request = EnvUpdateRequest(
|
|
tenant_id=test_tenant.id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
updates={"FRONTEND_URL": "https://new.domain.com"},
|
|
remove_keys=["OLD_KEY"],
|
|
)
|
|
|
|
task = await update_env(agent_id=test_agent.id, request=request, db=db)
|
|
|
|
assert task.id is not None
|
|
assert task.tenant_id == test_tenant.id
|
|
assert task.agent_id == test_agent.id
|
|
assert task.type == "ENV_UPDATE"
|
|
assert task.status == "pending"
|
|
assert task.payload["path"] == "/opt/letsbe/env/chatwoot.env"
|
|
assert task.payload["updates"] == {"FRONTEND_URL": "https://new.domain.com"}
|
|
assert task.payload["remove_keys"] == ["OLD_KEY"]
|
|
|
|
async def test_updates_only(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent: Agent
|
|
):
|
|
"""Request with only updates omits remove_keys from payload."""
|
|
request = EnvUpdateRequest(
|
|
tenant_id=test_tenant.id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
updates={"KEY": "VALUE"},
|
|
remove_keys=None,
|
|
)
|
|
|
|
task = await update_env(agent_id=test_agent.id, request=request, db=db)
|
|
|
|
assert task.type == "ENV_UPDATE"
|
|
assert task.payload["updates"] == {"KEY": "VALUE"}
|
|
assert "remove_keys" not in task.payload
|
|
|
|
async def test_remove_keys_only(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent: Agent
|
|
):
|
|
"""Request with only remove_keys omits updates from payload."""
|
|
request = EnvUpdateRequest(
|
|
tenant_id=test_tenant.id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
updates=None,
|
|
remove_keys=["OLD_KEY"],
|
|
)
|
|
|
|
task = await update_env(agent_id=test_agent.id, request=request, db=db)
|
|
|
|
assert task.type == "ENV_UPDATE"
|
|
assert "updates" not in task.payload
|
|
assert task.payload["remove_keys"] == ["OLD_KEY"]
|
|
|
|
async def test_tenant_not_found(self, db: AsyncSession, test_agent: Agent):
|
|
"""Returns 404 when tenant doesn't exist."""
|
|
fake_tenant_id = uuid.uuid4()
|
|
request = EnvUpdateRequest(
|
|
tenant_id=fake_tenant_id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
updates={"KEY": "VALUE"},
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await update_env(agent_id=test_agent.id, request=request, db=db)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert f"Tenant {fake_tenant_id} not found" in str(exc_info.value.detail)
|
|
|
|
async def test_agent_not_found(self, db: AsyncSession, test_tenant: Tenant):
|
|
"""Returns 404 when agent doesn't exist."""
|
|
fake_agent_id = uuid.uuid4()
|
|
request = EnvUpdateRequest(
|
|
tenant_id=test_tenant.id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
updates={"KEY": "VALUE"},
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await update_env(agent_id=fake_agent_id, request=request, db=db)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert f"Agent {fake_agent_id} not found" in str(exc_info.value.detail)
|
|
|
|
async def test_task_persisted_to_database(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent: Agent
|
|
):
|
|
"""Verify the task is actually persisted and can be retrieved."""
|
|
request = EnvUpdateRequest(
|
|
tenant_id=test_tenant.id,
|
|
path="/opt/letsbe/env/chatwoot.env",
|
|
updates={"KEY": "VALUE"},
|
|
)
|
|
|
|
task = await update_env(agent_id=test_agent.id, request=request, db=db)
|
|
|
|
# Query the task back from the database
|
|
result = await db.execute(select(Task).where(Task.id == task.id))
|
|
retrieved_task = result.scalar_one_or_none()
|
|
|
|
assert retrieved_task is not None
|
|
assert retrieved_task.type == "ENV_UPDATE"
|
|
assert retrieved_task.tenant_id == test_tenant.id
|