231 lines
8.2 KiB
Python
231 lines
8.2 KiB
Python
|
"""Tests for the Poste.io 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.poste import (
|
||
|
|
POSTE_STACK_DIR,
|
||
|
|
build_poste_initial_setup_step,
|
||
|
|
create_poste_initial_setup_task,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class TestBuildPosteInitialSetupStep:
|
||
|
|
"""Tests for the build_poste_initial_setup_step function."""
|
||
|
|
|
||
|
|
def test_returns_playwright_payload(self):
|
||
|
|
"""Verify that build_poste_initial_setup_step returns correct structure."""
|
||
|
|
payload = build_poste_initial_setup_step(
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert payload["scenario"] == "poste_initial_setup"
|
||
|
|
assert "inputs" in payload
|
||
|
|
assert "options" in payload
|
||
|
|
assert "timeout" in payload
|
||
|
|
|
||
|
|
def test_inputs_contain_required_fields(self):
|
||
|
|
"""Verify that inputs contain base_url and admin_email."""
|
||
|
|
payload = build_poste_initial_setup_step(
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
inputs = payload["inputs"]
|
||
|
|
assert inputs["base_url"] == "https://mail.example.com"
|
||
|
|
assert inputs["admin_email"] == "admin@example.com"
|
||
|
|
|
||
|
|
def test_password_included_when_provided(self):
|
||
|
|
"""Verify that admin_password is included in inputs when provided."""
|
||
|
|
payload = build_poste_initial_setup_step(
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
admin_password="secure-password-123",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert payload["inputs"]["admin_password"] == "secure-password-123"
|
||
|
|
|
||
|
|
def test_password_omitted_when_none(self):
|
||
|
|
"""Verify that admin_password is omitted when not provided."""
|
||
|
|
payload = build_poste_initial_setup_step(
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert "admin_password" not in payload["inputs"]
|
||
|
|
|
||
|
|
def test_allowed_domains_extracted_from_url(self):
|
||
|
|
"""Verify that allowed_domains is extracted from base_url."""
|
||
|
|
payload = build_poste_initial_setup_step(
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert payload["options"]["allowed_domains"] == ["mail.example.com"]
|
||
|
|
|
||
|
|
def test_allowed_domains_with_port(self):
|
||
|
|
"""Verify that allowed_domains handles URLs with ports."""
|
||
|
|
payload = build_poste_initial_setup_step(
|
||
|
|
base_url="https://mail.example.com:8443",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert payload["options"]["allowed_domains"] == ["mail.example.com:8443"]
|
||
|
|
|
||
|
|
def test_timeout_is_set(self):
|
||
|
|
"""Verify that timeout is 120 seconds."""
|
||
|
|
payload = build_poste_initial_setup_step(
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert payload["timeout"] == 120
|
||
|
|
|
||
|
|
def test_domain_url_with_subdomain(self):
|
||
|
|
"""Verify correct domain extraction from a subdomain URL."""
|
||
|
|
payload = build_poste_initial_setup_step(
|
||
|
|
base_url="https://mail.mycompany.io",
|
||
|
|
admin_email="postmaster@mycompany.io",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert payload["options"]["allowed_domains"] == ["mail.mycompany.io"]
|
||
|
|
assert payload["inputs"]["admin_email"] == "postmaster@mycompany.io"
|
||
|
|
|
||
|
|
def test_password_with_special_characters(self):
|
||
|
|
"""Verify that passwords with special characters are handled."""
|
||
|
|
special_password = "p@$$w0rd!#%^&*()"
|
||
|
|
payload = build_poste_initial_setup_step(
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
admin_password=special_password,
|
||
|
|
)
|
||
|
|
|
||
|
|
assert payload["inputs"]["admin_password"] == special_password
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
class TestCreatePosteInitialSetupTask:
|
||
|
|
"""Tests for the create_poste_initial_setup_task function."""
|
||
|
|
|
||
|
|
async def test_persists_playwright_task(self, db: AsyncSession, test_tenant: Tenant):
|
||
|
|
"""Verify that create_poste_initial_setup_task persists a PLAYWRIGHT task."""
|
||
|
|
agent_id = uuid.uuid4()
|
||
|
|
|
||
|
|
task = await create_poste_initial_setup_task(
|
||
|
|
db=db,
|
||
|
|
tenant_id=test_tenant.id,
|
||
|
|
agent_id=agent_id,
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert task.id is not None
|
||
|
|
assert task.tenant_id == test_tenant.id
|
||
|
|
assert task.agent_id == agent_id
|
||
|
|
assert task.type == "PLAYWRIGHT"
|
||
|
|
assert task.status == TaskStatus.PENDING.value
|
||
|
|
|
||
|
|
async def test_task_payload_contains_scenario(
|
||
|
|
self, db: AsyncSession, test_tenant: Tenant
|
||
|
|
):
|
||
|
|
"""Verify that the task payload contains the scenario field."""
|
||
|
|
task = await create_poste_initial_setup_task(
|
||
|
|
db=db,
|
||
|
|
tenant_id=test_tenant.id,
|
||
|
|
agent_id=uuid.uuid4(),
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert task.payload["scenario"] == "poste_initial_setup"
|
||
|
|
|
||
|
|
async def test_task_payload_contains_inputs(
|
||
|
|
self, db: AsyncSession, test_tenant: Tenant
|
||
|
|
):
|
||
|
|
"""Verify that the task payload contains the inputs field."""
|
||
|
|
task = await create_poste_initial_setup_task(
|
||
|
|
db=db,
|
||
|
|
tenant_id=test_tenant.id,
|
||
|
|
agent_id=uuid.uuid4(),
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="postmaster@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
inputs = task.payload["inputs"]
|
||
|
|
assert inputs["base_url"] == "https://mail.example.com"
|
||
|
|
assert inputs["admin_email"] == "postmaster@example.com"
|
||
|
|
|
||
|
|
async def test_task_with_password(self, db: AsyncSession, test_tenant: Tenant):
|
||
|
|
"""Verify that admin_password is included when provided."""
|
||
|
|
task = await create_poste_initial_setup_task(
|
||
|
|
db=db,
|
||
|
|
tenant_id=test_tenant.id,
|
||
|
|
agent_id=uuid.uuid4(),
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
admin_password="test-password-123",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert task.payload["inputs"]["admin_password"] == "test-password-123"
|
||
|
|
|
||
|
|
async def test_task_without_password(self, db: AsyncSession, test_tenant: Tenant):
|
||
|
|
"""Verify that admin_password is omitted when not provided."""
|
||
|
|
task = await create_poste_initial_setup_task(
|
||
|
|
db=db,
|
||
|
|
tenant_id=test_tenant.id,
|
||
|
|
agent_id=uuid.uuid4(),
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert "admin_password" not in task.payload["inputs"]
|
||
|
|
|
||
|
|
async def test_task_options_contain_allowed_domains(
|
||
|
|
self, db: AsyncSession, test_tenant: Tenant
|
||
|
|
):
|
||
|
|
"""Verify that task options include allowed_domains."""
|
||
|
|
task = await create_poste_initial_setup_task(
|
||
|
|
db=db,
|
||
|
|
tenant_id=test_tenant.id,
|
||
|
|
agent_id=uuid.uuid4(),
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert task.payload["options"]["allowed_domains"] == ["mail.example.com"]
|
||
|
|
|
||
|
|
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_poste_initial_setup_task(
|
||
|
|
db=db,
|
||
|
|
tenant_id=test_tenant.id,
|
||
|
|
agent_id=uuid.uuid4(),
|
||
|
|
base_url="https://mail.example.com",
|
||
|
|
admin_email="admin@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 == "PLAYWRIGHT"
|
||
|
|
assert retrieved_task.tenant_id == test_tenant.id
|
||
|
|
|
||
|
|
|
||
|
|
class TestPosteConstants:
|
||
|
|
"""Tests for module constants."""
|
||
|
|
|
||
|
|
def test_poste_stack_dir(self):
|
||
|
|
"""Verify POSTE_STACK_DIR is set correctly."""
|
||
|
|
assert POSTE_STACK_DIR == "/opt/letsbe/stacks/poste"
|