wangbo/ComfyUI-Manager
1
0
Fork 0
mirror of https://github.com/Comfy-Org/ComfyUI-Manager.git synced 2026-06-23 00:09:25 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

test(e2e): real-server coverage for dedicated install flags incl. URL-form pip

Browse Source 
- tests/e2e/test_e2e_secgate_legacy_flags.py: 18 tests across 8 server
  config fixtures proving both arms (allow/deny) of each flag against a
  running ComfyUI legacy server, batch composite gating, transitive-pip
  non-consultation, and denial-copy contract. Zero-install discipline:
  the designated do-not-install sample for direct git-URL rows, a
  synthetic nonexistent URL for batch unknown-URL rows.
- tests/e2e/test_e2e_pip_url_form.py: 4 tests proving the pip gate is
  argument-content-agnostic with a URL-form pip spec — deny arm (403,
  no install-scripts reservation, denial log names the flag) and
  real-install arm (200 + reservation, executed at next restart, the
  package importable in the isolated E2E venv, then fully self-cleaned).
  Uses the owned purpose-built fixture
  git+https://github.com/ltdrdata/pip-test1-do-not-install (public,
  zero-dep, pure Python, seconds-fast) instead of an external
  third-party repo, keeping the test's lifetime under our control.
This commit is contained in:
Dr.Lt.Data 2026-06-08 02:12:14 +09:00
parent 16ba874566
commit 385d0c18e1
2 changed files with 1049 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

408
tests/e2e/test_e2e_pip_url_form.py Normal file
View File

