ocdp-go/e2e_test.py

#!/usr/bin/env python3
"""
E2E Test: Admin creates user -> User deploys Helm Chart

Prerequisites:
1. Start services: ./start.sh
2. Install Playwright: pip install playwright && playwright install chromium
3. Run: python e2e_test.py

This test covers the complete OCDP workflow:
1. Admin login
2. Admin creates workspace
3. Admin creates user
4. User login
5. User browses charts from Harbor
6. User deploys a chart
7. Verify deployment success
"""

from playwright.sync_api import sync_playwright
import time
import random

BASE_URL = "http://localhost"
ADMIN_USER = "admin"
ADMIN_PASS = "admin123"

def screenshot(page, name):
    """Take a screenshot for debugging"""
    page.screenshot(path=f'/tmp/{name}.png', full_page=True)
    print(f"  [Screenshot] /tmp/{name}.png")

def login(page, username, password, expect_success=True):
    """Login with given credentials"""
    page.goto(f"{BASE_URL}/login")
    page.wait_for_load_state('networkidle')
    time.sleep(1)

    # Fill login form
    page.fill('#username', username)
    page.fill('#password', password)
    page.click('button[type="submit"]')

    # Wait for redirect
    time.sleep(3)
    try:
        page.wait_for_url("**/", timeout=8000)
    except:
        pass
    page.wait_for_load_state('networkidle')

    # Verify login
    token = page.evaluate('localStorage.getItem("access_token")')
    if token:
        print(f"  [OK] Logged in as {username}")
        return True
    else:
        if expect_success:
            screenshot(page, f'login_failed_{username}')
            print(f"  [FAIL] Login failed for {username}")
        return False

def logout(page):
    """Logout current user"""
    try:
        page.click('text=Logout', timeout=3000)
    except:
        # Clear localStorage manually
        page.evaluate('localStorage.clear()')
    time.sleep(1)
    page.wait_for_load_state('networkidle')

def create_workspace(page, name, description="Test workspace"):
    """Create a workspace via admin panel"""
    page.goto(f"{BASE_URL}/admin/workspaces")
    page.wait_for_load_state('networkidle')
    time.sleep(1)

    # Click Add Workspace button
    add_btn = page.locator('text=Add Workspace').first
    if add_btn:
        add_btn.click()
        time.sleep(1)

    # Fill form
    page.fill('input[required]', name)
    time.sleep(0.5)

    # Submit
    page.click('button[type="submit"]')
    time.sleep(3)
    page.wait_for_load_state('networkidle')
    screenshot(page, f'workspace_{name}')
    print(f"  [OK] Created workspace: {name}")
    return name

def create_user(page, username, password, role="user"):
    """Create a user via admin panel"""
    page.goto(f"{BASE_URL}/admin/users")
    page.wait_for_load_state('networkidle')
    time.sleep(1)

    # Click Add User button
    add_btn = page.locator('text=Add User').first
    if add_btn:
        add_btn.click()
        time.sleep(1)

    # Fill form
    inputs = page.locator('input[type="text"], input[type="password"]').all()
    if len(inputs) >= 2:
        inputs[0].fill(username)
        inputs[1].fill(password)
        print(f"  Filled form with username={username}")

    # Select role if dropdown exists
    role_select = page.locator('select').first
    if role_select.count() > 0:
        try:
            role_select.select_option(value=role)
        except:
            pass

    # Submit
    page.click('button[type="submit"]')
    time.sleep(3)
    page.wait_for_load_state('networkidle')
    screenshot(page, f'user_{username}')
    print(f"  [OK] Created user: {username} (role={role})")

