266 lines
16 KiB
Python
266 lines
16 KiB
Python
"""Fixed execution contracts; fake browser only, temporary SQLite / dedicated PG."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from apps.server.db import PostgresStore, SCHEMA, Store, digest, encoded
|
|
from apps.server.learning_contracts import EXECUTION_CONTRACT, EXECUTION_PROFILE, execution_metadata, execution_metadata_matches
|
|
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, learned_detail
|
|
from apps.server.worker import safe_failure_capture, safe_request_observations
|
|
|
|
META = execution_metadata(EXECUTION_PROFILE)
|
|
|
|
|
|
@pytest.fixture(params=["client", "pg_client"])
|
|
def profile_client(request):
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
def install_profile_worker(client):
|
|
"""Add actual worker echoes, with bounded corruption hooks for each RPC."""
|
|
worker = client.worker_reference
|
|
previous_call, previous_capture, previous_post = worker.call, worker.capture, client.post
|
|
profiles, state = {}, {"method": None, "place": "root", "mutation": None}
|
|
|
|
def capture(task_id):
|
|
result = previous_capture(task_id)
|
|
result.update(execution_metadata(profiles.get(task_id)))
|
|
return result
|
|
|
|
async def call(method, params):
|
|
if method == "open":
|
|
profiles[params["task_id"]] = params.get("execution_profile")
|
|
result = await previous_call(method, params)
|
|
if method != "close" and not (method == "status" and result.get("status") == "closed"):
|
|
result.update(execution_metadata(profiles.get(params["task_id"])))
|
|
if state["method"] == method:
|
|
target = result if state["place"] == "root" else result[state["place"]]
|
|
state["mutation"](target)
|
|
return result
|
|
|
|
def post(url, **kwargs):
|
|
if str(url) == "/api/learning/tasks" and "json" in kwargs:
|
|
kwargs["json"] = {"execution_profile": EXECUTION_PROFILE, **kwargs["json"]}
|
|
return previous_post(url, **kwargs)
|
|
|
|
worker.capture, worker.call, client.post = capture, call, post
|
|
return state, profiles
|
|
|
|
|
|
def raw_facts(store):
|
|
return {table: store.rows(f"SELECT * FROM {table} ORDER BY id") for table in ("observations", "learning_steps", "candidates", "validations")}
|
|
|
|
|
|
def test_fixed_descriptor_and_native_metadata_are_exact_not_caller_combinations():
|
|
assert META == {"execution_profile": EXECUTION_PROFILE, "execution_contract": EXECUTION_CONTRACT}
|
|
assert execution_metadata(None) == {}
|
|
assert execution_metadata_matches({}, None)
|
|
for value in ("legacy", "", {}, True):
|
|
with pytest.raises(ValueError):
|
|
execution_metadata(value)
|
|
for value in ({}, {"execution_profile": EXECUTION_PROFILE}, META | {"execution_contract": EXECUTION_CONTRACT | {"extra": True}}, META | {"execution_contract": EXECUTION_CONTRACT | {"engine": "browser_use"}}):
|
|
assert not execution_metadata_matches(value, EXECUTION_PROFILE)
|
|
request = {"method": "POST", "origin": "https://www.xiaohongshu.com", "pathname": "/api/telemetry", "resource_type": "xhr", "automating": True, "blocked": False,
|
|
"policy_version": "xhs-page-native-v1", "policy_rule": "native_page_traffic"}
|
|
assert safe_request_observations([request])[0]["policy_rule"] == "native_page_traffic"
|
|
for mutation in ({"policy_version": "xhs-readonly-ui-experimental-v1"}, {"policy_rule": "collect"}, {"blocked": True}, {"automating": False}):
|
|
assert "policy_rule" not in safe_request_observations([request | mutation])[0]
|
|
capture = {"url": "http://127.0.0.1/fixture", "elements": [], "screenshot_path": "synthetic.png", **META}
|
|
assert execution_metadata_matches(safe_failure_capture(capture), EXECUTION_PROFILE)
|
|
assert safe_failure_capture(capture | {"execution_contract": {}}) is None
|
|
legacy = {key: value for key, value in capture.items() if key not in META}
|
|
assert execution_metadata_matches(safe_failure_capture(legacy), None)
|
|
|
|
|
|
def test_nullable_legacy_and_explicit_profile_roundtrip_do_not_rewrite_history(profile_client):
|
|
client = profile_client
|
|
initialize(client)
|
|
old_task, old_candidate = working_candidate(client)
|
|
store = client.app_reference.state.store
|
|
facts = raw_facts(store)
|
|
assert store.one("SELECT execution_profile FROM learning_tasks WHERE id=$1", (old_task["id"],))["execution_profile"] is None
|
|
assert "execution_profile" not in client.get("/api/learning/tasks/" + old_task["id"]).json()
|
|
base = {key: old_task[key] for key in ("name", "platform", "entry_url", "goal", "account_key", "parameters")}
|
|
null = client.post("/api/learning/tasks", json=base | {"execution_profile": None})
|
|
assert null.status_code == 201 and "execution_profile" not in null.json()
|
|
for bad in ("legacy", "browser_use", "", {}, True):
|
|
assert client.post("/api/learning/tasks", json=base | {"execution_profile": bad}).status_code == 422
|
|
install_profile_worker(client)
|
|
task, candidate = working_candidate(client)
|
|
assert task["execution_profile"] == EXECUTION_PROFILE and "execution_contract" not in task
|
|
assert all(candidate["manifest"][key] == value for key, value in META.items())
|
|
assert digest(candidate["manifest"]) == candidate["digest"]
|
|
endpoint = "/api/learning/candidates/" + candidate["id"]
|
|
checked = client.post(endpoint + "/check").json()
|
|
assert checked["status"] == "passed" and "execution_profile" not in checked["environment"]
|
|
verify = client.post(endpoint + "/verify", json={"expected_outputs": {"results": "建材"}}).json()
|
|
assert verify["status"] == "passed" and all(verify["environment"][key] == value for key, value in META.items())
|
|
wire = [params for method, params in client.worker_reference.calls if method == "verify"][-1]
|
|
assert all(wire[key] == value for key, value in META.items())
|
|
detail = client.get("/api/learning/tasks/" + task["id"]).json()
|
|
assert all(detail["browser_readiness"][key] == value for key, value in META.items())
|
|
assert all(execution_metadata_matches(value, EXECUTION_PROFILE) for value in detail["observations"])
|
|
# Neither action nor candidate creation can override a task's contract.
|
|
for path, data in (("/browser/action", {"kind": "read", "selector": '[data-testid="result"]', "output_name": "results", "execution_profile": EXECUTION_PROFILE}),
|
|
("/candidates", {"execution_profile": EXECUTION_PROFILE})):
|
|
assert client.post("/api/learning/tasks/" + task["id"] + path, json=data).status_code == 422
|
|
old = client.get("/api/learning/candidates/" + old_candidate["id"]).json()
|
|
assert old["digest"] == old_candidate["digest"] and old["manifest"] == old_candidate["manifest"]
|
|
current = raw_facts(store)
|
|
for table, rows in facts.items():
|
|
assert [row for row in current[table] if row["id"] in {item["id"] for item in rows}] == rows
|
|
|
|
|
|
def test_profile_echo_missing_open_status_capture_and_action_never_become_success(profile_client):
|
|
client = profile_client
|
|
initialize(client)
|
|
state, _ = install_profile_worker(client)
|
|
task, _ = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
store = client.app_reference.state.store
|
|
old_steps = store.rows("SELECT * FROM learning_steps WHERE task_id=$1", (task["id"],))
|
|
old_obs = store.one("SELECT count(*) AS n FROM observations WHERE task_id=$1", (task["id"],))["n"]
|
|
def strip(value):
|
|
value.pop("execution_profile", None)
|
|
value.pop("execution_contract", None)
|
|
for method, place, path, data in (("capture", "root", "/browser/capture", {}),
|
|
("action", "root", "/browser/action", {"kind": "read", "selector": '[data-testid="result"]', "output_name": "results"}),
|
|
("action", "before", "/browser/action", {"kind": "read", "selector": '[data-testid="result"]', "output_name": "results"}),
|
|
("status", "root", "/browser/capture", {}), ("open", "root", "/browser/open", {})):
|
|
state.update(method=None)
|
|
assert client.post(prefix + "/browser/open").status_code == 200
|
|
assert client.post(prefix + "/browser/ready", json={"identity_label": "测试"}).status_code == 200
|
|
state.update(method=method, place=place, mutation=strip)
|
|
response = client.post(prefix + path, json=data)
|
|
assert response.status_code == 409, (method, place, response.text)
|
|
assert not client.app_reference.state.learning.browser_readiness(task["id"])["ready"]
|
|
assert store.rows("SELECT * FROM learning_steps WHERE task_id=$1", (task["id"],)) == old_steps
|
|
assert store.one("SELECT count(*) AS n FROM observations WHERE task_id=$1", (task["id"],))["n"] == old_obs
|
|
|
|
|
|
def test_new_profile_cannot_relabel_old_evidence_or_accept_mutated_manifest(profile_client):
|
|
client = profile_client
|
|
initialize(client)
|
|
old_task, candidate = working_candidate(client)
|
|
service = client.app_reference.state.learning
|
|
original = raw_facts(service.store)
|
|
service.store.execute("UPDATE learning_tasks SET execution_profile=$1 WHERE id=$2", (EXECUTION_PROFILE, old_task["id"]))
|
|
result = client.post("/api/learning/tasks/" + old_task["id"] + "/candidates", json={})
|
|
assert result.status_code == 409 and result.json()["detail"]["code"] == "EXECUTION_PROFILE_MISMATCH"
|
|
assert service.static_errors(candidate)
|
|
for change in ({}, {"execution_profile": "unknown", "execution_contract": EXECUTION_CONTRACT}, META):
|
|
changed = copy.deepcopy(candidate)
|
|
changed["manifest"].update(change)
|
|
changed["digest"] = digest(changed["manifest"])
|
|
assert service.static_errors(changed)
|
|
assert raw_facts(service.store) == original
|
|
|
|
|
|
def test_verify_bad_actual_profile_or_capture_records_failure_without_usable_outputs(profile_client):
|
|
client = profile_client
|
|
initialize(client)
|
|
state, _ = install_profile_worker(client)
|
|
task, candidate = working_candidate(client)
|
|
endpoint = "/api/learning/candidates/" + candidate["id"]
|
|
assert client.post(endpoint + "/check").json()["status"] == "passed"
|
|
for mutate in (lambda result: result.pop("execution_contract"),
|
|
lambda result: result.update(execution_profile="unknown"),
|
|
lambda result: result["evidence"][0].pop("execution_profile")):
|
|
state.update(method=None)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
client.post(prefix + "/browser/open")
|
|
client.post(prefix + "/browser/ready", json={"identity_label": "测试"})
|
|
state.update(method="verify", place="root", mutation=mutate)
|
|
response = client.post(endpoint + "/verify", json={"expected_outputs": {"results": "建材"}})
|
|
assert response.status_code == 200, response.text
|
|
result = response.json()
|
|
assert result["status"] == "failed" and "EXECUTION_PROFILE_MISMATCH" in result["error_codes"]
|
|
assert result["actual_outputs"] == {} and result["evidence"] == []
|
|
assert "execution_profile" not in result["environment"]
|
|
|
|
|
|
def test_new_discovery_and_reading_keep_actual_validation_profile_without_upgrading_old_search(profile_client):
|
|
client = profile_client
|
|
initialize(client)
|
|
_, old_search = search_candidate(client, [item("准备装修求推荐靠谱公司")])
|
|
old_run = client.post("/api/discovery/runs", json={"search_candidate_id": old_search["id"], "keyword": "装修", "region": "", "max_selected": 1}).json()
|
|
assert old_run["status"] == "completed" and "execution_profile" not in old_run
|
|
install_profile_worker(client)
|
|
_, candidate, _ = learned_detail(client)
|
|
result = client.post("/api/discovery/runs/" + old_run["id"] + "/readings", json={"detail_candidate_id": candidate["id"], "note_id": NOTE_ID})
|
|
assert result.status_code == 201, result.text
|
|
reading = result.json()
|
|
assert reading["status"] == "completed" and all(reading[key] == value for key, value in META.items())
|
|
assert client.get("/api/discovery/runs/" + old_run["id"]).json() == old_run
|
|
_, new_search = search_candidate(client, [item("准备装修求推荐靠谱公司")])
|
|
new_run = client.post("/api/discovery/runs", json={"search_candidate_id": new_search["id"], "keyword": "装修", "region": "", "max_selected": 1}).json()
|
|
assert new_run["status"] == "completed" and all(new_run[key] == value for key, value in META.items())
|
|
|
|
|
|
def old_sqlite(home):
|
|
home.mkdir(parents=True, exist_ok=True)
|
|
target = home / "learning.sqlite"
|
|
with sqlite3.connect(target) as conn:
|
|
conn.executescript(SCHEMA.replace(", execution_profile TEXT", ""))
|
|
conn.execute("INSERT INTO operators VALUES(?,?,?,?)", ("owner", "old", "not-used", "2026-10-25"))
|
|
conn.execute("INSERT INTO learning_tasks VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", ("old-task", "owner", "old", "demo", "http://127.0.0.1/demo", "", "old", "{}", "draft", None, "2026-10-25", "2026-10-25"))
|
|
manifest = {"schema_version": 2, "steps": [], "parameters": {}}
|
|
conn.execute("INSERT INTO candidates VALUES(?,?,?,?,?,?,?)", ("old-candidate", "old-task", "old", 1, digest(manifest), encoded(manifest), "2026-10-25"))
|
|
return target
|
|
|
|
|
|
def test_sqlite_existing_schema_upgrade_is_idempotent_and_preserves_old_manifest(tmp_path):
|
|
target = old_sqlite(tmp_path / "old")
|
|
with sqlite3.connect(target) as conn:
|
|
original = conn.execute("SELECT digest,manifest_json FROM candidates").fetchall()
|
|
store = Store(target.parent)
|
|
assert store.one("SELECT execution_profile FROM learning_tasks")["execution_profile"] is None
|
|
assert store.one("SELECT version FROM schema_versions WHERE version=10")
|
|
assert [(row["digest"], row["manifest_json"]) for row in store.rows("SELECT digest,manifest_json FROM candidates")] == original
|
|
assert Store(target.parent).one("SELECT count(*) AS n FROM schema_versions WHERE version=10")["n"] == 1
|
|
|
|
|
|
def test_postgres_existing_column_upgrade_and_old_sqlite_backup_migration(pg_client, tmp_path):
|
|
store = pg_client.app_reference.state.store
|
|
store.execute("ALTER TABLE learning_tasks DROP COLUMN execution_profile")
|
|
upgraded = PostgresStore(store.home, pg_client.test_database_url, schema=pg_client.test_schema)
|
|
assert upgraded.one("SELECT count(*) AS n FROM information_schema.columns WHERE table_schema=$1 AND table_name='learning_tasks' AND column_name='execution_profile'", (pg_client.test_schema,))["n"] == 1
|
|
assert upgraded.one("SELECT version FROM schema_versions WHERE version=10")
|
|
home = tmp_path / "legacy-import"
|
|
source = old_sqlite(home)
|
|
original_hash = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
schema = "assistant_learning_test_" + uuid4().hex
|
|
assert pg_client.test_database_url and "/assistant_learning_test" in pg_client.test_database_url
|
|
env = {**os.environ, "ASSISTANT_DATABASE_URL": pg_client.test_database_url, "ASSISTANT_DATABASE_SCHEMA": schema, "ASSISTANT_LEARNING_HOME": str(home), "PYTHONUTF8": "1"}
|
|
try:
|
|
completed = subprocess.run([sys.executable, "-m", "scripts.migrate_learning"], cwd=Path(__file__).resolve().parents[3], env=env, capture_output=True, timeout=60)
|
|
assert completed.returncode == 0, "isolated legacy migration failed"
|
|
summary = json.loads(completed.stdout)
|
|
assert summary["result"] == "copied"
|
|
imported = PostgresStore(home, pg_client.test_database_url, schema=schema)
|
|
assert imported.one("SELECT execution_profile FROM learning_tasks")["execution_profile"] is None
|
|
assert imported.one("SELECT manifest_json FROM candidates")["manifest_json"] == encoded({"schema_version": 2, "steps": [], "parameters": {}})
|
|
assert hashlib.sha256(source.read_bytes()).hexdigest() == original_hash
|
|
backups = list((home / "backups").glob("*.sqlite"))
|
|
assert len(backups) == 1
|
|
with sqlite3.connect(backups[0]) as conn:
|
|
assert "execution_profile" not in {row[1] for row in conn.execute("PRAGMA table_info(learning_tasks)")}
|
|
finally:
|
|
import psycopg
|
|
from psycopg import sql
|
|
with psycopg.connect(pg_client.test_database_url, connect_timeout=5) as conn:
|
|
conn.execute(sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format(sql.Identifier(schema)))
|