@ -0,0 +1,408 @@
"""[goal299 step3] E2E: URL-form pip install through the S-B endpoint.
CONTRACT UNDER TEST (SHIPPED — goal265; this module adds tests only):
the ``allow_pip_install`` gate on POST /v2/customnode/install/pip is
**argument-content-agnostic** (goal299-mm-addendum.md §1): a URL-form
argument (``git+https://github.com/...``) exercises the SAME
dedicated-flag gate as a bare package name — no code path inspects
the argument before the gate decision (legacy/manager_server.py:1574).
FIXTURE NOTE (goal329 — WHY the owned fixture): the install arm
originally exercised ``git+https://github.com/facebookresearch/sam2``;
that external, unpinned, heavy dependency (torch-ecosystem build at
restart time) made the test hostage to upstream maintainer drift
(maintainer-fragility decision, goal329). It is replaced by the OWNED,
purpose-built fixture ``git+https://github.com/ltdrdata/pip-test1-do-not-install``
(PUBLIC; package ``pip-test1-do-not-install``, module
``pip_test1_do_not_install``, ``MARKER = "pip-test1-do-not-install:ok"``,
zero deps, pure Python, no import side effects). The gate contract is
argument-content-agnostic, so the URL swap changes NOTHING about what
is being proven — only the install payload shrinks to a harmless,
owned package.
Two arms (goal299-spec.md §1.1, LOCKED):
DENY arm flag=false at security_level=weak -> 403, NO reservation
entry appended, denial log names allow_pip_install
(weak proves the FLAG, not the security_level, decides).
INSTALL arm flag=true at security_level=strong, loopback -> 200 =
gate-pass + RESERVATION (deferred-execution semantics,
addendum §1: '#FORCE' reserves into install-scripts.txt;
the real `uv pip install` runs at the NEXT server start)
-> restart -> fixture module REALLY importable (MARKER
verified) in the isolated E2E venv -> self-clean ->
absent again.
Placement: NEW SIBLING of tests/e2e/test_e2e_secgate_legacy_flags.py
(spec §1.2) — that module carries an engineered zero-install guarantee
(its header :96-100), so the real-install arm must NOT live there.
Helper reuse contract (spec §1.2, LOCKED): import-reuse is limited to
``_stage_flags_config`` / ``_stop_legacy`` / ``_make_flags_server_fixture``
/ ``_post_pip`` / ``_log_offset``; everything else needed here is a
sibling-local definition (provenance-commented copies where applicable).
Self-cleaning + idempotent (spec §1.3 install-arm steps 1/6): pre-guard
uninstall, teardown uninstall (runs even on failure), reservation-file
hygiene on both ends. The venv is uv-managed and has NO pip module
(addendum R4) — all (un)installs go through ``<venv>/bin/uv``.
Requires a pre-built E2E environment (setup_e2e_env.sh); skip-marked
otherwise. The install arm additionally requires github.com reachability
(addendum R5) and is skip-marked offline; the deny arm always runs.
"""
from __future__ import annotations
import ast
import os
import socket
import subprocess
import pytest
# Import-reuse: the LOCKED helper subset only (goal299-spec.md §1.2).
from test_e2e_secgate_legacy_flags import (
_log_offset,
_make_flags_server_fixture,
_post_pip,
_stop_legacy,
)
E2E_ROOT = os.environ.get("E2E_ROOT", "")
COMFYUI_PATH = os.path.join(E2E_ROOT, "comfyui") if E2E_ROOT else ""
SCRIPTS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scripts")
SERVER_LOG = os.path.join(E2E_ROOT, "logs", "comfyui.log") if E2E_ROOT else ""
# Same port as the flags module (its module constant is not in the locked
# import set; the value is part of the shared legacy-fixture contract).
PORT = 8199
# Owned fixture (goal329 — see FIXTURE NOTE in the module docstring).
# Dist name uses hyphens, import name uses underscores; MARKER gives a
# stronger installed-for-real check than a bare import.
FIXTURE_URL = "git+https://github.com/ltdrdata/pip-test1-do-not-install"
FIXTURE_DIST = "pip-test1-do-not-install"
FIXTURE_MODULE = "pip_test1_do_not_install"
FIXTURE_MARKER = "pip-test1-do-not-install:ok"
VENV_PY = os.path.join(E2E_ROOT, "venv", "bin", "python") if E2E_ROOT else ""
VENV_UV = os.path.join(E2E_ROOT, "venv", "bin", "uv") if E2E_ROOT else ""
# Reservation file (path shape per tests/e2e/test_e2e_legacy_real_ops.py:538;
# producer: reserve_script, legacy/manager_core.py:1836-1843; consumer:
# comfyui_manager/prestartup_script.py:485 at next server start).
SCRIPTS_PATH = (
os.path.join(
COMFYUI_PATH, "user", "__manager", "startup-scripts", "install-scripts.txt"
)
if E2E_ROOT
else ""
)
# Post-reservation restart budget (goal329): the owned fixture is pure
# Python with zero deps — its `uv pip install` completes in seconds, so
# the restart wall time is dominated by the plain server boot
# (+prestartup resolver), observed well under 60s in this harness. 180s
# equals the proven readiness budget the flags module's _start_legacy
# uses for a plain legacy start (>=3x headroom over observed boot); the
# python-side timeout is shell+60 per the helper invariant.
RESTART_BUDGET_S = 180
pytestmark = pytest.mark.skipif(
not E2E_ROOT
or not os.path.isfile(os.path.join(E2E_ROOT, ".e2e_setup_complete")),
reason="E2E_ROOT not set or E2E environment not ready",
)
# ---------------------------------------------------------------------------
# Sibling-local helpers (spec §1.2 — outside the locked import set)
# ---------------------------------------------------------------------------
def _start_legacy_install(extra_env: dict, shell_timeout: int = RESTART_BUDGET_S) -> int:
"""Parameterized variant of test_e2e_secgate_legacy_flags._start_legacy
(:147-158) — provenance copy per goal299-spec.md §1.2. The flags
module's helper hardcodes its env dict and python-side timeout, which
can neither inject per-restart env nor take a per-call readiness
budget. TIMEOUT is forwarded to start_comfyui_legacy.sh as the shell
readiness budget; the python-side subprocess timeout MUST exceed it
(+60s) so the start script, not python, owns the timeout."""
env = {
**os.environ,
"E2E_ROOT": E2E_ROOT,
"PORT": str(PORT),
"TIMEOUT": str(shell_timeout),
**extra_env,
}
r = subprocess.run(
["bash", os.path.join(SCRIPTS_DIR, "start_comfyui_legacy.sh")],
capture_output=True, text=True, timeout=shell_timeout + 60, env=env,
)
if r.returncode != 0:
raise RuntimeError(
f"Failed to start ComfyUI (legacy, install restart):\n{r.stderr}"
)
for part in r.stdout.strip().split():
if part.startswith("COMFYUI_PID="):
return int(part.split("=")[1])
raise RuntimeError(f"Could not parse PID:\n{r.stdout}")
def _log_slice(offset: int) -> str:
"""Provenance copy of test_e2e_secgate_legacy_flags._log_slice
(:241-246) — not in the locked import set (spec §1.2 fallback clause:
fixture semantics are the lock, import path is not)."""
if not os.path.isfile(SERVER_LOG):
return ""
with open(SERVER_LOG, "r", encoding="utf-8", errors="replace") as f:
f.seek(offset)
return f.read()
def _scripts_state() -> str:
"""Current install-scripts.txt content, or the sentinel 'ABSENT' —
the file may legitimately not exist (spec §1.3 deny step 2: it is
created lazily by reserve_script on first reservation)."""
if not os.path.isfile(SCRIPTS_PATH):
return "ABSENT"
with open(SCRIPTS_PATH, "r", encoding="utf-8", errors="replace") as f:
return f.read()
def _venv_fixture_probe() -> subprocess.CompletedProcess:
"""Import the fixture module in the isolated E2E venv and print its
MARKER (subprocess — the observable for really-installed /
really-absent). rc != 0 means absent; rc == 0 AND the MARKER value in
stdout means installed for real (stronger than a bare import)."""
return subprocess.run(
[VENV_PY, "-c", f"import {FIXTURE_MODULE} as m; print(m.MARKER)"],
capture_output=True, text=True, timeout=60,
)
def _uninstall_fixture() -> None:
"""Uninstall the fixture distribution from the E2E venv via uv.
The venv has NO pip module (`python -m pip` fails — addendum R4);
`<venv>/bin/uv pip uninstall --python <venv-python>` is the working
path. A no-op when the dist is absent (uv warns, exits 0)."""
subprocess.run(
[VENV_UV, "pip", "uninstall", FIXTURE_DIST, "--python", VENV_PY],
capture_output=True, text=True, timeout=180,
)
def _strip_fixture_reservation() -> None:
"""Remove any fixture residual line from install-scripts.txt
(addendum R8: a leftover reservation would silently mutate the venv
on the NEXT restart, possibly a different test's). Matches both the
hyphenated dist/URL form and the underscored module form."""
if not os.path.isfile(SCRIPTS_PATH):
return
with open(SCRIPTS_PATH, "r", encoding="utf-8", errors="replace") as f:
lines = f.readlines()
kept = [
ln for ln in lines
if FIXTURE_DIST not in ln.lower() and FIXTURE_MODULE not in ln.lower()
]
if kept != lines:
with open(SCRIPTS_PATH, "w", encoding="utf-8") as f:
f.writelines(kept)
def _github_reachable() -> bool:
"""Runtime network probe for the install arm (addendum R5)."""
try:
socket.create_connection(("github.com", 443), timeout=10).close()
return True
except OSError:
return False
# ===========================================================================
# DENY arm — flag=false, security_level=weak (weak proves flag-not-level
# decides; goal299-spec.md §1.3 TestPipUrlFormDeny)
# ===========================================================================
comfyui_pip_url_deny = _make_flags_server_fixture(False, False, "weak")
class TestPipUrlFormDeny:
"""URL-form POST is denied by the allow_pip_install flag gate exactly
like a bare package name (argument-content-agnostic, addendum §1):
403, no install-side-effect (no reservation entry), denial log names
the flag. Shared-POST shape (spec §1.3, LOCKED): exactly ONE POST via
a class-scoped record fixture; the three tests assert exclusively
against the record — independent of execution order, no re-reads of
live state."""
@pytest.fixture(scope="class")
def deny_post_record(self, comfyui_pip_url_deny):
offset = _log_offset()
snapshot_before = _scripts_state()
resp = _post_pip(FIXTURE_URL)
return {
"log_offset_before": offset,
"scripts_snapshot_before": snapshot_before,
"status_code": resp.status_code,
"log_slice_after": _log_slice(offset),
"scripts_state_after": _scripts_state(),
}
def test_url_form_denied_403(self, deny_post_record):
"""403 from the dedicated-flag gate despite security_level=weak
(deny-direction decoupling, mm §2 row 'deny arm, after POST')."""
assert deny_post_record["status_code"] == 403, (
f"URL-form pip POST: expected 403 (allow_pip_install=false "
f"overrides sl=weak), got {deny_post_record['status_code']} — "
f"the gate must be argument-content-agnostic."
)
def test_url_form_deny_no_reservation(self, deny_post_record):
"""No reservation entry appended on deny (mm §2: install-scripts.txt
'no entry written'; HTTP 200 would have meant reservation, so the
durable on-disk state is the decisive side-effect observable)."""
before = deny_post_record["scripts_snapshot_before"]
after = deny_post_record["scripts_state_after"]
assert after == before, (
f"deny arm: install-scripts.txt changed across the denied POST "
f"— a reservation leaked past the gate.\nbefore:\n{before}\n"
f"after:\n{after}"
)
if after != "ABSENT":
assert FIXTURE_DIST not in after.lower(), (
f"deny arm: a fixture line is present in install-scripts.txt "
f"after a denied POST:\n{after}"
)
def test_url_form_deny_log_names_flag(self, deny_post_record):
"""Denial log names allow_pip_install (SECURITY_MESSAGE_FLAG_PIP,
legacy/manager_server.py:47) and carries no security-level framing
for this denial (goal265 spec §1.1 invariant 6 — same assertion
shape as the flags module's SC-23)."""
log = deny_post_record["log_slice_after"]
assert "allow_pip_install" in log, (
"deny arm: denial log does not name allow_pip_install "
f"(SECURITY_MESSAGE_FLAG_PIP). Slice:\n{log[-1500:]}"
)
assert "security level to 'normal-'" not in log, (
"deny arm: denial log still carries the misleading "
"security-level copy (SECURITY_MESSAGE_NORMAL_MINUS)."
)
# ===========================================================================
# INSTALL arm — flag=true, security_level=strong (strong proves allow-side
# decoupling; goal299-spec.md §1.3 TestPipUrlFormInstall)
# ===========================================================================
comfyui_pip_url_install = _make_flags_server_fixture(False, True, "strong")
@pytest.mark.network
class TestPipUrlFormInstall:
"""Full deferred-install round trip (mm §2 observable-outcome table):
POST 200 = gate-pass + reservation (NOT installation) -> restart
executes the reserved `uv pip install -U git+...` at prestartup
-> fixture module imports with the expected MARKER in the isolated
venv -> self-clean -> absent.
Asserting importability right after the 200 would be a guaranteed
false-FAIL (mm §2 note) — the restart between POST and the import
assertion is the production execution path (addendum §4.1)."""
@pytest.fixture(scope="class", autouse=True)
def _network_guard(self):
if not _github_reachable():
pytest.skip("offline: github.com unreachable — install arm "
"requires network (deny arm unaffected)")
@pytest.fixture(autouse=True)
def _self_clean(self):
"""Teardown runs even on failure (spec §1.3 step 6): uninstall the
dist, strip any residual reservation line (addendum R8), and prove
the venv is back to baseline (import fails again)."""
yield
_uninstall_fixture()
_strip_fixture_reservation()
assert _venv_fixture_probe().returncode != 0, (
f"self-clean: {FIXTURE_MODULE} is still importable in the E2E "
f"venv after `uv pip uninstall {FIXTURE_DIST}` — venv NOT "
f"restored to baseline."
)
def test_url_form_install_end_to_end(self, comfyui_pip_url_install):
# Step 1 — pre-guard (idempotency, spec §1.3 step 1): a previous
# aborted run must not turn this into a false-PASS.
_uninstall_fixture()
_strip_fixture_reservation()
assert _venv_fixture_probe().returncode != 0, (
f"pre-guard: {FIXTURE_MODULE} already importable before the "
f"test — uninstall guard failed; venv not at baseline."
)
# Step 2 — POST the URL-form argument; 200 = gate-pass +
# reservation under the deferred-execution semantics (mm §1).
resp = _post_pip(FIXTURE_URL)
assert resp.status_code == 200, (
f"install arm: expected 200 (allow_pip_install=true overrides "
f"sl=strong), got {resp.status_code} — allow-side decoupling "
f"broken for the URL-form argument."
)
# Step 3 — reservation entry: a Python-list-repr line containing
# the fixture URL substring (producer reserve_script,
# legacy/manager_core.py:1836-1843; precedent assertion shape
# test_e2e_legacy_real_ops.py:516-538). NOT a shell-string match.
state = _scripts_state()
assert state != "ABSENT", (
"install arm: install-scripts.txt missing after 200 — "
"no reservation was written."
)
fixture_lines = [ln for ln in state.splitlines() if FIXTURE_URL in ln]
assert fixture_lines, (
f"install arm: no reservation line contains {FIXTURE_URL!r}.\n"
f"file content:\n{state}"
)
reserved = ast.literal_eval(fixture_lines[0])
assert isinstance(reserved, list) and "#FORCE" in reserved, (
f"install arm: reservation line is not the expected "
f"['..', '#FORCE', <pip cmd>...] list-repr shape: {reserved!r}"
)
# Step 4 — restart: the production execution path for the
# reservation (prestartup_script.py:485 consumes the file at
# server start). Budget rationale at RESTART_BUDGET_S — the
# owned zero-dep fixture installs in seconds; the budget covers
# the plain server boot, no build env vars needed (goal329
# retired the CUDA-build determinism knob with the heavy
# upstream package).
_stop_legacy()
_start_legacy_install({}, shell_timeout=RESTART_BUDGET_S)
# Step 5 — post-restart: reservation CONSUMED (prestartup removes
# the processed file) and the fixture REALLY importable in the
# venv, proven by its MARKER value (stronger than bare import:
# the marker ties the import to the owned fixture's content).
post_state = _scripts_state()
assert post_state == "ABSENT" or FIXTURE_URL not in post_state, (
f"install arm: reservation NOT consumed by the restart — "
f"install-scripts.txt still references the fixture:\n{post_state}"
)
probe = _venv_fixture_probe()
assert probe.returncode == 0, (
f"install arm: `import {FIXTURE_MODULE}` still fails in the "
f"E2E venv after the reservation-executing restart — install "
f"did not happen or did not target the venv.\n"
f"stderr:\n{probe.stderr[-500:]}"
)
assert FIXTURE_MARKER in probe.stdout, (
f"install arm: fixture imported but MARKER mismatch — "
f"expected {FIXTURE_MARKER!r} in stdout, got: {probe.stdout!r}"
)
# Version breadcrumb for triage (owned repo, so drift risk is
# retired — the breadcrumb now just documents what was installed).
show = subprocess.run(
[VENV_UV, "pip", "show", FIXTURE_DIST, "--python", VENV_PY],
capture_output=True, text=True, timeout=60,
)
print(f"\n[goal329 install-evidence] uv pip show {FIXTURE_DIST}:\n"
f"{show.stdout}")

