feat: add Nextcloud set-domain playbook v2
Add endpoint POST /tenants/{tenant_id}/nextcloud/set-domain that creates
a COMPOSITE task with two steps:
1. NEXTCLOUD_SET_DOMAIN - configures Nextcloud domain via occ commands
2. DOCKER_RELOAD - restarts the Nextcloud stack
Features:
- Auto-resolves first online agent for tenant
- Configurable pull flag for image updates
- Full test coverage (unit + integration tests)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
f40c5fcc69
commit
06f58ca18b
|
|
@ -9,9 +9,15 @@ from app.playbooks.chatwoot import (
|
|||
build_chatwoot_setup_steps,
|
||||
create_chatwoot_setup_task,
|
||||
)
|
||||
from app.playbooks.nextcloud import (
|
||||
build_nextcloud_set_domain_steps,
|
||||
create_nextcloud_set_domain_task,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"CompositeStep",
|
||||
"build_chatwoot_setup_steps",
|
||||
"create_chatwoot_setup_task",
|
||||
"build_nextcloud_set_domain_steps",
|
||||
"create_nextcloud_set_domain_task",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,99 @@
|
|||
"""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.
|
||||
|
||||
v2: Configures Nextcloud via NEXTCLOUD_SET_DOMAIN task, then reloads the stack.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.task import Task, TaskStatus
|
||||
|
||||
|
||||
class CompositeStep(BaseModel):
|
||||
"""A single step in a composite playbook."""
|
||||
|
||||
type: str = Field(..., description="Task type (e.g., ENV_UPDATE, DOCKER_RELOAD)")
|
||||
payload: dict[str, Any] = Field(
|
||||
default_factory=dict, description="Payload for this step"
|
||||
)
|
||||
|
||||
|
||||
# LetsBe standard paths
|
||||
NEXTCLOUD_STACK_DIR = "/opt/letsbe/stacks/nextcloud"
|
||||
|
||||
|
||||
def build_nextcloud_set_domain_steps(*, public_url: str, pull: bool) -> list[CompositeStep]:
|
||||
"""
|
||||
Build the sequence of steps required to set Nextcloud domain (v2).
|
||||
|
||||
Args:
|
||||
public_url: The public URL for Nextcloud (e.g., "https://cloud.example.com")
|
||||
pull: Whether to pull images before reloading the stack
|
||||
|
||||
Returns:
|
||||
List of 2 CompositeStep objects:
|
||||
1. NEXTCLOUD_SET_DOMAIN - configures Nextcloud via occ commands
|
||||
2. DOCKER_RELOAD - restarts the Nextcloud stack
|
||||
"""
|
||||
steps = [
|
||||
# Step 1: Configure Nextcloud domain via occ
|
||||
CompositeStep(
|
||||
type="NEXTCLOUD_SET_DOMAIN",
|
||||
payload={
|
||||
"public_url": public_url,
|
||||
},
|
||||
),
|
||||
# Step 2: Reload Docker stack
|
||||
CompositeStep(
|
||||
type="DOCKER_RELOAD",
|
||||
payload={
|
||||
"compose_dir": NEXTCLOUD_STACK_DIR,
|
||||
"pull": pull,
|
||||
},
|
||||
),
|
||||
]
|
||||
return steps
|
||||
|
||||
|
||||
async def create_nextcloud_set_domain_task(
|
||||
*,
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
agent_id: uuid.UUID,
|
||||
public_url: str,
|
||||
pull: bool,
|
||||
) -> Task:
|
||||
"""
|
||||
Create and persist a COMPOSITE task for Nextcloud set-domain.
|
||||
|
||||
Args:
|
||||
db: Async database session
|
||||
tenant_id: UUID of the tenant
|
||||
agent_id: UUID of the agent to assign the task to
|
||||
public_url: The public URL for Nextcloud
|
||||
pull: Whether to pull images before reloading
|
||||
|
||||
Returns:
|
||||
The created Task object with type="COMPOSITE"
|
||||
"""
|
||||
steps = build_nextcloud_set_domain_steps(public_url=public_url, pull=pull)
|
||||
|
||||
task = Task(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
type="COMPOSITE",
|
||||
payload={"steps": [step.model_dump() for step in steps]},
|
||||
status=TaskStatus.PENDING.value,
|
||||
)
|
||||
|
||||
db.add(task)
|
||||
await db.commit()
|
||||
await db.refresh(task)
|
||||
|
||||
return task
|
||||
|
|
@ -11,6 +11,7 @@ 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.schemas.task import TaskResponse
|
||||
|
||||
router = APIRouter(prefix="/tenants/{tenant_id}", tags=["Playbooks"])
|
||||
|
|
@ -27,6 +28,20 @@ class ChatwootSetupRequest(BaseModel):
|
|||
)
|
||||
|
||||
|
||||
class NextcloudSetDomainRequest(BaseModel):
|
||||
"""Request body for Nextcloud set-domain playbook."""
|
||||
|
||||
public_url: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="Public URL for Nextcloud (e.g., https://cloud.example.com)",
|
||||
)
|
||||
pull: bool = Field(
|
||||
False,
|
||||
description="Whether to pull images before reloading the stack",
|
||||
)
|
||||
|
||||
|
||||
# --- Helper functions ---
|
||||
|
||||
|
||||
|
|
@ -42,6 +57,19 @@ async def get_agent_by_id(db: AsyncSessionDep, agent_id: uuid.UUID) -> Agent | N
|
|||
return result.scalar_one_or_none()
|
||||
|
||||
|
||||
async def get_online_agent_for_tenant(
|
||||
db: AsyncSessionDep, tenant_id: uuid.UUID
|
||||
) -> Agent | None:
|
||||
"""Get the first online agent for a tenant."""
|
||||
result = await db.execute(
|
||||
select(Agent)
|
||||
.where(Agent.tenant_id == tenant_id)
|
||||
.where(Agent.status == "online")
|
||||
.limit(1)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
|
||||
# --- Route handlers ---
|
||||
|
||||
|
||||
|
|
@ -97,3 +125,59 @@ async def setup_chatwoot(
|
|||
)
|
||||
|
||||
return task
|
||||
|
||||
|
||||
@router.post(
|
||||
"/nextcloud/set-domain",
|
||||
response_model=TaskResponse,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
)
|
||||
async def nextcloud_set_domain(
|
||||
tenant_id: uuid.UUID,
|
||||
request: NextcloudSetDomainRequest,
|
||||
db: AsyncSessionDep,
|
||||
) -> Task:
|
||||
"""
|
||||
Set Nextcloud domain for a tenant (v1 - reload only).
|
||||
|
||||
Creates a COMPOSITE task with a single DOCKER_RELOAD step.
|
||||
Auto-resolves the agent from the tenant (first online agent).
|
||||
|
||||
**Note:** This is v1 of the playbook. It only reloads the Nextcloud stack.
|
||||
Future versions will add occ commands to configure config.php.
|
||||
|
||||
## Request Body
|
||||
- **public_url**: The public URL for Nextcloud (e.g., "https://cloud.example.com")
|
||||
- **pull**: Whether to pull images before reloading (default: false)
|
||||
|
||||
## Response
|
||||
Returns the created Task with type="COMPOSITE" and status="pending".
|
||||
|
||||
The SysAdmin Agent will pick up this task and execute the steps in sequence.
|
||||
"""
|
||||
# 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",
|
||||
)
|
||||
|
||||
# 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}",
|
||||
)
|
||||
|
||||
# Create the COMPOSITE task
|
||||
task = await create_nextcloud_set_domain_task(
|
||||
db=db,
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id,
|
||||
public_url=request.public_url,
|
||||
pull=request.pull,
|
||||
)
|
||||
|
||||
return task
|
||||
|
|
|
|||
|
|
@ -77,3 +77,20 @@ async def test_agent(db: AsyncSession) -> Agent:
|
|||
await db.commit()
|
||||
await db.refresh(agent)
|
||||
return agent
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def test_agent_for_tenant(db: AsyncSession, test_tenant: Tenant) -> Agent:
|
||||
"""Create a test agent linked to test_tenant with online status."""
|
||||
agent = Agent(
|
||||
id=uuid.uuid4(),
|
||||
tenant_id=test_tenant.id,
|
||||
name="test-agent-for-tenant",
|
||||
version="1.0.0",
|
||||
status="online",
|
||||
token="test-token-tenant",
|
||||
)
|
||||
db.add(agent)
|
||||
await db.commit()
|
||||
await db.refresh(agent)
|
||||
return agent
|
||||
|
|
|
|||
|
|
@ -0,0 +1,204 @@
|
|||
"""Tests for Nextcloud playbook 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.playbooks.nextcloud import NEXTCLOUD_STACK_DIR
|
||||
from app.routes.playbooks import (
|
||||
NextcloudSetDomainRequest,
|
||||
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
|
||||
|
|
@ -0,0 +1,163 @@
|
|||
"""Tests for the Nextcloud 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.nextcloud import (
|
||||
NEXTCLOUD_STACK_DIR,
|
||||
CompositeStep,
|
||||
build_nextcloud_set_domain_steps,
|
||||
create_nextcloud_set_domain_task,
|
||||
)
|
||||
|
||||
|
||||
class TestBuildNextcloudSetDomainSteps:
|
||||
"""Tests for the build_nextcloud_set_domain_steps function."""
|
||||
|
||||
def test_returns_two_steps(self):
|
||||
"""Verify that build_nextcloud_set_domain_steps returns exactly 2 steps."""
|
||||
steps = build_nextcloud_set_domain_steps(
|
||||
public_url="https://cloud.example.com", pull=False
|
||||
)
|
||||
assert len(steps) == 2
|
||||
assert all(isinstance(step, CompositeStep) for step in steps)
|
||||
|
||||
def test_first_step_is_nextcloud_set_domain(self):
|
||||
"""Verify the first step is NEXTCLOUD_SET_DOMAIN."""
|
||||
steps = build_nextcloud_set_domain_steps(
|
||||
public_url="https://cloud.example.com", pull=False
|
||||
)
|
||||
assert steps[0].type == "NEXTCLOUD_SET_DOMAIN"
|
||||
|
||||
def test_nextcloud_set_domain_payload(self):
|
||||
"""Verify NEXTCLOUD_SET_DOMAIN step has correct payload."""
|
||||
public_url = "https://cloud.example.com"
|
||||
steps = build_nextcloud_set_domain_steps(public_url=public_url, pull=False)
|
||||
assert steps[0].payload["public_url"] == public_url
|
||||
|
||||
def test_docker_reload_payload(self):
|
||||
"""Verify the DOCKER_RELOAD step has the correct payload structure."""
|
||||
steps = build_nextcloud_set_domain_steps(
|
||||
public_url="https://cloud.example.com", pull=False
|
||||
)
|
||||
|
||||
docker_step = steps[1]
|
||||
assert docker_step.type == "DOCKER_RELOAD"
|
||||
assert docker_step.payload["compose_dir"] == NEXTCLOUD_STACK_DIR
|
||||
assert docker_step.payload["pull"] is False
|
||||
|
||||
def test_pull_flag_true(self):
|
||||
"""Verify that pull=True is passed correctly."""
|
||||
steps = build_nextcloud_set_domain_steps(
|
||||
public_url="https://cloud.example.com", pull=True
|
||||
)
|
||||
|
||||
docker_step = steps[1]
|
||||
assert docker_step.payload["pull"] is True
|
||||
|
||||
def test_pull_flag_false(self):
|
||||
"""Verify that pull=False is passed correctly."""
|
||||
steps = build_nextcloud_set_domain_steps(
|
||||
public_url="https://cloud.example.com", pull=False
|
||||
)
|
||||
|
||||
docker_step = steps[1]
|
||||
assert docker_step.payload["pull"] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCreateNextcloudSetDomainTask:
|
||||
"""Tests for the create_nextcloud_set_domain_task function."""
|
||||
|
||||
async def test_persists_composite_task(self, db: AsyncSession, test_tenant: Tenant):
|
||||
"""Verify that create_nextcloud_set_domain_task persists a COMPOSITE task."""
|
||||
agent_id = uuid.uuid4()
|
||||
|
||||
task = await create_nextcloud_set_domain_task(
|
||||
db=db,
|
||||
tenant_id=test_tenant.id,
|
||||
agent_id=agent_id,
|
||||
public_url="https://cloud.example.com",
|
||||
pull=False,
|
||||
)
|
||||
|
||||
assert task.id is not None
|
||||
assert task.tenant_id == test_tenant.id
|
||||
assert task.agent_id == agent_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_nextcloud_set_domain_task(
|
||||
db=db,
|
||||
tenant_id=test_tenant.id,
|
||||
agent_id=uuid.uuid4(),
|
||||
public_url="https://cloud.example.com",
|
||||
pull=False,
|
||||
)
|
||||
|
||||
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_nextcloud_set_domain_task(
|
||||
db=db,
|
||||
tenant_id=test_tenant.id,
|
||||
agent_id=uuid.uuid4(),
|
||||
public_url="https://cloud.example.com",
|
||||
pull=True,
|
||||
)
|
||||
|
||||
steps = task.payload["steps"]
|
||||
|
||||
# First step should be NEXTCLOUD_SET_DOMAIN
|
||||
assert steps[0]["type"] == "NEXTCLOUD_SET_DOMAIN"
|
||||
assert steps[0]["payload"]["public_url"] == "https://cloud.example.com"
|
||||
|
||||
# Second step should be DOCKER_RELOAD
|
||||
assert steps[1]["type"] == "DOCKER_RELOAD"
|
||||
assert steps[1]["payload"]["compose_dir"] == NEXTCLOUD_STACK_DIR
|
||||
assert steps[1]["payload"]["pull"] is True
|
||||
|
||||
async def test_task_with_pull_false(self, db: AsyncSession, test_tenant: Tenant):
|
||||
"""Verify that pull=False is correctly stored in the task payload."""
|
||||
task = await create_nextcloud_set_domain_task(
|
||||
db=db,
|
||||
tenant_id=test_tenant.id,
|
||||
agent_id=uuid.uuid4(),
|
||||
public_url="https://cloud.example.com",
|
||||
pull=False,
|
||||
)
|
||||
|
||||
steps = task.payload["steps"]
|
||||
assert steps[1]["payload"]["pull"] is False
|
||||
|
||||
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_set_domain_task(
|
||||
db=db,
|
||||
tenant_id=test_tenant.id,
|
||||
agent_id=uuid.uuid4(),
|
||||
public_url="https://cloud.example.com",
|
||||
pull=False,
|
||||
)
|
||||
|
||||
# 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
|
||||
Loading…
Reference in New Issue