161 lines
11 KiB
Python
161 lines
11 KiB
Python
"""Viewport seek evidence and ordered selection of immutable recorded actions."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
|
|
from pydantic import ValidationError
|
|
import pytest
|
|
|
|
from apps.server.db import digest
|
|
from apps.server.learning import seek_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, task, working_candidate
|
|
from apps.server.tests.test_postgres import pg_client
|
|
from apps.server.tests.test_reading import NOTE_ID, TITLE, learned_detail
|
|
|
|
ROOT = "#seek"
|
|
TARGET = '[data-testid="detail-link"]'
|
|
VIEWPORT = {"width": 1280, "height": 840, "scroll_x": 0, "scroll_y": 320}
|
|
SEEK = {"kind": "scroll", "selector": ROOT, "direction": "down", "pixels": 320, "max_scrolls": 3, "timeout_ms": 5000,
|
|
"until_target": {"selector": TARGET, "field": "text", "parameter": "note_title"}}
|
|
|
|
|
|
@pytest.fixture(params=["client", "pg_client"])
|
|
def selection_client(request):
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
def test_viewport_scope_contract_and_finite_evidence_keep_old_container_semantics():
|
|
root = {"selector": ROOT, "tag": "div", "unique": True, "sensitive": False}
|
|
observed = {"viewport": VIEWPORT, "elements": [{"selector": TARGET, "tag": "a", "collection": True}]}
|
|
assert seek_evidence_valid(observed, root, SEEK["until_target"], "viewport")
|
|
assert not seek_evidence_valid(observed, root, SEEK["until_target"])
|
|
for can_scroll in (True, False):
|
|
assert seek_evidence_valid(observed, {**root, "scroll_y": {"can_scroll": can_scroll}}, SEEK["until_target"])
|
|
assert Action.model_validate({**SEEK, "scroll_scope": "viewport"}).scroll_scope == "viewport"
|
|
for changes in ({"scroll_scope": None}, {"scroll_scope": "container"}, {"selector": "body", "scroll_scope": "viewport"},
|
|
{"selector": "html", "scroll_scope": "viewport"}, {"until_target": None, "scroll_scope": "viewport"},
|
|
{"kind": "wait", "scroll_scope": "viewport"}):
|
|
with pytest.raises(ValidationError):
|
|
Action.model_validate(SEEK | changes)
|
|
with pytest.raises(ValidationError):
|
|
Action.model_validate({"kind": "scroll", "selector": ROOT, "scroll_scope": "viewport", "pixels": 320, "max_scrolls": 3, "direction": "down"})
|
|
for viewport in (None, {}, {**VIEWPORT, "width": 0}, {**VIEWPORT, "height": True}, {**VIEWPORT, "scroll_y": -1},
|
|
{**VIEWPORT, "scroll_x": float("inf")}, {**VIEWPORT, "scroll_y": 2 ** 53 + 1}, {**VIEWPORT, "scroll_y": 2 ** 2048}):
|
|
assert not seek_evidence_valid({**observed, "viewport": viewport}, root, SEEK["until_target"], "viewport")
|
|
for changes in ({"selector": "body"}, {"selector": "html"}, {"tag": "input"}, {"tag": None}, {"sensitive": True}, {"unique": False}):
|
|
assert not seek_evidence_valid(observed, root | changes, SEEK["until_target"], "viewport")
|
|
assert not seek_evidence_valid({"viewport": VIEWPORT, "elements": []}, root, SEEK["until_target"], "viewport")
|
|
assert seek_evidence_valid({"viewport": VIEWPORT, "elements": []}, root | {"read_hints": [{"selector": TARGET, "tag": "a"}]}, SEEK["until_target"], "viewport")
|
|
|
|
|
|
def test_viewport_seek_record_select_replacement_preserves_old_evidence_and_reading(selection_client):
|
|
client = selection_client
|
|
initialize(client)
|
|
_, search = search_candidate(client, [item(TITLE)])
|
|
run = client.post("/api/discovery/runs", json={"search_candidate_id": search["id"], "keyword": "准备装修"}).json()
|
|
created, legacy, _ = learned_detail(client, seek=True, content=True)
|
|
service, worker = client.app_reference.state.learning, client.worker_reference
|
|
prefix = "/api/learning/tasks/" + created["id"]
|
|
old_steps = client.get(prefix).json()["steps"]
|
|
frozen = copy.deepcopy(legacy["manifest"])
|
|
capture = worker.capture
|
|
mode = {"viewport": True}
|
|
def observed(task_id):
|
|
value = capture(task_id)
|
|
root = next(element for element in value["elements"] if element["selector"] == ROOT)
|
|
root.pop("scroll_y", None)
|
|
if mode["viewport"]:
|
|
value["viewport"] = copy.deepcopy(VIEWPORT)
|
|
return value
|
|
worker.capture = observed
|
|
assert client.post(prefix + "/browser/capture", json={}).status_code == 200
|
|
calls = len(worker.calls)
|
|
assert client.post(prefix + "/browser/action", json=SEEK).status_code == 422
|
|
assert not any(method == "action" for method, _ in worker.calls[calls:])
|
|
response = client.post(prefix + "/browser/action", json={**SEEK, "scroll_scope": "viewport"})
|
|
assert response.status_code == 200, response.text
|
|
replacement = response.json()["step"]
|
|
assert replacement["position"] == 9 and replacement["scroll_scope"] == "viewport"
|
|
wire = [params for method, params in worker.calls if method == "action"][-1]
|
|
assert wire["scroll_scope"] == "viewport" and wire["until_target"]["value"] == TITLE
|
|
chosen = [step["id"] for step in old_steps[:3]] + [replacement["id"]] + [step["id"] for step in old_steps[4:]]
|
|
candidate = client.post(prefix + "/candidates", json={"name": "观察到的新页面级定位", "step_ids": chosen})
|
|
assert candidate.status_code == 201, candidate.text
|
|
candidate = candidate.json()
|
|
assert candidate["version"] == legacy["version"] + 1 and candidate["id"] != legacy["id"]
|
|
assert candidate["manifest"]["source_step_ids"] == chosen and candidate["manifest"]["steps"][3]["scroll_scope"] == "viewport"
|
|
assert len(candidate["manifest"]["steps"]) == 8 and detail_contract(candidate["manifest"]) and candidate["status"] == "candidate"
|
|
assert service.validations(service.store.one("SELECT owner_id FROM learning_tasks WHERE id=$1", (created["id"],))["owner_id"], candidate_id=candidate["id"]) == []
|
|
references = step_evidence_references(candidate["manifest"]["steps"])
|
|
assert {item["observation_id"] for item in candidate["manifest"]["evidence_bindings"]} == references
|
|
assert set(candidate["manifest"]["steps"][-1]["batch_evidence_ids"]).issubset(references)
|
|
check = client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()
|
|
assert check["status"] == "passed", check
|
|
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 [params for method, params in worker.calls if method == "verify"][-1]["steps"][3]["scroll_scope"] == "viewport"
|
|
assert client.get(prefix).json()["steps"][:8] == old_steps
|
|
old = client.get("/api/learning/candidates/" + legacy["id"]).json()
|
|
assert old["manifest"] == frozen and old["digest"] == legacy["digest"]
|
|
default = client.post(prefix + "/candidates", json={}).json()
|
|
assert len(default["manifest"]["steps"]) == 9 and "source_step_ids" not in default["manifest"] and not detail_contract(default["manifest"])
|
|
# Rebinding the old container-only evidence as viewport evidence is rejected
|
|
# even though its original metadata was valid for a container scroll.
|
|
altered = copy.deepcopy(legacy)
|
|
altered["manifest"]["steps"][3]["scroll_scope"] = "viewport"
|
|
altered["digest"] = digest(altered["manifest"])
|
|
assert any("动态滚动" in error for error in service.static_errors(altered))
|
|
for key, value in (("scroll_scope", None), ("scroll_scope", "container"), ("selector", "body"), ("selector", "html")):
|
|
altered = copy.deepcopy(candidate["manifest"])
|
|
altered["steps"][3][key] = value
|
|
assert not detail_contract(altered)
|
|
mode["viewport"] = False
|
|
assert client.post(prefix + "/browser/capture", json={}).status_code == 200
|
|
count = len(client.get(prefix).json()["steps"])
|
|
assert client.post(prefix + "/browser/action", json=SEEK | {"scroll_scope": "viewport"}).status_code == 422
|
|
assert len(client.get(prefix).json()["steps"]) == count
|
|
|
|
|
|
def test_candidate_selected_ids_strict_order_closed_input_no_failed_creation(selection_client):
|
|
client = selection_client
|
|
initialize(client)
|
|
created, legacy = working_candidate(client)
|
|
service = client.app_reference.state.learning
|
|
prefix = "/api/learning/tasks/" + created["id"]
|
|
original_steps = client.get(prefix).json()["steps"]
|
|
foreign_task, _ = working_candidate(client)
|
|
foreign_id = client.get("/api/learning/tasks/" + foreign_task["id"]).json()["steps"][0]["id"]
|
|
chosen = [original_steps[index]["id"] for index in (2, 0)]
|
|
before = service.store.one("SELECT count(*) AS n FROM candidates WHERE task_id=$1", (created["id"],))["n"]
|
|
for body in ({"step_ids": []}, {"step_ids": None}, {"step_ids": [chosen[0], chosen[0]]}, {"step_ids": ["step_missing"]},
|
|
{"step_ids": [foreign_id]}, {"step_ids": [1]}, {"step_ids": [True]}, {"step_ids": "invalid"}, {"step_ids": ["bad/id"]},
|
|
{"step_ids": ["step_" + str(index) for index in range(31)]}, {"step_ids": chosen, "manifest": legacy["manifest"]},
|
|
{"step_ids": chosen, "steps": [{"kind": "click"}]}):
|
|
result = client.post(prefix + "/candidates", json=body)
|
|
assert result.status_code == 422, (body, result.text)
|
|
assert service.store.one("SELECT count(*) AS n FROM candidates WHERE task_id=$1", (created["id"],))["n"] == before
|
|
selected = client.post(prefix + "/candidates", json={"step_ids": chosen}).json()
|
|
assert selected["manifest"]["source_step_ids"] == chosen
|
|
assert selected["manifest"]["steps"] == [{key: value for key, value in original_steps[index].items() if key not in {"id", "created_at", "position"}} for index in (2, 0)]
|
|
assert not service.source_integrity_errors(selected)
|
|
for mutate in (lambda manifest: manifest["source_step_ids"].reverse(), lambda manifest: manifest["source_step_ids"].append(foreign_id),
|
|
lambda manifest: manifest["steps"][0].update(output_name="changed"), lambda manifest: manifest["evidence_bindings"].pop()):
|
|
altered = copy.deepcopy(selected)
|
|
mutate(altered["manifest"])
|
|
altered["digest"] = digest(altered["manifest"])
|
|
assert service.static_errors(altered)
|
|
old = client.get("/api/learning/candidates/" + legacy["id"]).json()
|
|
assert old["manifest"] == legacy["manifest"] and old["digest"] == legacy["digest"]
|
|
assert client.get(prefix).json()["steps"] == original_steps
|
|
empty = task(client)
|
|
endpoint = "/api/learning/tasks/" + empty["id"] + "/candidates"
|
|
assert client.post(endpoint, json={"step_ids": []}).status_code == 422
|
|
# The omitted-selection legacy path still freezes an empty candidate, which
|
|
# then fails static validation; no existing API behavior is silently changed.
|
|
candidate = client.post(endpoint, json={}).json()
|
|
assert client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()["status"] == "failed"
|