letsbe
/
letsbe-orchestrator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add Nextcloud initial setup endpoint via Playwright
Build and Push Docker Image / test (push) Successful in 55s Details
Build and Push Docker Image / build (push) Successful in 1m16s Details 

Add POST /api/v1/tenants/{tenant_id}/nextcloud/setup endpoint that:
- Creates a PLAYWRIGHT task for Nextcloud initial admin setup
- Validates tenant has domain configured
- Auto-resolves online agent for tenant
- Performs health check against Nextcloud before creating task
- Returns 409 if Nextcloud is unavailable

Changes:
- Add httpx for health checks
- Add build_nextcloud_initial_setup_step() and create_nextcloud_initial_setup_task()
- Add NextcloudInitialSetupRequest schema with username/password validation
- Add check_nextcloud_availability() helper for health checks
- Add comprehensive unit tests (42 tests total)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Matt 2025-12-08 20:20:31 +01:00
parent 4dac6d7e4e
commit 4bafefdfaf
5 changed files with 699 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

97
app/playbooks/nextcloud.py
View File

@ -1,13 +1,15 @@
"""Nextcloud deployment playbook.
Defines the steps required to set Nextcloud domain on a tenant server
that already has stacks and env templates under /opt/letsbe.
Defines the steps required to:
1. Set Nextcloud domain on a tenant server (v2: via NEXTCLOUD_SET_DOMAIN task)
2. Perform initial setup via Playwright automation (create admin account)
v2: Configures Nextcloud via NEXTCLOUD_SET_DOMAIN task, then reloads the stack.
Tenant servers must have stacks and env templates under /opt/letsbe.
"""
import uuid
from typing import Any
from urllib.parse import urlparse
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession
@ -28,6 +30,95 @@ class CompositeStep(BaseModel):
NEXTCLOUD_STACK_DIR = "/opt/letsbe/stacks/nextcloud"
# =============================================================================
# Initial Setup via Playwright
# =============================================================================
def build_nextcloud_initial_setup_step(
*,
base_url: str,
admin_username: str,
admin_password: str,
) -> dict[str, Any]:
"""
Build a PLAYWRIGHT task payload for Nextcloud initial setup.
This creates the admin account on a fresh Nextcloud installation.
Args:
base_url: The base URL for Nextcloud (e.g., "https://cloud.example.com")
admin_username: Username for the admin account
admin_password: Password for the admin account
Returns:
Task payload dict with type="PLAYWRIGHT"
"""
# Extract domain from URL for allowlist
parsed = urlparse(base_url)
allowed_domain = parsed.netloc # e.g., "cloud.example.com"
return {
"scenario": "nextcloud.initial_setup",
"inputs": {
"base_url": base_url,
"admin_username": admin_username,
"admin_password": admin_password,
"allowed_domains": [allowed_domain],
},
"timeout": 120,
}
async def create_nextcloud_initial_setup_task(
*,
db: AsyncSession,
tenant_id: uuid.UUID,
agent_id: uuid.UUID,
base_url: str,
admin_username: str,
admin_password: str,
) -> Task:
"""
Create and persist a PLAYWRIGHT task for Nextcloud initial setup.
Args:
db: Async database session
tenant_id: UUID of the tenant
agent_id: UUID of the agent to assign the task to
base_url: The base URL for Nextcloud
admin_username: Username for the admin account
admin_password: Password for the admin account
Returns:
The created Task object with type="PLAYWRIGHT"
"""
payload = build_nextcloud_initial_setup_step(
base_url=base_url,
admin_username=admin_username,
admin_password=admin_password,
)
task = Task(
tenant_id=tenant_id,
agent_id=agent_id,
type="PLAYWRIGHT",
payload=payload,
status=TaskStatus.PENDING.value,
)
db.add(task)
await db.commit()
await db.refresh(task)
return task
# =============================================================================
# Set Domain via NEXTCLOUD_SET_DOMAIN
# =============================================================================
def build_nextcloud_set_domain_steps(*, public_url: str, pull: bool) -> list[CompositeStep]:
"""
Build the sequence of steps required to set Nextcloud domain (v2).

166
app/routes/playbooks.py
View File

@ -1,9 +1,11 @@
"""Playbook endpoints for triggering infrastructure automation."""
import re
import uuid
import httpx
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import select
from app.db import AsyncSessionDep
@ -11,11 +13,17 @@ from app.models.agent import Agent
from app.models.task import Task
from app.models.tenant import Tenant
from app.playbooks.chatwoot import create_chatwoot_setup_task
from app.playbooks.nextcloud import create_nextcloud_set_domain_task
from app.playbooks.nextcloud import (
create_nextcloud_initial_setup_task,
create_nextcloud_set_domain_task,
)
from app.schemas.task import TaskResponse
router = APIRouter(prefix="/tenants/{tenant_id}", tags=["Playbooks"])
# Health check timeout for Nextcloud availability
NEXTCLOUD_HEALTH_CHECK_TIMEOUT = 10.0
class ChatwootSetupRequest(BaseModel):
"""Request body for Chatwoot setup playbook."""
@ -42,6 +50,32 @@ class NextcloudSetDomainRequest(BaseModel):
)
class NextcloudInitialSetupRequest(BaseModel):
"""Request body for Nextcloud initial setup via Playwright."""
admin_username: str = Field(
...,
min_length=3,
max_length=64,
description="Username for the Nextcloud admin account",
)
admin_password: str = Field(
...,
min_length=8,
description="Password for the Nextcloud admin account (minimum 8 characters)",
)
@field_validator("admin_username")
@classmethod
def validate_username(cls, v: str) -> str:
"""Validate that username is alphanumeric (with underscores allowed)."""
if not re.match(r"^[a-zA-Z][a-zA-Z0-9_]*$", v):
raise ValueError(
"Username must start with a letter and contain only letters, numbers, and underscores"
)
return v
# --- Helper functions ---
@ -70,6 +104,46 @@ async def get_online_agent_for_tenant(
return result.scalar_one_or_none()
async def check_nextcloud_availability(base_url: str) -> bool:
"""
Check if Nextcloud is available at the given URL.
Performs a health check against {base_url}/status.php which returns
JSON with installation status on a healthy Nextcloud instance.
Args:
base_url: The base URL for Nextcloud (e.g., "https://cloud.example.com")
Returns:
True if Nextcloud responds with a valid status, False otherwise
"""
status_url = f"{base_url.rstrip('/')}/status.php"
try:
async with httpx.AsyncClient(
timeout=NEXTCLOUD_HEALTH_CHECK_TIMEOUT,
follow_redirects=True,
) as client:
response = await client.get(status_url)
# Nextcloud status.php returns 200 with JSON when healthy
if response.status_code == 200:
# Try to parse as JSON - valid Nextcloud returns {"installed":true,...}
try:
data = response.json()
# If we get valid JSON with "installed" key, Nextcloud is available
return "installed" in data
except (ValueError, KeyError):
# Not valid JSON, but 200 OK might still indicate availability
return True
return False
except (httpx.RequestError, httpx.TimeoutException):
# Connection failed, timeout, or other network error
return False
# --- Route handlers ---
@ -181,3 +255,91 @@ async def nextcloud_set_domain(
)
return task
@router.post(
"/nextcloud/setup",
response_model=TaskResponse,
status_code=status.HTTP_201_CREATED,
)
async def nextcloud_initial_setup(
tenant_id: uuid.UUID,
request: NextcloudInitialSetupRequest,
db: AsyncSessionDep,
) -> Task:
"""
Perform initial Nextcloud setup via Playwright browser automation.
Creates a PLAYWRIGHT task that navigates to the Nextcloud installation page
and creates the admin account. This should only be called once on a fresh
Nextcloud installation.
**Prerequisites:**
- Tenant must have a domain configured
- Nextcloud must be deployed and accessible at `https://cloud.{domain}`
- An online agent must be registered for the tenant
**Health Check:**
Before creating the task, the endpoint checks if Nextcloud is available
at `https://cloud.{domain}/status.php`. If not reachable, returns 409.
## Request Body
- **admin_username**: Username for the admin account (3-64 chars, alphanumeric)
- **admin_password**: Password for the admin account (minimum 8 characters)
## Response
Returns the created Task with type="PLAYWRIGHT" and status="pending".
## Errors
- **404**: Tenant not found
- **400**: Tenant has no domain configured
- **404**: No online agent found for tenant
- **409**: Nextcloud not available (health check failed)
- **422**: Invalid username or password format
"""
# Validate tenant exists
tenant = await get_tenant_by_id(db, tenant_id)
if tenant is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Tenant {tenant_id} not found",
)
# Validate tenant has domain configured
if not tenant.domain:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Tenant {tenant_id} has no domain configured",
)
# Auto-resolve agent (first online agent for tenant)
agent = await get_online_agent_for_tenant(db, tenant_id)
if agent is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"No online agent found for tenant {tenant_id}",
)
# Construct Nextcloud base URL from tenant domain
base_url = f"https://cloud.{tenant.domain}"
# Health check: verify Nextcloud is accessible
is_available = await check_nextcloud_availability(base_url)
if not is_available:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Nextcloud not available at {base_url}. "
"Ensure Nextcloud is deployed and accessible before running setup.",
)
# Create the PLAYWRIGHT task
task = await create_nextcloud_initial_setup_task(
db=db,
tenant_id=tenant_id,
agent_id=agent.id,
base_url=base_url,
admin_username=request.admin_username,
admin_password=request.admin_password,
)
return task

3
requirements.txt
View File

@ -14,6 +14,9 @@ pydantic-settings>=2.1.0
# Utilities
python-dotenv>=1.0.0
# HTTP Client
httpx>=0.26.0
# Testing
pytest>=8.0.0
pytest-asyncio>=0.23.0

287
tests/routes/test_nextcloud_routes.py
View File

@ -1,9 +1,12 @@
"""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
@ -12,7 +15,9 @@ 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,
)
@ -202,3 +207,285 @@ class TestNextcloudSetDomainEndpoint:
# 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"
assert inputs["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

151
tests/test_playbooks_nextcloud.py
View File

@ -12,11 +12,162 @@ from app.models.tenant import Tenant
from app.playbooks.nextcloud import (
NEXTCLOUD_STACK_DIR,
CompositeStep,
build_nextcloud_initial_setup_step,
build_nextcloud_set_domain_steps,
create_nextcloud_initial_setup_task,
create_nextcloud_set_domain_task,
)
# =============================================================================
# Tests for Playwright Initial Setup
# =============================================================================
class TestBuildNextcloudInitialSetupStep:
"""Tests for the build_nextcloud_initial_setup_step function."""
def test_returns_playwright_payload(self):
"""Verify that build_nextcloud_initial_setup_step returns correct structure."""
payload = build_nextcloud_initial_setup_step(
base_url="https://cloud.example.com",
admin_username="admin",
admin_password="securepassword123",
)
assert payload["scenario"] == "nextcloud.initial_setup"
assert "inputs" in payload
assert "timeout" in payload
def test_inputs_contain_required_fields(self):
"""Verify that inputs contain all required fields."""
payload = build_nextcloud_initial_setup_step(
base_url="https://cloud.example.com",
admin_username="admin",
admin_password="securepassword123",
)
inputs = payload["inputs"]
assert inputs["base_url"] == "https://cloud.example.com"
assert inputs["admin_username"] == "admin"
assert inputs["admin_password"] == "securepassword123"
assert "allowed_domains" in inputs
def test_allowed_domains_extracted_from_url(self):
"""Verify that allowed_domains is extracted from base_url."""
payload = build_nextcloud_initial_setup_step(
base_url="https://cloud.example.com",
admin_username="admin",
admin_password="password",
)
assert payload["inputs"]["allowed_domains"] == ["cloud.example.com"]
def test_allowed_domains_with_port(self):
"""Verify that allowed_domains handles URLs with ports."""
payload = build_nextcloud_initial_setup_step(
base_url="https://cloud.example.com:8443",
admin_username="admin",
admin_password="password",
)
assert payload["inputs"]["allowed_domains"] == ["cloud.example.com:8443"]
def test_timeout_is_set(self):
"""Verify that timeout is set in the payload."""
payload = build_nextcloud_initial_setup_step(
base_url="https://cloud.example.com",
admin_username="admin",
admin_password="password",
)
assert payload["timeout"] == 120
@pytest.mark.asyncio
class TestCreateNextcloudInitialSetupTask:
"""Tests for the create_nextcloud_initial_setup_task function."""
async def test_persists_playwright_task(self, db: AsyncSession, test_tenant: Tenant):
"""Verify that create_nextcloud_initial_setup_task persists a PLAYWRIGHT task."""
agent_id = uuid.uuid4()
task = await create_nextcloud_initial_setup_task(
db=db,
tenant_id=test_tenant.id,
agent_id=agent_id,
base_url="https://cloud.example.com",
admin_username="admin",
admin_password="securepassword123",
)
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_nextcloud_initial_setup_task(
db=db,
tenant_id=test_tenant.id,
agent_id=uuid.uuid4(),
base_url="https://cloud.example.com",
admin_username="admin",
admin_password="password",
)
assert task.payload["scenario"] == "nextcloud.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_nextcloud_initial_setup_task(
db=db,
tenant_id=test_tenant.id,
agent_id=uuid.uuid4(),
base_url="https://cloud.example.com",
admin_username="testadmin",
admin_password="testpassword123",
)
inputs = task.payload["inputs"]
assert inputs["base_url"] == "https://cloud.example.com"
assert inputs["admin_username"] == "testadmin"
assert inputs["admin_password"] == "testpassword123"
assert inputs["allowed_domains"] == ["cloud.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_nextcloud_initial_setup_task(
db=db,
tenant_id=test_tenant.id,
agent_id=uuid.uuid4(),
base_url="https://cloud.example.com",
admin_username="admin",
admin_password="password",
)
# 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
# =============================================================================
# Tests for Set Domain Playbook
# =============================================================================
class TestBuildNextcloudSetDomainSteps:
"""Tests for the build_nextcloud_set_domain_steps function."""