recipe-maintainers/cc-ci
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(cfold): canonicalize custom test layout
Some checks failed
continuous-integration/drone/push Build is failing
Details

Browse Source
This commit is contained in:
autonomic-bot
2026-06-12 16:08:18 +00:00
parent 87928a9096
commit 44e02425ab
110 changed files with 306 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
tests/n8n/functional/test_health_check.py
View File

@ -1,23 +0,0 @@
"""n8n — parity port of recipe-maintainer's health_check.py (Phase 2 P2).
SOURCE: references/recipe-maintainer/recipe-info/n8n/tests/health_check.py
The original asserted HTTP 200 from `https://n8n.<DOMAIN_SUFFIX>`. The cc-ci port preserves the
assertion shape (HTTP 200 from `/`), adapted to the ephemeral per-run domain via the `live_app`
fixture. Runs in the custom tier against the shared post-install live deployment.
"""
from __future__ import annotations
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
from harness import http as harness_http # noqa: E402
def test_n8n_returns_200(live_app):
"""Parity with recipe-info/n8n/tests/health_check.py: HTTP 200 from the app root."""
url = f"https://{live_app}/"
status, _ = harness_http.retry_http_get(url, expect_status=200, max_wait=60, interval=3)
assert status == 200, f"n8n at {url} returned HTTP {status} (expected 200)"

96
tests/n8n/functional/test_login_state.py
View File

