400 lines
26 KiB
Python
400 lines
26 KiB
Python
"""Bounded learning failures and search Enter; fake browser + isolated stores only."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
import sqlite3
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
from apps.server.db import StoreConflict, digest
|
|
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 BrowserWorker, WorkerError, safe_diagnostics, safe_popup_observations, safe_request_observations, safe_request_summary
|
|
|
|
SEARCH = '[data-testid="search-input"]'
|
|
|
|
|
|
@pytest.fixture(params=["client", "pg_client"])
|
|
def failure_client(request):
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
def test_failed_action_appends_actual_before_after_without_step_or_retry(failure_client):
|
|
client = failure_client
|
|
initialize(client)
|
|
task, original_candidate = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
service, worker = client.app_reference.state.learning, client.worker_reference
|
|
old_call = worker.call
|
|
before, after = worker.capture(task["id"]), worker.capture(task["id"])
|
|
after["text"] = "动作已经产生局部效果，不能推断未发生"
|
|
before["headers"] = {"Authorization": "private-fixture"}
|
|
after["request_observations"] = [{"sequence": 1, "recorded_at": "2026-10-24T08:00:00.000Z", "method": "POST", "origin": "https://example.test", "pathname": "/api/token/raw-secret?query=private-fixture", "resource_type": "xhr", "automating": True, "blocked": True, "headers": {"Cookie": "private-fixture"}, "body": "private-fixture"}]
|
|
diagnostics = {"phase": "request_guard", "blocked_request_count": 1, "blocked_requests": [{"method": "POST", "origin": "https://example.test", "pathname": "/api/feed?credential=private-fixture#fragment", "resource_type": "xhr", "query": "private-fixture", "headers": "private-fixture", "body": "private-fixture"}], "stack": "private-fixture"}
|
|
attempts = []
|
|
|
|
async def failed(method, params):
|
|
if method == "action":
|
|
attempts.append(copy.deepcopy(params))
|
|
raise WorkerError("Raw error private-fixture", "WRITE_REQUEST_BLOCKED", before=before, after=after, diagnostics=diagnostics)
|
|
return await old_call(method, params)
|
|
worker.call = failed
|
|
response = client.post(prefix + "/browser/action", json={"kind": "click", "selector": '[data-testid="search-button"]'})
|
|
assert response.status_code == 409
|
|
detail = response.json()["detail"]
|
|
assert detail["code"] == "WRITE_REQUEST_BLOCKED" and detail["evidence_status"] == "recorded"
|
|
assert len(attempts) == 1 and len(service.steps(task["id"])) == 3
|
|
assert detail["action_failure"]["phase"] == "request_guard"
|
|
assert detail["action_failure"]["diagnostics"]["blocked_requests"] == [{"method": "POST", "origin": "https://example.test", "pathname": "/api/feed", "resource_type": "xhr"}]
|
|
observations = client.get(prefix).json()["observations"][-2:]
|
|
assert [item["id"] for item in observations] == detail["failure_observation_ids"]
|
|
assert detail["before_observation_id"] == observations[0]["id"]
|
|
assert detail["after_observation_id"] == observations[1]["id"]
|
|
assert observations[1]["text"] == after["text"]
|
|
for moment, item in zip(("before", "after"), observations):
|
|
assert item["source"] == "action_failure" and item["action_failure"]["moment"] == moment
|
|
assert item["action_failure"]["attempt_id"] == detail["action_failure"]["attempt_id"]
|
|
assert "screenshot_path" not in item and client.get(item["screenshot_url"]).status_code == 200
|
|
assert "private-fixture" not in json.dumps(item)
|
|
assert observations[1]["request_observations"][0]["pathname"] == "/api/:redacted/:redacted"
|
|
assert "private-fixture" not in response.text
|
|
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, (observations[0]["id"],))
|
|
frozen = client.get("/api/learning/candidates/" + original_candidate["id"]).json()
|
|
assert frozen["digest"] == original_candidate["digest"] and frozen["manifest"] == original_candidate["manifest"]
|
|
next_candidate = client.post(prefix + "/candidates", json={}).json()
|
|
assert len(next_candidate["manifest"]["steps"]) == 3
|
|
assert not set(detail["failure_observation_ids"]) & {binding["observation_id"] for binding in next_candidate["manifest"]["evidence_bindings"]}
|
|
owner = service.store.one("SELECT owner_id FROM learning_tasks WHERE id=$1", (task["id"],))["owner_id"]
|
|
with pytest.raises(HTTPException) as error:
|
|
service.detail(task["id"], owner + "-other")
|
|
assert error.value.status_code == 404
|
|
with TestClient(client.app_reference, base_url="http://127.0.0.1:8810") as anonymous:
|
|
assert anonymous.get(prefix).status_code == 401
|
|
assert anonymous.get(observations[0]["screenshot_url"]).status_code == 401
|
|
|
|
|
|
def test_failed_action_evidence_rejection_does_not_mask_original_error(client):
|
|
initialize(client)
|
|
task, _ = working_candidate(client)
|
|
worker = client.worker_reference
|
|
old_call = worker.call
|
|
after = worker.capture(task["id"])
|
|
after["screenshot_path"] = str(worker.home.parent / "outside.png")
|
|
|
|
async def fail(method, params):
|
|
if method == "action":
|
|
raise WorkerError("not returned", "WAIT_TIMEOUT", before=None, after=after, diagnostics={"phase": "settle"})
|
|
return await old_call(method, params)
|
|
worker.call = fail
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
previous_count = len(client.get(prefix).json()["observations"])
|
|
response = client.post(prefix + "/browser/action", json={"kind": "wait", "selector": "body"})
|
|
assert response.status_code == 409 and response.json()["detail"]["code"] == "WAIT_TIMEOUT"
|
|
assert response.json()["detail"]["failure_observation_ids"] == []
|
|
assert response.json()["detail"]["evidence_status"] == "unavailable"
|
|
assert len(client.get(prefix).json()["observations"]) == previous_count
|
|
assert len(client.app_reference.state.learning.steps(task["id"])) == 3
|
|
assert "outside.png" not in response.text
|
|
|
|
|
|
def test_search_enter_contract_static_evidence_and_legacy_scope(client):
|
|
initialize(client)
|
|
task, _ = working_candidate(client)
|
|
worker = client.worker_reference
|
|
capture = worker.capture
|
|
|
|
def search_capture(task_id):
|
|
value = capture(task_id)
|
|
for item in value["elements"]:
|
|
if item["selector"] == SEARCH:
|
|
item["search_input"] = True
|
|
return value
|
|
worker.capture = search_capture
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
client.post(prefix + "/browser/capture", json={})
|
|
response = client.post(prefix + "/browser/action", json={"kind": "press", "selector": SEARCH, "key": "Enter"})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["step"]["key"] == "Enter"
|
|
candidate = client.post(prefix + "/candidates", json={}).json()
|
|
service = client.app_reference.state.learning
|
|
assert not service.static_errors(candidate)
|
|
for value in (
|
|
{"kind": "press", "selector": SEARCH},
|
|
{"kind": "press", "selector": SEARCH, "key": "Control+Enter"},
|
|
{"kind": "press", "selector": SEARCH, "key": "Enter", "value": "send"},
|
|
{"kind": "click", "selector": SEARCH, "key": "Enter"},
|
|
{"kind": "press", "selector": '[data-testid="search-button"]', "key": "Enter"},
|
|
):
|
|
assert client.post(prefix + "/browser/action", json=value).status_code == 422
|
|
changed = copy.deepcopy(candidate)
|
|
changed["manifest"]["schema_version"] = 1
|
|
changed["digest"] = digest(changed["manifest"])
|
|
assert any("只读范围" in item for item in service.static_errors(changed))
|
|
worker.capture = capture
|
|
client.post(prefix + "/browser/capture", json={})
|
|
assert client.post(prefix + "/browser/action", json={"kind": "press", "selector": SEARCH, "key": "Enter"}).status_code == 422
|
|
|
|
|
|
def test_worker_rpc_preserves_only_safe_failure_details(tmp_path):
|
|
payload = {"code": "WRITE_REQUEST_BLOCKED", "message": "fixture failed", "before": {"url": "http://127.0.0.1/", "elements": [], "screenshot_path": str(tmp_path / "capture.png"), "headers": {"Authorization": "fixture-secret"}}, "diagnostics": {"phase": "execute", "blocked_requests": [], "blocked_request_count": 0, "token": "fixture-secret"}}
|
|
|
|
class Pipe:
|
|
def write(self, value):
|
|
self.request = json.loads(value)
|
|
async def drain(self):
|
|
pass
|
|
async def readline(self):
|
|
return json.dumps({"id": self.request["id"], "error": payload}).encode()
|
|
|
|
class Process:
|
|
returncode = None
|
|
stdin = stdout = Pipe()
|
|
|
|
async def check():
|
|
worker = BrowserWorker(tmp_path, tmp_path)
|
|
worker.process = Process()
|
|
with pytest.raises(WorkerError) as failure:
|
|
await worker.call("action", {"task_id": "fixture"})
|
|
error = failure.value
|
|
assert error.code == "WRITE_REQUEST_BLOCKED" and error.before["screenshot_path"].endswith("capture.png")
|
|
assert error.after is None and error.diagnostics["phase"] == "execute"
|
|
assert "fixture-secret" not in json.dumps({"before": error.before, "diagnostics": error.diagnostics})
|
|
asyncio.run(check())
|
|
assert safe_diagnostics({"phase": ["malformed"], "blocked_requests": [{"method": [], "resource_type": {}}]})["phase"] == "unknown"
|
|
|
|
|
|
def test_login_detection_precedes_confirmation_and_handles_session_changes(client):
|
|
initialize(client)
|
|
task, _ = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
worker, service = client.worker_reference, client.app_reference.state.learning
|
|
old_call = worker.call
|
|
current = {"state": "unknown", "basis": "visible_page", "observed_at": "2026-10-24T08:00:00.000Z", "evidence": ["no_conclusive_visible_evidence", "private-fixture"], "cookie": "private-fixture"}
|
|
|
|
async def detected(method, params):
|
|
result = await old_call(method, params)
|
|
if method in {"open", "status", "ready"}:
|
|
result["login_detection"] = copy.deepcopy(current)
|
|
return result
|
|
worker.call = detected
|
|
# An uncertain detector does not revoke an explicit, active confirmation.
|
|
readiness = client.get(prefix).json()["browser_readiness"]
|
|
assert readiness["ready"] and readiness["login_detection"]["evidence"] == ["no_conclusive_visible_evidence"]
|
|
assert "private-fixture" not in json.dumps(readiness)
|
|
current["state"] = "logged_out"
|
|
blocked = client.post(prefix + "/browser/ready", json={"identity_label": "fixture"})
|
|
assert blocked.status_code == 409 and blocked.json()["detail"]["code"] == "login_required"
|
|
assert not client.get(prefix).json()["browser_readiness"]["ready"]
|
|
count = len([call for call in worker.calls if call[0] == "action"])
|
|
assert client.post(prefix + "/browser/action", json={"kind": "read", "selector": '[data-testid="result"]', "output_name": "x"}).status_code == 409
|
|
assert len([call for call in worker.calls if call[0] == "action"]) == count
|
|
current["state"] = "verification_required"
|
|
assert client.post(prefix + "/browser/ready", json={"identity_label": "fixture"}).json()["detail"]["code"] == "verification_required"
|
|
current["state"] = "logged_in"
|
|
# Detection alone does not confirm a user or ask them to scan a second time.
|
|
owner = service.store.one("SELECT owner_id FROM learning_tasks WHERE id=$1", (task["id"],))["owner_id"]
|
|
assert not asyncio.run(service.refresh_browser(task["id"], owner))["ready"]
|
|
assert "无需重新扫码" in service.browser_readiness(task["id"])["message"]
|
|
assert client.post(prefix + "/browser/ready", json={"identity_label": "fixture"}).status_code == 200
|
|
current["state"] = "unknown"
|
|
assert asyncio.run(service.refresh_browser(task["id"], owner))["ready"]
|
|
|
|
|
|
@pytest.mark.parametrize("reason", ["popup_blocked", "scope_blocked"])
|
|
def test_blocked_session_overrides_logged_in_and_old_confirmation(client, reason):
|
|
initialize(client)
|
|
task, _ = working_candidate(client)
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
worker, service = client.worker_reference, client.app_reference.state.learning
|
|
original = worker.call
|
|
blocked = False
|
|
ready_attempts = 0
|
|
|
|
async def blocked_call(method, params):
|
|
nonlocal blocked, ready_attempts
|
|
if method == "ready":
|
|
ready_attempts += 1
|
|
blocked = True
|
|
raise WorkerError("fixture block", reason.upper())
|
|
result = await original(method, params)
|
|
if method == "status":
|
|
# Deliberately retain stale manual confirmation to test the API
|
|
# boundary, even though the updated worker also revokes it.
|
|
result.update(manual_confirmation=True, block_reason=reason if blocked else None,
|
|
login_detection={"state": "logged_in", "basis": "visible_page", "observed_at": "2026-10-24T08:00:00.000Z", "evidence": ["visible_current_account_navigation"]})
|
|
return result
|
|
worker.call = blocked_call
|
|
result = client.post(prefix + "/browser/ready", json={"identity_label": "fixture"})
|
|
assert result.status_code == 409 and result.json()["detail"]["code"] == reason
|
|
cached = service.browser_readiness(task["id"])
|
|
assert cached["ready"] is False and cached["code"] == reason
|
|
assert cached["next_action"] == "close_and_reopen_browser"
|
|
readiness = client.get(prefix).json()["browser_readiness"]
|
|
assert readiness["code"] == reason and readiness["block_reason"] == reason
|
|
assert readiness["login_detection"]["state"] == "logged_in" and readiness["ready"] is False
|
|
assert "关闭" in readiness["message"] and "重新打开" in readiness["message"]
|
|
assert "无需重新扫码" not in readiness["message"]
|
|
assert client.post(prefix + "/browser/ready", json={"identity_label": "fixture"}).json()["detail"]["code"] == reason
|
|
assert ready_attempts == 1
|
|
action_count = len([call for call in worker.calls if call[0] == "action"])
|
|
action = client.post(prefix + "/browser/action", json={"kind": "read", "selector": '[data-testid="result"]', "output_name": "x"})
|
|
assert action.status_code == 409 and action.json()["detail"]["code"] == reason
|
|
assert len([call for call in worker.calls if call[0] == "action"]) == action_count
|
|
|
|
|
|
def test_popup_events_are_bounded_safe_page_events_not_request_records(client):
|
|
initialize(client)
|
|
task, _ = working_candidate(client)
|
|
worker = client.worker_reference
|
|
original = worker.capture
|
|
events = [{"sequence": index, "recorded_at": "2026-10-24T08:00:00.000Z", "phase": "manual", "decision": "adopted", "reason": "same_origin_manual", "origin": "https://example.test", "pathname": "/explore/token/private-id?credential=fixture-secret#fragment", "method": "GET", "headers": {"Cookie": "fixture-secret"}, "body": "fixture-secret"} for index in range(12)]
|
|
|
|
def capture_with_events(task_id):
|
|
result = original(task_id)
|
|
result["popup_observations"] = copy.deepcopy(events)
|
|
return result
|
|
worker.capture = capture_with_events
|
|
observation = client.post("/api/learning/tasks/" + task["id"] + "/browser/capture", json={}).json()
|
|
records = observation["popup_observations"]
|
|
assert len(records) == 10 and records[0]["sequence"] == 2
|
|
assert records[-1]["pathname"] == "/explore/:redacted/private-id"
|
|
assert all(set(item) == {"sequence", "recorded_at", "phase", "decision", "reason", "origin", "pathname"} for item in records)
|
|
assert "fixture-secret" not in json.dumps(records)
|
|
error = WorkerError("fixture", before=capture_with_events(task["id"]))
|
|
assert error.before["popup_observations"] == records
|
|
assert safe_popup_observations([{**events[0], "phase": "raw-secret"}, {**events[0], "decision": []}]) == []
|
|
anonymous_origin = safe_popup_observations([{**events[0], "origin": "https://user:fixture-secret@example.test"}])[0]
|
|
assert "origin" not in anonymous_origin and "pathname" not in anonymous_origin
|
|
|
|
|
|
def test_experimental_request_policy_metadata_is_paired_exact_and_observation_only():
|
|
base = {"method": "POST", "origin": "https://example.test", "pathname": "/api/feed?token=fixture-secret", "resource_type": "xhr", "automating": True, "blocked": False, "policy_version": "xhs-search-experimental-v1", "policy_rule": "collect"}
|
|
policies = {
|
|
"xhs-search-experimental-v1": ("collect", "search_history", "search_notes", "search_onebox", "search_dqa"),
|
|
"xhs-search-experimental-v2": ("collect", "search_history", "search_notes", "search_onebox", "search_dqa", "apm_data"),
|
|
"xhs-search-experimental-v3": ("collect", "search_history", "search_notes", "search_onebox", "search_dqa", "apm_data", "apm_data_xhr"),
|
|
"xhs-note-open-experimental-v1": ("collect", "note_share_code", "note_widgets", "note_feed", "note_seo", "note_metrics", "apm_data"),
|
|
"xhs-note-open-experimental-v2": ("collect", "note_share_code", "note_widgets", "note_feed", "note_seo", "note_metrics", "apm_data", "note_history_report"),
|
|
"xhs-readonly-ui-experimental-v1": ("collect", "apm_data"),
|
|
"xhs-search-scroll-experimental-v1": ("collect", "apm_data", "search_notes"),
|
|
}
|
|
for version, rules in policies.items():
|
|
for rule in rules:
|
|
item = {**base, "policy_version": version, "policy_rule": rule}
|
|
observation = safe_request_observations([item])[0]
|
|
assert observation["policy_version"] == version and observation["policy_rule"] == rule
|
|
assert observation["pathname"] == "/api/feed"
|
|
assert set(safe_request_summary(item)) == {"method", "origin", "pathname", "resource_type"}
|
|
diagnostic = safe_diagnostics({"phase": "request_guard", "blocked_requests": [item], "blocked_request_count": 1})
|
|
assert set(diagnostic["blocked_requests"][0]) == {"method", "origin", "pathname", "resource_type"}
|
|
for changed in (
|
|
{"policy_version": "xhs-search-experimental-v999"}, {"policy_version": None}, {"policy_version": []}, {"policy_rule": "other"},
|
|
{"policy_version": "xhs-search-experimental-v1", "policy_rule": "apm_data"},
|
|
{"policy_version": "xhs-search-experimental-v2", "policy_rule": "note_feed"},
|
|
{"policy_version": "xhs-note-open-experimental-v1", "policy_rule": "search_notes"},
|
|
{"policy_version": "xhs-note-open-experimental-v1", "policy_rule": "note_history_report"},
|
|
{"policy_version": "xhs-note-open-experimental-v2", "policy_rule": "search_notes"},
|
|
{"policy_version": "xhs-search-experimental-v2", "policy_rule": "note_history_report"},
|
|
{"policy_version": "xhs-search-experimental-v2", "policy_rule": "apm_data_xhr"},
|
|
{"policy_version": "xhs-note-open-experimental-v2", "policy_rule": "apm_data_xhr"},
|
|
{"policy_version": "xhs-readonly-ui-experimental-v1", "policy_rule": "apm_data_xhr"},
|
|
{"policy_version": "xhs-readonly-ui-experimental-v1", "policy_rule": "search_notes"},
|
|
{"policy_version": "xhs-readonly-ui-experimental-v1", "policy_rule": "note_feed"},
|
|
{"policy_version": "xhs-readonly-ui-experimental-v1", "policy_rule": "note_history_report"},
|
|
{"policy_version": "xhs-search-scroll-experimental-v1", "policy_rule": "search_history"},
|
|
{"policy_version": "xhs-search-scroll-experimental-v1", "policy_rule": "apm_data_xhr"},
|
|
{"policy_version": "xhs-search-scroll-experimental-v1", "policy_rule": "note_feed"},
|
|
{"policy_rule": []}, {"automating": False}, {"automating": 1}, {"automating": "true"},
|
|
{"blocked": True}, {"blocked": 0}, {"blocked": "false"}, {"blocked": None},
|
|
):
|
|
observation = safe_request_observations([{**base, **changed}])[0]
|
|
assert "policy_version" not in observation and "policy_rule" not in observation
|
|
for missing in ("policy_version", "policy_rule", "automating", "blocked"):
|
|
observation = safe_request_observations([{key: value for key, value in base.items() if key != missing}])[0]
|
|
assert "policy_version" not in observation and "policy_rule" not in observation
|
|
|
|
|
|
def test_search_scroll_request_metadata_is_narrow_and_drops_forged_scope_and_private_data():
|
|
version = "xhs-search-scroll-experimental-v1"
|
|
base = {"method": "POST", "origin": "https://edith.xiaohongshu.com", "pathname": "/api/sns/web/v1/search/notes?keyword=fixture-secret#credential",
|
|
"resource_type": "fetch", "automating": True, "blocked": False, "policy_version": version, "policy_rule": "search_notes",
|
|
"scope": "note_open", "policy_scope": "search", "pageUrl": "https://example.test/?token=fixture-secret",
|
|
"headers": {"authorization": "fixture-secret"}, "body": {"keyword": "fixture-secret"}, "response_body": "fixture-secret"}
|
|
rules = ("collect", "apm_data", "search_notes")
|
|
for rule in rules:
|
|
result = safe_request_observations([{**base, "policy_rule": rule}])[0]
|
|
assert result == {"method": "POST", "origin": base["origin"], "pathname": "/api/sns/web/v1/search/notes", "resource_type": "fetch", "automating": True, "blocked": False,
|
|
"policy_version": version, "policy_rule": rule}
|
|
assert "fixture-secret" not in json.dumps(result)
|
|
assert set(safe_request_summary(base)) == {"method", "origin", "pathname", "resource_type"}
|
|
other_rules = ("search_history", "search_onebox", "search_dqa", "apm_data_xhr", "note_share_code", "note_widgets", "note_feed", "note_seo", "note_metrics", "note_history_report", "other")
|
|
for rule in other_rules:
|
|
result = safe_request_observations([{**base, "policy_rule": rule}])[0]
|
|
assert "policy_version" not in result and "policy_rule" not in result
|
|
for changes in ({"policy_version": version + "0"}, {"policy_version": "xhs-readonly-ui-experimental-v1"}, {"policy_version": "xhs-note-open-experimental-v2"},
|
|
{"policy_rule": None}, {"policy_rule": ["search_notes"]}, {"automating": False}, {"automating": 1}, {"blocked": True}, {"blocked": 0}):
|
|
result = safe_request_observations([{**base, **changes}])[0]
|
|
assert "policy_version" not in result and "policy_rule" not in result
|
|
sensitive = safe_request_observations([{**base, "pathname": "/api/token/13812345678?credential=fixture-secret"}])[0]
|
|
assert sensitive["pathname"] == "/api/:redacted/:id" and "fixture-secret" not in json.dumps(sensitive)
|
|
|
|
|
|
def test_failed_capture_keeps_only_bounded_read_discovery_metadata():
|
|
capture = {"url": "https://example.test/", "screenshot_path": "fixture.png", "elements": [{"selector": "#comments", "tag": "div", "kind": "read", "stable_label": '{"name":"search"}', "selector_basis": "scroll_container", "collection": True, "source_selectors": ["#first", "#second", "#third", "#fourth"],
|
|
"scroll_y": {"can_scroll": True, "scroll_top": 0, "scroll_height": 1400, "client_height": 700, "headers": "fixture-secret"},
|
|
"read_hints": [{"selector": ".body", "tag": "div", "label": "x" * 200, "unique": True, "count": 1, "in_viewport": True, "link": False, "source": "attribute", "attribute": "datetime", "body": "fixture-secret"}] * 15, "cookie": "fixture-secret"}]}
|
|
result = WorkerError("fixture", before=capture).before["elements"][0]
|
|
assert result["selector_basis"] == "scroll_container" and result["collection"] is True
|
|
assert result["stable_label"] == '{"name":"search"}'
|
|
assert result["scroll_y"] == {"can_scroll": True, "scroll_top": 0, "scroll_height": 1400, "client_height": 700}
|
|
assert len(result["read_hints"]) == 12 and len(result["read_hints"][0]["label"]) == 100
|
|
assert result["source_selectors"] == ["#first", "#second", "#third"]
|
|
assert result["read_hints"][0]["source"] == "attribute" and result["read_hints"][0]["attribute"] == "datetime"
|
|
assert "fixture-secret" not in json.dumps(result)
|
|
capture["elements"][0].update(selector_basis="arbitrary", collection="true", source_selectors=[None, "#actual"], scroll_y={"scroll_top": float("nan"), "can_scroll": "true"}, read_hints=[None, {"selector": "div", "count": "1", "unique": 1, "link": [], "source": "text", "attribute": "data-token"}])
|
|
result = WorkerError("fixture", before=capture).before["elements"][0]
|
|
assert "selector_basis" not in result and "collection" not in result
|
|
assert result["scroll_y"] == {} and result["read_hints"] == [{"selector": "div", "source": "text"}]
|
|
assert result["source_selectors"] == ["#actual"]
|
|
for invalid in ({"name": "search"}, "x" * 501, None, True):
|
|
capture["elements"][0]["stable_label"] = invalid
|
|
assert "stable_label" not in WorkerError("fixture", before=capture).before["elements"][0]
|
|
|
|
|
|
def test_search_v3_request_metadata_persists_without_widening_old_versions(failure_client):
|
|
client = failure_client
|
|
initialize(client)
|
|
task, _ = working_candidate(client)
|
|
worker = client.worker_reference
|
|
original = worker.capture
|
|
version = "xhs-search-experimental-v3"
|
|
rules = ("collect", "search_history", "search_notes", "search_onebox", "search_dqa", "apm_data", "apm_data_xhr")
|
|
base = {"method": "POST", "origin": "https://apm-fe.xiaohongshu.com", "pathname": "/api/data?token=fixture-secret", "resource_type": "xhr", "automating": True, "blocked": False, "policy_version": version, "policy_rule": "apm_data_xhr"}
|
|
raw = [{**base, "policy_rule": rule} for rule in rules]
|
|
old_versions = ("xhs-search-experimental-v1", "xhs-search-experimental-v2", "xhs-note-open-experimental-v1", "xhs-note-open-experimental-v2", "xhs-readonly-ui-experimental-v1")
|
|
raw.extend({**base, "policy_version": old} for old in old_versions)
|
|
raw.append({**base, "blocked": True})
|
|
|
|
def capture_with_policy(task_id):
|
|
capture = original(task_id)
|
|
capture["request_observations"] = copy.deepcopy(raw)
|
|
return capture
|
|
worker.capture = capture_with_policy
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
result = client.post(prefix + "/browser/capture", json={})
|
|
assert result.status_code == 200, result.text
|
|
records = result.json()["request_observations"]
|
|
assert [item["policy_rule"] for item in records[:7]] == list(rules)
|
|
assert all(item["policy_version"] == version for item in records[:7])
|
|
assert all("policy_rule" not in item and "policy_version" not in item for item in records[7:])
|
|
assert "fixture-secret" not in json.dumps(records)
|
|
stored = client.get(prefix).json()["observations"][-1]
|
|
assert stored["id"] == result.json()["id"] and stored["request_observations"] == records
|
|
assert set(safe_request_summary(base)) == {"method", "origin", "pathname", "resource_type"}
|