rose-ash/shared/tests/test_sx_app_pages.py

"""Integration tests for SX docs app — page rendering + interactive API endpoints.

Runs inside the test container, hitting the sx_docs service over the internal
network.  Uses ``SX-Request: true`` header to bypass the silent-SSO OAuth
redirect on page requests.

Tested:
  - All 27 example pages render with 200 and contain meaningful content
  - All 23 attribute detail pages render and mention the attribute name
  - All 35+ interactive API endpoints return 200 with expected content
"""
from __future__ import annotations

import os
import re

import httpx
import pytest

SX_BASE = os.environ.get("INTERNAL_URL_SX", "http://sx_docs:8000")
HEADERS = {"SX-Request": "true"}
TIMEOUT = 15.0


def _get(path: str, **kw) -> httpx.Response:
    return httpx.get(
        f"{SX_BASE}{path}",
        headers=HEADERS,
        timeout=TIMEOUT,
        follow_redirects=True,
        **kw,
    )


def _post(path: str, **kw) -> httpx.Response:
    return httpx.post(
        f"{SX_BASE}{path}",
        headers=HEADERS,
        timeout=TIMEOUT,
        follow_redirects=True,
        **kw,
    )


def _put(path: str, **kw) -> httpx.Response:
    return httpx.put(
        f"{SX_BASE}{path}",
        headers=HEADERS,
        timeout=TIMEOUT,
        follow_redirects=True,
        **kw,
    )


def _patch(path: str, **kw) -> httpx.Response:
    return httpx.patch(
        f"{SX_BASE}{path}",
        headers=HEADERS,
        timeout=TIMEOUT,
        follow_redirects=True,
        **kw,
    )


def _delete(path: str, **kw) -> httpx.Response:
    return httpx.delete(
        f"{SX_BASE}{path}",
        headers=HEADERS,
        timeout=TIMEOUT,
        follow_redirects=True,
        **kw,
    )


# ── Check that the sx_docs service is reachable ──────────────────────────

def _sx_reachable() -> bool:
    try:
        r = httpx.get(f"{SX_BASE}/", timeout=5, follow_redirects=False)
        return r.status_code in (200, 302)
    except Exception:
        return False


pytestmark = pytest.mark.skipif(
    not _sx_reachable(),
    reason=f"sx_docs service not reachable at {SX_BASE}",
)


# ═════════════════════════════════════════════════════════════════════════
# Example pages — rendering
# ═════════════════════════════════════════════════════════════════════════

EXAMPLES = [
    "click-to-load",
    "form-submission",
    "polling",
    "delete-row",
    "inline-edit",
    "oob-swaps",
    "lazy-loading",
    "infinite-scroll",
    "progress-bar",
    "active-search",
    "inline-validation",
    "value-select",
    "reset-on-submit",
    "edit-row",
    "bulk-update",
    "swap-positions",
    "select-filter",
    "tabs",
    "animations",
    "dialogs",
    "keyboard-shortcuts",
    "put-patch",
    "json-encoding",
    "vals-and-headers",
    "loading-states",
    "sync-replace",
    "retry",
]


@pytest.mark.parametrize("slug", EXAMPLES)
def test_example_page_renders(slug: str):
    """Each example page must render successfully on two consecutive loads."""
    for attempt in (1, 2):
        r = _get(f"/examples/{slug}")
        assert r.status_code == 200, (
            f"/examples/{slug} returned {r.status_code} on attempt {attempt}"
        )
        assert len(r.text) > 500, (
            f"/examples/{slug} response too short ({len(r.text)} bytes) on attempt {attempt}"
        )
    # Every example page should have a demo section
    assert "demo" in r.text.lower() or "example" in r.text.lower(), (
        f"/examples/{slug} missing demo/example content"
    )


# ═════════════════════════════════════════════════════════════════════════
# Attribute detail pages — rendering
# ═════════════════════════════════════════════════════════════════════════

ATTRIBUTES = [
    "sx-get",
    "sx-post",
    "sx-put",
    "sx-delete",
    "sx-patch",
    "sx-trigger",
    "sx-target",
    "sx-swap",
    "sx-swap-oob",
    "sx-select",
    "sx-confirm",
    "sx-push-url",
    "sx-sync",
    "sx-encoding",
    "sx-headers",
    "sx-include",
    "sx-vals",
    "sx-media",
    "sx-disable",
    "sx-on",       # URL slug for sx-on:*
    "sx-boost",
    "sx-preload",
    "sx-preserve",
    "sx-indicator",
    "sx-validate",
    "sx-ignore",
    "sx-optimistic",
    "sx-replace-url",
    "sx-disabled-elt",
    "sx-prompt",
    "sx-params",
    "sx-sse",
    "sx-sse-swap",
    "sx-retry",
    "data-sx",
    "data-sx-env",
]


