493 lines
17 KiB
Python
493 lines
17 KiB
Python
"""Tests for Nextcloud playbook routes."""
|
|
|
|
import uuid
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from fastapi import HTTPException
|
|
from pydantic import ValidationError
|
|
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.playbooks.nextcloud import NEXTCLOUD_STACK_DIR
|
|
from app.routes.playbooks import (
|
|
NextcloudInitialSetupRequest,
|
|
NextcloudSetDomainRequest,
|
|
nextcloud_initial_setup,
|
|
nextcloud_set_domain,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestNextcloudSetDomainEndpoint:
|
|
"""Tests for the nextcloud_set_domain endpoint."""
|
|
|
|
async def test_happy_path_creates_task(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent_for_tenant: Agent
|
|
):
|
|
"""POST /tenants/{id}/nextcloud/set-domain returns 201 with COMPOSITE task."""
|
|
request = NextcloudSetDomainRequest(
|
|
public_url="https://cloud.example.com",
|
|
pull=False,
|
|
)
|
|
|
|
task = await nextcloud_set_domain(
|
|
tenant_id=test_tenant.id, request=request, db=db
|
|
)
|
|
|
|
assert task.id is not None
|
|
assert task.tenant_id == test_tenant.id
|
|
assert task.agent_id == test_agent_for_tenant.id
|
|
assert task.type == "COMPOSITE"
|
|
assert task.status == "pending"
|
|
|
|
async def test_task_has_both_steps(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent_for_tenant: Agent
|
|
):
|
|
"""Verify response task payload contains both steps in correct order."""
|
|
request = NextcloudSetDomainRequest(
|
|
public_url="https://cloud.example.com",
|
|
pull=True,
|
|
)
|
|
|
|
task = await nextcloud_set_domain(
|
|
tenant_id=test_tenant.id, request=request, db=db
|
|
)
|
|
|
|
assert "steps" in task.payload
|
|
assert len(task.payload["steps"]) == 2
|
|
|
|
# First step: NEXTCLOUD_SET_DOMAIN
|
|
step0 = task.payload["steps"][0]
|
|
assert step0["type"] == "NEXTCLOUD_SET_DOMAIN"
|
|
assert step0["payload"]["public_url"] == "https://cloud.example.com"
|
|
|
|
# Second step: DOCKER_RELOAD
|
|
step1 = task.payload["steps"][1]
|
|
assert step1["type"] == "DOCKER_RELOAD"
|
|
assert step1["payload"]["compose_dir"] == NEXTCLOUD_STACK_DIR
|
|
assert step1["payload"]["pull"] is True
|
|
|
|
async def test_pull_flag_default_false(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent_for_tenant: Agent
|
|
):
|
|
"""Verify pull defaults to False when not specified."""
|
|
request = NextcloudSetDomainRequest(
|
|
public_url="https://cloud.example.com",
|
|
)
|
|
|
|
task = await nextcloud_set_domain(
|
|
tenant_id=test_tenant.id, request=request, db=db
|
|
)
|
|
|
|
step = task.payload["steps"][1]
|
|
assert step["payload"]["pull"] is False
|
|
|
|
async def test_tenant_not_found_returns_404(self, db: AsyncSession):
|
|
"""Non-existent tenant_id returns 404."""
|
|
fake_tenant_id = uuid.uuid4()
|
|
request = NextcloudSetDomainRequest(
|
|
public_url="https://cloud.example.com",
|
|
pull=False,
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await nextcloud_set_domain(
|
|
tenant_id=fake_tenant_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_no_online_agent_returns_404(
|
|
self, db: AsyncSession, test_tenant: Tenant
|
|
):
|
|
"""Tenant with no online agent returns 404."""
|
|
request = NextcloudSetDomainRequest(
|
|
public_url="https://cloud.example.com",
|
|
pull=False,
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await nextcloud_set_domain(
|
|
tenant_id=test_tenant.id, request=request, db=db
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "No online agent found" in str(exc_info.value.detail)
|
|
|
|
async def test_offline_agent_not_resolved(
|
|
self, db: AsyncSession, test_tenant: Tenant
|
|
):
|
|
"""Offline agent should not be auto-resolved."""
|
|
# Create an offline agent for the tenant
|
|
offline_agent = Agent(
|
|
id=uuid.uuid4(),
|
|
tenant_id=test_tenant.id,
|
|
name="offline-agent",
|
|
version="1.0.0",
|
|
status="offline",
|
|
token="offline-token",
|
|
)
|
|
db.add(offline_agent)
|
|
await db.commit()
|
|
|
|
request = NextcloudSetDomainRequest(
|
|
public_url="https://cloud.example.com",
|
|
pull=False,
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await nextcloud_set_domain(
|
|
tenant_id=test_tenant.id, request=request, db=db
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "No online agent found" in str(exc_info.value.detail)
|
|
|
|
async def test_task_persisted_to_database(
|
|
self, db: AsyncSession, test_tenant: Tenant, test_agent_for_tenant: Agent
|
|
):
|
|
"""Verify the task is actually persisted and can be retrieved."""
|
|
request = NextcloudSetDomainRequest(
|
|
public_url="https://cloud.example.com",
|
|
pull=False,
|
|
)
|
|
|
|
task = await nextcloud_set_domain(
|
|
tenant_id=test_tenant.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 == "COMPOSITE"
|
|
assert retrieved_task.tenant_id == test_tenant.id
|
|
assert retrieved_task.agent_id == test_agent_for_tenant.id
|
|
|
|
async def test_auto_resolves_first_online_agent(
|
|
self, db: AsyncSession, test_tenant: Tenant
|
|
):
|
|
"""Verify that the first online agent is auto-resolved."""
|
|
# Create two agents for the tenant - one offline, one online
|
|
offline_agent = Agent(
|
|
id=uuid.uuid4(),
|
|
tenant_id=test_tenant.id,
|
|
name="offline-agent",
|
|
version="1.0.0",
|
|
status="offline",
|
|
token="offline-token",
|
|
)
|
|
online_agent = Agent(
|
|
id=uuid.uuid4(),
|
|
tenant_id=test_tenant.id,
|
|
name="online-agent",
|
|
version="1.0.0",
|
|
status="online",
|
|
token="online-token",
|
|
)
|
|
db.add(offline_agent)
|
|
db.add(online_agent)
|
|
await db.commit()
|
|
|
|
request = NextcloudSetDomainRequest(
|
|
public_url="https://cloud.example.com",
|
|
pull=False,
|
|
)
|
|
|
|
task = await nextcloud_set_domain(
|
|
tenant_id=test_tenant.id, request=request, db=db
|
|
)
|
|
|
|
# Should have resolved to the online agent
|
|
assert task.agent_id == online_agent.id
|
|
|
|
|
|
# =============================================================================
|
|
# Fixtures for Initial Setup Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_tenant_with_domain(db: AsyncSession) -> Tenant:
|
|
"""Create a test tenant with domain configured."""
|
|
tenant = Tenant(
|
|
id=uuid.uuid4(),
|
|
name="Test Tenant With Domain",
|
|
domain="example.com",
|
|
)
|
|
db.add(tenant)
|
|
await db.commit()
|
|
await db.refresh(tenant)
|
|
return tenant
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_agent_for_tenant_with_domain(
|
|
db: AsyncSession, test_tenant_with_domain: Tenant
|
|
) -> Agent:
|
|
"""Create a test agent linked to test_tenant_with_domain with online status."""
|
|
agent = Agent(
|
|
id=uuid.uuid4(),
|
|
tenant_id=test_tenant_with_domain.id,
|
|
name="test-agent-for-tenant-domain",
|
|
version="1.0.0",
|
|
status="online",
|
|
token="test-token-domain",
|
|
)
|
|
db.add(agent)
|
|
await db.commit()
|
|
await db.refresh(agent)
|
|
return agent
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for Nextcloud Initial Setup Endpoint
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestNextcloudInitialSetupEndpoint:
|
|
"""Tests for the nextcloud_initial_setup endpoint."""
|
|
|
|
@patch("app.routes.playbooks.check_nextcloud_availability")
|
|
async def test_happy_path_creates_task(
|
|
self,
|
|
mock_health_check: AsyncMock,
|
|
db: AsyncSession,
|
|
test_tenant_with_domain: Tenant,
|
|
test_agent_for_tenant_with_domain: Agent,
|
|
):
|
|
"""POST /tenants/{id}/nextcloud/setup returns 201 with PLAYWRIGHT task."""
|
|
mock_health_check.return_value = True
|
|
|
|
request = NextcloudInitialSetupRequest(
|
|
admin_username="admin",
|
|
admin_password="securepassword123",
|
|
)
|
|
|
|
task = await nextcloud_initial_setup(
|
|
tenant_id=test_tenant_with_domain.id, request=request, db=db
|
|
)
|
|
|
|
assert task.id is not None
|
|
assert task.tenant_id == test_tenant_with_domain.id
|
|
assert task.agent_id == test_agent_for_tenant_with_domain.id
|
|
assert task.type == "PLAYWRIGHT"
|
|
assert task.status == "pending"
|
|
|
|
@patch("app.routes.playbooks.check_nextcloud_availability")
|
|
async def test_task_has_correct_payload(
|
|
self,
|
|
mock_health_check: AsyncMock,
|
|
db: AsyncSession,
|
|
test_tenant_with_domain: Tenant,
|
|
test_agent_for_tenant_with_domain: Agent,
|
|
):
|
|
"""Verify response task payload contains scenario and inputs."""
|
|
mock_health_check.return_value = True
|
|
|
|
request = NextcloudInitialSetupRequest(
|
|
admin_username="myadmin",
|
|
admin_password="mypassword123",
|
|
)
|
|
|
|
task = await nextcloud_initial_setup(
|
|
tenant_id=test_tenant_with_domain.id, request=request, db=db
|
|
)
|
|
|
|
assert task.payload["scenario"] == "nextcloud_initial_setup"
|
|
inputs = task.payload["inputs"]
|
|
assert inputs["base_url"] == "https://cloud.example.com"
|
|
assert inputs["admin_username"] == "myadmin"
|
|
assert inputs["admin_password"] == "mypassword123"
|
|
# allowed_domains should be in options, not inputs
|
|
assert task.payload["options"]["allowed_domains"] == ["cloud.example.com"]
|
|
|
|
async def test_tenant_not_found_returns_404(self, db: AsyncSession):
|
|
"""Non-existent tenant_id returns 404."""
|
|
fake_tenant_id = uuid.uuid4()
|
|
request = NextcloudInitialSetupRequest(
|
|
admin_username="admin",
|
|
admin_password="password123",
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await nextcloud_initial_setup(
|
|
tenant_id=fake_tenant_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_tenant_without_domain_returns_400(
|
|
self, db: AsyncSession, test_tenant: Tenant
|
|
):
|
|
"""Tenant without domain configured returns 400."""
|
|
request = NextcloudInitialSetupRequest(
|
|
admin_username="admin",
|
|
admin_password="password123",
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await nextcloud_initial_setup(
|
|
tenant_id=test_tenant.id, request=request, db=db
|
|
)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "no domain configured" in str(exc_info.value.detail)
|
|
|
|
async def test_no_online_agent_returns_404(
|
|
self, db: AsyncSession, test_tenant_with_domain: Tenant
|
|
):
|
|
"""Tenant with no online agent returns 404."""
|
|
request = NextcloudInitialSetupRequest(
|
|
admin_username="admin",
|
|
admin_password="password123",
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await nextcloud_initial_setup(
|
|
tenant_id=test_tenant_with_domain.id, request=request, db=db
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "No online agent found" in str(exc_info.value.detail)
|
|
|
|
@patch("app.routes.playbooks.check_nextcloud_availability")
|
|
async def test_nextcloud_unavailable_returns_409(
|
|
self,
|
|
mock_health_check: AsyncMock,
|
|
db: AsyncSession,
|
|
test_tenant_with_domain: Tenant,
|
|
test_agent_for_tenant_with_domain: Agent,
|
|
):
|
|
"""Health check failure returns 409."""
|
|
mock_health_check.return_value = False
|
|
|
|
request = NextcloudInitialSetupRequest(
|
|
admin_username="admin",
|
|
admin_password="password123",
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await nextcloud_initial_setup(
|
|
tenant_id=test_tenant_with_domain.id, request=request, db=db
|
|
)
|
|
|
|
assert exc_info.value.status_code == 409
|
|
assert "Nextcloud not available" in str(exc_info.value.detail)
|
|
|
|
@patch("app.routes.playbooks.check_nextcloud_availability")
|
|
async def test_health_check_called_with_correct_url(
|
|
self,
|
|
mock_health_check: AsyncMock,
|
|
db: AsyncSession,
|
|
test_tenant_with_domain: Tenant,
|
|
test_agent_for_tenant_with_domain: Agent,
|
|
):
|
|
"""Verify health check is called with correct URL."""
|
|
mock_health_check.return_value = True
|
|
|
|
request = NextcloudInitialSetupRequest(
|
|
admin_username="admin",
|
|
admin_password="password123",
|
|
)
|
|
|
|
await nextcloud_initial_setup(
|
|
tenant_id=test_tenant_with_domain.id, request=request, db=db
|
|
)
|
|
|
|
mock_health_check.assert_called_once_with("https://cloud.example.com")
|
|
|
|
def test_username_too_short_raises_validation_error(self):
|
|
"""Username less than 3 characters raises ValidationError."""
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
NextcloudInitialSetupRequest(
|
|
admin_username="ab", # Too short
|
|
admin_password="password123",
|
|
)
|
|
|
|
assert "admin_username" in str(exc_info.value)
|
|
|
|
def test_username_invalid_chars_raises_validation_error(self):
|
|
"""Username with invalid characters raises ValidationError."""
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
NextcloudInitialSetupRequest(
|
|
admin_username="admin@user", # Invalid char @
|
|
admin_password="password123",
|
|
)
|
|
|
|
assert "admin_username" in str(exc_info.value)
|
|
|
|
def test_username_starting_with_number_raises_validation_error(self):
|
|
"""Username starting with number raises ValidationError."""
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
NextcloudInitialSetupRequest(
|
|
admin_username="123admin", # Starts with number
|
|
admin_password="password123",
|
|
)
|
|
|
|
assert "admin_username" in str(exc_info.value)
|
|
|
|
def test_password_too_short_raises_validation_error(self):
|
|
"""Password less than 8 characters raises ValidationError."""
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
NextcloudInitialSetupRequest(
|
|
admin_username="admin",
|
|
admin_password="short", # Too short
|
|
)
|
|
|
|
assert "admin_password" in str(exc_info.value)
|
|
|
|
def test_valid_username_with_underscore(self):
|
|
"""Username with underscore is valid."""
|
|
request = NextcloudInitialSetupRequest(
|
|
admin_username="admin_user",
|
|
admin_password="password123",
|
|
)
|
|
assert request.admin_username == "admin_user"
|
|
|
|
def test_valid_username_with_numbers(self):
|
|
"""Username with numbers (not at start) is valid."""
|
|
request = NextcloudInitialSetupRequest(
|
|
admin_username="admin123",
|
|
admin_password="password123",
|
|
)
|
|
assert request.admin_username == "admin123"
|
|
|
|
@patch("app.routes.playbooks.check_nextcloud_availability")
|
|
async def test_task_persisted_to_database(
|
|
self,
|
|
mock_health_check: AsyncMock,
|
|
db: AsyncSession,
|
|
test_tenant_with_domain: Tenant,
|
|
test_agent_for_tenant_with_domain: Agent,
|
|
):
|
|
"""Verify the task is actually persisted and can be retrieved."""
|
|
mock_health_check.return_value = True
|
|
|
|
request = NextcloudInitialSetupRequest(
|
|
admin_username="admin",
|
|
admin_password="password123",
|
|
)
|
|
|
|
task = await nextcloud_initial_setup(
|
|
tenant_id=test_tenant_with_domain.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 == "PLAYWRIGHT"
|
|
assert retrieved_task.tenant_id == test_tenant_with_domain.id
|
|
assert retrieved_task.agent_id == test_agent_for_tenant_with_domain.id
|