cc-ci/tests/n8n/functional/test_login_state.py

"""n8n — recipe-specific functional test (Phase 2 P3, ≥2 beyond parity).

n8n's `/rest/login` reports the auth/owner state — what the editor SPA calls right after settings
to decide between "show the owner-setup wizard" vs "show the login page" vs "we already have a
session". A fresh-deploy n8n with user-management enabled MUST return a JSON body indicating no
owner exists yet (status=200 + a no-owner / no-session shape). A still-starting n8n returns the
"n8n is starting up" placeholder HTML with the same 200; this test polls until a real JSON shape
arrives and asserts the auth subsystem responded — proving the user-management layer initialized
on top of the public settings (distinct from test_rest_settings.py, which only proves the public
settings endpoint).

Runs in the custom tier against the shared post-install deployment.
"""

from __future__ import annotations

import json
import os
import ssl
import sys
import urllib.request

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
from harness import http as harness_http  # noqa: E402


def _raw_get(url: str, timeout: int = 15):
    """Same shape as test_rest_settings._raw_get_json: returns (status, content_type, json_or_None,
    raw_excerpt_or_err)."""
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    req = urllib.request.Request(url, method="GET")
    try:
        resp = urllib.request.urlopen(req, timeout=timeout, context=ctx)
    except urllib.error.HTTPError as e:
        # n8n's /rest/login may return 4xx on a fresh install (no session); still informative
        ct = e.headers.get("Content-Type", "") if e.headers else ""
        try:
            raw = e.read()
        except Exception:  # noqa: BLE001
            return e.code, ct, None, "<read-failed>"
        if "application/json" not in ct:
            return e.code, ct, None, raw.decode(errors="replace")[:80]
        try:
            return e.code, ct, json.loads(raw), None
        except (json.JSONDecodeError, ValueError) as je:
            return e.code, ct, None, str(je)
    except Exception as e:  # noqa: BLE001
        return 0, "", None, str(e)
    ct = resp.headers.get("Content-Type", "")
    raw = resp.read()
    resp.close()
    if "application/json" not in ct:
        return resp.status, ct, None, raw.decode(errors="replace")[:80]
    try:
        return resp.status, ct, json.loads(raw), None
    except (json.JSONDecodeError, ValueError) as je:
        return resp.status, ct, None, str(je)


def test_login_endpoint_returns_json(live_app):
    """Poll /rest/login until it returns a real JSON response (not the SPA placeholder). Assert the
    response indicates the auth subsystem is up — either a JSON error envelope (no session) or a
    user / state object. n8n's exact response shape evolves: pre-1.x returned a flat error; recent
    versions return {"code":..., "message":..., "hint":...} for the no-session case. We accept any
    JSON object — but reject the placeholder HTML."""
    url = f"https://{live_app}/rest/login"

    state: dict = {}

    def _ready():
        status, ct, body, err = _raw_get(url)
        state.update({"status": status, "ct": ct, "err": err, "body": body})
        if status == 0:
            return None
        # The "starting up" placeholder is text/html with 200 -> body is None, ct doesn't contain
        # application/json. Reject until ct says JSON.
        if "application/json" not in ct:
            return None
        return body if body is not None else None

    body = harness_http.assert_converges(
        _ready,
        f"GET {url} returns JSON (n8n auth subsystem ready)",
        max_wait=180,
        interval=5,
    )

    # The auth endpoint returned a JSON body — that's the proof. The shape varies, but JSON it is.
    assert body is not None, f"/rest/login returned no parseable JSON: state={state}"
    # If it's a dict, it's the expected envelope; if it's a list, n8n shouldn't do that on this
    # endpoint, but accept either; only reject obvious non-shapes.
    assert isinstance(body, (dict, list)), (
        f"/rest/login returned unexpected JSON type {type(body).__name__}: {body!r}"
    )