def browse_and_deploy_chart(page, chart_repo="charts/nginx"):
    """Browse charts and deploy"""
    page.goto(f"{BASE_URL}/charts")
    page.wait_for_load_state('networkidle')
    time.sleep(2)
    screenshot(page, 'charts_page')

    # Select first available registry
    registry_btns = page.locator('button:has(svg), button:has-text("harbor")').all()
    for btn in registry_btns:
        try:
            txt = btn.text_content() or ''
            if txt.strip() and len(txt.strip()) > 0:
                btn.click()
                print(f"  [OK] Selected registry: {txt.strip()}")
                time.sleep(2)
                break
        except:
            pass

    # Find and click the chart repository
    time.sleep(2)
    repo_btns = page.locator('button:has-text("charts/")').all()
    for btn in repo_btns:
        try:
            txt = btn.text_content() or ''
            if chart_repo in txt:
                btn.click()
                print(f"  [OK] Selected repo: {chart_repo}")
                time.sleep(2)
                break
        except:
            pass

    # Look for Deploy button
    time.sleep(2)
    deploy_btns = page.locator('button:has-text("Deploy")').all()
    for btn in deploy_btns:
        try:
            if btn.is_visible():
                btn.click()
                print(f"  [OK] Opened deploy modal")
                time.sleep(2)
                break
        except:
            pass

    # Fill deploy form
    time.sleep(1)
    inputs = page.locator('input[type="text"]').all()
    for inp in inputs:
        try:
            if inp.is_visible():
                placeholder = (inp.get_attribute('placeholder') or '').lower()
                if 'release' in placeholder or 'name' in placeholder:
                    instance_name = f"test-{int(time.time())}"
                    inp.fill(instance_name)
                    print(f"  Filled release name: {instance_name}")
                    break
        except:
            pass

    # Submit deployment
    time.sleep(1)
    submit_btns = page.locator('button:has-text("Deploy")').all()
    for btn in submit_btns:
        try:
            if btn.is_visible() and 'Deploying' not in (btn.text_content() or ''):
                btn.click()
                print(f"  [OK] Submitted deployment")
                time.sleep(3)
                screenshot(page, 'deployment_result')
                break
        except:
            pass

    return True

def main():
    print("=" * 60)
    print("OCDP E2E Test: Admin -> User Deployment Flow")
    print("=" * 60)

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False, args=['--no-sandbox'])
        page = browser.new_page()
        page.set_viewport_size({"width": 1920, "height": 1080})

        try:
            # Step 1: Login as admin
            print("\n[Step 1] Admin login")
            if not login(page, ADMIN_USER, ADMIN_PASS):
                print("[FAIL] Admin login failed")
                return

            # Step 2: Create workspace
            print("\n[Step 2] Create workspace")
            workspace_name = f"test-ws-{int(time.time())}"
            try:
                create_workspace(page, workspace_name)
            except Exception as e:
                print(f"  [WARN] Workspace creation: {e}")

            # Step 3: Create user
            print("\n[Step 3] Create user")
            test_user = f"e2euser_{int(time.time())}"
            test_pass = "test123456"
            try:
                create_user(page, test_user, test_pass, "user")
            except Exception as e:
                print(f"  [WARN] User creation: {e}")
                # Use existing user if creation fails
                test_user = "test"

            # Step 4: Logout and login as test user
            print(f"\n[Step 4] Login as {test_user}")
            logout(page)
            time.sleep(1)
            login(page, test_user, test_pass, expect_success=True)

            # Step 5: Browse charts
            print("\n[Step 5] Browse and deploy chart")
            try:
                browse_and_deploy_chart(page, "charts/nginx")
                print("  [OK] Chart deployed")
            except Exception as e:
                print(f"  [WARN] Chart deployment: {e}")
                screenshot(page, 'deploy_error')

            print("\n" + "=" * 60)
            print("[PASS] E2E Test Complete!")
            print("=" * 60)
            screenshot(page, 'e2e_complete')

        except Exception as e:
            print(f"\n[FAIL] Error: {e}")
            screenshot(page, 'e2e_error')
        finally:
            time.sleep(2)
            browser.close()

if __name__ == "__main__":
    main()