641
tests/e2e/test_e2e_secgate_legacy_flags.py Normal file
View File

@ -0,0 +1,641 @@
"""[goal265 step4 — RED] E2E endpoint binding for the dedicated install flags
``allow_git_url_install`` / ``allow_pip_install`` on the LEGACY server.
TARGET CONTRACT (NOT YET IMPLEMENTED — goal265-spec.md §1, LOCKED):
S-A POST /v2/customnode/install/git_url gated by allow_git_url_install
S-B POST /v2/customnode/install/pip gated by allow_pip_install
S-C batch unknown-URL install: retained middle+ entry gate AND the
allow_git_url_install full predicate replaces the high+ risky check;
unknown-pip 'block' branch stays UNCONDITIONAL (Q1).
The security_level term is fully REMOVED from S-A/S-B (spec §1.1 inv. 1);
denial logs name the responsible flag, not the security level (inv. 6).
These tests are EXPECTED TO FAIL against current code (today S-A/S-B are
``is_allowed_security_level('high+')``-gated) — RED confirmation is Step 5.
Do NOT weaken them to pass against today's code.
This module delivers the LGU2/LPP2 legacy-fixture follow-up explicitly
deferred by tests/e2e/test_e2e_secgate_default.py (its header, lines 31-34).
SC rows covered (goal265-scenarios.md), grouped by server config combination
(one legacy-server launch per combination, config.ini staged per flags):
CFG-A (git=T, pip=T, sl=strong, nm=public): SC-01, SC-11, SC-08
CFG-B (git=T, pip=F, sl=normal, nm=public): SC-02, SC-04, SC-10, SC-17,
SC-18 (transitive-pip arm)
CFG-C (git=F, pip=T, sl=normal, nm=public): SC-16
CFG-D (git=F, pip=F, sl=weak, nm=public): SC-05, SC-09, SC-13,
SC-23 (denial-log copy arm)
CFG-E (git=T, pip=T, sl=weak, nm=public): SC-24 (Q1 unknown-pip block)
CFG-F (git=T, pip=T, sl=normal, nm=public): SC-25A/B (guards, flags-on)
CFG-G (git=F, pip=F, sl=normal, nm=public): SC-25A/B (guards, flags-off)
CFG-H (git=T, pip=F, sl=normal, nm=personal_cloud): SC-30 (note below)
SC-30 limitation (deliberate, documented): the E2E harness listens on
127.0.0.1, so the personal_cloud arm of the middle+ entry gate is satisfied
via is_local_mode as well; the public-listener arm of personal_cloud is
proven at predicate level (tests/common/test_install_flag_predicate.py
SC-03/SC-12) — this row re-proves the endpoint binding with
network_mode=personal_cloud staged.
SC-18 strategy: endpoint-level arm — git-URL install transaction completes
(200) with allow_pip_install=false and the captured log slice contains NO
allow_pip_install denial, proving the pip flag is never consulted inside a
git transaction (MM §1.4). The spec's integration-level alternative
(in-process gitclone_install with execute_install_script mocked) remains
open to Step-7 if this arm proves under-discriminating.
Side-effect control: batch rows pre-seed the target node directory so the
queue worker's gitclone_install short-circuits on "Already exists" — the
GATE decision (the contract under test) happens synchronously in
_install_custom_node BEFORE the worker runs, so the batch response's
``failed`` list is a complete gate observable without real clones.
Requires a pre-built E2E environment (setup_e2e_env.sh); skip-marked
otherwise (harness precedent: tests/e2e/test_e2e_secgate_default.py).
"""
from __future__ import annotations
import configparser
import os
import shutil
import subprocess
import pytest
import requests
E2E_ROOT = os.environ.get("E2E_ROOT", "")
COMFYUI_PATH = os.path.join(E2E_ROOT, "comfyui") if E2E_ROOT else ""
SCRIPTS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scripts")
PORT = 8199
BASE_URL = f"http://127.0.0.1:{PORT}"
CONFIG_PATH = os.path.join(COMFYUI_PATH, "user", "__manager", "config.ini")
CONFIG_BACKUP = CONFIG_PATH + ".before-flags"
SERVER_LOG = os.path.join(E2E_ROOT, "logs", "comfyui.log") if E2E_ROOT else ""
# Purpose-built test-fixture repo (same constant as
# tests/e2e/test_e2e_legacy_real_ops.py TRUSTED_GIT_URL) — the DESIGNATED
# do-not-install sample for every S-A surface test that performs a REAL
# clone (direct /v2/customnode/install/git_url).
UNKNOWN_GIT_URL = "https://github.com/ltdrdata/nodepack-test1-do-not-install"
UNKNOWN_GIT_DIRNAME = "nodepack-test1-do-not-install"
# Batch S-C "unknown-URL" rows need a files URL genuinely NOT in the
# custom-node DB (scenario preconditions SC-04/08/09/30: "file URL NOT in
# node DB" -> get_risky_level returns 'high+' -> the dedicated-flag gate
# engages). DISCOVERY (task #297 real-server run): the do-not-install
# sample above IS registered in the Comfy-Org main-channel
# custom-node-list (served from the manager cache), so get_risky_level
# returns 'middle+' for it and the flag gate is bypassed — using it for
# batch rows silently tests the wrong path (observed as an SC-09
# false-FAIL: flag=false yet the item queued via middle+ at sl=weak).
# The batch rows therefore use a SYNTHETIC nonexistent URL. ZERO-INSTALL
# guarantee holds by construction: deny rows (SC-08/09) never reach the
# worker, and allow rows (SC-04/30) pre-seed the matching placeholder dir
# so the worker's gitclone_install short-circuits on "Already exists" —
# the URL is never cloned (and could not be: the repo does not exist).
BATCH_UNKNOWN_GIT_URL = "https://github.com/ltdrdata/goal265-nonexistent-unknown-node"
BATCH_UNKNOWN_DIRNAME = "goal265-nonexistent-unknown-node"
# Long-standing custom-node-list.json entry — used by SC-24/SC-25B rows that
# need a files URL the DB KNOWS (get_risky_level must not short-circuit at
# high+ on the URL check).
KNOWN_GIT_URL = "https://github.com/ltdrdata/ComfyUI-Impact-Pack"
KNOWN_GIT_DIRNAME = "ComfyUI-Impact-Pack"
UNKNOWN_PIP_PKG = "cm-goal265-nonexistent-pip-pkg-xyz"
pytestmark = pytest.mark.skipif(
not E2E_ROOT
or not os.path.isfile(os.path.join(E2E_ROOT, ".e2e_setup_complete")),
reason="E2E_ROOT not set or E2E environment not ready",
)
# ---------------------------------------------------------------------------
# Config staging + server lifecycle helpers
# ---------------------------------------------------------------------------
def _stage_flags_config(git: bool, pip: bool, security_level: str,
network_mode: str) -> None:
"""Patch config.ini with the row preconditions (backup-once pattern,
mirrors start_comfyui_strict.sh)."""
if not os.path.isfile(CONFIG_BACKUP):
shutil.copy(CONFIG_PATH, CONFIG_BACKUP)
cfg = configparser.ConfigParser(strict=False)
cfg.read(CONFIG_PATH)
if "default" not in cfg:
cfg["default"] = {}
cfg["default"]["allow_git_url_install"] = "true" if git else "false"
cfg["default"]["allow_pip_install"] = "true" if pip else "false"
cfg["default"]["security_level"] = security_level
cfg["default"]["network_mode"] = network_mode
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
cfg.write(f)
def _restore_config() -> None:
if os.path.isfile(CONFIG_BACKUP):
shutil.move(CONFIG_BACKUP, CONFIG_PATH)
def _start_legacy() -> int:
env = {**os.environ, "E2E_ROOT": E2E_ROOT, "PORT": str(PORT)}
r = subprocess.run(
["bash", os.path.join(SCRIPTS_DIR, "start_comfyui_legacy.sh")],
capture_output=True, text=True, timeout=180, env=env,
)
if r.returncode != 0:
raise RuntimeError(f"Failed to start ComfyUI (legacy):\n{r.stderr}")
for part in r.stdout.strip().split():
if part.startswith("COMFYUI_PID="):
return int(part.split("=")[1])
raise RuntimeError(f"Could not parse PID:\n{r.stdout}")
def _stop_legacy() -> None:
env = {**os.environ, "E2E_ROOT": E2E_ROOT, "PORT": str(PORT)}
subprocess.run(
["bash", os.path.join(SCRIPTS_DIR, "stop_comfyui.sh")],
capture_output=True, text=True, timeout=30, env=env,
)
def _make_flags_server_fixture(git: bool, pip: bool, security_level: str,
network_mode: str = "public"):
"""Class-scoped fixture factory: stage config -> start legacy server ->
yield -> stop server -> restore config."""
@pytest.fixture(scope="class")
def _server():
_stage_flags_config(git, pip, security_level, network_mode)
try:
pid = _start_legacy()
try:
yield pid
finally:
_stop_legacy()
finally:
_restore_config()
return _server
comfyui_flags_a = _make_flags_server_fixture(True, True, "strong")
comfyui_flags_b = _make_flags_server_fixture(True, False, "normal")
comfyui_flags_c = _make_flags_server_fixture(False, True, "normal")
comfyui_flags_d = _make_flags_server_fixture(False, False, "weak")
comfyui_flags_e = _make_flags_server_fixture(True, True, "weak")
comfyui_flags_f = _make_flags_server_fixture(True, True, "normal")
comfyui_flags_g = _make_flags_server_fixture(False, False, "normal")
comfyui_flags_h = _make_flags_server_fixture(
True, False, "normal", network_mode="personal_cloud")
# ---------------------------------------------------------------------------
# Request / observation helpers
# ---------------------------------------------------------------------------
def _post_git_url(url: str) -> requests.Response:
return requests.post(
f"{BASE_URL}/v2/customnode/install/git_url", data=url, timeout=120,
)
def _post_pip(packages: str) -> requests.Response:
return requests.post(
f"{BASE_URL}/v2/customnode/install/pip", data=packages, timeout=60,
)
def _batch_install_item(ui_id: str, files: list[str],
pip: list[str] | None = None) -> dict:
"""Unknown-version batch install item shape consumed by
legacy _install_custom_node (version='unknown' -> files URL path)."""
return {
"id": ui_id,
"ui_id": ui_id,
"version": "unknown",
"files": files,
"pip": pip or [],
"channel": "default",
"mode": "cache",
}
def _post_batch(payload: dict) -> requests.Response:
return requests.post(
f"{BASE_URL}/v2/manager/queue/batch", json=payload, timeout=120,
)
def _log_offset() -> int:
return os.path.getsize(SERVER_LOG) if os.path.isfile(SERVER_LOG) else 0
def _log_slice(offset: int) -> str:
if not os.path.isfile(SERVER_LOG):
return ""
with open(SERVER_LOG, "r", encoding="utf-8", errors="replace") as f:
f.seek(offset)
return f.read()
def _custom_nodes_dir() -> str:
return os.path.join(COMFYUI_PATH, "custom_nodes")
def _preseed_node_dir(dirname: str) -> str:
"""Create a placeholder node dir so the queue worker's gitclone_install
short-circuits on 'Already exists' (no real clone; the gate decision
under test already happened synchronously)."""
target = os.path.join(_custom_nodes_dir(), dirname)
os.makedirs(target, exist_ok=True)
marker = os.path.join(target, ".goal265-preseed")
with open(marker, "w", encoding="utf-8") as f:
f.write("goal265 step4 gate-test placeholder\n")
return target
def _remove_node_dir(dirname: str) -> None:
for candidate in (
os.path.join(_custom_nodes_dir(), dirname),
os.path.join(_custom_nodes_dir(), ".disabled", dirname),
os.path.join(_custom_nodes_dir(), dirname + ".disabled"),
):
if os.path.isdir(candidate):
shutil.rmtree(candidate, ignore_errors=True)
# ===========================================================================
# CFG-A: git=T, pip=T, sl=strong, nm=public
# ===========================================================================
class TestFlagsOnStrongLevel:
"""Allow-direction decoupling: flags=true must allow S-A/S-B even at
security_level=strong (today: 403). Batch entry gate stays
security_level-coupled (middle+ fails at strong)."""
def test_sc01_git_url_allowed_at_strong_when_flag_true(
self, comfyui_flags_a):
"""SC-01: git=true, sl=strong, loopback -> POST git_url returns 200
(today 403 — proves security_level irrelevance, allow direction).
Pre-removes the fixture node so a real (tiny) clone proceeds; a
200 via res.action=='skip' is equally a gate-pass observable."""
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
try:
resp = _post_git_url(UNKNOWN_GIT_URL)
assert resp.status_code == 200, (
f"SC-01: expected 200 (flag=true overrides sl=strong), got "
f"{resp.status_code} — security_level still gates S-A. "
f"Body: {resp.text[:200]}"
)
finally:
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
def test_sc11_pip_allowed_at_strong_when_flag_true(self, comfyui_flags_a):
"""SC-11: pip=true, sl=strong, loopback -> POST pip returns 200
(today 403). text-unidecode: pure-Python, tiny, idempotent — same
trusted constant as test_e2e_legacy_real_ops.py."""
resp = _post_pip("text-unidecode")
assert resp.status_code == 200, (
f"SC-11: expected 200 (flag=true overrides sl=strong), got "
f"{resp.status_code} — security_level still gates S-B."
)
def test_sc08_batch_unknown_url_denied_at_strong(self, comfyui_flags_a):
"""SC-08: git=true, sl=strong -> batch unknown-URL install DENIED —
the retained middle+ entry gate fails first (composite gate:
middle+ AND flag; spec §1.2 S-C row, ':1427' gate UNCHANGED)."""
item = _batch_install_item("sc08-unknown", [BATCH_UNKNOWN_GIT_URL])
resp = _post_batch({"install": [item]})
assert resp.status_code == 200, resp.text[:200]
failed = resp.json().get("failed", [])
assert "sc08-unknown" in failed, (
"SC-08: batch unknown-URL install must stay denied at "
"sl=strong (middle+ entry gate) even with "
f"allow_git_url_install=true — failed={failed!r}"
)
# ===========================================================================
# CFG-B: git=T, pip=F, sl=normal, nm=public
# ===========================================================================
class TestGitFlagOnDefaultLevel:
"""Default security_level: git flag alone opens S-A; pip stays closed."""
def test_sc02_git_url_allowed_at_default_level(self, comfyui_flags_b):
"""SC-02: git=true, sl=normal (default), loopback -> 200
(today 403). Self-cleaning (pre+post remove): a leftover clone
would turn the NEXT install of the same repo into an
'Already exists' 400 (gitclone_install unknown-repo path has no
skip arm — observed as an SC-17 false-FAIL in the first #297 run)."""
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
try:
resp = _post_git_url(UNKNOWN_GIT_URL)
assert resp.status_code == 200, (
f"SC-02: expected 200 at sl=normal with flag=true, got "
f"{resp.status_code}. Body: {resp.text[:200]}"
)
finally:
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
def test_sc10_invalid_url_reaches_validation_400(self, comfyui_flags_b):
"""SC-10: git=true, sl=normal -> POST an INVALID url -> 400 from
installer validation, NOT 403 from the gate. Distinguishes gate-403
from install-400 (today the gate 403s before URL validation)."""
resp = _post_git_url("not-a-url")
assert resp.status_code == 400, (
f"SC-10: expected 400 (installer validation), got "
f"{resp.status_code} — a 403 means the flag gate did not open."
)
def test_sc17_git_open_pip_closed_independent(self, comfyui_flags_b):
"""SC-17: git=true, pip=false, sl=normal -> git_url 200, pip 403
(flags fully independent, [D1][D2]). Self-cleaning like SC-02 —
a stale clone from an earlier S-A test would 400 the git arm."""
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
try:
git_resp = _post_git_url(UNKNOWN_GIT_URL)
pip_resp = _post_pip("text-unidecode")
assert git_resp.status_code == 200, (
f"SC-17 git arm: expected 200, got {git_resp.status_code}"
)
assert pip_resp.status_code == 403, (
f"SC-17 pip arm: expected 403, got {pip_resp.status_code}"
)
finally:
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
def test_sc18_git_transaction_never_consults_pip_flag(
self, comfyui_flags_b):
"""SC-18: git=true, pip=false -> a git-URL install transaction
(clone + execute_install_script dependency step) COMPLETES, and the
captured log slice contains no allow_pip_install denial — the pip
flag governs ONLY the standalone S-B surface (MM §1.4 transaction
scope; spec §1.1 invariant 4).
Fresh install forced (pre-remove) so execute_install_script actually
runs inside the transaction window."""
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
offset = _log_offset()
try:
resp = _post_git_url(UNKNOWN_GIT_URL)
log = _log_slice(offset)
assert resp.status_code == 200, (
f"SC-18: git transaction failed ({resp.status_code}) with "
"pip=false — transitive scope broken? Body: "
f"{resp.text[:200]}"
)
assert "allow_pip_install" not in log, (
"SC-18: the git-URL install transaction consulted/logged "
"allow_pip_install — per-surface scope violated (MM §1.4):\n"
+ log[-2000:]
)
finally:
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
def test_sc04_batch_unknown_url_queued_at_default_level(
self, comfyui_flags_b):
"""SC-04: git=true, sl=normal -> batch unknown-URL install is
QUEUED (failed list empty): middle+ entry passes at normal local
AND the flag replaces the high+ risky check (today 403: risky
high+ fails at normal). Target dir pre-seeded — see module
docstring 'Side-effect control'."""
_preseed_node_dir(BATCH_UNKNOWN_DIRNAME)
try:
item = _batch_install_item("sc04-unknown", [BATCH_UNKNOWN_GIT_URL])
resp = _post_batch({"install": [item]})
assert resp.status_code == 200, resp.text[:200]
failed = resp.json().get("failed", [])
assert "sc04-unknown" not in failed, (
"SC-04: batch unknown-URL install still denied at "
"sl=normal with allow_git_url_install=true — the risky "
f"high+ check was not replaced by the flag. failed={failed!r}"
)
finally:
_remove_node_dir(BATCH_UNKNOWN_DIRNAME)
# ===========================================================================
# CFG-C: git=F, pip=T, sl=normal, nm=public
# ===========================================================================
class TestPipFlagOnDefaultLevel:
def test_sc16_git_closed_pip_open_independent(self, comfyui_flags_c):
"""SC-16: git=false, pip=true, sl=normal -> git_url 403, pip 200
(flags fully independent, [D1][D2])."""
git_resp = _post_git_url(UNKNOWN_GIT_URL)
pip_resp = _post_pip("text-unidecode")
assert git_resp.status_code == 403, (
f"SC-16 git arm: expected 403, got {git_resp.status_code}"
)
assert pip_resp.status_code == 200, (
f"SC-16 pip arm: expected 200, got {pip_resp.status_code}"
)
# ===========================================================================
# CFG-D: git=F, pip=F, sl=weak, nm=public
# ===========================================================================
class TestFlagsOffWeakLevel:
"""Deny-direction decoupling: flags=false must deny S-A/S-B even at
security_level=weak (today: 200) — and the denial log must name the
responsible flag (SC-23 log arm, spec §1.3)."""
def test_sc05_sc23_git_url_denied_at_weak_log_names_flag(
self, comfyui_flags_d):
"""SC-05: git=false, sl=weak, loopback -> 403 (today 200 — deny
direction of security_level irrelevance).
SC-23 (log arm): the denial log names allow_git_url_install and
drops the security-level framing (SECURITY_MESSAGE_NORMAL_MINUS)."""
offset = _log_offset()
resp = _post_git_url(UNKNOWN_GIT_URL)
log = _log_slice(offset)
assert resp.status_code == 403, (
f"SC-05: expected 403 (flag=false overrides sl=weak), got "
f"{resp.status_code} — security_level still opens S-A."
)
assert "allow_git_url_install" in log, (
"SC-23: denial log does not name allow_git_url_install "
f"(spec §1.3 SECURITY_MESSAGE_FLAG_GIT_URL). Slice:\n{log[-1500:]}"
)
assert "security level to 'normal-'" not in log, (
"SC-23: denial log still carries the misleading "
"security-level copy (SECURITY_MESSAGE_NORMAL_MINUS)."
)
def test_sc13_sc23_pip_denied_at_weak_log_names_flag(
self, comfyui_flags_d):
"""SC-13: pip=false, sl=weak, loopback -> 403 (today 200).
SC-23 (log arm): denial log names allow_pip_install
(spec §1.3 SECURITY_MESSAGE_FLAG_PIP)."""
offset = _log_offset()
resp = _post_pip("text-unidecode")
log = _log_slice(offset)
assert resp.status_code == 403, (
f"SC-13: expected 403 (flag=false overrides sl=weak), got "
f"{resp.status_code} — security_level still opens S-B."
)
assert "allow_pip_install" in log, (
"SC-23: denial log does not name allow_pip_install "
f"(spec §1.3 SECURITY_MESSAGE_FLAG_PIP). Slice:\n{log[-1500:]}"
)
def test_sc09_batch_unknown_url_denied_when_flag_false(
self, comfyui_flags_d):
"""SC-09: git=false, sl=weak -> batch unknown-URL install denied —
the flag gate fails despite weak (today: allowed). Composite gate's
flag term is the decider here (middle+ passes at weak local)."""
item = _batch_install_item("sc09-unknown", [BATCH_UNKNOWN_GIT_URL])
resp = _post_batch({"install": [item]})
assert resp.status_code == 200, resp.text[:200]
failed = resp.json().get("failed", [])
assert "sc09-unknown" in failed, (
"SC-09: batch unknown-URL install must be denied with "
f"allow_git_url_install=false even at sl=weak. failed={failed!r}"
)
# ===========================================================================
# CFG-E: git=T, pip=T, sl=weak, nm=public — Q1 guard
# ===========================================================================
class TestUnknownPipStaysBlocked:
def test_sc24_batch_unknown_pip_block_is_unconditional(
self, comfyui_flags_e):
"""SC-24 (Q1 guard): batch item whose files are ALL DB-known (no
high+ short-circuit on the URL check) but whose pip list names an
UNKNOWN package -> still denied. get_risky_level's 'block' branch
stays UNCONDITIONAL — the flags do NOT open it (spec §5 freeze
item 7), even with both flags true at sl=weak."""
item = _batch_install_item(
"sc24-known-url-unknown-pip",
[KNOWN_GIT_URL],
pip=[UNKNOWN_PIP_PKG],
)
resp = _post_batch({"install": [item]})
assert resp.status_code == 200, resp.text[:200]
failed = resp.json().get("failed", [])
assert "sc24-known-url-unknown-pip" in failed, (
"SC-24: unknown-pip 'block' branch opened — it must remain "
f"unconditional regardless of the new flags. failed={failed!r}"
)
# ===========================================================================
# CFG-F / CFG-G: out-of-scope legacy guards (SC-25A/B) — flags have NO effect
# ===========================================================================
def _batch_op_item(ui_id: str) -> dict:
"""Item for middle-gated batch ops (fix/uninstall/update). The GATE
check precedes node resolution, so a placeholder spec is sufficient:
the queue worker's later failure on the nonexistent node is irrelevant
to the synchronous gate observable (the response's failed list)."""
return {
"id": ui_id,
"ui_id": ui_id,
"version": "unknown",
"files": [f"https://example.com/{ui_id}"],
}
class _LegacyGuardRows:
"""Shared assertions for SC-25A/B — executed under BOTH flag combos
((t,t) via CFG-F and (f,f) via CFG-G); outcomes must be IDENTICAL,
proving the flags have zero effect on middle/middle+ surfaces."""
flags_label = ""
def _assert_sc25a(self):
"""SC-25A: legacy middle-gated ops (fix/uninstall/update) stay
allowed at sl=normal local for this flag combination."""
payload = {
"fix": [_batch_op_item(f"sc25a-fix-{self.flags_label}")],
"uninstall": [_batch_op_item(f"sc25a-unin-{self.flags_label}")],
"update": [_batch_op_item(f"sc25a-upd-{self.flags_label}")],
}
resp = _post_batch(payload)
assert resp.status_code == 200, resp.text[:200]
failed = resp.json().get("failed", [])
gate_denied = [x for x in failed if x.startswith("sc25a-")]
assert gate_denied == [], (
f"SC-25A({self.flags_label}): middle-gated batch ops were "
f"gate-denied at sl=normal — flags must have ZERO effect on "
f"out-of-scope surfaces. failed={failed!r}"
)
def _assert_sc25b(self):
"""SC-25B: batch KNOWN-node install (middle+ path, DB-resolved
files URL) stays allowed at sl=normal local for this flag
combination. Known-node dir pre-seeded — no real install."""
_preseed_node_dir(KNOWN_GIT_DIRNAME)
try:
item = _batch_install_item(
f"sc25b-known-{self.flags_label}", [KNOWN_GIT_URL])
resp = _post_batch({"install": [item]})
assert resp.status_code == 200, resp.text[:200]
failed = resp.json().get("failed", [])
assert f"sc25b-known-{self.flags_label}" not in failed, (
f"SC-25B({self.flags_label}): KNOWN-node batch install was "
f"denied at sl=normal — middle+ path must be unaffected by "
f"the flags. failed={failed!r}"
)
finally:
_remove_node_dir(KNOWN_GIT_DIRNAME)
class TestGuardsFlagsOn(_LegacyGuardRows):
flags_label = "flags-on"
def test_sc25a_middle_ops_allowed_with_flags_on(self, comfyui_flags_f):
self._assert_sc25a()
def test_sc25b_known_install_allowed_with_flags_on(self, comfyui_flags_f):
self._assert_sc25b()
class TestGuardsFlagsOff(_LegacyGuardRows):
flags_label = "flags-off"
def test_sc25a_middle_ops_allowed_with_flags_off(self, comfyui_flags_g):
self._assert_sc25a()
def test_sc25b_known_install_allowed_with_flags_off(self, comfyui_flags_g):
self._assert_sc25b()
# ===========================================================================
# CFG-H: git=T, pip=F, sl=normal, nm=personal_cloud
# ===========================================================================
class TestPersonalCloudBatch:
def test_sc30_batch_unknown_url_queued_under_personal_cloud(
self, comfyui_flags_h):
"""SC-30: git=true, sl=normal, nm=personal_cloud -> batch
unknown-URL install queued: the S-C composite gate's middle+ arm
passes (is_local_mode OR is_personal_cloud) AND the flag is true.
See module docstring for the loopback-harness limitation; the
public-listener personal_cloud arm is proven at predicate level."""
_preseed_node_dir(BATCH_UNKNOWN_DIRNAME)
try:
item = _batch_install_item("sc30-unknown", [BATCH_UNKNOWN_GIT_URL])
resp = _post_batch({"install": [item]})
assert resp.status_code == 200, resp.text[:200]
failed = resp.json().get("failed", [])
assert "sc30-unknown" not in failed, (
"SC-30: batch unknown-URL install denied under "
f"network_mode=personal_cloud with flag=true. "
f"failed={failed!r}"
)
finally:
_remove_node_dir(BATCH_UNKNOWN_DIRNAME)