DashBoard/tests/e2e/test_wip_hold_pages_e2e.py

# -*- coding: utf-8 -*-
"""E2E coverage for WIP Overview / WIP Detail / Hold Detail pages."""

from __future__ import annotations

import time
from urllib.parse import parse_qs, quote, urlparse
import re

import pytest
import requests
from playwright.sync_api import Page, expect


def _pick_workcenter(app_server: str) -> str:
    """Pick a real workcenter to reduce flaky E2E failures."""
    try:
        response = requests.get(f"{app_server}/api/wip/meta/workcenters", timeout=10)
        payload = response.json() if response.ok else {}
        items = payload.get("data") or []
        if items:
            return items[0].get("name") or "TMTT"
    except Exception:
        pass
    return "TMTT"


def _pick_hold_reason(app_server: str) -> str:
    """Pick a real hold reason to reduce flaky E2E failures."""
    try:
        response = requests.get(f"{app_server}/api/wip/overview/hold", timeout=10)
        payload = response.json() if response.ok else {}
        items = (payload.get("data") or {}).get("items") or []
        if items:
            return items[0].get("reason") or "YieldLimit"
    except Exception:
        pass
    return "YieldLimit"


def _get_with_retry(url: str, attempts: int = 3, timeout: float = 10.0):
    """Best-effort GET helper to reduce transient test flakiness."""
    last_exc = None
    for _ in range(max(attempts, 1)):
        try:
            return requests.get(url, timeout=timeout, allow_redirects=False)
        except requests.RequestException as exc:
            last_exc = exc
            time.sleep(0.5)
    if last_exc:
        raise last_exc
    raise RuntimeError("request retry exhausted without exception")


def _wait_for_response_url_tokens(page: Page, tokens: list[str], timeout_seconds: float = 30.0):
    """Wait until a response URL contains all tokens."""
    matched = []

    def handle_response(resp):
        if all(token in resp.url for token in tokens):
            matched.append(resp)

    page.on("response", handle_response)
    deadline = time.time() + timeout_seconds
    while time.time() < deadline and not matched:
        page.wait_for_timeout(200)
    return matched[0] if matched else None


def _wait_for_response(page: Page, predicate, timeout_seconds: float = 30.0):
    """Wait until a response satisfies the predicate."""
    matched = []

    def handle_response(resp):
        try:
            if predicate(resp):
                matched.append(resp)
        except Exception:
            return

    page.on("response", handle_response)
    deadline = time.time() + timeout_seconds
    while time.time() < deadline and not matched:
        page.wait_for_timeout(200)
    return matched[0] if matched else None


