218 lines
13 KiB
Python
218 lines
13 KiB
Python
"""Explicit loading waits freeze ordinary root evidence, never absent rows.
|
|
|
|
Browser behavior is separately tested by the worker's local-page scenarios.
|
|
These API checks use fake captures and isolated SQLite / dedicated PG schemas.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import sqlite3
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from apps.server.db import StoreConflict, digest
|
|
from apps.server.learning import wait_scope_evidence_valid
|
|
from apps.server.learning_checks import step_evidence_references
|
|
from apps.server.learning_contracts import Action
|
|
from apps.server.reading import detail_contract
|
|
from apps.server.tests.test_discovery import item, search_candidate
|
|
from apps.server.tests.test_learning import client, initialize, working_candidate
|
|
from apps.server.tests.test_postgres import pg_client
|
|
from apps.server.tests.test_reading import NOTE_ID, TITLE, learned_detail
|
|
|
|
ROOT = "#search-results"
|
|
COLLECTION = '[data-testid="result"]'
|
|
WAIT = {"kind": "wait", "selector": COLLECTION, "wait_scope": ROOT, "state": "visible", "min_count": 1, "timeout_ms": 1000}
|
|
ROOT_EVIDENCE = {"selector": ROOT, "tag": "section", "kind": "read", "unique": True, "sensitive": False, "in_viewport": True}
|
|
|
|
|
|
@pytest.fixture(params=["client", "pg_client"])
|
|
def scoped_wait_client(request):
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
def install_loading_capture(worker):
|
|
capture = worker.capture
|
|
state = {"root": copy.deepcopy(ROOT_EVIDENCE), "duplicate": False}
|
|
|
|
def loading(task_id):
|
|
value = capture(task_id)
|
|
value["elements"] = [item for item in value["elements"] if item["selector"] != COLLECTION]
|
|
if state["root"] is not None:
|
|
value["elements"].append(copy.deepcopy(state["root"]))
|
|
if state["duplicate"]:
|
|
value["elements"].append(copy.deepcopy(state["root"]))
|
|
return value
|
|
|
|
worker.capture = loading
|
|
return state
|
|
|
|
|
|
def test_scoped_wait_closed_contract_leaves_legacy_wait_unchanged():
|
|
assert Action.model_validate(WAIT).model_dump(exclude_none=True) == WAIT
|
|
assert Action.model_validate(WAIT | {"selector": "#global a.title", "wait_scope": "#global"}).wait_scope == "#global"
|
|
assert Action.model_validate(WAIT | {"selector": ":scope > .result"}).selector == ":scope > .result"
|
|
for changes in ({"wait_scope": None}, {"wait_scope": ""}, {"wait_scope": "body"}, {"wait_scope": "html"},
|
|
{"wait_scope": "html #ordinary"}, {"wait_scope": ":root"}, {"wait_scope": "xpath=//section"},
|
|
{"selector": "body .result"}, {"selector": "html"}, {"selector": "text=hello"},
|
|
{"state": "hidden"}, {"state": "text", "text": "ready"}, {"state": None},
|
|
{"min_count": None}, {"min_count": True}, {"min_count": 0}, {"min_count": 101},
|
|
{"timeout_ms": 15001}, {"target": {"field": "text", "value": "ready"}},
|
|
{"kind": "read", "output_name": "results"}, {"row_selector": ".row"}, {"scroll_scope": "viewport"}):
|
|
with pytest.raises(ValidationError):
|
|
Action.model_validate(WAIT | changes)
|
|
for missing in ("state", "min_count"):
|
|
with pytest.raises(ValidationError):
|
|
Action.model_validate({key: value for key, value in WAIT.items() if key != missing})
|
|
old = {"kind": "wait", "selector": "body", "state": "visible"}
|
|
assert Action.model_validate(old).model_dump(exclude_none=True) == old
|
|
old = {"kind": "wait", "selector": COLLECTION, "state": "hidden"}
|
|
assert Action.model_validate(old).model_dump(exclude_none=True) == old
|
|
|
|
|
|
def test_scoped_wait_root_evidence_is_unique_visible_ordinary_read_not_a_row_claim():
|
|
assert wait_scope_evidence_valid({"elements": [ROOT_EVIDENCE]}, ROOT)
|
|
assert not wait_scope_evidence_valid({"elements": []}, ROOT)
|
|
assert not wait_scope_evidence_valid({"elements": [ROOT_EVIDENCE, ROOT_EVIDENCE]}, ROOT)
|
|
for changes in ({"kind": "click"}, {"kind": "fill"}, {"kind": None}, {"unique": False}, {"unique": 1},
|
|
{"sensitive": True}, {"sensitive": None}, {"in_viewport": False}, {"in_viewport": None},
|
|
{"tag": "body"}, {"tag": "html"}, {"tag": "form"}, {"tag": "button"}, {"tag": "iframe"},
|
|
{"tag": "input"}, {"tag": "script"}, {"tag": None}, {"tag": "not a tag"}):
|
|
assert not wait_scope_evidence_valid({"elements": [ROOT_EVIDENCE | changes]}, ROOT)
|
|
for selector in (None, "", "body", "html", "xpath=//section"):
|
|
assert not wait_scope_evidence_valid({"elements": [ROOT_EVIDENCE]}, selector)
|
|
|
|
|
|
def test_scoped_wait_records_zero_row_before_freezes_root_replays_without_mutating_history(scoped_wait_client):
|
|
client = scoped_wait_client
|
|
initialize(client)
|
|
task, old = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
worker = client.worker_reference
|
|
state = install_loading_capture(worker)
|
|
assert client.post(prefix + "/browser/capture", json={}).status_code == 200
|
|
original_steps = client.get(prefix).json()["steps"]
|
|
calls = len(worker.calls)
|
|
# The unscoped path still needs observed collection rows.
|
|
assert client.post(prefix + "/browser/action", json={key: value for key, value in WAIT.items() if key != "wait_scope"}).status_code == 422
|
|
assert not any(method == "action" for method, _ in worker.calls[calls:])
|
|
result = client.post(prefix + "/browser/action", json=WAIT)
|
|
assert result.status_code == 200, result.text
|
|
result = result.json()
|
|
step = result["step"]
|
|
assert step["wait_scope"] == ROOT and step["selector"] == COLLECTION and step["position"] == 4
|
|
assert "selector_evidence_id" not in step
|
|
assert not any(item["selector"] == COLLECTION for item in result["before"]["elements"])
|
|
assert wait_scope_evidence_valid(result["before"], ROOT)
|
|
assert step["evidence_id"] == result["before"]["id"]
|
|
assert step["after_evidence_id"] == result["after"]["id"]
|
|
wire = [params for method, params in worker.calls if method == "action"][-1]
|
|
assert all(wire[key] == value for key, value in WAIT.items())
|
|
chosen = [step["id"], original_steps[-1]["id"]]
|
|
candidate = client.post(prefix + "/candidates", json={"step_ids": chosen}).json()
|
|
assert candidate["manifest"]["source_step_ids"] == chosen
|
|
assert candidate["manifest"]["steps"][0]["wait_scope"] == ROOT
|
|
refs = {item["observation_id"] for item in candidate["manifest"]["evidence_bindings"]}
|
|
assert refs == step_evidence_references(candidate["manifest"]["steps"])
|
|
endpoint = "/api/learning/candidates/" + candidate["id"]
|
|
assert client.post(endpoint + "/check").json()["status"] == "passed"
|
|
# Future observations cannot change the evidence already frozen by the wait.
|
|
state["root"] = None
|
|
client.post(prefix + "/browser/capture", json={})
|
|
assert client.post(endpoint + "/check").json()["status"] == "passed"
|
|
verification = client.post(endpoint + "/verify", json={"expected_outputs": {"results": "建材"}})
|
|
assert verification.status_code == 200 and verification.json()["status"] == "passed", verification.text
|
|
assert [params for method, params in worker.calls if method == "verify"][-1]["steps"][0]["wait_scope"] == ROOT
|
|
service = client.app_reference.state.learning
|
|
for query in ("UPDATE observations SET data_json='{}' WHERE id=$1", "DELETE FROM observations WHERE id=$1"):
|
|
with pytest.raises((StoreConflict, sqlite3.IntegrityError)):
|
|
service.store.execute(query, (step["evidence_id"],))
|
|
unchanged = client.get("/api/learning/candidates/" + old["id"]).json()
|
|
assert unchanged["manifest"] == old["manifest"] and unchanged["digest"] == old["digest"]
|
|
assert client.get(prefix).json()["steps"][:3] == original_steps
|
|
|
|
|
|
def test_scoped_wait_rejects_latest_invalid_root_before_worker_despite_prior_valid_root(scoped_wait_client):
|
|
client = scoped_wait_client
|
|
initialize(client)
|
|
task, _ = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
worker = client.worker_reference
|
|
state = install_loading_capture(worker)
|
|
assert client.post(prefix + "/browser/capture", json={}).status_code == 200
|
|
for changes in ({"kind": "click"}, {"unique": False}, {"sensitive": True}, {"in_viewport": False},
|
|
{"tag": "body"}, {"tag": "html"}, {"tag": "button"}, {"tag": "form"}, {"tag": "input"},
|
|
{"tag": "iframe"}, {"tag": "script"}, {"tag": None}):
|
|
state["root"] = ROOT_EVIDENCE | changes
|
|
assert client.post(prefix + "/browser/capture", json={}).status_code == 200
|
|
calls = len(worker.calls)
|
|
response = client.post(prefix + "/browser/action", json=WAIT)
|
|
assert response.status_code == 422, (changes, response.text)
|
|
assert not any(method == "action" for method, _ in worker.calls[calls:])
|
|
state["root"] = None
|
|
client.post(prefix + "/browser/capture", json={})
|
|
assert client.post(prefix + "/browser/action", json=WAIT).status_code == 422
|
|
state.update(root=copy.deepcopy(ROOT_EVIDENCE), duplicate=True)
|
|
client.post(prefix + "/browser/capture", json={})
|
|
assert client.post(prefix + "/browser/action", json=WAIT).status_code == 422
|
|
assert len(client.get(prefix).json()["steps"]) == 3
|
|
|
|
|
|
def test_scoped_wait_worker_before_cannot_silently_replace_root_evidence(scoped_wait_client):
|
|
client = scoped_wait_client
|
|
initialize(client)
|
|
task, _ = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
state = install_loading_capture(client.worker_reference)
|
|
assert client.post(prefix + "/browser/capture", json={}).status_code == 200
|
|
state["root"] = None
|
|
response = client.post(prefix + "/browser/action", json=WAIT)
|
|
assert response.status_code == 409, response.text
|
|
assert response.json()["detail"]["code"] == "WAIT_SCOPE_CHANGED"
|
|
current = client.get(prefix).json()
|
|
assert len(current["steps"]) == 3 and current["observations"][-1]["source"] == "action_failure"
|
|
|
|
|
|
def test_scoped_wait_static_check_cannot_relabel_old_collection_evidence(scoped_wait_client):
|
|
client = scoped_wait_client
|
|
initialize(client)
|
|
task, old = working_candidate(client)
|
|
service = client.app_reference.state.learning
|
|
altered = copy.deepcopy(old)
|
|
# Even though this old evidence has result rows, it has never shown the scope.
|
|
altered["manifest"]["steps"].insert(0, WAIT | {"evidence_id": old["manifest"]["steps"][0]["evidence_id"], "after_evidence_id": old["manifest"]["steps"][0]["after_evidence_id"]})
|
|
altered["digest"] = digest(altered["manifest"])
|
|
assert any("集合等待" in error for error in service.static_errors(altered))
|
|
assert client.get("/api/learning/candidates/" + old["id"]).json()["digest"] == old["digest"]
|
|
|
|
|
|
@pytest.mark.parametrize("seek", [False, True], ids=["seven_steps", "eight_steps"])
|
|
def test_scoped_wait_keeps_note_binding_reading_shapes_and_independent_readback(scoped_wait_client, seek):
|
|
client = scoped_wait_client
|
|
initialize(client)
|
|
_, search = search_candidate(client, [item(TITLE)])
|
|
run = client.post("/api/discovery/runs", json={"search_candidate_id": search["id"], "keyword": "装修"}).json()
|
|
task, old, _ = learned_detail(client, seek=seek)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
steps = client.get(prefix).json()["steps"]
|
|
state = install_loading_capture(client.worker_reference)
|
|
client.post(prefix + "/browser/capture", json={})
|
|
result = client.post(prefix + "/browser/action", json=WAIT)
|
|
assert result.status_code == 200, result.text
|
|
replacement = result.json()["step"]
|
|
assert not any(item["selector"] == COLLECTION for item in result.json()["before"]["elements"])
|
|
selected = [step["id"] for step in steps]
|
|
selected[2] = replacement["id"]
|
|
candidate = client.post(prefix + "/candidates", json={"step_ids": selected}).json()
|
|
assert detail_contract(candidate["manifest"]) and len(candidate["manifest"]["steps"]) == (8 if seek else 7)
|
|
assert client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()["status"] == "passed"
|
|
reading = client.post("/api/discovery/runs/" + run["id"] + "/readings", json={"detail_candidate_id": candidate["id"], "note_id": NOTE_ID})
|
|
assert reading.status_code == 201 and reading.json()["status"] == "completed", reading.text
|
|
assert reading.json()["note_binding_result"]["href_checked"] and reading.json()["note_binding_result"]["url_checked"]
|
|
actual = [params for method, params in client.worker_reference.calls if method == "verify"][-1]
|
|
assert actual["steps"][2]["wait_scope"] == ROOT and actual["note_binding"] == {"note_id": NOTE_ID}
|
|
unchanged = client.get("/api/learning/candidates/" + old["id"]).json()
|
|
assert unchanged["manifest"] == old["manifest"] and unchanged["digest"] == old["digest"]
|
|
assert client.get(prefix).json()["steps"][:len(steps)] == steps
|