220 lines
14 KiB
Python
220 lines
14 KiB
Python
"""Bounded request-denial reasons and new immutable failure readback.
|
|
|
|
No provider or platform is contacted: API checks use a fake worker and either
|
|
temporary SQLite or the opt-in dedicated PostgreSQL test DB's random schema.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
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, TITLE, detail_output, learned_detail
|
|
from apps.server.worker import (
|
|
ACTIVE_REQUEST_POLICIES, REQUEST_POLICY_REASONS, WorkerError,
|
|
safe_diagnostics, safe_failure_diagnostics, safe_request_observations, safe_request_summary,
|
|
)
|
|
|
|
VERSION = "xhs-readonly-ui-experimental-v1"
|
|
REQUEST = {"method": "POST", "origin": "https://apm-fe.xiaohongshu.com", "pathname": "/api/data?query=PRIVATE_FIXTURE#PRIVATE_FIXTURE",
|
|
"resource_type": "xhr", "policy_reason": "resource_type_not_allowed", "active_policy_version": VERSION,
|
|
"expected_resource_types": ["fetch", "fetch"], "headers": {"Cookie": "PRIVATE_FIXTURE"}, "body": "PRIVATE_FIXTURE", "query": "PRIVATE_FIXTURE"}
|
|
DIAGNOSTICS = {"phase": "execute", "blocked_requests": [REQUEST], "blocked_request_count": 1, "stack": "PRIVATE_FIXTURE"}
|
|
|
|
|
|
def failure(position=2, kind="click", code="WRITE_REQUEST_BLOCKED"):
|
|
return {"step_position": position, "kind": kind, "code": code, "diagnostics": copy.deepcopy(DIAGNOSTICS)}
|
|
|
|
|
|
@pytest.fixture(params=["client", "pg_client"])
|
|
def diagnostics_client(request):
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
def test_request_denials_keep_only_closed_reasons_current_versions_and_safe_types():
|
|
expected_base = {"method": "POST", "origin": "https://apm-fe.xiaohongshu.com", "pathname": "/api/data", "resource_type": "xhr"}
|
|
assert safe_request_summary(REQUEST) == expected_base
|
|
actual = safe_request_summary(REQUEST, blocked=True)
|
|
assert actual == expected_base | {"policy_reason": "resource_type_not_allowed", "active_policy_version": VERSION, "expected_resource_types": ["fetch"]}
|
|
for reason in REQUEST_POLICY_REASONS:
|
|
value = safe_request_summary(REQUEST | {"policy_reason": reason}, blocked=True)
|
|
assert value["policy_reason"] == reason
|
|
assert ("expected_resource_types" in value) == (reason == "resource_type_not_allowed")
|
|
assert len(REQUEST_POLICY_REASONS) == 15 and len(ACTIVE_REQUEST_POLICIES) == 4
|
|
for version in ACTIVE_REQUEST_POLICIES:
|
|
assert safe_request_summary(REQUEST | {"active_policy_version": version}, blocked=True)["active_policy_version"] == version
|
|
for version in ("xhs-search-experimental-v1", "xhs-search-experimental-v2", "xhs-note-open-experimental-v1", "PRIVATE_FIXTURE", [], None):
|
|
assert "active_policy_version" not in safe_request_summary(REQUEST | {"active_policy_version": version}, blocked=True)
|
|
for reason in ("PRIVATE_FIXTURE", [], None, 2):
|
|
value = safe_request_summary(REQUEST | {"policy_reason": reason}, blocked=True)
|
|
assert "policy_reason" not in value and "expected_resource_types" not in value
|
|
for types in (None, "fetch", {}, [], [[], {}, 1, True, "PRIVATE_FIXTURE"]):
|
|
assert "expected_resource_types" not in safe_request_summary(REQUEST | {"expected_resource_types": types}, blocked=True)
|
|
value = safe_request_summary(REQUEST | {"expected_resource_types": ["fetch", "fetch", "xhr", "image", "other", "font", "PRIVATE_FIXTURE", {}]}, blocked=True)
|
|
assert value["expected_resource_types"] == ["fetch", "xhr", "image", "other"]
|
|
assert "PRIVATE_FIXTURE" not in json.dumps(value)
|
|
|
|
|
|
def test_observation_denials_require_actual_boolean_block_and_keep_allowed_rule_contract():
|
|
base = REQUEST | {"automating": True, "blocked": True, "sequence": 2, "recorded_at": "2026-10-25T09:00:00.000Z"}
|
|
blocked = safe_request_observations([base])[0]
|
|
assert blocked["policy_reason"] == "resource_type_not_allowed" and blocked["active_policy_version"] == VERSION
|
|
assert blocked["expected_resource_types"] == ["fetch"]
|
|
for block in (False, "true", 1, None):
|
|
value = safe_request_observations([base | {"blocked": block}])[0]
|
|
assert not {"policy_reason", "active_policy_version", "expected_resource_types"} & value.keys()
|
|
allowed = safe_request_observations([base | {"blocked": False, "resource_type": "fetch", "policy_version": VERSION, "policy_rule": "apm_data"}])[0]
|
|
assert allowed["policy_version"] == VERSION and allowed["policy_rule"] == "apm_data"
|
|
assert not {"policy_reason", "active_policy_version", "expected_resource_types"} & allowed.keys()
|
|
# Adding denial diagnostics cannot add an allowed label to a blocked request.
|
|
rejected = safe_request_observations([base | {"policy_version": VERSION, "policy_rule": "apm_data"}])[0]
|
|
assert "policy_version" not in rejected and "policy_rule" not in rejected
|
|
assert "PRIVATE_FIXTURE" not in json.dumps([blocked, allowed, rejected])
|
|
|
|
|
|
def test_diagnostics_bounds_and_repeated_sanitization_never_restore_raw_payload():
|
|
value = safe_diagnostics(DIAGNOSTICS | {"blocked_requests": [REQUEST] * 30, "blocked_request_count": 2_000_000})
|
|
assert len(value["blocked_requests"]) == 12 and value["blocked_request_count"] == 1_000_000
|
|
assert safe_diagnostics(value) == value
|
|
assert value["phase"] == "execute" and "PRIVATE_FIXTURE" not in json.dumps(value)
|
|
assert safe_diagnostics({"phase": [], "blocked_requests": [None, {"method": []}]}) == {"phase": "unknown", "blocked_requests": [], "blocked_request_count": 0}
|
|
e = WorkerError("Not persisted", "WRITE_REQUEST_BLOCKED", diagnostics=DIAGNOSTICS)
|
|
assert e.diagnostics == safe_diagnostics(DIAGNOSTICS)
|
|
|
|
|
|
def test_failure_diagnostics_needs_real_failed_code_and_exact_step_not_new_outcome():
|
|
steps = [{"kind": "fill"}, {"kind": "click"}, {"kind": "read"}]
|
|
codes = ["WRITE_REQUEST_BLOCKED"]
|
|
cleaned = safe_failure_diagnostics(failure(), steps, codes, passed=False)
|
|
assert cleaned == failure() | {"diagnostics": safe_diagnostics(DIAGNOSTICS)}
|
|
assert "PRIVATE_FIXTURE" not in json.dumps(cleaned)
|
|
for passed in (True, None, "false", 0):
|
|
assert safe_failure_diagnostics(failure(), steps, codes, passed=passed) is None
|
|
for changes in ({"step_position": True}, {"step_position": 0}, {"step_position": 4}, {"step_position": "2"},
|
|
{"kind": "read"}, {"kind": []}, {"code": "WAIT_TIMEOUT"}, {"code": []},
|
|
{"diagnostics": None}, {"extra": "PRIVATE_FIXTURE"}):
|
|
assert safe_failure_diagnostics(failure() | changes, steps, codes, passed=False) is None
|
|
assert safe_failure_diagnostics(failure(), steps, [], passed=False) is None
|
|
assert safe_failure_diagnostics(None, steps, codes, passed=False) is None
|
|
|
|
|
|
def test_failed_recording_keeps_denial_reasons_through_both_error_boundaries(diagnostics_client):
|
|
client = diagnostics_client
|
|
initialize(client)
|
|
task, old = working_candidate(client)
|
|
worker, service = client.worker_reference, client.app_reference.state.learning
|
|
prefix = "/api/learning/tasks/" + task["id"]
|
|
original = worker.call
|
|
steps = client.get(prefix).json()["steps"]
|
|
async def failed(method, params):
|
|
if method == "action":
|
|
after = worker.capture(task["id"])
|
|
after["request_observations"] = [REQUEST | {"automating": True, "blocked": True}]
|
|
raise WorkerError("not returned", "WRITE_REQUEST_BLOCKED", before=worker.capture(task["id"]), after=after, diagnostics=DIAGNOSTICS)
|
|
return await original(method, params)
|
|
worker.call = failed
|
|
result = client.post(prefix + "/browser/action", json={"kind": "click", "selector": '[data-testid="search-button"]'})
|
|
assert result.status_code == 409, result.text
|
|
assert result.json()["detail"]["action_failure"]["diagnostics"] == safe_diagnostics(DIAGNOSTICS)
|
|
assert client.get(prefix).json()["steps"] == steps
|
|
observed = client.get(prefix).json()["observations"][-1]
|
|
assert observed["request_observations"][0]["policy_reason"] == "resource_type_not_allowed"
|
|
assert "PRIVATE_FIXTURE" not in result.text and "PRIVATE_FIXTURE" not in json.dumps(observed)
|
|
current = client.get("/api/learning/candidates/" + old["id"]).json()
|
|
assert current["manifest"] == old["manifest"] and current["digest"] == old["digest"]
|
|
|
|
|
|
def test_verification_appends_only_valid_failure_diagnostics_keeps_legacy_and_success(diagnostics_client):
|
|
client = diagnostics_client
|
|
initialize(client)
|
|
task, candidate = working_candidate(client)
|
|
worker, store = client.worker_reference, client.app_reference.state.store
|
|
endpoint = "/api/learning/candidates/" + candidate["id"]
|
|
assert client.post(endpoint + "/check").json()["status"] == "passed"
|
|
values = {"expected_outputs": {"results": "建材"}}
|
|
legacy = client.post(endpoint + "/verify", json=values).json()
|
|
assert legacy["status"] == "passed" and "failure_diagnostics" not in legacy
|
|
legacy_raw = store.one("SELECT data_json FROM validations WHERE id=$1", (legacy["id"],))
|
|
frozen = store.one("SELECT manifest_json,digest FROM candidates WHERE id=$1", (candidate["id"],))
|
|
mode = {"value": "valid"}
|
|
original = worker.call
|
|
async def diagnosed(method, params):
|
|
value = await original(method, params)
|
|
if method == "verify":
|
|
value.update(passed=False, errors=["本机合成的请求阻断"], error_codes=["WRITE_REQUEST_BLOCKED"], failure_diagnostics=failure())
|
|
value["evidence"][0]["request_observations"] = [REQUEST | {"automating": True, "blocked": True}]
|
|
if mode["value"] == "invalid":
|
|
value["failure_diagnostics"]["kind"] = "read"
|
|
elif mode["value"] == "missing":
|
|
value.pop("failure_diagnostics")
|
|
elif mode["value"] == "passed":
|
|
value.update(passed=True, errors=[], error_codes=[])
|
|
return value
|
|
worker.call = diagnosed
|
|
for current in ("valid", "invalid", "missing", "passed"):
|
|
mode["value"] = current
|
|
response = client.post(endpoint + "/verify", json=values)
|
|
assert response.status_code == 200, response.text
|
|
record = response.json()
|
|
assert record["status"] == ("passed" if current == "passed" else "failed")
|
|
assert ("failure_diagnostics" in record) == (current == "valid")
|
|
if current == "valid":
|
|
assert record["failure_diagnostics"] == failure() | {"diagnostics": safe_diagnostics(DIAGNOSTICS)}
|
|
assert "PRIVATE_FIXTURE" not in response.text
|
|
assert any(value["id"] == record["id"] and value == record for value in client.get(endpoint).json()["validations"])
|
|
assert store.one("SELECT data_json FROM validations WHERE id=$1", (legacy["id"],)) == legacy_raw
|
|
assert store.one("SELECT manifest_json,digest FROM candidates WHERE id=$1", (candidate["id"],)) == frozen
|
|
assert next(value for value in client.get(endpoint).json()["validations"] if value["id"] == legacy["id"]) == legacy
|
|
|
|
|
|
def test_reading_copies_new_failure_top_level_and_preserves_old_records(diagnostics_client):
|
|
client = diagnostics_client
|
|
initialize(client)
|
|
_, search = search_candidate(client, [item(TITLE)])
|
|
run = client.post("/api/discovery/runs", json={"search_candidate_id": search["id"], "keyword": "装修"}).json()
|
|
task, candidate, _ = learned_detail(client)
|
|
worker, store = client.worker_reference, client.app_reference.state.store
|
|
endpoint = "/api/discovery/runs/" + run["id"] + "/readings"
|
|
payload = {"detail_candidate_id": candidate["id"], "note_id": NOTE_ID}
|
|
legacy = client.post(endpoint, json=payload).json()
|
|
assert legacy["status"] == "completed" and "failure_diagnostics" not in legacy
|
|
originals = {"reading": store.one("SELECT data_json FROM reading_runs WHERE id=$1", (legacy["id"],)),
|
|
"validation": store.one("SELECT data_json FROM validations WHERE id=$1", (legacy["validation_id"],)),
|
|
"candidate": store.one("SELECT digest,manifest_json FROM candidates WHERE id=$1", (candidate["id"],)),
|
|
"discovery": store.one("SELECT data_json FROM discovery_runs WHERE id=$1", (run["id"],))}
|
|
mode = {"partial": False}
|
|
original = worker.call
|
|
async def diagnosed(method, params):
|
|
value = await original(method, params)
|
|
if method == "verify":
|
|
value.update(passed=False, errors=["本机合成的请求阻断"], error_codes=["WRITE_REQUEST_BLOCKED"])
|
|
if mode["partial"]:
|
|
value["outputs"] = {"note_detail": detail_output()}
|
|
value["failure_diagnostics"] = failure(7, "extract")
|
|
else:
|
|
value["outputs"] = {}
|
|
value["note_binding_result"] = {"note_id": NOTE_ID, "href_checked": False, "url_checked": False, "source_url": None, "title": None}
|
|
value["failure_diagnostics"] = failure(3, "wait")
|
|
return value
|
|
worker.call = diagnosed
|
|
for partial in (False, True):
|
|
mode["partial"] = partial
|
|
response = client.post(endpoint, json=payload)
|
|
assert response.status_code == 201, response.text
|
|
reading = response.json()
|
|
assert reading["status"] == ("partial" if partial else "failed") and reading["comments"] is None
|
|
assert reading["failure_diagnostics"]["kind"] == ("extract" if partial else "wait")
|
|
assert "PRIVATE_FIXTURE" not in response.text
|
|
validation = next(value for value in client.get("/api/learning/candidates/" + candidate["id"]).json()["validations"] if value["id"] == reading["validation_id"])
|
|
assert validation["status"] == "failed" and validation["failure_diagnostics"] == reading["failure_diagnostics"]
|
|
assert client.get("/api/discovery/readings/" + reading["id"]).json() == reading
|
|
assert store.one("SELECT data_json FROM reading_runs WHERE id=$1", (legacy["id"],)) == originals["reading"]
|
|
assert store.one("SELECT data_json FROM validations WHERE id=$1", (legacy["validation_id"],)) == originals["validation"]
|
|
assert store.one("SELECT digest,manifest_json FROM candidates WHERE id=$1", (candidate["id"],)) == originals["candidate"]
|
|
assert store.one("SELECT data_json FROM discovery_runs WHERE id=$1", (run["id"],)) == originals["discovery"]
|
|
assert client.get("/api/discovery/readings/" + legacy["id"]).json() == legacy
|