@pytest.mark.e2e
class TestWipAndHoldPagesE2E:
    """E2E tests for WIP/Hold page URL + API behavior."""

    def test_wip_overview_restores_status_from_url(self, page: Page, app_server: str):
        page.goto(
            f"{app_server}/wip-overview?type=PJA3460&status=queue",
            wait_until="commit",
            timeout=60000,
        )
        response = _wait_for_response_url_tokens(
            page,
            ["/api/wip/overview/matrix", "type=PJA3460", "status=QUEUE"],
            timeout_seconds=30.0,
        )
        assert response is not None, "Did not observe expected matrix request with URL filters"
        assert response.ok
        expect(page.locator("body")).to_be_visible()

    def test_wip_detail_reads_status_and_back_link_keeps_filters(self, page: Page, app_server: str):
        workcenter = _pick_workcenter(app_server)
        page.goto(
            f"{app_server}/wip-detail?workcenter={quote(workcenter)}&type=PJA3460&status=queue",
            wait_until="commit",
            timeout=60000,
        )

        response = _wait_for_response(
            page,
            lambda resp: (
                "/api/wip/detail/" in resp.url
                and (
                    parse_qs(urlparse(resp.url).query).get("type", [None])[0] == "PJA3460"
                    or parse_qs(urlparse(resp.url).query).get("pj_type", [None])[0] == "PJA3460"
                )
                and parse_qs(urlparse(resp.url).query).get("status", [None])[0] in {"QUEUE", "queue"}
            ),
            timeout_seconds=30.0,
        )
        assert response is not None, "Did not observe expected detail request with URL filters"
        assert response.ok

        back_href = page.locator("a.btn-back").get_attribute("href") or ""
        parsed = urlparse(back_href)
        params = parse_qs(parsed.query)
        assert parsed.path in {"/wip-overview", "/portal-shell/wip-overview"}
        assert params.get("type", [None])[0] == "PJA3460"
        assert params.get("status", [None])[0] in {"queue", "QUEUE"}

    def test_hold_detail_without_reason_redirects_to_overview(self, page: Page, app_server: str):
        nav_resp = _get_with_retry(f"{app_server}/api/portal/navigation", attempts=3, timeout=10.0)
        nav_payload = nav_resp.json() if nav_resp.ok else {}
        spa_enabled = bool(nav_payload.get("portal_spa_enabled"))

        response = _get_with_retry(f"{app_server}/hold-detail", attempts=3, timeout=10.0)
        assert response.status_code == 302
        if spa_enabled:
            assert response.headers.get("Location") == "/portal-shell/wip-overview"
        else:
            assert response.headers.get("Location") == "/wip-overview"

    def test_hold_detail_calls_summary_distribution_and_lots(self, page: Page, app_server: str):
        reason = _pick_hold_reason(app_server)
        seen = set()

        def handle_response(resp):
            parsed = urlparse(resp.url)
            query = parse_qs(parsed.query)
            if query.get("reason", [None])[0] != reason:
                return
            if parsed.path.endswith("/api/wip/hold-detail/summary"):
                seen.add("summary")
            elif parsed.path.endswith("/api/wip/hold-detail/distribution"):
                seen.add("distribution")
            elif parsed.path.endswith("/api/wip/hold-detail/lots"):
                seen.add("lots")

        page.on("response", handle_response)
        page.goto(
            f"{app_server}/hold-detail?reason={quote(reason)}",
            wait_until="commit",
            timeout=60000,
        )

        deadline = time.time() + 30
        while time.time() < deadline and len(seen) < 3:
            page.wait_for_timeout(200)

        assert seen == {"summary", "distribution", "lots"}

    def test_portal_shell_deep_links_keep_detail_routes(self, page: Page, app_server: str):
        workcenter = _pick_workcenter(app_server)
        page.goto(
            f"{app_server}/portal-shell/wip-detail?workcenter={quote(workcenter)}&status=queue",
            wait_until="commit",
            timeout=60000,
        )
        expect(page).to_have_url(re.compile(r".*/portal-shell/wip-detail\\?.*workcenter=.*"))
        detail_response = _wait_for_response(
            page,
            lambda resp: (
                "/api/wip/detail/" in resp.url
                and parse_qs(urlparse(resp.url).query).get("status", [None])[0] in {"QUEUE", "queue"}
            ),
            timeout_seconds=30.0,
        )
        assert detail_response is not None
        assert detail_response.ok

        reason = _pick_hold_reason(app_server)
        page.goto(
            f"{app_server}/portal-shell/hold-detail?reason={quote(reason)}",
            wait_until="commit",
            timeout=60000,
        )
        expect(page).to_have_url(re.compile(r".*/portal-shell/hold-detail\\?.*reason=.*"))
        summary_response = _wait_for_response(
            page,
            lambda resp: (
                "/api/wip/hold-detail/summary" in resp.url
                and parse_qs(urlparse(resp.url).query).get("reason", [None])[0] == reason
            ),
            timeout_seconds=30.0,
        )
        assert summary_response is not None
        assert summary_response.ok

    def test_portal_shell_wip_overview_drilldown_routes_to_detail_pages(self, page: Page, app_server: str):
        page.goto(
            f"{app_server}/portal-shell/wip-overview",
            wait_until="commit",
            timeout=60000,
        )
        page.wait_for_timeout(3000)

        matrix_links = page.locator("td.clickable")
        if matrix_links.count() == 0:
            pytest.skip("No matrix rows available for WIP drilldown")
        matrix_links.first.click()
        expect(page).to_have_url(re.compile(r".*/portal-shell/wip-detail\\?.*workcenter=.*"))

        page.goto(
            f"{app_server}/portal-shell/wip-overview",
            wait_until="commit",
            timeout=60000,
        )
        page.wait_for_timeout(3000)

        reason_links = page.locator("a.reason-link")
        if reason_links.count() == 0:
            pytest.skip("No pareto reason links available for HOLD drilldown")
        reason_links.first.click()
        expect(page).to_have_url(re.compile(r".*/portal-shell/hold-detail\\?.*reason=.*"))