import json import subprocess import sys from datetime import datetime, timezone from tests.test_pixelrag_rag_candidate_replay_service import _write_receipt def _patch_receipt(path, **updates): payload = json.loads(path.read_text(encoding="utf-8")) for key, value in updates.items(): if key == "page_metrics": payload.setdefault("page_metrics", {}).update(value) else: payload[key] = value path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") return payload def test_pixelrag_platform_probe_builds_machine_actions_for_shopee_and_coupang(tmp_path): from services.pixelrag_platform_probe_service import ( POLICY, build_pixelrag_platform_probe_readiness, ) shopee_receipt = _write_receipt( tmp_path, platform="shopee_tw", manifest_id="shopee-traffic", title="蝦皮購物 | 花得更少買得更好", url="https://shopee.tw/search?keyword=sunscreen", ) _patch_receipt( shopee_receipt, page_metrics={ "final_url": "https://shopee.tw/verify/traffic/error?home_url=https%3A%2F%2Fshopee.tw", "title": "蝦皮購物 | 花得更少買得更好", }, ) _write_receipt( tmp_path, platform="coupang_tw", manifest_id="coupang-403", title="Access Denied", url="https://www.tw.coupang.com/search?q=iphone", http_status=403, ) payload = build_pixelrag_platform_probe_readiness(artifact_root=tmp_path) by_platform = {item["platform"]: item for item in payload["probe_items"]} assert payload["policy"] == POLICY assert payload["status"] == "ok" assert payload["summary"]["probe_candidate_count"] == 2 assert payload["summary"]["ready_for_probe_count"] == 2 assert payload["summary"]["traffic_verification_count"] == 1 assert payload["summary"]["access_denied_count"] == 1 assert payload["summary"]["writes_database_count"] == 0 assert payload["controlled_apply"]["primary_human_gate_count"] == 0 assert by_platform["shopee_tw"]["probe_status"] == "ready_for_public_context_probe" assert by_platform["shopee_tw"]["next_machine_action"] == ( "run_shopee_public_context_probe_then_structured_source_fallback" ) assert by_platform["shopee_tw"]["public_browser_context"]["locale"] == "zh-TW" assert by_platform["shopee_tw"]["public_browser_context"]["storage_state_allowed"] is False assert by_platform["shopee_tw"]["capture_manifest_preview"]["success"] is True assert by_platform["coupang_tw"]["probe_status"] == "structured_source_or_backoff_required" assert by_platform["coupang_tw"]["next_machine_action"] == ( "use_structured_source_or_platform_backoff_policy" ) assert by_platform["coupang_tw"]["structured_source_fallback"]["available"] is True assert payload["probe_contract"]["raw_cookie_or_session_read"] is False assert payload["probe_contract"]["blocked_pages_are_not_product_data"] is True def test_pixelrag_platform_probe_reads_vlm_interstitial_receipts(tmp_path): from services.pixelrag_platform_probe_service import build_pixelrag_platform_probe_readiness capture_path = _write_receipt( tmp_path / "capture", platform="shopee_tw", manifest_id="shopee-language", title="Shopee", url="https://shopee.tw/search?keyword=sunscreen", ) vlm_dir = tmp_path / "vlm" / "shopee_tw" / "shopee-language" vlm_dir.mkdir(parents=True) (vlm_dir / "vlm_replay_receipt.json").write_text( json.dumps({ "generated_at": datetime.now(timezone.utc).isoformat(), "platform": "shopee_tw", "manifest_id": "shopee-language", "source_receipt_path": str(capture_path), "worker_status": "executed_warning", "next_machine_action": "run_platform_probe_or_use_structured_api", "parsed_output": { "blocked_page_detected": False, "fields": { "title": { "value": "蝦皮購物 | 花得更少買得更好", "confidence": 0.95, "evidence_refs": ["tile:1"], }, "price": {"value": None, "confidence": 0.0, "evidence_refs": []}, }, "notes": [{"note": "Language selection page"}], }, "validation": { "blocked_page_detected": False, "non_product_or_interstitial_detected": True, "interstitial_signal_detected": True, "generic_marketplace_title_detected": True, "present_field_count": 1, "missing_required_fields": ["price"], }, }, ensure_ascii=False), encoding="utf-8", ) payload = build_pixelrag_platform_probe_readiness( artifact_root=tmp_path / "capture", vlm_receipt_root=tmp_path / "vlm", platform="shopee_tw", ) assert payload["status"] == "ok" assert payload["summary"]["probe_candidate_count"] == 1 assert payload["summary"]["vlm_source_count"] == 1 assert payload["summary"]["language_or_region_interstitial_count"] == 1 item = payload["probe_items"][0] assert item["source_type"] == "vlm_replay_receipt" assert item["barrier_type"] == "language_or_region_interstitial" assert item["probe_status"] == "ready_for_public_context_probe" assert item["primary_human_gate_count"] == 0 def test_pixelrag_platform_probe_cli_outputs_machine_readable_json(tmp_path): _write_receipt( tmp_path, platform="coupang_tw", manifest_id="coupang-403", title="Access Denied", url="https://www.tw.coupang.com/search?q=iphone", http_status=403, ) completed = subprocess.run( [ sys.executable, "scripts/ops/report_pixelrag_platform_probe.py", "--artifact-root", str(tmp_path), "--platform", "coupang_tw", ], capture_output=True, check=False, text=True, ) assert completed.returncode == 0 payload = json.loads(completed.stdout) assert payload["summary"]["probe_candidate_count"] == 1 assert payload["probe_items"][0]["platform"] == "coupang_tw" assert payload["probe_contract"]["writes_database"] is False def test_pixelrag_capture_worker_dry_run_uses_safe_public_context(tmp_path): from services.pixelrag_crawler_integration_service import ( build_pixelrag_marketplace_search_manifest, ) manifest = build_pixelrag_marketplace_search_manifest( platform="shopee_tw", keyword="sunscreen", crawler="PixelRAGPlatformProbe.public_context_visual_fallback", trigger_reason="platform_interstitial_or_blocked_page_probe", ) manifest["public_browser_context"] = { "locale": "zh-TW", "timezone_id": "Asia/Taipei", "extra_http_headers": { "Accept-Language": "zh-TW,zh-Hant;q=0.95,zh;q=0.9,en;q=0.7", "Cookie": "must_not_survive", "Authorization": "must_not_survive", }, } completed = subprocess.run( [ sys.executable, "scripts/ops/capture_pixelrag_visual_evidence.py", "--manifest-json", json.dumps(manifest, ensure_ascii=False), "--output-dir", str(tmp_path / "artifacts"), "--dry-run", ], capture_output=True, check=False, text=True, ) assert completed.returncode == 0 payload = json.loads(completed.stdout) context = payload["public_browser_context"] assert context["locale"] == "zh-TW" assert context["timezone_id"] == "Asia/Taipei" assert context["extra_http_headers"]["Accept-Language"].startswith("zh-TW") assert "Cookie" not in context["extra_http_headers"] assert "Authorization" not in context["extra_http_headers"] assert context["raw_cookie_or_session_read_allowed"] is False def test_pixelrag_platform_probe_route_returns_readback(tmp_path, monkeypatch): from flask import Flask from routes import system_public_routes as routes from services import pixelrag_platform_probe_service as service _write_receipt( tmp_path, platform="coupang_tw", manifest_id="coupang-403", title="Access Denied", url="https://www.tw.coupang.com/search?q=iphone", http_status=403, ) monkeypatch.setattr(service, "DEFAULT_ARTIFACT_ROOT", str(tmp_path)) app = Flask(__name__) with app.test_request_context( "/api/ai-automation/pixelrag-platform-probe?platform=coupang_tw" ): response = routes.ai_automation_pixelrag_platform_probe_api.__wrapped__() payload = response.get_json() assert payload["policy"] == "read_only_pixelrag_platform_probe_readiness_v1" assert payload["summary"]["probe_candidate_count"] == 1 assert payload["probe_items"][0]["next_machine_action"] == ( "use_structured_source_or_platform_backoff_policy" )