ocdp-go/test/instance_card_action_layout_playwright.py

#!/usr/bin/env python3
# Covers InstanceCard action layout. It prefers a harmless failed metadata
# instance when the API preserves one; if chart validation rejects before DB
# persistence, it falls back to mocking only the instance list/delete API so the
# visual overflow assertion remains independent from deployment behavior.

import json
import os
import time
import uuid
from dataclasses import dataclass
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urljoin
from urllib.request import Request, urlopen

from playwright.sync_api import expect, sync_playwright


RAW_BASE_URL = os.environ.get("BASE_URL", "http://localhost:18081/api/v1").rstrip("/")
BASE_URL = RAW_BASE_URL + "/"
FRONTEND_URL = os.environ.get("FRONTEND_URL", "http://localhost:18080")
ADMIN_USER = os.environ.get("ADMIN_USER", "admin")
ADMIN_PASS = os.environ["ADMIN_PASS"]


@dataclass
class Response:
    status: int
    body: str
    json: Any


def parse_json(body: str) -> Any:
    try:
        return json.loads(body) if body else None
    except json.JSONDecodeError:
        return None


def request(method: str, path: str, token: str | None = None, payload: Any = None) -> Response:
    data = None
    headers = {"Accept": "application/json"}
    if payload is not None:
        data = json.dumps(payload).encode("utf-8")
        headers["Content-Type"] = "application/json"
    if token:
        headers["Authorization"] = f"Bearer {token}"
    try:
        with urlopen(Request(urljoin(BASE_URL, path.lstrip("/")), data=data, headers=headers, method=method), timeout=40) as res:
            body = res.read().decode("utf-8", errors="replace")
            return Response(res.status, body, parse_json(body))
    except HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        return Response(exc.code, body, parse_json(body))
    except URLError as exc:
        raise AssertionError(f"Cannot reach {BASE_URL}: {exc}") from exc


def login_api(username: str, password: str) -> str:
    resp = request("POST", "/auth/login", payload={"username": username, "password": password})
    if resp.status != 200:
        raise AssertionError(f"login failed: HTTP {resp.status} {resp.body[:300]}")
    return str(resp.json["accessToken"])


def first_id(items: Any, name: str | None = None) -> str:
    if not isinstance(items, list):
        raise AssertionError(f"expected list, got {items!r}")
    for item in items:
        if isinstance(item, dict) and item.get("id") and (name is None or item.get("name") == name):
            return str(item["id"])
    raise AssertionError(f"could not find id for {name or 'first item'}")


def list_instances(cluster_id: str, token: str) -> list[dict[str, Any]]:
    resp = request("GET", f"/clusters/{cluster_id}/instances", token)
    if resp.status != 200 or not isinstance(resp.json, dict):
        raise AssertionError(f"list instances failed: HTTP {resp.status} {resp.body[:300]}")
    return [item for item in resp.json.get("instances", []) if isinstance(item, dict)]


def login_ui(page) -> None:
    page.goto(FRONTEND_URL, wait_until="networkidle")
    if page.locator("input[type='password']").count() == 0:
        return
    page.locator("input:not([type='password'])").first.fill(ADMIN_USER)
    page.locator("input[type='password']").first.fill(ADMIN_PASS)
    page.get_by_role("button", name="Login").last.click()
    page.wait_for_url("**/home", timeout=15000)
    page.wait_for_load_state("networkidle")


def main() -> int:
    token = login_api(ADMIN_USER, ADMIN_PASS)
    clusters = request("GET", "/clusters", token)
    registries = request("GET", "/registries", token)
    if clusters.status != 200 or registries.status != 200:
        raise AssertionError("clusters/registries must be available")
    cluster_id = first_id(clusters.json, "k3s") if any(item.get("name") == "k3s" for item in clusters.json) else first_id(clusters.json)
    registry_id = first_id(registries.json)
    suffix = uuid.uuid4().hex[:8]
    release = f"ocdp-ui-overflow-{suffix}"
    namespace = f"ocdp-ui-overflow-{suffix}"
    instance_id = ""
    synthetic = False
    try:
        create = request(
            "POST",
            f"/clusters/{cluster_id}/instances",
            token,
            {
                "name": release,
                "namespace": namespace,
                "registryId": registry_id,
                "repository": f"charts/nonexistent-ui-overflow-{suffix}",
                "tag": "0.0.0",
                "valuesYaml": "replicaCount: 1\n",
            },
        )
        if create.status not in {201, 400}:
            raise AssertionError(f"expected create to succeed or fail after DB insert, got HTTP {create.status}: {create.body[:500]}")
        for _ in range(20):
            matches = [item for item in list_instances(cluster_id, token) if item.get("name") == release]
            if matches:
                instance_id = str(matches[0]["id"])
                break
            time.sleep(0.5)
        if not instance_id:
            synthetic = True

        with sync_playwright() as p:
            browser = p.chromium.launch(headless=True)
            page = browser.new_page(viewport={"width": 920, "height": 760})
            page.on("dialog", lambda dialog: dialog.accept())
            login_ui(page)
            if synthetic:
                synthetic_instance = {
                    "id": f"synthetic-{suffix}",
                    "name": release,
                    "namespace": namespace,
                    "clusterId": cluster_id,
                    "registryId": registry_id,
                    "repository": "charts/nonexistent",
                    "chart": "nonexistent",
                    "version": "0.0.0",
                    "status": "failed",
                    "ownerUsername": ADMIN_USER,
                    "values": {},
                    "createdAt": "2026-05-15T00:00:00Z",
                    "updatedAt": "2026-05-15T00:00:00Z",
                }

                def fulfill_instances(route):
                    if route.request.method == "GET":
                        route.fulfill(status=200, content_type="application/json", body=json.dumps({"instances": [synthetic_instance], "total": 1}))
                        return
                    if route.request.method == "DELETE":
                        route.fulfill(status=204, body="")
                        return
                    route.continue_()

                page.route("**/api/v1/clusters/*/instances", fulfill_instances)
                page.route("**/api/v1/clusters/*/instances/*", fulfill_instances)
            page.get_by_role("button", name="Instances", exact=True).click()
            page.wait_for_load_state("networkidle")
            heading = page.get_by_role("heading", name=release, exact=True).first
            expect(heading).to_be_visible(timeout=15000)
            card = heading.locator("xpath=ancestor::div[contains(@class, 'group')][1]")
            delete_button = card.get_by_role("button", name="Delete", exact=True)
            expect(delete_button).to_be_visible()
            card_box = card.bounding_box()
            button_box = delete_button.bounding_box()
            viewport = page.viewport_size or {"width": 920, "height": 760}
            assert card_box and button_box, "card and delete button must have layout boxes"
            assert button_box["x"] >= card_box["x"] - 1, "Delete button overflows left edge of card"
            assert button_box["x"] + button_box["width"] <= card_box["x"] + card_box["width"] + 1, "Delete button overflows right edge of card"
            assert button_box["x"] + button_box["width"] <= viewport["width"], "Delete button overflows viewport"
            delete_button.click()
            page.wait_for_timeout(500)
            browser.close()
        print("PASS: instance card action layout")
        return 0
    finally:
        if instance_id:
            request("DELETE", f"/clusters/{cluster_id}/instances/{instance_id}", token)


if __name__ == "__main__":
    raise SystemExit(main())