youngkingdom/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-03-19 05:37:51 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix : make playwright tests idempotent (#13332)

Browse Source 
### What problem does this PR solve?

Playwright tests previously depended on cross-file execution order
(`auth -> provider -> dataset -> chat`).
This change makes setup explicit and idempotent via fixtures so tests
can run independently.

- Added/standardized prerequisite fixtures in
`test/playwright/conftest.py`:
- `ensure_auth_context`, `ensure_model_provider_configured`,
`ensure_dataset_ready`, `ensure_chat_ready`
- Made provisioning reusable/idempotent with `RUN_ID`-based resource
naming.
- Synced auth envs (`E2E_ADMIN_EMAIL`, `E2E_ADMIN_PASSWORD`) into seeded
creds.
- Fixed provider cache freshness (`auth_header`/`page` refresh on cache
hit).

Also included minimal stability fixes:
- dataset create stale-element click handling,
- search wait logic for results/empty-state,
- agent create-menu handling,
- agent run-step retry when run UI doesn’t open first click.

### Type of change

- [x] Test fix
- [x] Refactoring

---------

Co-authored-by: Liu An <asiro@qq.com>
This commit is contained in:
Idriss Sbaaoui
2026-03-04 10:07:14 +08:00
committed by GitHub
parent 1c87f97dde
commit 2f4ca38adf
10 changed files with 500 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
README_ar.md
View File

@ -5,7 +5,7 @@
</div>
<p align="center">
<a href="./README.md"><img alt="README in English" src="https://img.shields.io/badge/English-DBEDFA"></a>
<a href="./README.md"><img alt="README in English" src="https://img.shields.io/badge/English-DFE0E5"></a>
<a href="./README_zh.md"><img alt="简体中文版自述文件" src="https://img.shields.io/badge/简体中文-DFE0E5"></a>
<a href="./README_tzh.md"><img alt="繁體版中文自述文件" src="https://img.shields.io/badge/繁體中文-DFE0E5"></a>
<a href="./README_ja.md"><img alt="日本語のREADME" src="https://img.shields.io/badge/日本語-DFE0E5"></a>
@ -13,7 +13,7 @@
<a href="./README_id.md"><img alt="Bahasa Indonesia" src="https://img.shields.io/badge/Bahasa Indonesia-DFE0E5"></a>
<a href="./README_pt_br.md"><img alt="Português(Brasil)" src="https://img.shields.io/badge/Português(Brasil)-DFE0E5"></a>
<a href="./README_fr.md"><img alt="README en Français" src="https://img.shields.io/badge/Français-DFE0E5"></a>
<a href="./README_ar.md"><img alt="README in Arabic" src="https://img.shields.io/badge/Arabic-DFE0E5"></a>
<a href="./README_ar.md"><img alt="README in Arabic" src="https://img.shields.io/badge/Arabic-DBEDFA"></a>
</p>
<p align="center">

129
test/playwright/README.md
View File

@ -1,122 +1,59 @@
# Playwright auth UI tests
# Playwright Test README
## Quick start
Smoke test (always runs at least one test):
## One-line command (run everything)
```bash
pytest -q test/playwright -m smoke
BASE_URL=http://localhost:9222 E2E_ADMIN_EMAIL=admin@ragflow.io E2E_ADMIN_PASSWORD=admin PW_FIXTURE_DEBUG=1 uv run pytest -q test/playwright -s --junitxml=/tmp/playwright-full.xml
```
Run all auth UI tests:
## Common commands
Run smoke subset:
```bash
pytest -q test/playwright -m auth
BASE_URL=http://localhost:9222 E2E_ADMIN_EMAIL=admin@ragflow.io E2E_ADMIN_PASSWORD=admin uv run pytest -q test/playwright -m smoke -s --junitxml=/tmp/playwright-smoke.xml
```
If you use `uv`:
Run full suite:
```bash
uv run pytest -q test/playwright -m smoke
BASE_URL=http://localhost:9222 E2E_ADMIN_EMAIL=admin@ragflow.io E2E_ADMIN_PASSWORD=admin uv run pytest -q test/playwright -s --junitxml=/tmp/playwright-full.xml
```
## Environment variables
Required/optional:
- `BASE_URL` (default: `http://127.0.0.1`)
- Example dev UI: `http://localhost:9222`.
- For Docker (`SVR_WEB_HTTP_PORT=80`), set `BASE_URL=http://localhost`.
- `LOGIN_PATH` (default: `/login`)
- `SEEDED_USER_EMAIL` and `SEEDED_USER_PASSWORD` (optional; enables login success test)
- `DEMO_CREDS=1` (optional; uses demo credentials `qa@infiniflow.com` / `123` for login success test)
- `REG_EMAIL_BASE` (default: `qa@infiniflow.org`)
- `REG_EMAIL_UNIQUE=1` (optional; enables unique registration emails like `qa_1700000000000_123456@infiniflow.org`)
- `POST_LOGIN_PATH` (optional; expected path after login success, e.g. `/`)
- `REGISTER_ENABLED_EXPECTED` (optional; reserved for future gating checks)
Diagnostics and debugging:
- `PW_STEP_LOG=1` enable step logging
- `PW_NET_LOG=1` log `requestfailed` + console errors during the run
- `PW_TRACE=1` save a Playwright trace on failure
- `PW_BROWSER` (default: `chromium`)
- `PW_HEADLESS` (default: `1`, set `0` to see the browser)
- `PLAYWRIGHT_ACTION_TIMEOUT_MS` (default: `30000`)
- Legacy: `PW_TIMEOUT_MS`
- `PW_SLOWMO_MS` (default: `0`)
- `PLAYWRIGHT_HANG_TIMEOUT_S` (default: `1800`, set `0` to disable)
- Legacy: `HANG_TIMEOUT_S`
- `PLAYWRIGHT_AUTH_READY_TIMEOUT_MS` (default: `15000`)
- Legacy: `AUTH_READY_TIMEOUT_MS`
## What runs without credentials
- `auth/test_smoke_auth_page.py` (marker: `smoke`, always runs)
- `auth/test_toggle_login_register.py` (skips if register toggle is gated off)
- `auth/test_validation_presence.py`
- `auth/test_sso_optional.py` (skips if no SSO providers are rendered)
- `auth/test_register_success_optional.py` (skips if register toggle is gated off)
- `auth/test_register_then_login_flow.py` (skips unless `REG_EMAIL_UNIQUE=1`)
`auth/test_login_success_optional.py` only runs if `DEMO_CREDS=1` or `SEEDED_USER_EMAIL` and `SEEDED_USER_PASSWORD` are set.
## Login success examples
Run with demo credentials:
Run one file in isolation:
```bash
DEMO_CREDS=1 BASE_URL=http://localhost:9222 \
pytest -q test/playwright/auth/test_login_success_optional.py::test_login_success_optional -s -vv
BASE_URL=http://localhost:9222 E2E_ADMIN_EMAIL=admin@ragflow.io E2E_ADMIN_PASSWORD=admin uv run pytest -q test/playwright/e2e/test_next_apps_agent.py -s --junitxml=/tmp/playwright-agent.xml
```
Run with env credentials:
Run one test case in isolation:
```bash
SEEDED_USER_EMAIL=user@yourdomain.com SEEDED_USER_PASSWORD=secret BASE_URL=http://localhost:9222 \
pytest -q test/playwright/auth/test_login_success_optional.py::test_login_success_optional -s -vv
BASE_URL=http://localhost:9222 E2E_ADMIN_EMAIL=admin@ragflow.io E2E_ADMIN_PASSWORD=admin uv run pytest -q test/playwright/e2e/test_next_apps_chat.py::test_chat_create_select_dataset_and_receive_answer_flow -s -x --junitxml=/tmp/playwright-chat-one.xml
```
## Registration examples
## Argument reference
Registration rejects plus-addressing; the backend only allows local-part characters `[A-Za-z0-9_.-]`.
- `uv run`: run `pytest` inside the project-managed Python environment.
- `pytest`: test runner.
- `-q`: quieter output.
- `test/playwright`: run the whole Playwright suite folder.
- `test/playwright/...py`: run one file only.
- `::test_name`: run one test function only.
- `-m smoke`: run tests with `@pytest.mark.smoke`.
- `-s`: show `print()` and fixture logs live.
- `-x`: stop at first failure.
- `--junitxml=/tmp/<name>.xml`: write machine-readable results to XML.
Register only:
## Environment variables used in commands
```bash
REG_EMAIL_UNIQUE=1 BASE_URL=http://localhost:9222 \
pytest -q test/playwright/auth/test_register_success_optional.py::test_register_success_optional -s -vv
```
- `BASE_URL`: app URL (this suite is currently run against `http://localhost:9222`).
- `E2E_ADMIN_EMAIL`: login email for authenticated flows.
- `E2E_ADMIN_PASSWORD`: login password for authenticated flows.
- `PW_FIXTURE_DEBUG=1`: optional; prints fixture provisioning details.
Register then login (single test):
## Output and artifacts
```bash
REG_EMAIL_UNIQUE=1 BASE_URL=http://localhost:9222 \
pytest -q test/playwright/auth/test_register_then_login_flow.py::test_register_then_login_flow -s -vv
```
Run the end-to-end demo script:
```bash
BASE_URL=http://localhost:9222 \
scripts/run_auth_demo.sh
```
## Artifacts on failure
Artifacts are written to:
- `test/playwright/artifacts/`
- per-test screenshots are stored under `test/playwright/artifacts/<testname>/`
On failure, the suite writes:
- a full-page screenshot (`.png`)
- a full HTML dump (`.html`)
- a diagnostics log (`.log`)
- an optional trace (`.zip`) if `PW_TRACE=1`
## Hang investigation
- Automatic stack dump after `PLAYWRIGHT_HANG_TIMEOUT_S` seconds.
- Manual dump: `kill -USR1 <pytest_pid>` (writes traceback to stderr).
- JUnit XML files are written to `/tmp/...` from `--junitxml`.
- Screenshots and diagnostics are written under:
- `test/playwright/artifacts/`

365
test/playwright/conftest.py
View File

@ -50,6 +50,8 @@ AUTH_SUBMIT_SELECTOR = (
_PUBLIC_KEY_CACHE = None
_RSA_CIPHER_CACHE = None
_HANG_WATCHDOG_INSTALLED = False
_PROVIDER_READY_CACHE: dict[str, dict] = {}
_DATASET_READY_CACHE: dict[str, dict] = {}
class _RegisterDisabled(RuntimeError):
@ -85,6 +87,15 @@ def _env_int_with_fallback(primary: str, fallback: str | None, default: int) ->
return default
def _sync_seeded_credentials_from_admin_env() -> None:
admin_email = os.getenv("E2E_ADMIN_EMAIL")
admin_password = os.getenv("E2E_ADMIN_PASSWORD")
if admin_email and not os.getenv("SEEDED_USER_EMAIL"):
os.environ["SEEDED_USER_EMAIL"] = admin_email
if admin_password and not os.getenv("SEEDED_USER_PASSWORD"):
os.environ["SEEDED_USER_PASSWORD"] = admin_password
def _sanitize_timeout_ms(value: int | None, fallback: int | None) -> int | None:
if value is None or value <= 0:
return fallback
@ -274,6 +285,73 @@ def _api_post_json(url: str, payload: dict, timeout_s: int = 10) -> tuple[int, d
raise RuntimeError(f"URLError: {exc}") from exc
def _api_request_json(
url: str,
method: str = "GET",
payload: dict | None = None,
headers: dict | None = None,
timeout_s: int = 10,
) -> tuple[int, dict | None]:
data = None
if payload is not None:
data = json.dumps(payload).encode("utf-8")
req_headers = {"Content-Type": "application/json"}
if headers:
req_headers.update(headers)
req = Request(url, data=data, headers=req_headers, method=method)
try:
with urlopen(req, timeout=timeout_s) as resp:
body = resp.read()
if body:
try:
return resp.status, json.loads(body.decode("utf-8"))
except Exception:
return resp.status, None
return resp.status, None
except HTTPError as exc:
body = exc.read()
parsed = None
if body:
try:
parsed = json.loads(body.decode("utf-8"))
except Exception:
parsed = None
raise RuntimeError(
f"{method} {url} failed with HTTPError {exc.code}: {parsed or body!r}"
) from exc
except URLError as exc:
raise RuntimeError(f"{method} {url} failed with URLError: {exc}") from exc
def _response_data(payload: dict | None) -> dict:
if not isinstance(payload, dict):
return {}
if payload.get("code") not in (0, None):
raise RuntimeError(f"API returned failure payload: {payload}")
data = payload.get("data")
return data if isinstance(data, dict) else {}
def _extract_auth_header_from_page(page) -> str:
token = page.evaluate(
"""
() => {
const auth = localStorage.getItem('Authorization');
if (auth && auth.length) return auth;
const token = localStorage.getItem('Token');
if (token && token.length) return token;
return '';
}
"""
)
if not token:
raise AssertionError(
"Missing Authorization/Token in localStorage after login. "
"Cannot provision prerequisites via API."
)
return str(token)
def _rsa_encrypt_password(password: str) -> str:
global _PUBLIC_KEY_CACHE
global _RSA_CIPHER_CACHE
@ -762,6 +840,7 @@ def reg_password() -> str:
@pytest.fixture(scope="session")
def seeded_user_credentials(base_url: str, login_url: str, browser) -> tuple[str, str]:
_sync_seeded_credentials_from_admin_env()
env_email = os.getenv("SEEDED_USER_EMAIL")
env_password = os.getenv("SEEDED_USER_PASSWORD")
if env_email and env_password:
@ -836,6 +915,253 @@ def reg_nickname() -> str:
return REG_NICKNAME_DEFAULT
@pytest.fixture(scope="session")
def run_id() -> str:
value = os.getenv("RUN_ID")
if not value:
value = f"{int(time.time())}_{secrets.token_hex(2)}"
safe = _sanitize_filename(value) or f"{int(time.time())}_{secrets.token_hex(2)}"
os.environ["RUN_ID"] = safe
return safe
@pytest.fixture(scope="module")
def ensure_auth_context(
flow_page,
login_url: str,
seeded_user_credentials,
):
from test.playwright.helpers.auth_waits import wait_for_login_complete
page_instance = flow_page
email, password = seeded_user_credentials
timeout_ms = _playwright_auth_ready_timeout_ms() or DEFAULT_TIMEOUT_MS
token_wait_js = """
() => {
const token = localStorage.getItem('Token');
const auth = localStorage.getItem('Authorization');
return Boolean((token && token.length) || (auth && auth.length));
}
"""
try:
if "/login" not in page_instance.url:
page_instance.wait_for_function(token_wait_js, timeout=1500)
return page_instance
except Exception:
pass
page_instance.goto(login_url, wait_until="domcontentloaded")
active_form = page_instance.locator(AUTH_ACTIVE_FORM_SELECTOR)
expect(active_form).to_have_count(1, timeout=timeout_ms)
email_input = active_form.locator(AUTH_EMAIL_INPUT_SELECTOR).first
password_input = active_form.locator(AUTH_PASSWORD_INPUT_SELECTOR).first
submit_button = active_form.locator(AUTH_SUBMIT_SELECTOR).first
expect(email_input).to_be_visible(timeout=timeout_ms)
expect(password_input).to_be_visible(timeout=timeout_ms)
email_input.fill(email)
password_input.fill(password)
password_input.blur()
try:
submit_button.click(timeout=timeout_ms)
except PlaywrightTimeoutError:
submit_button.click(force=True, timeout=timeout_ms)
wait_for_login_complete(page_instance, timeout_ms=timeout_ms)
return page_instance
def _ensure_model_provider_ready_via_api(base_url: str, auth_header: str) -> dict:
headers = {"Authorization": auth_header}
_, my_llms_payload = _api_request_json(
_build_url(base_url, "/v1/llm/my_llms"), headers=headers
)
my_llms_data = _response_data(my_llms_payload)
has_provider = bool(my_llms_data)
created_provider = False
zhipu_key = os.getenv("ZHIPU_AI_API_KEY")
if not has_provider and zhipu_key:
_, set_key_payload = _api_request_json(
_build_url(base_url, "/v1/llm/set_api_key"),
method="POST",
payload={"llm_factory": "ZHIPU-AI", "api_key": zhipu_key},
headers=headers,
)
_response_data(set_key_payload)
has_provider = True
created_provider = True
_, my_llms_payload = _api_request_json(
_build_url(base_url, "/v1/llm/my_llms"), headers=headers
)
my_llms_data = _response_data(my_llms_payload)
if not has_provider:
pytest.skip("No model provider configured and ZHIPU_AI_API_KEY is not set.")
_, tenant_payload = _api_request_json(
_build_url(base_url, "/v1/user/tenant_info"), headers=headers
)
tenant_data = _response_data(tenant_payload)
tenant_id = tenant_data.get("tenant_id")
if not tenant_id:
raise RuntimeError(f"tenant_info missing tenant_id: {tenant_data}")
if not tenant_data.get("llm_id"):
llm_id = "glm-4-flash@ZHIPU-AI" if "ZHIPU-AI" in my_llms_data else None
if not llm_id:
pytest.skip(
"Provider exists but no default llm_id could be inferred for tenant setup."
)
tenant_payload = {
"tenant_id": tenant_id,
"llm_id": llm_id,
"embd_id": tenant_data.get("embd_id") or "BAAI/bge-small-en-v1.5@Builtin",
"img2txt_id": tenant_data.get("img2txt_id") or "",
"asr_id": tenant_data.get("asr_id") or "",
"tts_id": tenant_data.get("tts_id"),
}
_, set_tenant_payload = _api_request_json(
_build_url(base_url, "/v1/user/set_tenant_info"),
method="POST",
payload=tenant_payload,
headers=headers,
)
_response_data(set_tenant_payload)
return {
"tenant_id": tenant_id,
"has_provider": True,
"created_provider": created_provider,
"llm_factories": list(my_llms_data.keys()) if isinstance(my_llms_data, dict) else [],
}
@pytest.fixture(scope="module")
def ensure_model_provider_configured(
ensure_auth_context,
base_url: str,
seeded_user_credentials,
):
page_instance = ensure_auth_context
auth_header = _extract_auth_header_from_page(page_instance)
email = seeded_user_credentials[0] if seeded_user_credentials else "unknown"
cache_key = f"{base_url}|{email}|provider"
cached = _PROVIDER_READY_CACHE.get(cache_key)
if cached:
cached["page"] = page_instance
cached["auth_header"] = auth_header
return cached
provider_info = _ensure_model_provider_ready_via_api(base_url, auth_header)
payload = {
"page": page_instance,
"auth_header": auth_header,
"email": email,
**provider_info,
}
if _env_bool("PW_FIXTURE_DEBUG", False):
print(
"[prereq] provider_ready "
f"email={email} created_provider={payload.get('created_provider', False)} "
f"llm_factories={payload.get('llm_factories', [])}",
flush=True,
)
_PROVIDER_READY_CACHE[cache_key] = payload
return payload
def _find_dataset_by_name(kbs_payload: dict | None, dataset_name: str) -> dict | None:
data = _response_data(kbs_payload)
kbs = data.get("kbs")
if not isinstance(kbs, list):
return None
for item in kbs:
if isinstance(item, dict) and item.get("name") == dataset_name:
return item
return None
def _ensure_dataset_ready_via_api(
base_url: str, auth_header: str, dataset_name: str
) -> dict:
headers = {"Authorization": auth_header}
list_url = _build_url(base_url, "/v1/kb/list?page=1&page_size=200")
_, list_payload = _api_request_json(list_url, method="POST", payload={}, headers=headers)
existing = _find_dataset_by_name(list_payload, dataset_name)
if existing:
return {
"kb_id": existing.get("id"),
"kb_name": dataset_name,
"reused": True,
}
_, create_payload = _api_request_json(
_build_url(base_url, "/v1/kb/create"),
method="POST",
payload={"name": dataset_name},
headers=headers,
)
created_data = _response_data(create_payload)
kb_id = created_data.get("id")
if kb_id:
return {"kb_id": kb_id, "kb_name": dataset_name, "reused": False}
_, list_payload_after = _api_request_json(
list_url, method="POST", payload={}, headers=headers
)
existing_after = _find_dataset_by_name(list_payload_after, dataset_name)
if not existing_after:
raise RuntimeError(
f"Dataset {dataset_name!r} not found after kb/create response={create_payload}"
)
return {
"kb_id": existing_after.get("id"),
"kb_name": dataset_name,
"reused": False,
}
@pytest.fixture(scope="module")
def ensure_dataset_ready(
ensure_model_provider_configured,
base_url: str,
run_id: str,
):
provider_state = ensure_model_provider_configured
dataset_name = f"e2e-dataset-{run_id}"
cache_key = f"{base_url}|{provider_state.get('email', 'unknown')}|{dataset_name}"
cached = _DATASET_READY_CACHE.get(cache_key)
if cached:
return cached
dataset_info = _ensure_dataset_ready_via_api(
base_url,
provider_state["auth_header"],
dataset_name,
)
payload = {
**dataset_info,
"run_id": run_id,
}
if _env_bool("PW_FIXTURE_DEBUG", False):
print(
"[prereq] dataset_ready "
f"kb_name={payload.get('kb_name')} reused={payload.get('reused')} "
f"kb_id={payload.get('kb_id')}",
flush=True,
)
_DATASET_READY_CACHE[cache_key] = payload
return payload
@pytest.fixture(scope="module")
def ensure_chat_ready(ensure_dataset_ready):
return ensure_dataset_ready
@pytest.fixture
def snap(page, request):
if "flow_page" in request.fixturenames:
@ -1215,17 +1541,36 @@ def _locator_is_topmost(locator) -> bool:
def auth_click():
def _click(locator, label: str = "click") -> None:
timeout_ms = _playwright_auth_ready_timeout_ms()
try:
locator.click(timeout=timeout_ms)
except PlaywrightTimeoutError as exc:
if "intercepts pointer events" in str(exc) and _locator_is_topmost(
locator
):
if _env_bool("PW_FIXTURE_DEBUG", False):
print(f"[auth-click] forcing {label}", flush=True)
locator.click(force=True, timeout=timeout_ms)
attempts = 3
for idx in range(attempts):
try:
locator.click(timeout=timeout_ms)
return
raise
except PlaywrightTimeoutError as exc:
message = str(exc).lower()
can_force = (
"intercepts pointer events" in message
or "element was detached" in message
or "element is not stable" in message
)
if not can_force:
raise
if "intercepts pointer events" in message and not _locator_is_topmost(
locator
):
if idx >= attempts - 1:
raise
time.sleep(0.15)
continue
try:
if _env_bool("PW_FIXTURE_DEBUG", False):
print(f"[auth-click] forcing {label} attempt={idx + 1}", flush=True)
locator.click(force=True, timeout=timeout_ms)
return
except PlaywrightTimeoutError:
if idx >= attempts - 1:
raise
time.sleep(0.15)
return _click

33
test/playwright/e2e/test_dataset_upload_parse.py
View File

@ -7,9 +7,8 @@ from urllib.parse import urljoin
import pytest
from playwright.sync_api import expect
from test.playwright.helpers._auth_helpers import ensure_authed
from test.playwright.helpers.flow_steps import flow_params, require
from test.playwright.helpers.auth_selectors import EMAIL_INPUT, PASSWORD_INPUT, SUBMIT_BUTTON
from test.playwright.helpers.auth_waits import wait_for_login_complete
from test.playwright.helpers.response_capture import capture_response
from test.playwright.helpers.datasets import (
delete_uploaded_file,
@ -37,8 +36,6 @@ def step_01_login(
auth_click,
seeded_user_credentials,
):
email, password = seeded_user_credentials
repo_root = Path(__file__).resolve().parents[3]
file_paths = [
repo_root / "test/benchmark/test_docs/Doc1.pdf",
@ -52,25 +49,14 @@ def step_01_login(
flow_state["filenames"] = [path.name for path in file_paths]
with step("open login page"):
flow_page.goto(login_url, wait_until="domcontentloaded")
form, _ = active_auth_context()
email_input = form.locator(EMAIL_INPUT)
password_input = form.locator(PASSWORD_INPUT)
with step("fill credentials"):
expect(email_input).to_have_count(1)
expect(password_input).to_have_count(1)
email_input.fill(email)
password_input.fill(password)
password_input.blur()
with step("submit login"):
submit_button = form.locator(SUBMIT_BUTTON)
expect(submit_button).to_have_count(1)
auth_click(submit_button, "submit_login")
with step("wait for login"):
wait_for_login_complete(flow_page, timeout_ms=RESULT_TIMEOUT_MS)
ensure_authed(
flow_page,
login_url,
active_auth_context,
auth_click,
seeded_user_credentials=seeded_user_credentials,
timeout_ms=RESULT_TIMEOUT_MS,
)
flow_state["logged_in"] = True
snap("login_complete")
@ -276,6 +262,7 @@ def test_dataset_upload_parse_and_delete_flow(
flow_state,
base_url,
login_url,
ensure_model_provider_configured,
active_auth_context,
step,
snap,

43
test/playwright/e2e/test_next_apps_agent.py
View File

@ -269,25 +269,31 @@ def step_06_run_agent(
else:
run_button = run_root
expect(run_button).to_be_visible(timeout=RESULT_TIMEOUT_MS)
try:
auth_click(run_button, "agent_run")
except Exception:
page.wait_for_timeout(500)
auth_click(run_button, "agent_run_retry")
run_attempts = max(1, int(os.getenv("PW_AGENT_RUN_ATTEMPTS", "2")))
last_error = None
for attempt in range(run_attempts):
if attempt > 0:
page.wait_for_timeout(500)
try:
auth_click(run_button, f"agent_run_attempt_{attempt + 1}")
except Exception as exc:
last_error = exc
continue
try:
run_ui_locator.first.wait_for(state="visible", timeout=run_ui_timeout_ms)
flow_state["agent_running"] = True
snap("agent_run_started")
return
except Exception as exc:
last_error = exc
try:
run_ui_locator.first.wait_for(state="visible", timeout=run_ui_timeout_ms)
except Exception:
_raise_with_diagnostics(
page,
"Agent run UI did not open after clicking Run.",
snap=snap,
snap_name="agent_run_missing",
)
flow_state["agent_running"] = True
snap("agent_run_started")
return
suffix = f" last_error={last_error}" if last_error else ""
_raise_with_diagnostics(
page,
f"Agent run UI did not open after clicking Run ({run_attempts} attempts).{suffix}",
snap=snap,
snap_name="agent_run_missing",
)
def step_07_send_chat(
@ -378,6 +384,7 @@ def test_agent_create_then_import_json_then_run_and_wait_idle_flow(
flow_state,
base_url,
login_url,
ensure_dataset_ready,
active_auth_context,
step,
snap,

1
test/playwright/e2e/test_next_apps_chat.py
View File

@ -108,6 +108,7 @@ def test_chat_create_select_dataset_and_receive_answer_flow(
flow_state,
base_url,
login_url,
ensure_chat_ready,
active_auth_context,
step,
snap,

17
test/playwright/e2e/test_next_apps_search.py
View File

@ -9,6 +9,7 @@ from test.playwright.helpers._next_apps_helpers import (
_goto_home,
_nav_click,
_open_create_from_list,
_search_query_input,
_select_first_dataset_and_save,
_unique_name,
_wait_for_url_or_testid,
@ -20,7 +21,9 @@ def _wait_for_results_navigation(page, timeout_ms: int = RESULT_TIMEOUT_MS) -> N
() => {
const top = document.querySelector("[data-testid='top-nav']");
const navs = Array.from(document.querySelectorAll('[role="navigation"]'));
return navs.some((nav) => !top || !top.contains(nav));
if (navs.some((nav) => !top || !top.contains(nav))) return true;
const body = (document.body && document.body.innerText || '').toLowerCase();
return body.includes('no results found');
}
"""
page.wait_for_function(wait_js, timeout=timeout_ms)
@ -38,7 +41,10 @@ def _wait_for_results_navigation(page, timeout_ms: int = RESULT_TIMEOUT_MS) -> N
)
navs = page.locator("[role='navigation']")
target = navs.first if index < 0 else navs.nth(index)
expect(target).to_be_visible(timeout=timeout_ms)
if index >= 0:
expect(target).to_be_visible(timeout=timeout_ms)
return
expect(page.locator("text=/no results found/i").first).to_be_visible(timeout=timeout_ms)
def step_01_ensure_authed(
@ -144,9 +150,7 @@ def step_05_select_dataset(
require(flow_state, "search_created")
page = flow_page
with step("select dataset"):
search_input = page.locator(
"input[placeholder*='How can I help you today']"
).first
search_input = _search_query_input(page)
_select_first_dataset_and_save(
page,
timeout_ms=RESULT_TIMEOUT_MS,
@ -169,7 +173,7 @@ def step_06_run_query(
):
require(flow_state, "search_input_ready")
page = flow_page
search_input = page.locator("input[placeholder*='How can I help you today']").first
search_input = _search_query_input(page)
with step("run search query"):
expect(search_input).to_be_visible(timeout=RESULT_TIMEOUT_MS)
search_input.fill("ragflow")
@ -197,6 +201,7 @@ def test_search_create_select_dataset_and_results_nav_appears_flow(
flow_state,
base_url,
login_url,
ensure_dataset_ready,
active_auth_context,
step,
snap,

22
test/playwright/helpers/_auth_helpers.py
View File

@ -30,8 +30,10 @@ def ensure_authed(
if seeded_user_credentials:
email, password = seeded_user_credentials
else:
email = os.getenv("SEEDED_USER_EMAIL")
password = os.getenv("SEEDED_USER_PASSWORD")
email = os.getenv("SEEDED_USER_EMAIL") or os.getenv("E2E_ADMIN_EMAIL")
password = os.getenv("SEEDED_USER_PASSWORD") or os.getenv(
"E2E_ADMIN_PASSWORD"
)
if not email or not password:
pytest.skip("SEEDED_USER_EMAIL/SEEDED_USER_PASSWORD not set.")
@ -45,22 +47,16 @@ def ensure_authed(
try:
if "/login" not in page.url:
if (
page.locator(
"input[data-testid='auth-email'], [data-testid='auth-email'] input"
).count()
== 0
):
try:
page.wait_for_function(token_wait_js, timeout=2000)
return
except Exception:
pass
page.wait_for_function(token_wait_js, timeout=2000)
return
except Exception:
pass
page.goto(login_url, wait_until="domcontentloaded")
if "/login" not in page.url:
return
form, _ = active_auth_context()
email_input = form.locator(
"input[data-testid='auth-email'], [data-testid='auth-email'] input"

22
test/playwright/helpers/_next_apps_helpers.py
View File

@ -111,6 +111,16 @@ def _open_create_from_list(
).first
expect(fallback_card).to_be_visible(timeout=RESULT_TIMEOUT_MS)
fallback_card.click()
if modal_testid == "agent-create-modal":
menu = page.locator("[data-testid='agent-create-menu']")
if menu.count() > 0 and menu.first.is_visible():
create_blank = menu.locator("text=/create from blank/i")
if create_blank.count() > 0 and create_blank.first.is_visible():
create_blank.first.click()
else:
first_item = menu.locator("[role='menuitem']").first
expect(first_item).to_be_visible(timeout=RESULT_TIMEOUT_MS)
first_item.click()
modal = page.locator(f"[data-testid='{modal_testid}']")
expect(modal).to_be_visible(timeout=RESULT_TIMEOUT_MS)
return modal
@ -134,6 +144,18 @@ def _fill_and_save_create_modal(
expect(modal).not_to_be_visible(timeout=RESULT_TIMEOUT_MS)
def _search_query_input(page):
candidates = [
page.locator("[data-testid='search-query-input']"),
page.locator("input[placeholder*='How can I help you today']"),
page.locator("input[placeholder*='help you today']"),
]
for candidate in candidates:
if candidate.count() > 0:
return candidate.first
return page.locator("input[type='text']").first
def _select_first_dataset_and_save(
page,
timeout_ms: int = RESULT_TIMEOUT_MS,

78
test/playwright/helpers/datasets.py
View File

@ -394,54 +394,24 @@ def open_create_dataset_modal(page, expect, timeout_ms: int):
debug(f"[dataset] entrypoint_wait_timeout url={url} snippet={snippet!r}")
raise
empty_text = page.locator("text=/no dataset created yet/i").first
if empty_text.count() > 0:
debug("[dataset] using empty-state entrypoint")
expect(empty_text).to_be_visible(timeout=5000)
element_handle = empty_text.element_handle()
if element_handle is None:
debug("[dataset] empty-state text element handle not available")
dump_clickable_candidates(page)
raise AssertionError("Empty-state text element not available for click.")
handle = page.evaluate_handle(
"""
(el) => {
const closest = el.closest('button, a, [role="button"]');
if (closest) return closest;
let node = el;
for (let i = 0; i < 6 && node; i += 1) {
if (node.nodeType !== Node.ELEMENT_NODE) {
node = node.parentElement;
continue;
}
const element = node;
const hasOnClick = typeof element.onclick === 'function' || element.hasAttribute('onclick');
const tabIndex = element.getAttribute('tabindex');
const hasTab = tabIndex === '0';
const cursor = window.getComputedStyle(element).cursor;
if (hasOnClick || hasTab || cursor === 'pointer') {
return element;
}
node = element.parentElement;
}
return null;
}
""",
element_handle,
)
element = handle.as_element()
if element is None:
debug("[dataset] empty-state clickable ancestor not found")
dump_clickable_candidates(page)
raise AssertionError("No clickable ancestor found for empty dataset state.")
element.click()
else:
def _click_entrypoint(locator) -> None:
try:
locator.click()
except Exception as exc:
message = str(exc).lower()
if (
"not attached to the dom" not in message
and "intercepts pointer events" not in message
and "element is not stable" not in message
):
raise
locator.click(force=True)
def _click_create_button_entrypoint() -> None:
debug("[dataset] using create button entrypoint")
create_btn = None
if hasattr(page, "get_by_role"):
create_btn = page.get_by_role(
"button", name=re.compile(r"create dataset", re.I)
)
create_btn = page.get_by_role("button", name=re.compile(r"create dataset", re.I))
if create_btn is None or create_btn.count() == 0:
create_btn = page.locator(
"button", has_text=re.compile(r"create dataset", re.I)
@ -470,7 +440,23 @@ def open_create_dataset_modal(page, expect, timeout_ms: int):
snippet = "\n".join(lines[:20])[:500]
debug(f"[dataset] entrypoint_not_found url={url} snippet={snippet!r}")
raise
create_btn.click()
_click_entrypoint(create_btn)
empty_text = page.locator("text=/no dataset created yet/i").first
if empty_text.count() > 0:
debug("[dataset] using empty-state entrypoint")
expect(empty_text).to_be_visible(timeout=5000)
entrypoint = empty_text.locator(
"xpath=ancestor-or-self::*[self::button or self::a or @role='button'][1]"
)
if entrypoint.count() > 0:
expect(entrypoint.first).to_be_visible(timeout=5000)
_click_entrypoint(entrypoint.first)
else:
debug("[dataset] empty-state clickable ancestor not found; falling back")
_click_create_button_entrypoint()
else:
_click_create_button_entrypoint()
modal = page.locator("[role='dialog']").filter(has_text=re.compile("create dataset", re.I))
expect(modal).to_be_visible(timeout=timeout_ms)