142 lines
5.1 KiB
Python
142 lines
5.1 KiB
Python
"""Tests for the Chatwoot playbook module."""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.task import Task, TaskStatus
|
|
from app.models.tenant import Tenant
|
|
from app.playbooks.chatwoot import (
|
|
CHATWOOT_ENV_PATH,
|
|
CHATWOOT_STACK_DIR,
|
|
CompositeStep,
|
|
build_chatwoot_setup_steps,
|
|
create_chatwoot_setup_task,
|
|
)
|
|
|
|
|
|
class TestBuildChatwootSetupSteps:
|
|
"""Tests for the build_chatwoot_setup_steps function."""
|
|
|
|
def test_returns_two_steps(self):
|
|
"""Verify that build_chatwoot_setup_steps returns exactly 2 steps."""
|
|
steps = build_chatwoot_setup_steps(domain="support.example.com")
|
|
assert len(steps) == 2
|
|
assert all(isinstance(step, CompositeStep) for step in steps)
|
|
|
|
def test_env_update_payload(self):
|
|
"""Verify the ENV_UPDATE step has the correct payload structure."""
|
|
domain = "support.example.com"
|
|
steps = build_chatwoot_setup_steps(domain=domain)
|
|
|
|
env_step = steps[0]
|
|
assert env_step.type == "ENV_UPDATE"
|
|
assert env_step.payload["path"] == CHATWOOT_ENV_PATH
|
|
assert env_step.payload["updates"]["FRONTEND_URL"] == f"https://{domain}"
|
|
assert env_step.payload["updates"]["BACKEND_URL"] == f"https://{domain}"
|
|
|
|
def test_docker_reload_payload(self):
|
|
"""Verify the DOCKER_RELOAD step has the correct payload structure."""
|
|
steps = build_chatwoot_setup_steps(domain="support.example.com")
|
|
|
|
docker_step = steps[1]
|
|
assert docker_step.type == "DOCKER_RELOAD"
|
|
assert docker_step.payload["compose_dir"] == CHATWOOT_STACK_DIR
|
|
assert docker_step.payload["pull"] is True
|
|
|
|
def test_domain_url_formatting(self):
|
|
"""Verify that domain URLs are properly formatted with https."""
|
|
domain = "chat.mycompany.io"
|
|
steps = build_chatwoot_setup_steps(domain=domain)
|
|
|
|
env_step = steps[0]
|
|
assert env_step.payload["updates"]["FRONTEND_URL"] == "https://chat.mycompany.io"
|
|
assert env_step.payload["updates"]["BACKEND_URL"] == "https://chat.mycompany.io"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestCreateChatwootSetupTask:
|
|
"""Tests for the create_chatwoot_setup_task function."""
|
|
|
|
async def test_persists_composite_task(self, db: AsyncSession, test_tenant: Tenant):
|
|
"""Verify that create_chatwoot_setup_task persists a COMPOSITE task."""
|
|
task = await create_chatwoot_setup_task(
|
|
db=db,
|
|
tenant_id=test_tenant.id,
|
|
agent_id=None,
|
|
domain="support.example.com",
|
|
)
|
|
|
|
assert task.id is not None
|
|
assert task.tenant_id == test_tenant.id
|
|
assert task.type == "COMPOSITE"
|
|
assert task.status == TaskStatus.PENDING.value
|
|
|
|
async def test_task_payload_contains_steps(self, db: AsyncSession, test_tenant: Tenant):
|
|
"""Verify that the task payload contains the steps array."""
|
|
task = await create_chatwoot_setup_task(
|
|
db=db,
|
|
tenant_id=test_tenant.id,
|
|
agent_id=None,
|
|
domain="support.example.com",
|
|
)
|
|
|
|
assert "steps" in task.payload
|
|
assert len(task.payload["steps"]) == 2
|
|
|
|
async def test_task_steps_structure(self, db: AsyncSession, test_tenant: Tenant):
|
|
"""Verify that the steps in the payload have the correct structure."""
|
|
task = await create_chatwoot_setup_task(
|
|
db=db,
|
|
tenant_id=test_tenant.id,
|
|
agent_id=None,
|
|
domain="support.example.com",
|
|
)
|
|
|
|
steps = task.payload["steps"]
|
|
|
|
# First step should be ENV_UPDATE
|
|
assert steps[0]["type"] == "ENV_UPDATE"
|
|
assert "path" in steps[0]["payload"]
|
|
assert "updates" in steps[0]["payload"]
|
|
|
|
# Second step should be DOCKER_RELOAD
|
|
assert steps[1]["type"] == "DOCKER_RELOAD"
|
|
assert "compose_dir" in steps[1]["payload"]
|
|
assert steps[1]["payload"]["pull"] is True
|
|
|
|
async def test_task_with_agent_id(self, db: AsyncSession, test_tenant: Tenant):
|
|
"""Verify that agent_id is properly assigned when provided."""
|
|
agent_id = uuid.uuid4()
|
|
|
|
# Note: In a real scenario, the agent would need to exist in the DB
|
|
# For this test, we're just verifying the task stores the agent_id
|
|
task = await create_chatwoot_setup_task(
|
|
db=db,
|
|
tenant_id=test_tenant.id,
|
|
agent_id=agent_id,
|
|
domain="support.example.com",
|
|
)
|
|
|
|
assert task.agent_id == agent_id
|
|
|
|
async def test_task_persisted_to_database(self, db: AsyncSession, test_tenant: Tenant):
|
|
"""Verify the task is actually persisted and can be retrieved."""
|
|
task = await create_chatwoot_setup_task(
|
|
db=db,
|
|
tenant_id=test_tenant.id,
|
|
agent_id=None,
|
|
domain="support.example.com",
|
|
)
|
|
|
|
# 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 == "COMPOSITE"
|
|
assert retrieved_task.tenant_id == test_tenant.id
|