import json import shutil import subprocess import sys from datetime import datetime, timezone import pytest def test_pixelrag_assessment_starts_visual_evidence_fallback_not_price_write(): from services.pixelrag_crawler_integration_service import ( RESULT_PHASE1_READY, build_pixelrag_crawler_integration_assessment, ) payload = build_pixelrag_crawler_integration_assessment() assert payload["success"] is True assert payload["result"] == RESULT_PHASE1_READY assert payload["feasible"] is True assert payload["can_start_now"] is True assert payload["recommended_mode"] == "pixelrag_inspired_visual_evidence_fallback" assert payload["full_pixelrag_ready"] is False assert payload["controlled_apply"] == { "network_call": False, "db_write": False, "secret_read": False, "github_dependency": False, "production_price_write": False, "rollback": "Disable the visual fallback selector or ignore generated artifacts.", } phase_ids = {phase["id"]: phase for phase in payload["phases"]} assert phase_ids["PXR-1"]["status"] == "ready_to_start" assert phase_ids["PXR-2"]["writes"] == ["artifact_file_only"] assert phase_ids["PXR-3"]["status"] == "deferred_model_not_verified" assert phase_ids["PXR-5"]["writes"] == ["review_diagnostics_only"] assert any( guard["code"] == "no_price_write_from_pixels" for guard in payload["safety_guards"] ) def test_pixelrag_assessment_blocks_when_visual_capture_is_not_ready(): from services.pixelrag_crawler_integration_service import ( RESULT_BLOCKED_NO_CAPTURE, build_pixelrag_crawler_integration_assessment, ) payload = build_pixelrag_crawler_integration_assessment( capabilities={"playwright_artifact_pipeline": False} ) assert payload["result"] == RESULT_BLOCKED_NO_CAPTURE assert payload["feasible"] is False assert payload["can_start_now"] is False assert payload["recommended_mode"] == "prepare_visual_capture_prerequisites" assert payload["phases"][0]["status"] == "blocked_prerequisite" def test_pixelrag_assessment_keeps_github_and_hosted_embedding_out_of_runtime(): from services.pixelrag_crawler_integration_service import ( build_pixelrag_crawler_integration_assessment, ) payload = build_pixelrag_crawler_integration_assessment( capabilities={ "ollama_multimodal_embedding_ready": True, "pgvector_visual_index_ready": True, } ) findings = {finding["code"]: finding for finding in payload["source_findings"]} guards = {guard["code"]: guard for guard in payload["safety_guards"]} assert payload["full_pixelrag_ready"] is True assert "Ollama" in findings["qwen3_vl_embedding_optional"]["adoption"] assert guards["no_external_embedding_api"]["status"] == "enforced" assert guards["no_github_runtime_dependency"]["status"] == "enforced" assert payload["controlled_apply"]["github_dependency"] is False def test_pixelrag_report_cli_outputs_machine_readable_json(): completed = subprocess.run( [ sys.executable, "scripts/ops/report_pixelrag_crawler_integration.py", "--platform", "momo", ], capture_output=True, check=False, text=True, ) assert completed.returncode == 0 payload = json.loads(completed.stdout) assert payload["success"] is True assert payload["target_platforms"] == ["momo"] assert payload["next_actions"][0]["action"] == "add_visual_evidence_manifest_lane" def test_visual_fallback_selector_routes_parser_empty_and_low_confidence_cases(): from services.pixelrag_crawler_integration_service import ( should_emit_visual_evidence_fallback, ) parser_empty = should_emit_visual_evidence_fallback( parser_success=False, parsed_item_count=0, missing_fields=["price"], failure_reason="HTML selector did not find rendered price block", ) clean_parser = should_emit_visual_evidence_fallback( parser_success=True, parsed_item_count=8, confidence_band="high", ) assert parser_empty["should_emit"] is True assert "parser_failed" in parser_empty["triggers"] assert "critical_field_missing" in parser_empty["triggers"] assert clean_parser["should_emit"] is False assert clean_parser["fallback_reason"] == "structured_evidence_sufficient" def test_pixelrag_ecommerce_expansion_plan_includes_shopee_and_coupang(): from services.pixelrag_crawler_integration_service import ( ECOMMERCE_EXPANSION_POLICY, build_pixelrag_ecommerce_expansion_plan, ) payload = build_pixelrag_ecommerce_expansion_plan() platforms = {item["platform"]: item for item in payload["platforms"]} assert payload["policy"] == ECOMMERCE_EXPANSION_POLICY assert payload["status"] == "marketplace_profiles_ready" assert platforms["shopee_tw"]["manifest_ready"] is True assert platforms["coupang_tw"]["manifest_ready"] is True assert "price" in platforms["shopee_tw"]["data_targets"] assert payload["next_actions"][0]["target_platforms"] == ["shopee_tw", "coupang_tw"] assert payload["controlled_apply"]["production_price_write"] is False def test_marketplace_search_manifest_builds_shopee_visual_capture_target(): from services.pixelrag_crawler_integration_service import build_pixelrag_marketplace_search_manifest payload = build_pixelrag_marketplace_search_manifest( platform="shopee_tw", keyword="防曬乳", page_size={"width": 1440, "height": 1024}, tile_size={"width": 512, "height": 512}, ) assert payload["success"] is True assert payload["capture_target"]["platform"] == "shopee_tw" assert payload["capture_target"]["url"].startswith("https://shopee.tw/search?keyword=") assert "%E9%98%B2%E6%9B%AC%E4%B9%B3" in payload["capture_target"]["url"] assert payload["marketplace_profile"]["display_name"] == "Shopee Taiwan" assert payload["tile_plan"]["tile_count"] == 6 def test_crawler_failure_action_persists_manifest_queue(tmp_path): from services.pixelrag_crawler_integration_service import build_pixelrag_crawler_failure_action payload = build_pixelrag_crawler_failure_action( url="https://24h.pchome.com.tw/search/?q=test", platform="pchome", crawler="PChomeCrawler.search_products", parser_success=False, parsed_item_count=0, missing_fields=["price", "product_id"], failure_reason="search API returned no product IDs", persist_manifest=True, queue_root=tmp_path, ) assert payload["success"] is True assert payload["status"] == "manifest_persisted" assert payload["next_machine_action"] == "run_pixelrag_visual_capture_worker" assert payload["controlled_apply"]["artifact_write"] is True manifest_path = payload["persistence"]["path"] queued = json.loads(open(manifest_path, encoding="utf-8").read()) assert queued["capture_target"]["platform"] == "pchome" assert queued["queue_policy"] == "read_only_pixelrag_crawler_failure_action_v1" def test_visual_evidence_manifest_builds_tile_plan_without_runtime_writes(): from services.pixelrag_crawler_integration_service import ( RESULT_MANIFEST_READY, build_pixelrag_visual_evidence_manifest, ) payload = build_pixelrag_visual_evidence_manifest( url="https://m.momoshop.com.tw/search.momo?searchKeyword=test", platform="momo", crawler="MomoCrawler.search_products", trigger_reason="parser_empty", viewport={"name": "desktop-1440", "width": 1440, "height": 950}, page_size={"width": 1440, "height": 1900}, tile_size={"width": 512, "height": 512}, ) assert payload["success"] is True assert payload["result"] == RESULT_MANIFEST_READY assert payload["status"] == "capture_requested" assert payload["tile_plan"] == { "tiles_x": 3, "tiles_y": 4, "tile_count": 12, "overlap_px": 0, } assert payload["artifact"]["writes"] == ["artifact_file_only"] assert payload["artifact"]["suggested_path"].startswith( "runtime_artifacts/pixelrag_visual_evidence/momo/" ) assert payload["controlled_apply"]["db_write"] is False assert payload["controlled_apply"]["network_call"] is False assert payload["controlled_apply"]["production_price_write"] is False def test_visual_evidence_manifest_rejects_urls_with_credentials(): from services.pixelrag_crawler_integration_service import ( RESULT_MANIFEST_REJECTED, build_pixelrag_visual_evidence_manifest, ) payload = build_pixelrag_visual_evidence_manifest( url="https://user:pass@example.test/product", platform="momo", crawler="MomoCrawler", trigger_reason="parser_empty", ) assert payload["success"] is False assert payload["result"] == RESULT_MANIFEST_REJECTED assert "URL credentials are not allowed" in payload["errors"][0] def test_pixelrag_report_cli_outputs_visual_manifest(): completed = subprocess.run( [ sys.executable, "scripts/ops/report_pixelrag_crawler_integration.py", "--platform", "pchome", "--manifest-url", "https://24h.pchome.com.tw/prod/TEST-000000001", "--crawler", "PChomeCrawler.search_products", "--trigger-reason", "parser_empty", "--page-size", "page=1024x1024", "--tile-size", "tile=512x512", ], capture_output=True, check=False, text=True, ) assert completed.returncode == 0 payload = json.loads(completed.stdout) assert payload["result"] == "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_READY" assert payload["capture_target"]["platform"] == "pchome" assert payload["tile_plan"]["tile_count"] == 4 def test_pixelrag_ai_automation_route_returns_assessment_and_manifest(): from flask import Flask from routes import system_public_routes as routes app = Flask(__name__) with app.test_request_context( "/api/ai-automation/pixelrag-crawler-integration?platform=momo" ): assessment_response = routes.ai_automation_pixelrag_crawler_integration_api.__wrapped__() assessment = assessment_response.get_json() with app.test_request_context( "/api/ai-automation/pixelrag-crawler-integration" "?platform=pchome" "&manifest_url=https://24h.pchome.com.tw/prod/TEST-000000001" "&crawler=PChomeCrawler.search_products" "&trigger_reason=parser_empty" ): manifest_response = routes.ai_automation_pixelrag_crawler_integration_api.__wrapped__() manifest = manifest_response.get_json() with app.test_request_context( "/api/ai-automation/pixelrag-crawler-integration?expansion_plan=true" ): expansion_response = routes.ai_automation_pixelrag_crawler_integration_api.__wrapped__() expansion = expansion_response.get_json() with app.test_request_context( "/api/ai-automation/pixelrag-crawler-integration" "?platform=shopee_tw&marketplace_keyword=防曬乳" ): marketplace_response = routes.ai_automation_pixelrag_crawler_integration_api.__wrapped__() marketplace = marketplace_response.get_json() assert assessment["policy"] == "read_only_pixelrag_crawler_integration_assessment_v1" assert assessment["target_platforms"] == ["momo"] assert assessment["controlled_apply"]["db_write"] is False assert manifest["policy"] == "read_only_pixelrag_visual_evidence_manifest_v1" assert manifest["result"] == "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_READY" assert manifest["capture_target"]["platform"] == "pchome" assert expansion["policy"] == "read_only_pixelrag_ecommerce_expansion_plan_v1" assert expansion["next_actions"][0]["target_platforms"] == ["shopee_tw", "coupang_tw"] assert marketplace["capture_target"]["platform"] == "shopee_tw" assert marketplace["marketplace_profile"]["keyword"] == "防曬乳" def _write_capture_receipt(root, *, platform="pchome", manifest_id="manifest-1"): target_dir = root / platform / manifest_id tiles_dir = target_dir / "tiles" tiles_dir.mkdir(parents=True) screenshot = target_dir / "fullpage.png" tile_0 = tiles_dir / "tile_000.png" tile_1 = tiles_dir / "tile_001.png" screenshot.write_bytes(b"png") tile_0.write_bytes(b"png") tile_1.write_bytes(b"png") receipt = { "success": True, "policy": "read_only_pixelrag_visual_capture_artifact_v1", "status": "captured", "generated_at": datetime.now(timezone.utc).isoformat(), "manifest_id": manifest_id, "capture_target": { "platform": platform, "crawler": "PChomeCrawler.visual_fallback", "trigger_reason": "parser_empty", "url": "https://24h.pchome.com.tw/", }, "http_status": 200, "page_metrics": { "final_url": "https://24h.pchome.com.tw/", "title": "PChome 24h購物", }, "tile_plan": { "planned_tile_count": 6, "emitted_tile_count": 2, }, "files": [ {"kind": "fullpage_screenshot", "path": str(screenshot)}, {"kind": "tile", "path": str(tile_0)}, {"kind": "tile", "path": str(tile_1)}, ], } (target_dir / "capture_receipt.json").write_text( json.dumps(receipt, ensure_ascii=False), encoding="utf-8", ) return target_dir / "capture_receipt.json" def test_pixelrag_visual_evidence_readback_reports_latest_receipt(tmp_path): from services.pixelrag_crawler_integration_service import ( READBACK_POLICY, build_pixelrag_visual_evidence_readback, ) receipt_path = _write_capture_receipt(tmp_path) payload = build_pixelrag_visual_evidence_readback( artifact_root=tmp_path, platform="pchome", manifest_id="manifest-1", ) assert payload["policy"] == READBACK_POLICY assert payload["status"] == "ok" assert payload["summary"]["receipt_count"] == 1 assert payload["summary"]["tile_file_count"] == 2 assert payload["summary"]["missing_file_count"] == 0 assert payload["summary"]["http_status"] == 200 assert payload["latest_receipt"]["path"] == str(receipt_path) assert payload["next_machine_action"] == "keep_pixelrag_visual_evidence_receipt" assert payload["controlled_apply"]["db_write"] is False def test_pixelrag_visual_evidence_readback_warns_on_access_denied_page(tmp_path): from services.pixelrag_crawler_integration_service import build_pixelrag_visual_evidence_readback receipt_path = _write_capture_receipt(tmp_path, platform="coupang_tw", manifest_id="denied") receipt = json.loads(receipt_path.read_text(encoding="utf-8")) receipt["http_status"] = 403 receipt["page_metrics"] = { "final_url": "https://www.tw.coupang.com/search?q=iphone", "title": "Access Denied", } receipt_path.write_text(json.dumps(receipt), encoding="utf-8") payload = build_pixelrag_visual_evidence_readback( artifact_root=tmp_path, platform="coupang_tw", manifest_id="denied", ) assert payload["status"] == "warning" assert payload["summary"]["visual_barrier_detected"] is True assert payload["summary"]["visual_barrier_reason"] == "http_status_403" assert payload["next_machine_action"] == "run_platform_probe_or_use_structured_api" def test_pixelrag_visual_evidence_readback_warns_when_missing(tmp_path): from services.pixelrag_crawler_integration_service import build_pixelrag_visual_evidence_readback payload = build_pixelrag_visual_evidence_readback(artifact_root=tmp_path, platform="momo") assert payload["success"] is False assert payload["status"] == "warning" assert payload["summary"]["receipt_count"] == 0 assert payload["next_machine_action"] == "run_pixelrag_visual_capture_worker" def test_pixelrag_visual_evidence_readback_relocates_copied_receipt_paths(tmp_path): from services.pixelrag_crawler_integration_service import build_pixelrag_visual_evidence_readback receipt_path = _write_capture_receipt(tmp_path, platform="pchome", manifest_id="copied") receipt = json.loads(receipt_path.read_text(encoding="utf-8")) receipt["files"] = [ {"kind": "fullpage_screenshot", "path": "/host/runtime/fullpage.png"}, {"kind": "tile", "path": "/host/runtime/tiles/tile_000.png"}, {"kind": "tile", "path": "/host/runtime/tiles/tile_001.png"}, ] receipt_path.write_text(json.dumps(receipt), encoding="utf-8") payload = build_pixelrag_visual_evidence_readback( artifact_root=tmp_path, platform="pchome", manifest_id="copied", ) assert payload["status"] == "ok" assert payload["summary"]["missing_file_count"] == 0 assert all(item["exists"] for item in payload["latest_receipt"]["files"]) def test_pixelrag_visual_evidence_readback_route_returns_payload(tmp_path, monkeypatch): from flask import Flask from routes import system_public_routes as routes from services import pixelrag_crawler_integration_service as service _write_capture_receipt(tmp_path, platform="pchome", manifest_id="route-manifest") monkeypatch.setattr(service, "DEFAULT_ARTIFACT_ROOT", str(tmp_path)) app = Flask(__name__) with app.test_request_context( "/api/ai-automation/pixelrag-visual-evidence-readback" "?platform=pchome&manifest_id=route-manifest&max_age_hours=999" ): response = routes.ai_automation_pixelrag_visual_evidence_readback_api.__wrapped__() payload = response.get_json() assert payload["policy"] == "read_only_pixelrag_visual_evidence_readback_v1" assert payload["status"] == "ok" assert payload["manifest_id"] == "route-manifest" assert payload["summary"]["tile_file_count"] == 2 def test_pixelrag_capture_worker_dry_run_plans_artifact_outputs(tmp_path): if not shutil.which("node"): pytest.skip("node is required for the dry-run capture worker test") from services.pixelrag_crawler_integration_service import ( build_pixelrag_visual_evidence_manifest, ) manifest = build_pixelrag_visual_evidence_manifest( url="https://m.momoshop.com.tw/search.momo?searchKeyword=test", platform="momo", crawler="MomoCrawler.search_products", trigger_reason="parser_empty", page_size={"width": 1024, "height": 1024}, tile_size={"width": 512, "height": 512}, ) completed = subprocess.run( [ "node", "scripts/ops/capture_pixelrag_visual_evidence.js", "--manifest-json", json.dumps(manifest), "--output-dir", str(tmp_path), "--dry-run", ], capture_output=True, check=False, text=True, ) assert completed.returncode == 0 payload = json.loads(completed.stdout) assert payload["policy"] == "read_only_pixelrag_visual_capture_artifact_v1" assert payload["status"] == "dry_run_ready" assert payload["controlled_apply"]["artifact_write"] is False assert payload["tile_plan"]["planned_tile_count"] == 4 assert payload["planned_output"]["screenshot"].endswith("fullpage.png") def test_pixelrag_capture_worker_rejects_wrong_platform_host(tmp_path): if not shutil.which("node"): pytest.skip("node is required for the dry-run capture worker test") from services.pixelrag_crawler_integration_service import ( build_pixelrag_visual_evidence_manifest, ) manifest = build_pixelrag_visual_evidence_manifest( url="https://example.test/product", platform="momo", crawler="MomoCrawler.search_products", trigger_reason="parser_empty", ) completed = subprocess.run( [ "node", "scripts/ops/capture_pixelrag_visual_evidence.js", "--manifest-json", json.dumps(manifest), "--output-dir", str(tmp_path), "--dry-run", ], capture_output=True, check=False, text=True, ) assert completed.returncode == 1 payload = json.loads(completed.stdout) assert payload["status"] == "rejected" assert any("not allowed for platform momo" in error for error in payload["errors"]) def test_pixelrag_python_capture_worker_dry_run_plans_artifact_outputs(tmp_path): from services.pixelrag_crawler_integration_service import ( build_pixelrag_visual_evidence_manifest, ) manifest = build_pixelrag_visual_evidence_manifest( url="https://24h.pchome.com.tw/", platform="pchome", crawler="PChomeCrawler.visual_fallback", trigger_reason="parser_empty", page_size={"width": 1024, "height": 1024}, tile_size={"width": 512, "height": 512}, ) completed = subprocess.run( [ sys.executable, "scripts/ops/capture_pixelrag_visual_evidence.py", "--manifest-json", json.dumps(manifest), "--output-dir", str(tmp_path), "--dry-run", ], capture_output=True, check=False, text=True, ) assert completed.returncode == 0 payload = json.loads(completed.stdout) assert payload["policy"] == "read_only_pixelrag_visual_capture_artifact_v1" assert payload["status"] == "dry_run_ready" assert payload["controlled_apply"]["artifact_write"] is False assert payload["tile_plan"]["planned_tile_count"] == 4 assert payload["capture_target"]["platform"] == "pchome" def test_pixelrag_python_capture_worker_accepts_shopee_manifest(tmp_path): from services.pixelrag_crawler_integration_service import build_pixelrag_marketplace_search_manifest manifest = build_pixelrag_marketplace_search_manifest( platform="shopee_tw", keyword="防曬乳", page_size={"width": 1024, "height": 1024}, tile_size={"width": 512, "height": 512}, ) completed = subprocess.run( [ sys.executable, "scripts/ops/capture_pixelrag_visual_evidence.py", "--manifest-json", json.dumps(manifest), "--output-dir", str(tmp_path), "--dry-run", ], capture_output=True, check=False, text=True, ) assert completed.returncode == 0 payload = json.loads(completed.stdout) assert payload["status"] == "dry_run_ready" assert payload["capture_target"]["platform"] == "shopee_tw" assert payload["tile_plan"]["planned_tile_count"] == 4