@pytest.mark.parametrize("slug", ATTRIBUTES)
def test_attribute_page_renders(slug: str):
    """Each attribute page must render successfully on two consecutive loads."""
    for attempt in (1, 2):
        r = _get(f"/reference/attributes/{slug}")
        assert r.status_code == 200, (
            f"/reference/attributes/{slug} returned {r.status_code} on attempt {attempt}"
        )
        assert len(r.text) > 500, (
            f"/reference/attributes/{slug} response too short on attempt {attempt}"
        )
    # The attribute name (or a prefix of it) should appear somewhere
    check = slug.rstrip("*").rstrip(":")
    assert check.lower() in r.text.lower(), (
        f"/reference/attributes/{slug} does not mention '{check}'"
    )


# ═════════════════════════════════════════════════════════════════════════
# Example API endpoints — interactive demos
# ═════════════════════════════════════════════════════════════════════════

class TestExampleAPIs:
    """Test the interactive demo API endpoints."""

    def test_click_to_load(self):
        r = _get("/examples/api/click")
        assert r.status_code == 200

    def test_form_submission(self):
        r = _post("/examples/api/form", data={"name": "Alice"})
        assert r.status_code == 200
        assert "Alice" in r.text

    def test_polling(self):
        r = _get("/examples/api/poll")
        assert r.status_code == 200

    def test_delete_row(self):
        r = _delete("/examples/api/delete/1")
        assert r.status_code == 200

    def test_inline_edit_get(self):
        r = _get("/examples/api/edit")
        assert r.status_code == 200

    def test_inline_edit_post(self):
        r = _post("/examples/api/edit", data={"name": "New Name"})
        assert r.status_code == 200

    def test_inline_edit_cancel(self):
        r = _get("/examples/api/edit/cancel")
        assert r.status_code == 200

    def test_oob_swap(self):
        r = _get("/examples/api/oob")
        assert r.status_code == 200

    def test_lazy_loading(self):
        r = _get("/examples/api/lazy")
        assert r.status_code == 200

    def test_infinite_scroll(self):
        r = _get("/examples/api/scroll", params={"page": "1"})
        assert r.status_code == 200

    def test_progress_start(self):
        r = _post("/examples/api/progress/start")
        assert r.status_code == 200

    def test_progress_status(self):
        r = _get("/examples/api/progress/status")
        assert r.status_code == 200

    def test_active_search(self):
        r = _get("/examples/api/search", params={"q": "py"})
        assert r.status_code == 200
        assert "Python" in r.text

    def test_inline_validation(self):
        r = _get("/examples/api/validate", params={"email": "test@example.com"})
        assert r.status_code == 200

    def test_validation_submit(self):
        r = _post("/examples/api/validate/submit", data={"email": "test@example.com"})
        assert r.status_code == 200

    def test_value_select(self):
        r = _get("/examples/api/values", params={"category": "Languages"})
        assert r.status_code == 200

    def test_reset_on_submit(self):
        r = _post("/examples/api/reset-submit", data={"message": "hello"})
        assert r.status_code == 200

    def test_edit_row_get(self):
        r = _get("/examples/api/editrow/1")
        assert r.status_code == 200

    def test_edit_row_post(self):
        r = _post("/examples/api/editrow/1", data={"name": "X", "price": "10", "stock": "5"})
        assert r.status_code == 200

    def test_edit_row_cancel(self):
        r = _get("/examples/api/editrow/1/cancel")
        assert r.status_code == 200

    def test_bulk_update(self):
        r = _post("/examples/api/bulk", data={"ids": ["1", "2"], "status": "active"})
        assert r.status_code == 200

    def test_swap_positions(self):
        r = _post("/examples/api/swap-log")
        assert r.status_code == 200

    def test_dashboard_filter(self):
        r = _get("/examples/api/dashboard", params={"region": "all"})
        assert r.status_code == 200

    def test_tabs(self):
        r = _get("/examples/api/tabs/overview")
        assert r.status_code == 200

    def test_animate(self):
        r = _get("/examples/api/animate")
        assert r.status_code == 200

    def test_dialog_open(self):
        r = _get("/examples/api/dialog")
        assert r.status_code == 200

    def test_dialog_close(self):
        r = _get("/examples/api/dialog/close")
        assert r.status_code == 200

    def test_keyboard(self):
        r = _get("/examples/api/keyboard")
        assert r.status_code == 200

    def test_put_patch_edit(self):
        r = _get("/examples/api/putpatch/edit-all")
        assert r.status_code == 200

    def test_put_request(self):
        r = _put("/examples/api/putpatch", data={"name": "X", "email": "x@x.com", "role": "Dev"})
        assert r.status_code == 200

    def test_put_patch_cancel(self):
        r = _get("/examples/api/putpatch/cancel")
        assert r.status_code == 200

    def test_json_encoding(self):
        r = httpx.post(
            f"{SX_BASE}/examples/api/json-echo",
            content='{"key":"val"}',
            headers={**HEADERS, "Content-Type": "application/json"},
            timeout=TIMEOUT,
        )
        assert r.status_code == 200

    def test_echo_vals(self):
        r = _get("/examples/api/echo-vals", params={"source": "test"})
        assert r.status_code == 200

    def test_echo_headers(self):
        r = httpx.get(
            f"{SX_BASE}/examples/api/echo-headers",
            headers={**HEADERS, "X-Custom": "hello"},
            timeout=TIMEOUT,
        )
        assert r.status_code == 200

    def test_slow_endpoint(self):
        r = _get("/examples/api/slow")
        assert r.status_code == 200

    def test_slow_search(self):
        r = _get("/examples/api/slow-search", params={"q": "test"})
        assert r.status_code == 200

    def test_flaky_endpoint(self):
        # May fail 2/3 times — just check it returns *something*
        r = _get("/examples/api/flaky")
        assert r.status_code in (200, 503)


