242 lines
15 KiB
Python
242 lines
15 KiB
Python
"""Isolated API contracts. Browser actions here use FakeWorker, not a platform."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import sqlite3
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
from apps.server.db import StoreConflict, digest, encoded, identifier, now
|
|
from apps.server.learning_checks import output_assertion_checks
|
|
from apps.server.main import create_app
|
|
from apps.server.tests.test_learning import FakeWorker, client, initialize, working_candidate
|
|
from apps.server.tests.test_postgres import pg_client
|
|
|
|
|
|
def extract_output(text="建材评论", *, href="http://127.0.0.1/demo"):
|
|
return {"items": [{"body": text, "link": href}], "count": 1, "truncated": False, "source_url": "http://127.0.0.1/demo", "coverage": "bounded"}
|
|
|
|
|
|
EXTRACT = {"kind": "extract", "selector": '[data-testid="result"]', "output_name": "structured", "limit": 2,
|
|
"fields": [{"name": "body", "source": "text", "required": True}, {"name": "link", "selector": "a", "source": "href"}]}
|
|
|
|
|
|
def test_action_budgets_nested_fields_and_parameterized_target(client):
|
|
initialize(client)
|
|
task, _ = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
worker = client.worker_reference
|
|
original = worker.call
|
|
|
|
async def structured_call(method, params):
|
|
result = await original(method, params)
|
|
if method == "action" and params["kind"] == "extract":
|
|
result["result"] = extract_output()
|
|
return result
|
|
worker.call = structured_call
|
|
actions = [{"kind": "scroll", "selector": "body", "pixels": 600, "max_scrolls": 2},
|
|
{"kind": "wait", "selector": "body", "state": "visible", "timeout_ms": 2000},
|
|
{**EXTRACT, "target": {"field": "text", "parameter": "keyword"}}]
|
|
for action in actions:
|
|
response = client.post(prefix + "/browser/action", json=action)
|
|
assert response.status_code == 200, response.text
|
|
wire = worker.calls[-1][1]
|
|
assert wire["target"] == {"field": "text", "value": "建材"}
|
|
candidate = client.post(prefix + "/candidates", json={}).json()
|
|
assert candidate["manifest"]["schema_version"] == 2
|
|
assert candidate["manifest"]["steps"][-1]["target"] == {"field": "text", "parameter": "keyword"}
|
|
assert client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()["status"] == "passed"
|
|
for bad in [
|
|
{"kind": "scroll", "selector": "body", "max_scrolls": 11},
|
|
{"kind": "scroll", "selector": "body", "pixels": True},
|
|
{"kind": "wait", "selector": "body", "state": "text"},
|
|
{"kind": "wait", "selector": "body", "timeout_ms": 15001},
|
|
{**EXTRACT, "limit": 101},
|
|
{**EXTRACT, "fields": [{"name": "bad", "source": "attribute", "attribute": "data-token"}]},
|
|
{**EXTRACT, "fields": [{"name": "constructor", "source": "text"}]},
|
|
{**EXTRACT, "fields": [{"name": "bad", "selector": "xpath=//input", "source": "text"}]},
|
|
{**EXTRACT, "fields": [{"name": "bad", "selector": "body", "source": "text"}]},
|
|
{**EXTRACT, "fields": [{"name": "bad", "source": "attribute", "attribute": "data-1bad"}]},
|
|
{**EXTRACT, "fields": [{"name": "bad", "source": "text", "required": "true"}]},
|
|
{**EXTRACT, "target": {"field": "text", "value": " "}},
|
|
{"kind": "wait", "selector": "body", "state": "text", "text": " "},
|
|
{"kind": "scroll", "selector": "body", "target": {"field": "text", "value": "unexpected"}},
|
|
{**EXTRACT, "target": {"field": "attribute", "attribute": "value", "value": "secret"}},
|
|
{**EXTRACT, "target": {"field": "text", "parameter": "missing"}},
|
|
{**EXTRACT, "target": {"field": "text", "parameter": "keyword", "value": "ambiguous"}},
|
|
{"kind": "extract", "selector": "body", "output_name": "all", "fields": [{"name": "body", "source": "text"}]},
|
|
]:
|
|
result = client.post(prefix + "/browser/action", json=bad)
|
|
assert result.status_code == 422, result.text
|
|
|
|
|
|
def test_hidden_wait_freezes_prior_same_task_selector_evidence(client):
|
|
initialize(client)
|
|
task, _ = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
worker = client.worker_reference
|
|
original = worker.capture
|
|
def without_results(task_id):
|
|
result = original(task_id)
|
|
result["elements"] = [item for item in result["elements"] if item["selector"] != '[data-testid="result"]']
|
|
return result
|
|
worker.capture = without_results
|
|
client.post(prefix + "/browser/capture", json={})
|
|
result = client.post(prefix + "/browser/action", json={"kind": "wait", "selector": '[data-testid="result"]', "state": "hidden"})
|
|
assert result.status_code == 200, result.text
|
|
assert result.json()["step"]["selector_evidence_id"]
|
|
assert client.post(prefix + "/browser/action", json={"kind": "wait", "selector": "#never-seen", "state": "hidden"}).status_code == 422
|
|
candidate = client.post(prefix + "/candidates", json={}).json()
|
|
assert client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()["status"] == "passed"
|
|
|
|
|
|
def test_runner_budget_and_invalid_recording_output_do_not_freeze_success(client, monkeypatch):
|
|
initialize(client)
|
|
task, candidate = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
service = client.app_reference.state.learning
|
|
result = client.post(prefix + "/browser/action", json=EXTRACT)
|
|
assert result.status_code == 409, "FakeWorker intentionally returns an invalid extract shape"
|
|
assert len(service.steps(task["id"])) == 3
|
|
changed = copy.deepcopy(candidate)
|
|
changed["manifest"]["steps"] = candidate["manifest"]["steps"] * 11
|
|
assert any("步骤预算" in error for error in service.static_errors(changed))
|
|
monkeypatch.setattr(service, "steps", lambda task_id: [{}] * 30)
|
|
assert client.post(prefix + "/browser/action", json={"kind": "scroll", "selector": "body"}).status_code == 409
|
|
|
|
|
|
def test_structured_output_assertions_check_fields_counts_and_exclude_urls():
|
|
contracts = {"structured": EXTRACT}
|
|
good = extract_output()
|
|
assert all(item["passed"] for item in output_assertion_checks({"structured": good}, {"structured": "建材"}, contracts))
|
|
for changed in [
|
|
{**good, "count": 2}, {**good, "truncated": "false"},
|
|
{**good, "items": [{"body": None, "link": "http://127.0.0.1/"}]},
|
|
{**good, "items": [{"body": "建材", "link": None, "undeclared": "extra"}]},
|
|
{**good, "coverage": "empty"},
|
|
]:
|
|
checks = output_assertion_checks({"structured": changed}, {"structured": "建材"}, contracts)
|
|
assert not next(item for item in checks if item["name"].startswith("output_schema"))["passed"]
|
|
only_link = extract_output("无关文本", href="http://127.0.0.1/?q=建材")
|
|
assert not all(item["passed"] for item in output_assertion_checks({"structured": only_link}, {"structured": "建材"}, contracts))
|
|
clipped = {**good, "truncated": True}
|
|
assert all(item["passed"] for item in output_assertion_checks({"structured": clipped}, {"structured": "建材"}, contracts))
|
|
empty = {"items": [], "count": 0, "truncated": True, "source_url": good["source_url"], "coverage": "empty"}
|
|
checks = output_assertion_checks({"structured": empty}, {"structured": "建材"}, contracts)
|
|
assert next(item for item in checks if item["name"].startswith("output_schema"))["passed"]
|
|
assert not all(item["passed"] for item in checks), "empty is a valid shape but not a successful text assertion"
|
|
|
|
|
|
def test_legacy_v1_frozen_candidate_still_checks_and_replays(client):
|
|
initialize(client)
|
|
task, current = working_candidate(client)
|
|
legacy = copy.deepcopy(current["manifest"])
|
|
legacy["schema_version"] = 1
|
|
candidate_id = identifier("candidate")
|
|
client.app_reference.state.store.execute("INSERT INTO candidates(id,task_id,name,version,digest,manifest_json,created_at) VALUES($1,$2,$3,$4,$5,$6,$7)", (candidate_id, task["id"], "历史 V1", 2, digest(legacy), encoded(legacy), now()))
|
|
prefix = "/api/learning/candidates/" + candidate_id
|
|
assert client.post(prefix + "/check").json()["status"] == "passed"
|
|
response = client.post(prefix + "/verify", json={"expected_outputs": {"results": "建材"}})
|
|
assert response.status_code == 200 and response.json()["status"] == "passed"
|
|
assert client.worker_reference.calls[-1][1]["schema_version"] == 1
|
|
assert client.get(prefix).json()["manifest"] == legacy
|
|
|
|
|
|
@pytest.fixture(params=["client", "pg_client"])
|
|
def quality_client(request):
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
def test_quality_deadlines_late_feedback_recovery_immutability_and_restart(quality_client, monkeypatch):
|
|
client = quality_client
|
|
initialize(client)
|
|
task, candidate = working_candidate(client)
|
|
prefix = "/api/learning/candidates/" + candidate["id"]
|
|
client.post(prefix + "/check")
|
|
clock = [datetime(2026, 10, 24, 1, 0, tzinfo=timezone.utc)]
|
|
monkeypatch.setattr("apps.server.learning.now", lambda: clock[0].isoformat(timespec="milliseconds"))
|
|
payload = {"expected_outputs": {"results": "建材"}}
|
|
exploration = client.post(prefix + "/verify", json=payload).json()
|
|
assert exploration["mode"] == "exploration" and exploration["observation_due_at"] is None
|
|
clear = {"state": "clear", "note": "人工核对通知与功能", "process_review": "passed", "source": "manual"}
|
|
def append(record, data):
|
|
return client.post("/api/learning/validations/" + record["id"] + "/quality-observations", json=data)
|
|
assert append(exploration, clear).status_code == 409
|
|
assert client.post(prefix + "/verify", json={**payload, "mode": "runtime"}).status_code == 422
|
|
runtime_input = {**payload, "mode": "runtime", "baseline_note": "人工检查官方消息及功能，开始前无已知异常", "observation_window_minutes": 1}
|
|
runtime = client.post(prefix + "/verify", json=runtime_input).json()
|
|
assert runtime["quality"]["state"] == "awaiting_observation"
|
|
assert datetime.fromisoformat(runtime["observation_due_at"]) == clock[0] + timedelta(minutes=1)
|
|
assert append(runtime, clear).status_code == 409
|
|
assert append(runtime, {**clear, "state": "waiting", "note": "等待窗口到期", "process_review": "not_reviewed"}).status_code == 201
|
|
assert append(runtime, {**clear, "occurred_at": "2026-10-24T01:00:00"}).status_code == 422
|
|
assert append(runtime, {**clear, "occurred_at": "2099-01-01T00:00:00Z"}).status_code == 422
|
|
clock[0] += timedelta(minutes=2)
|
|
assert client.get(prefix).json()["quality"]["state"] == "awaiting_observation", "elapsed time is not an automatic review"
|
|
assert append(runtime, {**clear, "process_review": "not_reviewed"}).status_code == 409
|
|
early_check = (datetime.fromisoformat(runtime["started_at"]) + timedelta(seconds=30)).isoformat()
|
|
assert append(runtime, {**clear, "occurred_at": early_check}).status_code == 409
|
|
assert client.get(prefix).json()["quality"]["state"] == "awaiting_observation", "late registration cannot turn a pre-deadline check into clearance"
|
|
observation = append(runtime, {**clear, "occurred_at": runtime["observation_due_at"]})
|
|
assert observation.status_code == 201 and observation.json()["verification"] == "operator_reported"
|
|
assert client.get(prefix).json()["quality"]["state"] == "verified_in_scope"
|
|
assert not client.post(prefix + "/publish-check").json()["published"]
|
|
store = client.app_reference.state.store
|
|
original_validation = store.one("SELECT data_json FROM validations WHERE id=$1", (runtime["id"],))["data_json"]
|
|
clock[0] += timedelta(minutes=1)
|
|
alert = append(runtime, {**clear, "state": "alert", "source": "official_message", "notice_ref": "本机合成反馈-1", "note": "延后收到的测试异常，需复核关联"})
|
|
assert alert.status_code == 201
|
|
assert client.get(prefix).json()["quality"]["state"] == "review_required"
|
|
assert append(runtime, clear).status_code == 201
|
|
assert client.get(prefix).json()["quality"]["state"] == "review_required", "old run cannot clear a later alert"
|
|
clock[0] += timedelta(seconds=1)
|
|
recovered = client.post(prefix + "/verify", json=runtime_input).json()
|
|
clock[0] += timedelta(minutes=2)
|
|
assert append(recovered, clear).status_code == 201
|
|
assert client.get(prefix).json()["quality"]["state"] == "verified_in_scope"
|
|
clock[0] += timedelta(seconds=1)
|
|
assert append(runtime, {**clear, "state": "uncertain", "note": "旧运行又出现关联未知的后续反馈"}).status_code == 201
|
|
assert client.get(prefix).json()["quality"]["state"] == "review_required"
|
|
assert store.one("SELECT data_json FROM validations WHERE id=$1", (runtime["id"],))["data_json"] == original_validation
|
|
conflict = StoreConflict if store.backend == "postgresql" else sqlite3.IntegrityError
|
|
for statement in ("UPDATE quality_observations SET data_json='{}'", "DELETE FROM quality_observations"):
|
|
with pytest.raises(conflict):
|
|
store.execute(statement)
|
|
if store.backend == "postgresql":
|
|
with pytest.raises(StoreConflict):
|
|
store.execute("TRUNCATE quality_observations")
|
|
restarted = create_app(store.home, FakeWorker(store.home), bootstrap_admin=False,
|
|
database_url=getattr(client, "test_database_url", ""), database_schema=getattr(client, "test_schema", None))
|
|
with TestClient(restarted, base_url="http://127.0.0.1:8810") as second:
|
|
second.cookies.update(client.cookies)
|
|
dto = second.get(prefix).json()
|
|
assert dto["quality"]["state"] == "review_required"
|
|
recorded = next(row for row in dto["validations"] if row["id"] == runtime["id"])
|
|
assert len(recorded["quality_observations"]) == 5
|
|
assert recorded["quality"]["state"] == "review_required"
|
|
assert not second.get("/api/learning/tasks/" + task["id"]).json()["browser_readiness"]["ready"]
|
|
|
|
|
|
def test_quality_auth_scopes_csrf_and_failed_runtime_cannot_pass(client):
|
|
initialize(client)
|
|
_, candidate = working_candidate(client)
|
|
prefix = "/api/learning/candidates/" + candidate["id"]
|
|
static = client.post(prefix + "/check").json()
|
|
payload = {"mode": "runtime", "baseline_note": "本机合成基线", "expected_outputs": {"results": "不会出现"}, "observation_window_minutes": 1}
|
|
run = client.post(prefix + "/verify", json=payload).json()
|
|
assert run["status"] == "failed" and run["quality"]["state"] == "review_required"
|
|
url = "/api/learning/validations/" + run["id"] + "/quality-observations"
|
|
body = {"state": "clear", "note": "人工测试", "process_review": "passed"}
|
|
assert client.post(url, json=body, headers={"X-CSRF-Token": "invalid"}).status_code == 403
|
|
assert client.post(url, json=body).status_code == 409
|
|
assert client.post(url.replace(run["id"], static["id"]), json={**body, "state": "waiting"}).status_code == 422
|
|
with pytest.raises(HTTPException) as caught:
|
|
client.app_reference.state.learning.append_quality_observation(run["id"], "different-maintainer", {**body, "source": "manual"})
|
|
assert caught.value.status_code == 404
|
|
with TestClient(client.app_reference, base_url="http://127.0.0.1:8810") as outsider:
|
|
outsider.headers["X-CSRF-Token"] = outsider.get("/api/auth/status").json()["csrf_token"]
|
|
assert outsider.post(url, json=body).status_code == 401
|