94 lines
5.5 KiB
Python
94 lines
5.5 KiB
Python
"""Completion contracts use fake browser evidence in isolated SQLite/PG stores."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
|
|
import pytest
|
|
|
|
from apps.server.db import 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_learning_extensions import EXTRACT, extract_output
|
|
from apps.server.tests.test_postgres import pg_client
|
|
|
|
|
|
@pytest.fixture(params=["client", "pg_client"])
|
|
def completion_client(request):
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
def test_scroll_completion_contract_freezes_and_replays_without_rewriting_legacy(completion_client):
|
|
client = completion_client
|
|
initialize(client)
|
|
task, old_candidate = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
worker = client.worker_reference
|
|
original = worker.call
|
|
|
|
async def completed(method, params):
|
|
result = await original(method, params)
|
|
if method == "action" and params["kind"] == "extract":
|
|
result["result"] = {**extract_output(), "truncated": True, "skipped_incomplete_count": 1}
|
|
elif method == "verify":
|
|
result["outputs"]["structured"] = {**extract_output(), "truncated": True, "skipped_incomplete_count": 1}
|
|
return result
|
|
worker.call = completed
|
|
scroll = {"kind": "scroll", "selector": "body", "pixels": 320, "max_scrolls": 4, "until_selector": '[data-testid="comment"]'}
|
|
extract = {**EXTRACT, "complete_only": True}
|
|
for action in (scroll, extract):
|
|
response = client.post(prefix + "/browser/action", json=action)
|
|
assert response.status_code == 200, response.text
|
|
candidate = client.post(prefix + "/candidates", json={}).json()
|
|
assert candidate["manifest"]["steps"][-2]["until_selector"] == scroll["until_selector"]
|
|
assert candidate["manifest"]["steps"][-1]["complete_only"] is True
|
|
assert client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()["status"] == "passed"
|
|
verify = client.post("/api/learning/candidates/" + candidate["id"] + "/verify", json={"mode": "exploration", "expected_outputs": {"structured": "建材评论"}})
|
|
assert verify.status_code == 200 and verify.json()["status"] == "passed", verify.text
|
|
assert verify.json()["actual_outputs"]["structured"]["skipped_incomplete_count"] == 1
|
|
wire = worker.calls[-1][1]
|
|
assert wire["steps"][-2]["until_selector"] == scroll["until_selector"] and wire["steps"][-1]["complete_only"] is True
|
|
# Existing frozen versions neither gain fields nor change their digest.
|
|
stored = client.get("/api/learning/candidates/" + old_candidate["id"]).json()
|
|
assert stored["manifest"] == old_candidate["manifest"] and stored["digest"] == old_candidate["digest"]
|
|
assert client.post("/api/learning/candidates/" + old_candidate["id"] + "/check").json()["status"] == "passed"
|
|
for bad in (
|
|
{**scroll, "until_selector": ""}, {**scroll, "until_selector": " "},
|
|
{**scroll, "until_selector": "body .body"}, {**scroll, "until_selector": "html"},
|
|
{**scroll, "until_selector": ":root .body"}, {**scroll, "until_selector": "xpath=//div"},
|
|
{**scroll, "until_selector": "javascript:alert(1)"}, {**scroll, "until_selector": "x" * 1001},
|
|
{**scroll, "until_selector": None},
|
|
{**EXTRACT, "until_selector": ".body"}, {**scroll, "complete_only": True},
|
|
{**extract, "complete_only": 1}, {**extract, "complete_only": "true"},
|
|
{**extract, "complete_only": None},
|
|
):
|
|
count = len([call for call in worker.calls if call[0] == "action"])
|
|
assert client.post(prefix + "/browser/action", json=bad).status_code == 422
|
|
assert len([call for call in worker.calls if call[0] == "action"]) == count
|
|
service = client.app_reference.state.learning
|
|
malformed = copy.deepcopy(candidate)
|
|
malformed["manifest"]["steps"][-2]["until_selector"] = "body .body"
|
|
malformed["digest"] = digest(malformed["manifest"])
|
|
assert any("契约" in error for error in service.static_errors(malformed))
|
|
|
|
|
|
def test_complete_only_shape_is_explicit_and_never_passes_empty_or_missing_fields():
|
|
contract = {**EXTRACT, "complete_only": True}
|
|
good = {**extract_output(), "skipped_incomplete_count": 2, "truncated": True}
|
|
def checks(value, current=contract):
|
|
return output_assertion_checks({"structured": value}, {"structured": "建材评论"}, {"structured": current})
|
|
assert all(item["passed"] for item in checks(good))
|
|
assert all(item["passed"] for item in checks({**good, "skipped_incomplete_count": 0, "truncated": False}))
|
|
# Legacy extracts continue to use exactly the old five-field shape.
|
|
assert all(item["passed"] for item in checks(extract_output(), EXTRACT))
|
|
invalid = [extract_output(), {**good, "skipped_incomplete_count": -1}, {**good, "skipped_incomplete_count": True},
|
|
{**good, "skipped_incomplete_count": 1.5}, {**good, "truncated": False},
|
|
{**good, "items": [{"body": None, "link": None}]}, {**good, "items": [{"link": None}]},
|
|
{**good, "count": 2}, {**good, "unexpected": "payload"}]
|
|
for value in invalid:
|
|
assert not checks(value)[0]["passed"]
|
|
assert not checks(good, EXTRACT)[0]["passed"]
|
|
empty = {"items": [], "count": 0, "truncated": True, "coverage": "empty", "source_url": good["source_url"], "skipped_incomplete_count": 2}
|
|
result = checks(empty)
|
|
assert result[0]["passed"] and not all(item["passed"] for item in result)
|
|
assert not next(item for item in result if item["name"] == "output_present:structured")["passed"]
|