# ═════════════════════════════════════════════════════════════════════════
# Reference API endpoints — attribute demos
# ═════════════════════════════════════════════════════════════════════════

class TestReferenceAPIs:
    """Test the reference attribute demo API endpoints."""

    def test_time(self):
        r = _get("/reference/api/time")
        assert r.status_code == 200
        # Should contain a time string (HH:MM:SS pattern)
        assert re.search(r"\d{2}:\d{2}:\d{2}", r.text), "No time found in response"

    def test_greet(self):
        r = _post("/reference/api/greet", data={"name": "Bob"})
        assert r.status_code == 200
        assert "Bob" in r.text

    def test_status_put(self):
        r = _put("/reference/api/status", data={"status": "published"})
        assert r.status_code == 200
        assert "published" in r.text.lower()

    def test_theme_patch(self):
        r = _patch("/reference/api/theme", data={"theme": "dark"})
        assert r.status_code == 200
        assert "dark" in r.text.lower()

    def test_delete_item(self):
        r = _delete("/reference/api/item/42")
        assert r.status_code == 200

    def test_trigger_search(self):
        r = _get("/reference/api/trigger-search", params={"q": "hello"})
        assert r.status_code == 200
        assert "hello" in r.text.lower()

    def test_swap_item(self):
        r = _get("/reference/api/swap-item")
        assert r.status_code == 200

    def test_oob(self):
        r = _get("/reference/api/oob")
        assert r.status_code == 200
        # OOB response should contain sx-swap-oob attribute
        assert "oob" in r.text.lower()

    def test_select_page(self):
        r = _get("/reference/api/select-page")
        assert r.status_code == 200
        assert "the-content" in r.text

    def test_slow_echo(self):
        r = _get("/reference/api/slow-echo", params={"q": "sync"})
        assert r.status_code == 200
        assert "sync" in r.text.lower()

    def test_upload_name(self):
        r = _post(
            "/reference/api/upload-name",
            files={"file": ("test.txt", b"hello", "text/plain")},
        )
        assert r.status_code == 200
        assert "test.txt" in r.text

    def test_echo_headers(self):
        r = httpx.get(
            f"{SX_BASE}/reference/api/echo-headers",
            headers={**HEADERS, "X-Custom-Token": "abc123"},
            timeout=TIMEOUT,
        )
        assert r.status_code == 200

    def test_echo_vals_get(self):
        r = _get("/reference/api/echo-vals", params={"category": "books"})
        assert r.status_code == 200

    def test_echo_vals_post(self):
        r = _post("/reference/api/echo-vals", data={"source": "demo", "page": "3"})
        assert r.status_code == 200

    def test_flaky(self):
        r = _get("/reference/api/flaky")
        assert r.status_code in (200, 503)

    def test_prompt_echo(self):
        r = httpx.get(
            f"{SX_BASE}/reference/api/prompt-echo",
            headers={**HEADERS, "SX-Prompt": "Alice"},
            timeout=TIMEOUT,
        )
        assert r.status_code == 200
        assert "Alice" in r.text

    def test_sse_time(self):
        """SSE endpoint returns event-stream content type."""
        with httpx.stream("GET", f"{SX_BASE}/reference/api/sse-time",
                          headers=HEADERS, timeout=TIMEOUT) as r:
            assert r.status_code == 200
            ct = r.headers.get("content-type", "")
            assert "text/event-stream" in ct
            # Read just the first chunk to verify format
            for chunk in r.iter_text():
                assert "event:" in chunk or "data:" in chunk
                break  # only need the first chunk