ewoooc/tests/test_pixelrag_platform_probe_service.py

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"
    )