235 lines
13 KiB
Python
235 lines
13 KiB
Python
"""Scoped batch/empty contracts, using fake DOM captures and isolated stores."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from apps.server.db import StoreConflict, digest
|
|
from apps.server.learning_checks import output_assertion_checks
|
|
from apps.server.tests.test_learning import client, initialize, working_candidate
|
|
from apps.server.tests.test_postgres import pg_client
|
|
from apps.server.worker import WorkerError
|
|
|
|
|
|
FIELDS = [{"name": "id", "source": "attribute", "attribute": "id", "required": True}, {"name": "body", "selector": ".content", "source": "text", "required": True}]
|
|
SCOPED = {"kind": "extract", "selector": "#scope", "row_selector": ".row", "fields": FIELDS, "limit": 3, "complete_only": True, "output_name": "scoped"}
|
|
BATCH = {**SCOPED, "batch": {"scroll_selector": "#scope", "pixels": 320, "max_batches": 2, "dedupe_field": "id"}}
|
|
EMPTY = {**SCOPED, "allow_empty": True, "empty_state": {"selector": ".empty", "text": "暂无评论"}}
|
|
|
|
|
|
@pytest.fixture(params=["client", "pg_client"])
|
|
def scoped_client(request):
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
def prepare(client):
|
|
initialize(client)
|
|
task, legacy = working_candidate(client)
|
|
worker = client.worker_reference
|
|
capture = worker.capture
|
|
def scoped_capture(task_id):
|
|
value = capture(task_id)
|
|
value["elements"].append({"selector": "#scope", "tag": "main", "kind": "read", "label": "内容区域", "unique": True, "sensitive": False, "in_viewport": True,
|
|
"scroll_y": {"can_scroll": True, "scroll_top": 0, "scroll_height": 1200, "client_height": 500}})
|
|
return value
|
|
worker.capture = scoped_capture
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
assert client.post(prefix + "/browser/capture", json={}).status_code == 200
|
|
return task, legacy, prefix, worker
|
|
|
|
|
|
def batch_output():
|
|
return {"items": [{"id": "row-a", "body": "建材采购甲"}, {"id": "row-b", "body": "建材采购乙"}], "count": 2, "truncated": True,
|
|
"coverage": "bounded", "source_url": "http://127.0.0.1/demo", "skipped_incomplete_count": 0,
|
|
"batch_summary": {"batches_read": 2, "duplicates_removed": 1, "stop_reason": "batch_budget"}}
|
|
|
|
|
|
def empty_output():
|
|
return {"items": [], "count": 0, "truncated": False, "coverage": "empty", "source_url": "http://127.0.0.1/demo", "skipped_incomplete_count": 0,
|
|
"empty_state": {"confirmed": True, "selector": ".empty", "text": "暂无评论"}}
|
|
|
|
|
|
def batches(worker, task_id):
|
|
return [{"batch_index": index, "added_count": 1, "duplicate_count": index - 1, "capture": worker.capture(task_id)} for index in (1, 2)]
|
|
|
|
|
|
def test_scoped_batch_evidence_freezes_and_replays_with_immutable_history(scoped_client):
|
|
client = scoped_client
|
|
task, legacy, prefix, worker = prepare(client)
|
|
original = worker.call
|
|
async def batch_call(method, params):
|
|
value = await original(method, params)
|
|
if method == "action" and params["kind"] == "extract":
|
|
value.update(result=batch_output(), batch_evidence=batches(worker, task["id"]))
|
|
if method == "verify":
|
|
value["outputs"]["scoped"] = batch_output()
|
|
value["evidence"].extend(item["capture"] for item in batches(worker, task["id"]))
|
|
value["error_codes"] = []
|
|
return value
|
|
worker.call = batch_call
|
|
response = client.post(prefix + "/browser/action", json=BATCH)
|
|
assert response.status_code == 200, response.text
|
|
ids = response.json()["step"]["batch_evidence_ids"]
|
|
assert len(ids) == 2
|
|
observations = client.get(prefix).json()["observations"]
|
|
assert [item["id"] for item in observations[-3:-1]] == ids
|
|
assert observations[-1]["id"] == response.json()["after"]["id"]
|
|
for index, item in enumerate(observations[-3:-1], 1):
|
|
assert item["source"] == "action_batch" and item["batch"]["batch_index"] == index
|
|
assert client.get(item["screenshot_url"]).status_code == 200
|
|
candidate = client.post(prefix + "/candidates", json={}).json()
|
|
assert set(ids) <= {item["observation_id"] for item in candidate["manifest"]["evidence_bindings"]}
|
|
assert client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()["status"] == "passed"
|
|
record = client.post("/api/learning/candidates/" + candidate["id"] + "/verify", json={"expected_outputs": {"scoped": "建材采购乙"}}).json()
|
|
assert record["status"] == "passed" and len(record["evidence"]) == 3 and record["error_codes"] == []
|
|
old = client.get("/api/learning/candidates/" + legacy["id"]).json()
|
|
assert old["digest"] == legacy["digest"] and old["manifest"] == legacy["manifest"]
|
|
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, (ids[0],))
|
|
broken = copy.deepcopy(candidate)
|
|
broken["manifest"]["steps"][-1]["batch_evidence_ids"] = ["observation_other_task"]
|
|
broken["digest"] = digest(broken["manifest"])
|
|
assert service.static_errors(broken)
|
|
|
|
|
|
def test_scoped_batch_failure_keeps_partial_evidence_without_success_step(scoped_client):
|
|
client = scoped_client
|
|
task, _, prefix, worker = prepare(client)
|
|
original = worker.call
|
|
async def fail(method, params):
|
|
if method == "action":
|
|
raise WorkerError("fixture error", "WAIT_TIMEOUT", before=worker.capture(task["id"]), after=worker.capture(task["id"]), batch_evidence=batches(worker, task["id"])[:1], error_codes=["WAIT_TIMEOUT", "bad code private-fixture"])
|
|
return await original(method, params)
|
|
worker.call = fail
|
|
result = client.post(prefix + "/browser/action", json=BATCH)
|
|
assert result.status_code == 409 and result.json()["detail"]["code"] == "WAIT_TIMEOUT"
|
|
detail = result.json()["detail"]
|
|
assert len(detail["batch_evidence_ids"]) == 1 and detail["error_codes"] == ["WAIT_TIMEOUT"]
|
|
current = client.get(prefix).json()
|
|
assert len(current["steps"]) == 3
|
|
assert current["observations"][-2]["source"] == "action_failure_batch"
|
|
assert current["observations"][-1]["action_failure"]["moment"] == "after"
|
|
assert "private-fixture" not in result.text
|
|
|
|
|
|
def test_scoped_empty_requires_exact_evidence_and_explicit_expectation(scoped_client):
|
|
client = scoped_client
|
|
task, _, prefix, worker = prepare(client)
|
|
original = worker.call
|
|
error_mode = False
|
|
async def empty_call(method, params):
|
|
value = await original(method, params)
|
|
if method == "action" and params["kind"] == "extract":
|
|
value["result"] = empty_output()
|
|
if method == "verify":
|
|
value["outputs"]["scoped"] = empty_output()
|
|
if error_mode:
|
|
value.update(passed=False, outputs={}, errors=["需要重新登录"], error_codes=["LOGIN_REQUIRED", "bad code private-fixture"])
|
|
return value
|
|
worker.call = empty_call
|
|
raw_empty = {**EMPTY, "empty_state": {**EMPTY["empty_state"], "text": " 暂无评论 \n"}}
|
|
result = client.post(prefix + "/browser/action", json=raw_empty)
|
|
assert result.status_code == 200, result.text
|
|
assert result.json()["step"]["empty_state"]["text"] == "暂无评论"
|
|
candidate = client.post(prefix + "/candidates", json={}).json()
|
|
assert candidate["manifest"]["steps"][-1]["empty_state"]["text"] == "暂无评论"
|
|
endpoint = "/api/learning/candidates/" + candidate["id"]
|
|
assert client.post(endpoint + "/check").json()["status"] == "passed"
|
|
expected = {"expected_outputs": {"scoped": {"state": "empty"}}}
|
|
verification = client.post(endpoint + "/verify", json=expected)
|
|
assert verification.status_code == 200 and verification.json()["status"] == "passed", verification.text
|
|
assert all(item["passed"] for item in verification.json()["assertion_checks"])
|
|
content = client.post(endpoint + "/verify", json={"expected_outputs": {"scoped": "暂无评论"}})
|
|
assert content.status_code == 200 and content.json()["status"] == "failed"
|
|
for invalid in ({"state": "empty", "extra": True}, {"state": "anything"}, []):
|
|
assert client.post(endpoint + "/verify", json={"expected_outputs": {"scoped": invalid}}).status_code == 422
|
|
error_mode = True
|
|
failure = client.post(endpoint + "/verify", json=expected)
|
|
assert failure.status_code == 200 and failure.json()["status"] == "failed"
|
|
assert failure.json()["error_codes"] == ["LOGIN_REQUIRED"]
|
|
assert "private-fixture" not in failure.text
|
|
|
|
|
|
def test_scoped_contract_rejects_unsafe_inputs_before_worker(client):
|
|
_, _, prefix, worker = prepare(client)
|
|
invalid = [
|
|
{**SCOPED, "row_selector": "body .row"}, {**SCOPED, "row_selector": ""},
|
|
{**SCOPED, "target": {"field": "text", "value": "x"}},
|
|
{**BATCH, "complete_only": False}, {**BATCH, "allow_empty": True, "empty_state": EMPTY["empty_state"]},
|
|
{**BATCH, "batch": {**BATCH["batch"], "max_batches": 6}}, {**BATCH, "batch": {**BATCH["batch"], "pixels": True}},
|
|
{**BATCH, "batch": {**BATCH["batch"], "dedupe_field": "body"}},
|
|
{**BATCH, "batch": {**BATCH["batch"], "scroll_selector": "#not-observed"}},
|
|
{**EMPTY, "empty_state": {"selector": "body .empty", "text": "暂无评论"}},
|
|
{**EMPTY, "empty_state": {"selector": ".empty", "text": " "}},
|
|
{**EMPTY, "allow_empty": "true"}, {**EMPTY, "allow_empty": None},
|
|
{**SCOPED, "empty_state": EMPTY["empty_state"]},
|
|
{"kind": "wait", "selector": '[data-testid="result"]', "state": "hidden", "min_count": 1},
|
|
{"kind": "wait", "selector": '[data-testid="result"]', "state": "visible", "min_count": True},
|
|
]
|
|
for action in invalid:
|
|
calls = len([item for item in worker.calls if item[0] == "action"])
|
|
result = client.post(prefix + "/browser/action", json=action)
|
|
assert result.status_code == 422, result.text
|
|
assert len([item for item in worker.calls if item[0] == "action"]) == calls
|
|
result = client.post(prefix + "/browser/action", json={"kind": "wait", "selector": '[data-testid="result"]', "state": "visible", "min_count": 1})
|
|
assert result.status_code == 200, result.text
|
|
candidate = client.post(prefix + "/candidates", json={}).json()
|
|
assert client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()["status"] == "passed"
|
|
calls = len([item for item in worker.calls if item[0] == "verify"])
|
|
assert client.post("/api/learning/candidates/" + candidate["id"] + "/verify", json={"expected_outputs": {"results": {"state": "empty"}}}).status_code == 422
|
|
assert len([item for item in worker.calls if item[0] == "verify"]) == calls
|
|
|
|
|
|
def test_scoped_shape_rejects_pseudo_empty_and_bad_deduplication():
|
|
def checks(value, contract=EMPTY, expected={"state": "empty"}):
|
|
return output_assertion_checks({"scoped": value}, {"scoped": expected}, {"scoped": contract})
|
|
assert all(item["passed"] for item in checks(empty_output()))
|
|
for changed in (
|
|
{**empty_output(), "truncated": True}, {**empty_output(), "skipped_incomplete_count": 1},
|
|
{**empty_output(), "empty_state": {"confirmed": False, "selector": ".empty", "text": "暂无评论"}},
|
|
{**empty_output(), "empty_state": {"confirmed": True, "selector": ".other", "text": "暂无评论"}},
|
|
{**empty_output(), "empty_state": {"confirmed": True, "selector": ".empty", "text": "加载中"}},
|
|
{key: value for key, value in empty_output().items() if key != "empty_state"},
|
|
):
|
|
assert not all(item["passed"] for item in checks(changed))
|
|
assert not all(item["passed"] for item in checks(empty_output(), {**SCOPED, "allow_empty": False}))
|
|
assert not all(item["passed"] for item in checks(empty_output(), expected="暂无评论"))
|
|
good = batch_output()
|
|
assert all(item["passed"] for item in checks(good, BATCH, "采购"))
|
|
for changed in (
|
|
{**good, "truncated": False},
|
|
{**good, "items": [good["items"][0], good["items"][0]]},
|
|
{**good, "batch_summary": {**good["batch_summary"], "batches_read": 3}},
|
|
{**good, "batch_summary": {**good["batch_summary"], "duplicates_removed": True}},
|
|
{**good, "batch_summary": {**good["batch_summary"], "stop_reason": "unknown"}},
|
|
):
|
|
assert not checks(changed, BATCH, "采购")[0]["passed"]
|
|
|
|
|
|
def test_scoped_boundary_container_can_read_first_batch(client):
|
|
task, _, prefix, worker = prepare(client)
|
|
old_capture, old_call = worker.capture, worker.call
|
|
def capture(task_id):
|
|
value = old_capture(task_id)
|
|
value["elements"][-1]["scroll_y"]["can_scroll"] = False
|
|
return value
|
|
worker.capture = capture
|
|
assert client.post(prefix + "/browser/capture", json={}).status_code == 200
|
|
async def first_batch(method, params):
|
|
value = await old_call(method, params)
|
|
if method == "action":
|
|
output = batch_output()
|
|
output.update(items=output["items"][:1], count=1, batch_summary={"batches_read": 1, "duplicates_removed": 0, "stop_reason": "boundary"})
|
|
value.update(result=output, batch_evidence=[{"batch_index": 1, "added_count": 1, "duplicate_count": 0, "capture": capture(task["id"])}])
|
|
return value
|
|
worker.call = first_batch
|
|
result = client.post(prefix + "/browser/action", json=BATCH)
|
|
assert result.status_code == 200, result.text
|
|
assert result.json()["result"]["batch_summary"]["stop_reason"] == "boundary"
|
|
candidate = client.post(prefix + "/candidates", json={}).json()
|
|
assert client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()["status"] == "passed"
|