"""Explicit navigation profile metadata; isolated stores and fake worker only.""" import copy import json import pytest from apps.server.learning_contracts import ( EXECUTION_PROFILE, NAVIGATION_PROFILE, NAVIGATION_CONTRACT, execution_metadata, execution_metadata_matches, ) from apps.server.tests.test_execution_profile import install_profile_worker 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 ( WorkerError, safe_diagnostics, safe_failure_capture, safe_failure_diagnostics, safe_navigation_comparison, safe_request_observations, ) COMPARISON = { "expected_kind": "profile", "actual_kind": "profile", "same_origin": True, "same_pathname": True, "same_object_id": True, "same_query": False, "same_query_pairs": True, "same_hash": True, "other_query_equal": True, "xsec_token": {"expected_count": 1, "actual_count": 1, "values_equal": True}, "xsec_source": {"expected_count": 1, "actual_count": 0, "values_equal": False}, } def navigation_diagnostics(comparison=None): return {"phase": "execute", "blocked_requests": [], "blocked_request_count": 0, "navigation_comparison": copy.deepcopy(COMPARISON if comparison is None else comparison)} def navigation_failure(comparison=None, code="NAVIGATION_BINDING_MISMATCH"): return {"step_position": 2, "kind": "click", "code": code, "diagnostics": navigation_diagnostics(comparison)} @pytest.fixture(params=["client", "pg_client"]) def nav_client(request): return request.getfixturevalue(request.param) def test_navigation_descriptor_cannot_be_mixed_with_existing_contract(): expected = {"execution_profile": NAVIGATION_PROFILE, "execution_contract": NAVIGATION_CONTRACT} assert execution_metadata(NAVIGATION_PROFILE) == expected assert execution_metadata_matches(expected, NAVIGATION_PROFILE) assert not execution_metadata_matches(expected, EXECUTION_PROFILE) assert not execution_metadata_matches(execution_metadata(EXECUTION_PROFILE), NAVIGATION_PROFILE) for contract in ({}, NAVIGATION_CONTRACT | {"extra": True}, NAVIGATION_CONTRACT | {"network_policy": "xhs-page-native-v1"}): assert not execution_metadata_matches(expected | {"execution_contract": contract}, NAVIGATION_PROFILE) captured = {"url": "https://www.xiaohongshu.com/notification", "elements": [], "screenshot_path": "synthetic.png", **expected} assert execution_metadata_matches(safe_failure_capture(captured), NAVIGATION_PROFILE) assert safe_failure_capture(captured | {"execution_contract": {}}) is None summary = safe_request_observations([{"method": "POST", "origin": "https://www.xiaohongshu.com", "pathname": "/api/telemetry", "resource_type": "xhr", "automating": True, "blocked": False, "policy_version": "xhs-page-navigation-v1", "policy_rule": "native_page_traffic"}]) assert summary[0]["policy_rule"] == "native_page_traffic" def test_navigation_profile_is_explicit_frozen_and_checked_at_actual_worker_echo(nav_client): initialize(nav_client) state, _ = install_profile_worker(nav_client) previous = nav_client.post def post(url, **kwargs): if str(url) == "/api/learning/tasks" and "json" in kwargs: kwargs["json"] = {**kwargs["json"], "execution_profile": NAVIGATION_PROFILE} return previous(url, **kwargs) nav_client.post = post task, candidate = working_candidate(nav_client) metadata = execution_metadata(NAVIGATION_PROFILE) assert task["execution_profile"] == NAVIGATION_PROFILE assert execution_metadata_matches(candidate["manifest"], NAVIGATION_PROFILE) endpoint = "/api/learning/candidates/" + candidate["id"] assert nav_client.post(endpoint + "/check").json()["status"] == "passed" result = nav_client.post(endpoint + "/verify", json={"expected_outputs": {"results": "建材"}}).json() assert result["status"] == "passed" and execution_metadata_matches(result["environment"], NAVIGATION_PROFILE) wire = [p for m, p in nav_client.worker_reference.calls if m == "verify"][-1] assert all(wire[key] == value for key, value in metadata.items()) facts = copy.deepcopy(nav_client.app_reference.state.learning.steps(task["id"])) state.update(method="capture", place="root", mutation=lambda value: value.update(execution_metadata(EXECUTION_PROFILE))) rejected = nav_client.post("/api/learning/tasks/" + task["id"] + "/browser/capture", json={}) assert rejected.status_code == 409 assert rejected.json()["detail"]["code"] == "EXECUTION_PROFILE_MISMATCH" assert nav_client.app_reference.state.learning.steps(task["id"]) == facts def test_navigation_comparison_only_keeps_eleven_fixed_relationship_fields(): original = copy.deepcopy(COMPARISON) injected = copy.deepcopy(COMPARISON) injected.update(expected_url="PRIVATE_URL_SENTINEL", actual_id="PRIVATE_ID_SENTINEL", query_keys=["PRIVATE_QUERY_SENTINEL"]) injected["xsec_token"].update(expected_value="PRIVATE_TOKEN_SENTINEL", actual_value="PRIVATE_TOKEN_SENTINEL") injected["xsec_source"].update(url="PRIVATE_SOURCE_SENTINEL") cleaned = safe_navigation_comparison(injected) assert cleaned == original and len(cleaned) == 11 assert all(set(cleaned[key]) == {"expected_count", "actual_count", "values_equal"} for key in ("xsec_token", "xsec_source")) assert "PRIVATE_" not in json.dumps(cleaned) and COMPARISON == original cleaned["xsec_token"]["actual_count"] = 2 assert injected["xsec_token"]["actual_count"] == original["xsec_token"]["actual_count"] == 1 for kind in ("note", "profile", "notification", "explore", "search", "invalid"): assert safe_navigation_comparison(COMPARISON | {"expected_kind": kind, "actual_kind": kind}) is not None boundary = copy.deepcopy(COMPARISON) boundary["xsec_token"].update(expected_count=0, actual_count=1000) assert safe_navigation_comparison(boundary) == boundary def test_navigation_comparison_rejects_missing_or_mistyped_fields_as_a_whole(): assert safe_navigation_comparison(None) is None assert safe_navigation_comparison([]) is None for field in COMPARISON: value = copy.deepcopy(COMPARISON) value.pop(field) assert safe_navigation_comparison(value) is None for field in ("expected_kind", "actual_kind"): for invalid in (None, [], {}, True, "private_unknown_kind"): assert safe_navigation_comparison(COMPARISON | {field: invalid}) is None for field in ("same_origin", "same_pathname", "same_object_id", "same_query", "same_query_pairs", "same_hash", "other_query_equal"): for invalid in (0, 1, "true", "false", None, [], {}): assert safe_navigation_comparison(COMPARISON | {field: invalid}) is None for field in ("xsec_token", "xsec_source"): for invalid in (None, [], 1): assert safe_navigation_comparison(COMPARISON | {field: invalid}) is None for key in ("expected_count", "actual_count", "values_equal"): value = copy.deepcopy(COMPARISON) value[field].pop(key) assert safe_navigation_comparison(value) is None for key in ("expected_count", "actual_count"): for invalid in (-1, 1001, True, 1.0, "1", None, []): value = copy.deepcopy(COMPARISON) value[field][key] = invalid assert safe_navigation_comparison(value) is None for invalid in (0, 1, "true", None): value = copy.deepcopy(COMPARISON) value[field]["values_equal"] = invalid assert safe_navigation_comparison(value) is None def test_navigation_comparison_requires_failed_bound_code_and_new_profile(): raw = navigation_diagnostics() cleaned = safe_diagnostics(raw, code="NAVIGATION_BINDING_MISMATCH", execution_profile=NAVIGATION_PROFILE) assert cleaned["navigation_comparison"] == COMPARISON for code, profile in ((None, NAVIGATION_PROFILE), ("SOURCE_CHANGED", NAVIGATION_PROFILE), ("NAVIGATION_BINDING_MISMATCH", None), ("NAVIGATION_BINDING_MISMATCH", EXECUTION_PROFILE)): assert "navigation_comparison" not in safe_diagnostics(raw, code=code, execution_profile=profile) steps, codes = [{"kind": "fill"}, {"kind": "click"}], ["NAVIGATION_BINDING_MISMATCH"] assert safe_failure_diagnostics(navigation_failure(), steps, codes, passed=False, execution_profile=NAVIGATION_PROFILE)["diagnostics"] == cleaned for passed in (True, None, 0, "false"): assert safe_failure_diagnostics(navigation_failure(), steps, codes, passed=passed, execution_profile=NAVIGATION_PROFILE) is None assert safe_failure_diagnostics(navigation_failure(), steps, [], passed=False, execution_profile=NAVIGATION_PROFILE) is None assert "navigation_comparison" not in safe_failure_diagnostics(navigation_failure(), steps, codes, passed=False)["diagnostics"] capture = {"url": "http://127.0.0.1/fixture", "elements": [], "screenshot_path": "synthetic.png", **execution_metadata(NAVIGATION_PROFILE)} error = WorkerError("not persisted", "NAVIGATION_BINDING_MISMATCH", before=capture, diagnostics=raw) assert error.diagnostics == cleaned for before, after in ((None, None), (capture | execution_metadata(EXECUTION_PROFILE), None), (capture, capture | execution_metadata(EXECUTION_PROFILE))): rejected = WorkerError("not persisted", "NAVIGATION_BINDING_MISMATCH", before=before, after=after, diagnostics=raw) assert "navigation_comparison" not in rejected.diagnostics @pytest.mark.parametrize("profile", [NAVIGATION_PROFILE, EXECUTION_PROFILE, None]) def test_navigation_failure_diagnostics_record_and_verify_are_append_only_and_profile_bound(nav_client, profile): initialize(nav_client) install_profile_worker(nav_client) previous_post = nav_client.post def post(url, **kwargs): if str(url) == "/api/learning/tasks" and "json" in kwargs: kwargs["json"] = {**kwargs["json"], "execution_profile": profile} return previous_post(url, **kwargs) nav_client.post = post task, candidate = working_candidate(nav_client) worker, service = nav_client.worker_reference, nav_client.app_reference.state.learning store, task_prefix = service.store, "/api/learning/tasks/" + task["id"] endpoint = "/api/learning/candidates/" + candidate["id"] assert nav_client.post(endpoint + "/check").json()["status"] == "passed" inputs = {"expected_outputs": {"results": "建材"}} old = nav_client.post(endpoint + "/verify", json=inputs).json() assert old["status"] == "passed" and "failure_diagnostics" not in old old_raw = store.one("SELECT data_json FROM validations WHERE id=$1", (old["id"],)) original_candidate = store.one("SELECT manifest_json,digest FROM candidates WHERE id=$1", (candidate["id"],)) original_steps = copy.deepcopy(service.steps(task["id"])) mode = {"value": "valid"} original_call = worker.call async def diagnosed(method, params): code = "SOURCE_CHANGED" if mode["value"] == "wrong_code" else "NAVIGATION_BINDING_MISMATCH" comparison = copy.deepcopy(COMPARISON) comparison["expected_url"] = "PRIVATE_URL_SENTINEL" comparison["xsec_token"]["expected_value"] = "PRIVATE_TOKEN_SENTINEL" if mode["value"] == "invalid": comparison["same_hash"] = "true" if method == "action" and mode["value"] != "passed": raise WorkerError("not persisted", code, before=worker.capture(task["id"]), after=worker.capture(task["id"]), diagnostics=navigation_diagnostics(comparison)) result = await original_call(method, params) if method == "verify": result.update(passed=mode["value"] == "passed", errors=[] if mode["value"] == "passed" else ["合成导航来源不一致"], error_codes=[] if mode["value"] == "passed" else [code], failure_diagnostics=navigation_failure(comparison, code)) return result worker.call = diagnosed for value in ("valid", "wrong_code", "invalid"): mode["value"] = value response = nav_client.post(task_prefix + "/browser/action", json={"kind": "click", "selector": '[data-testid="search-button"]'}) assert response.status_code == 409, response.text expected = profile == NAVIGATION_PROFILE and value == "valid" diagnostics = response.json()["detail"]["action_failure"]["diagnostics"] assert ("navigation_comparison" in diagnostics) is expected if expected: assert diagnostics["navigation_comparison"] == COMPARISON observations = nav_client.get(task_prefix).json()["observations"] failure_observations = [item for item in observations if item.get("action_failure", {}).get("attempt_id") == response.json()["detail"]["action_failure"]["attempt_id"]] assert len(failure_observations) == 2 assert all(item["action_failure"]["diagnostics"] == diagnostics for item in failure_observations) assert service.steps(task["id"]) == original_steps and "PRIVATE_" not in response.text recorded = nav_client.post(endpoint + "/verify", json=inputs) assert recorded.status_code == 200, recorded.text record = recorded.json() assert record["status"] == "failed" assert ("navigation_comparison" in record["failure_diagnostics"]["diagnostics"]) is expected if expected: assert record["failure_diagnostics"]["diagnostics"]["navigation_comparison"] == COMPARISON assert "PRIVATE_" not in recorded.text assert next(item for item in nav_client.get(endpoint).json()["validations"] if item["id"] == record["id"]) == record mode["value"] = "passed" passed = nav_client.post(endpoint + "/verify", json=inputs).json() assert passed["status"] == "passed" and "failure_diagnostics" not in passed assert store.one("SELECT data_json FROM validations WHERE id=$1", (old["id"],)) == old_raw assert store.one("SELECT manifest_json,digest FROM candidates WHERE id=$1", (candidate["id"],)) == original_candidate assert service.steps(task["id"]) == original_steps assert next(item for item in nav_client.get(endpoint).json()["validations"] if item["id"] == old["id"]) == old