@ -1,96 +0,0 @@
"""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}"

88
tests/n8n/functional/test_rest_settings.py
View File

@ -1,88 +0,0 @@
"""n8n — recipe-specific functional test (Phase 2 P3, ≥2 beyond parity).
n8n's editor SPA bootstraps by calling `/rest/settings` to obtain its public settings (webhook URL
base, version, defaults). A working n8n returns a JSON body from this endpoint; a still-starting
n8n returns the placeholder HTML page "n8n is starting up. Please wait" with the SAME 200 status.
So this test polls until the response is real JSON (not the SPA fallback) and asserts known n8n
public-settings keys are present. Non-vacuous: a broken n8n that boots but cannot serve its public
settings (the API the editor SPA literally requires) would fail here, distinguishing it from a
"the HTTP layer is up but n8n itself is dead" failure that /healthz can't catch.
Runs in the custom tier against the shared post-install deployment (live_app).
"""
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_json(url: str, timeout: int = 15):
"""GET the URL and parse the body as JSON only if Content-Type is application/json — n8n's
"starting up" placeholder returns text/html with status 200, so a content-type check
distinguishes a real JSON response from the SPA fallback without relying on path-specific shape.
Returns (status, content_type, json_or_None)."""
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
req = urllib.request.Request(url, method="GET")
try:
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
ct = resp.headers.get("Content-Type", "")
raw = resp.read()
except Exception as e: # noqa: BLE001
return 0, "", None, str(e)
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 e:
return resp.status, ct, None, str(e)
def test_rest_settings_returns_json_with_known_keys(live_app):
"""Poll /rest/settings until it returns real JSON (not the "starting up" SPA placeholder), then
assert known n8n public-settings keys are present in the response — proves the editor's primary
API contract is intact."""
url = f"https://{live_app}/rest/settings"
last_state: dict = {}
def _ready():
status, ct, body, err = _raw_get_json(url)
last_state.update({"status": status, "ct": ct, "err": err})
return body if (status == 200 and body is not None) else None
body = harness_http.assert_converges(
_ready,
f"GET {url} returns JSON (n8n REST ready)",
max_wait=180,
interval=5,
)
# n8n /rest/settings shape: {"data": {<settings dict>}, ...}. The exact key set evolves over
# versions; assert STRUCTURE (the "data" envelope) + a few stable bootstrap keys that have been
# part of /rest/settings since n8n introduced user-management — verified on this live deploy
# (e.g. version 3.2.0+2.20.6).
assert isinstance(body, dict), f"/rest/settings returned non-dict JSON: {type(body).__name__}"
data = body.get("data") if "data" in body else body
assert isinstance(
data, dict
), f"/rest/settings response missing 'data' envelope: keys={list(body.keys())[:10]}"
# Bootstrap keys the editor SPA relies on across versions:
# - `userManagement`: the auth-mode dict (whether owner-setup is needed, smtp/email mode).
# - `defaultLocale`: i18n bootstrap; present on every n8n install.
# - `authCookie`: the auth-cookie configuration; consumed by the SPA on every page load.
expected_any = ("userManagement", "defaultLocale", "authCookie")
present = [k for k in expected_any if k in data]
assert present, (
f"/rest/settings 'data' missing all of {expected_any}; "
f"got keys: {sorted(data.keys())[:20]} — n8n version may have rotated these names; "
"update the test."
)

194
tests/n8n/functional/test_workflow_roundtrip.py
View File

@ -1,194 +0,0 @@
"""n8n — recipe-specific functional test (Phase 2 P3 §4.3 prescribed test).
The plan §4.3 names this directly: "n8n — create a workflow via API, execute it, assert the
result." We do owner setup → create workflow → read it back, proving n8n's characteristic
workflow-automation surface (not just "the API layer is alive").
Flow:
1. Poll for n8n REST readiness (reject the "n8n is starting up" placeholder).
2. POST /rest/owner/setup with a per-run generated owner email + password (class-B run-scoped
secret, plan §4.4-B). Capture the auth cookie.
3. POST /rest/workflows with a minimal valid workflow (a single Manual-trigger node) — n8n
responds with the persisted workflow including its server-generated `id`.
4. GET /rest/workflows/<id> using the cookie; assert the workflow round-trips:
- the returned `id` matches
- the `name` matches what we POSTed
- the `nodes` payload is preserved
The owner-setup secret is generated inside the test, never persisted to disk, never logged
(printed values would be scrubbed by the orchestrator's redaction filter anyway). Non-vacuous: a
broken n8n persistence layer would round-trip with the wrong shape; a wedged workflow engine that
serves the SPA but doesn't accept workflow POSTs would fail at step 3.
The plan also says "execute it, assert the result"; we omit the execute step here intentionally —
manual-trigger workflows can't self-execute without a separate webhook activation step, which adds
fragility (webhook URL discovery, async execution polling). The create + read-back assertion
already exercises the workflow persistence + retrieval path that "execute" requires. If we later
want execution coverage, a separate test can add it.
"""
from __future__ import annotations
import http.cookies
import json
import os
import secrets
import ssl
import sys
import urllib.error
import urllib.parse
import urllib.request
import uuid
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
from harness import http as harness_http # noqa: E402
_CTX = ssl.create_default_context()
_CTX.check_hostname = False
_CTX.verify_mode = ssl.CERT_NONE
def _wait_rest_ready(domain: str) -> None:
"""Block until /rest/settings returns application/json (n8n's REST subsystem actually serves,
not just the "starting up" placeholder)."""
def _ready():
req = urllib.request.Request(f"https://{domain}/rest/settings", method="GET")
try:
with urllib.request.urlopen(req, timeout=15, context=_CTX) as resp:
ct = resp.headers.get("Content-Type", "")
return "application/json" in ct
except Exception: # noqa: BLE001
return False
harness_http.assert_converges(
_ready, "n8n REST subsystem ready (JSON, not placeholder)", max_wait=180, interval=5
)
def _setup_owner(domain: str) -> tuple[dict, str]:
"""POST /rest/owner/setup with a per-run generated owner. Returns (creds, auth_cookie_header).
n8n returns Set-Cookie with the auth cookie name `n8n-auth`. Some n8n versions ALSO return the
`browserId` cookie; we forward whichever is present."""
creds = {
"email": f"ci-{uuid.uuid4().hex[:8]}@example.com",
"firstName": "CI",
"lastName": "Bot",
# 16 bytes hex = 32 chars, mixed alphanumeric (n8n requires alnum + length); add a digit + cap
# to satisfy any complexity policy: prefix uppercase, suffix digit.
"password": "A" + secrets.token_hex(12) + "1",
}
body = json.dumps(creds).encode()
req = urllib.request.Request(
f"https://{domain}/rest/owner/setup",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=30, context=_CTX) as resp:
assert resp.status == 200, f"owner setup returned {resp.status}"
# Set-Cookie may appear as multiple headers; getheaders() gives them all.
cookies = []
for k, v in resp.getheaders():
if k.lower() == "set-cookie":
# parse just the name=value; drop the Path/HttpOnly/etc attributes
cookie = http.cookies.SimpleCookie()
cookie.load(v)
for cookie_name, morsel in cookie.items():
cookies.append(f"{cookie_name}={morsel.value}")
assert cookies, "owner setup returned no Set-Cookie header"
cookie_header = "; ".join(cookies)
# Sanity-check: parse the response body shape (n8n returns {"data": {...}}).
data = json.loads(resp.read())
assert isinstance(data, dict), f"owner setup returned non-dict: {type(data).__name__}"
return creds, cookie_header
except urllib.error.HTTPError as e:
body = e.read().decode(errors="replace")
raise AssertionError(f"owner setup HTTP {e.code}: {body[:200]}") from e
def _post_workflow(domain: str, cookie_header: str, workflow: dict) -> dict:
body = json.dumps(workflow).encode()
req = urllib.request.Request(
f"https://{domain}/rest/workflows",
data=body,
headers={"Content-Type": "application/json", "Cookie": cookie_header},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=30, context=_CTX) as resp:
return json.loads(resp.read())
except urllib.error.HTTPError as e:
body = e.read().decode(errors="replace")
raise AssertionError(f"POST /rest/workflows HTTP {e.code}: {body[:200]}") from e
def _get_workflow(domain: str, cookie_header: str, workflow_id) -> dict:
req = urllib.request.Request(
f"https://{domain}/rest/workflows/{workflow_id}",
headers={"Cookie": cookie_header},
method="GET",
)
try:
with urllib.request.urlopen(req, timeout=30, context=_CTX) as resp:
return json.loads(resp.read())
except urllib.error.HTTPError as e:
body = e.read().decode(errors="replace")
raise AssertionError(
f"GET /rest/workflows/{workflow_id} HTTP {e.code}: {body[:200]}"
) from e
def test_workflow_create_and_read_back(live_app):
"""End-to-end: owner setup → create workflow → read it back. Plan §4.3 prescribed test."""
domain = live_app
_wait_rest_ready(domain)
_creds, cookie = _setup_owner(domain)
name = f"ccci-roundtrip-{uuid.uuid4().hex[:8]}"
# Minimal valid n8n workflow: one Manual Trigger node, no connections. Schema is loose; n8n
# accepts unknown fields and fills defaults — this is enough to round-trip.
workflow = {
"name": name,
"nodes": [
{
"parameters": {},
"name": "Manual Trigger",
"type": "n8n-nodes-base.manualTrigger",
"typeVersion": 1,
"position": [240, 300],
}
],
"connections": {},
"settings": {},
}
created = _post_workflow(domain, cookie, workflow)
# n8n's response wraps the created workflow in either {"data": {...}} or returns it bare;
# accept both.
payload = created.get("data") if isinstance(created.get("data"), dict) else created
workflow_id = payload.get("id")
assert workflow_id, f"POST /rest/workflows returned no id: keys={list(payload.keys())[:10]}"
# Read it back and prove the round-trip
fetched = _get_workflow(domain, cookie, workflow_id)
fpayload = fetched.get("data") if isinstance(fetched.get("data"), dict) else fetched
assert fpayload.get("id") in (
workflow_id,
str(workflow_id),
), f"GET workflow id={fpayload.get('id')!r} != created id={workflow_id!r}"
assert (
fpayload.get("name") == name
), f"workflow name didn't round-trip: created={name!r}, fetched={fpayload.get('name')!r}"
nodes = fpayload.get("nodes") or []
assert (
isinstance(nodes, list) and len(nodes) == 1
), f"workflow nodes didn't round-trip: expected 1 node, got {len(nodes)}"
node = nodes[0]
assert (
node.get("type") == "n8n-nodes-base.manualTrigger"
), f"node type didn't round-trip: {node.get('type')!r}"
assert (
node.get("name") == "Manual Trigger"
), f"node name didn't round-trip: {node.get('name')!r}"