265 lines
9.3 KiB
Python
265 lines
9.3 KiB
Python
|
"""n8n initial setup scenario.
|
||
|
|
|
||
|
|
Automates the first-time setup for a fresh n8n installation.
|
||
|
|
This scenario:
|
||
|
|
1. Navigates to the n8n setup page
|
||
|
|
2. Creates the owner account with email and password
|
||
|
|
3. Skips optional setup steps
|
||
|
|
"""
|
||
|
|
|
||
|
|
import secrets
|
||
|
|
import string
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from playwright.async_api import Page
|
||
|
|
|
||
|
|
from app.playwright_scenarios import register_scenario
|
||
|
|
from app.playwright_scenarios.base import BaseScenario, ScenarioOptions, ScenarioResult
|
||
|
|
|
||
|
|
|
||
|
|
def generate_secure_password(length: int = 24) -> str:
|
||
|
|
"""Generate a cryptographically secure password.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
length: Password length (default: 24)
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
A secure random password with mixed characters
|
||
|
|
"""
|
||
|
|
alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
|
||
|
|
password = [
|
||
|
|
secrets.choice(string.ascii_lowercase),
|
||
|
|
secrets.choice(string.ascii_uppercase),
|
||
|
|
secrets.choice(string.digits),
|
||
|
|
secrets.choice("!@#$%^&*"),
|
||
|
|
]
|
||
|
|
password.extend(secrets.choice(alphabet) for _ in range(length - 4))
|
||
|
|
password_list = list(password)
|
||
|
|
secrets.SystemRandom().shuffle(password_list)
|
||
|
|
return "".join(password_list)
|
||
|
|
|
||
|
|
|
||
|
|
@register_scenario
|
||
|
|
class N8nInitialSetup(BaseScenario):
|
||
|
|
"""Automate n8n first-time owner account setup.
|
||
|
|
|
||
|
|
This scenario handles the initial owner account creation when
|
||
|
|
n8n is freshly installed. It fills in the account details
|
||
|
|
and completes the setup wizard.
|
||
|
|
|
||
|
|
Required inputs:
|
||
|
|
base_url: The n8n instance URL (e.g., https://n8n.example.com)
|
||
|
|
admin_email: Email address for the owner account
|
||
|
|
|
||
|
|
Optional inputs:
|
||
|
|
admin_password: Password for owner account (auto-generated if not provided)
|
||
|
|
admin_first_name: First name for the owner (default: "Admin")
|
||
|
|
admin_last_name: Last name for the owner (default: "User")
|
||
|
|
|
||
|
|
Result data:
|
||
|
|
setup_completed: Whether initial setup was completed
|
||
|
|
admin_email: The configured owner email address
|
||
|
|
admin_password: The password (generated or provided) - STORE SECURELY
|
||
|
|
already_configured: True if n8n was already set up
|
||
|
|
"""
|
||
|
|
|
||
|
|
@property
|
||
|
|
def name(self) -> str:
|
||
|
|
return "n8n_initial_setup"
|
||
|
|
|
||
|
|
@property
|
||
|
|
def required_inputs(self) -> list[str]:
|
||
|
|
return ["base_url", "admin_email"]
|
||
|
|
|
||
|
|
@property
|
||
|
|
def optional_inputs(self) -> list[str]:
|
||
|
|
return ["admin_password", "admin_first_name", "admin_last_name"]
|
||
|
|
|
||
|
|
@property
|
||
|
|
def description(self) -> str:
|
||
|
|
return "Automate n8n first-time owner account setup"
|
||
|
|
|
||
|
|
async def execute(
|
||
|
|
self,
|
||
|
|
page: Page,
|
||
|
|
inputs: dict[str, Any],
|
||
|
|
options: ScenarioOptions,
|
||
|
|
) -> ScenarioResult:
|
||
|
|
"""Execute the n8n initial setup.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
page: Playwright Page object
|
||
|
|
inputs: Scenario inputs (base_url, admin_email)
|
||
|
|
options: Scenario options
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
ScenarioResult with setup status and credentials
|
||
|
|
"""
|
||
|
|
base_url = inputs["base_url"].rstrip("/")
|
||
|
|
admin_email = inputs["admin_email"]
|
||
|
|
admin_password = inputs.get("admin_password") or generate_secure_password()
|
||
|
|
admin_first_name = inputs.get("admin_first_name", "Admin")
|
||
|
|
admin_last_name = inputs.get("admin_last_name", "User")
|
||
|
|
|
||
|
|
screenshots = []
|
||
|
|
result_data = {
|
||
|
|
"setup_completed": False,
|
||
|
|
"admin_email": admin_email,
|
||
|
|
"admin_password": admin_password,
|
||
|
|
"already_configured": False,
|
||
|
|
}
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Navigate to n8n
|
||
|
|
await page.goto(base_url, wait_until="networkidle")
|
||
|
|
|
||
|
|
current_url = page.url
|
||
|
|
|
||
|
|
# Check if already configured (redirects to signin)
|
||
|
|
if "/signin" in current_url:
|
||
|
|
result_data["already_configured"] = True
|
||
|
|
result_data["setup_completed"] = True
|
||
|
|
return ScenarioResult(
|
||
|
|
success=True,
|
||
|
|
data=result_data,
|
||
|
|
screenshots=screenshots,
|
||
|
|
error=None,
|
||
|
|
)
|
||
|
|
|
||
|
|
# n8n setup page should show the owner setup form
|
||
|
|
# Look for setup form elements
|
||
|
|
email_input = page.locator(
|
||
|
|
'input[name="email"], '
|
||
|
|
'input[type="email"], '
|
||
|
|
'input[placeholder*="email" i], '
|
||
|
|
'input[autocomplete="email"]'
|
||
|
|
)
|
||
|
|
|
||
|
|
if await email_input.count() == 0:
|
||
|
|
# Try navigating to setup URL
|
||
|
|
await page.goto(f"{base_url}/setup", wait_until="networkidle")
|
||
|
|
|
||
|
|
if "/signin" in page.url:
|
||
|
|
result_data["already_configured"] = True
|
||
|
|
result_data["setup_completed"] = True
|
||
|
|
return ScenarioResult(
|
||
|
|
success=True,
|
||
|
|
data=result_data,
|
||
|
|
screenshots=screenshots,
|
||
|
|
error=None,
|
||
|
|
)
|
||
|
|
|
||
|
|
# Fill in the owner setup form
|
||
|
|
# First name
|
||
|
|
first_name_input = page.locator(
|
||
|
|
'input[name="firstName"], '
|
||
|
|
'input[name="first_name"], '
|
||
|
|
'input[placeholder*="first" i], '
|
||
|
|
'input[autocomplete="given-name"]'
|
||
|
|
).first
|
||
|
|
if await first_name_input.count() > 0:
|
||
|
|
await first_name_input.wait_for(state="visible", timeout=10000)
|
||
|
|
await first_name_input.fill(admin_first_name)
|
||
|
|
|
||
|
|
# Last name
|
||
|
|
last_name_input = page.locator(
|
||
|
|
'input[name="lastName"], '
|
||
|
|
'input[name="last_name"], '
|
||
|
|
'input[placeholder*="last" i], '
|
||
|
|
'input[autocomplete="family-name"]'
|
||
|
|
).first
|
||
|
|
if await last_name_input.count() > 0:
|
||
|
|
await last_name_input.fill(admin_last_name)
|
||
|
|
|
||
|
|
# Email
|
||
|
|
email_input = page.locator(
|
||
|
|
'input[name="email"], '
|
||
|
|
'input[type="email"], '
|
||
|
|
'input[placeholder*="email" i]'
|
||
|
|
).first
|
||
|
|
await email_input.wait_for(state="visible", timeout=10000)
|
||
|
|
await email_input.fill(admin_email)
|
||
|
|
|
||
|
|
# Password
|
||
|
|
password_input = page.locator(
|
||
|
|
'input[name="password"], '
|
||
|
|
'input[type="password"]'
|
||
|
|
).first
|
||
|
|
await password_input.fill(admin_password)
|
||
|
|
|
||
|
|
# Take screenshot before submitting
|
||
|
|
if options.screenshot_on_success and options.artifacts_dir:
|
||
|
|
pre_submit_path = options.artifacts_dir / "n8n_pre_submit.png"
|
||
|
|
await page.screenshot(path=str(pre_submit_path))
|
||
|
|
screenshots.append(str(pre_submit_path))
|
||
|
|
|
||
|
|
# Click Next / Create Account button
|
||
|
|
submit_button = page.locator(
|
||
|
|
'button:has-text("Next"), '
|
||
|
|
'button:has-text("Create"), '
|
||
|
|
'button:has-text("Get started"), '
|
||
|
|
'button[type="submit"]'
|
||
|
|
).first
|
||
|
|
await submit_button.click()
|
||
|
|
|
||
|
|
# Wait for next step or dashboard
|
||
|
|
await page.wait_for_timeout(3000)
|
||
|
|
|
||
|
|
# n8n may show additional setup steps (personalization, usage, etc.)
|
||
|
|
# Skip through them
|
||
|
|
for _ in range(3):
|
||
|
|
skip_button = page.locator(
|
||
|
|
'button:has-text("Skip"), '
|
||
|
|
'a:has-text("Skip"), '
|
||
|
|
'button:has-text("Get started"), '
|
||
|
|
'button:has-text("Next")'
|
||
|
|
)
|
||
|
|
if await skip_button.count() > 0:
|
||
|
|
await skip_button.first.click()
|
||
|
|
await page.wait_for_timeout(2000)
|
||
|
|
else:
|
||
|
|
break
|
||
|
|
|
||
|
|
# Check if we reached the workflow editor or dashboard
|
||
|
|
await page.wait_for_timeout(2000)
|
||
|
|
current_url = page.url
|
||
|
|
|
||
|
|
if any(kw in current_url for kw in ["/workflow", "/home", "/dashboard"]):
|
||
|
|
result_data["setup_completed"] = True
|
||
|
|
else:
|
||
|
|
# Check for indicators of successful setup
|
||
|
|
canvas = page.locator(
|
||
|
|
'.workflow-canvas, '
|
||
|
|
'[class*="workflow"], '
|
||
|
|
'[class*="canvas"], '
|
||
|
|
'#app'
|
||
|
|
)
|
||
|
|
if await canvas.count() > 0:
|
||
|
|
result_data["setup_completed"] = True
|
||
|
|
|
||
|
|
# Take final screenshot
|
||
|
|
if options.screenshot_on_success and options.artifacts_dir:
|
||
|
|
final_path = options.artifacts_dir / "n8n_setup_complete.png"
|
||
|
|
await page.screenshot(path=str(final_path))
|
||
|
|
screenshots.append(str(final_path))
|
||
|
|
|
||
|
|
return ScenarioResult(
|
||
|
|
success=result_data["setup_completed"],
|
||
|
|
data=result_data,
|
||
|
|
screenshots=screenshots,
|
||
|
|
error=None if result_data["setup_completed"] else "Setup may not have completed",
|
||
|
|
)
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
if options.screenshot_on_failure and options.artifacts_dir:
|
||
|
|
error_path = options.artifacts_dir / "n8n_setup_error.png"
|
||
|
|
await page.screenshot(path=str(error_path))
|
||
|
|
screenshots.append(str(error_path))
|
||
|
|
|
||
|
|
return ScenarioResult(
|
||
|
|
success=False,
|
||
|
|
data=result_data,
|
||
|
|
screenshots=screenshots,
|
||
|
|
error=f"n8n setup failed: {str(e)}",
|
||
|
|
)
|