wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(ai): 接入 PixelRAG RAG replay smoke

Browse Source
This commit is contained in:
ogt
2026-07-09 22:40:51 +08:00
parent cc3c0697a7
commit 1bbf73a176
13 changed files with 886 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
tests/test_ai_automation_smoke_service.py
View File

@@ -1431,11 +1431,13 @@ def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch):
monkeypatch.setattr(smoke, "_ai_surface_html_readback_check", lambda: smoke._check("surface", "ok", "ok"))
monkeypatch.setattr(smoke, "_sitewide_ui_ux_agent_check", lambda: smoke._check("sitewide", "ok", "ok"))
monkeypatch.setattr(smoke, "_sitewide_visual_qa_check", lambda: smoke._check("visual", "ok", "ok"))
monkeypatch.setattr(smoke, "_external_mcp_rag_integration_check", lambda: smoke._check("external mcp rag", "ok", "ok"))
monkeypatch.setattr(smoke, "_pixelrag_rag_candidate_replay_check", lambda: smoke._check("pixelrag replay", "ok", "ok"))
result = smoke.collect_ai_automation_smoke(record_history=False)
assert result["status"] == "critical"
assert result["summary"] == {"ok": 28, "warning": 1, "critical": 1, "total": 30}
assert result["summary"] == {"ok": 31, "warning": 1, "critical": 1, "total": 33}
def test_pchome_controlled_apply_drift_monitor_reports_verified_zero_drift(monkeypatch):
@@ -3962,6 +3964,8 @@ def test_collect_ai_automation_smoke_persists_recent_history(tmp_path, monkeypat
monkeypatch.setattr(smoke, "_ai_surface_html_readback_check", lambda: smoke._check("surface", "ok", "ok"))
monkeypatch.setattr(smoke, "_sitewide_ui_ux_agent_check", lambda: smoke._check("sitewide", "ok", "ok"))
monkeypatch.setattr(smoke, "_sitewide_visual_qa_check", lambda: smoke._check("visual", "ok", "ok"))
monkeypatch.setattr(smoke, "_external_mcp_rag_integration_check", lambda: smoke._check("external mcp rag", "ok", "ok"))
monkeypatch.setattr(smoke, "_pixelrag_rag_candidate_replay_check", lambda: smoke._check("pixelrag replay", "ok", "ok"))
first = smoke.collect_ai_automation_smoke(history_limit=5)
second = smoke.collect_ai_automation_smoke(history_limit=5)
@@ -4017,7 +4021,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
json.dumps({
"generated_at": datetime.now().isoformat(timespec="seconds"),
"status": "ok",
"summary": {"ok": 30, "warning": 0, "critical": 0, "total": 30},
"summary": {"ok": 33, "warning": 0, "critical": 0, "total": 33},
"checks": [
{
"name": "PChome 受控落地 drift monitor",
@@ -4078,10 +4082,10 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
"repair_action_count": 0,
},
},
{
"name": "Sitewide visual QA readback",
"status": "ok",
"summary": "Sitewide visual QA ok，12 routes x 3 viewports 無 overflow 退化",
{
"name": "Sitewide visual QA readback",
"status": "ok",
"summary": "Sitewide visual QA ok，12 routes x 3 viewports 無 overflow 退化",
"details": {
"policy": "read_only_sitewide_visual_qa_readback_v1",
"result_count": 36,
@@ -4092,11 +4096,42 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
"visual_offender_count": 0,
"artifact_path": "data/ai_automation/sitewide_visual_qa_latest.json",
"artifact_generated_at": datetime.now().isoformat(timespec="seconds"),
"stale": False,
"stale": False,
},
},
}
],
}, ensure_ascii=False) + "\n",
{
"name": "External MCP/RAG integration readback",
"status": "ok",
"summary": "External MCP/RAG registry 9/9 absorbed; unresolved=0; runtime mcp=True rag=True",
"details": {
"policy": "read_only_external_mcp_rag_integration_readback_v1",
"integration_status": "fully_integrated",
"total_capabilities": 9,
"absorbed_count": 9,
"unresolved_count": 0,
"mcp_router_enabled": True,
"rag_enabled": True,
"gemini_hard_disabled": True,
"next_machine_action": "keep_external_mcp_rag_runtime_monitoring",
},
},
{
"name": "PixelRAG RAG candidate replay",
"status": "ok",
"summary": "PixelRAG RAG candidate replay receipts=1, eligible=1, blocked=0, invalid=0, barriers=0",
"details": {
"policy": "read_only_pixelrag_rag_candidate_replay_v1",
"receipt_count": 1,
"eligible_count": 1,
"blocked_count": 0,
"invalid_count": 0,
"visual_barrier_count": 0,
"blocked_pages_are_not_product_data": True,
"next_machine_action": "run_ocr_vlm_replay_then_promotion_gate",
},
}
],
}, ensure_ascii=False) + "\n",
encoding="utf-8",
)
monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path))
@@ -4112,7 +4147,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
)
assert summary["policy"] == "read_only_ai_automation_scheduled_health_summary"
assert summary["status"] == "ok"
assert summary["summary"]["total"] == 27
assert summary["summary"]["total"] == 30
assert summary["summary"]["primary_human_gate_count"] == 0
assert summary["summary"]["writes_database_count"] == 0
assert pchome_family["status"] == "ok"
@@ -5776,6 +5811,18 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
assert visual_family["status"] == "ok"
assert visual_family["details"]["route_count"] == 12
assert visual_family["details"]["failed_count"] == 0
external_mcp_rag_family = next(
item for item in summary["families"]
if item["key"] == "external_mcp_rag_integration"
)
assert external_mcp_rag_family["status"] == "ok"
assert external_mcp_rag_family["details"]["unresolved_count"] == 0
pixelrag_rag_replay_family = next(
item for item in summary["families"]
if item["key"] == "pixelrag_rag_candidate_replay"
)
assert pixelrag_rag_replay_family["status"] == "ok"
assert pixelrag_rag_replay_family["details"]["eligible_count"] == 1
assert summary["scheduled_outputs"]["telegram_send_in_preview"] is False
assert summary["scheduled_outputs"]["writes_database"] is False
assert summary["automation_policy"]["primary_flow"] == "ai_controlled"
@@ -6382,6 +6429,16 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch):
"visual ok",
{"route_count": 12, "viewport_count": 3, "failed_count": 0},
))
monkeypatch.setattr(smoke, "_external_mcp_rag_integration_check", lambda: smoke._check(
"External MCP/RAG integration readback",
"ok",
"external mcp rag ok",
))
monkeypatch.setattr(smoke, "_pixelrag_rag_candidate_replay_check", lambda: smoke._check(
"PixelRAG RAG candidate replay",
"ok",
"pixelrag replay ok",
))
result = smoke.collect_ai_automation_smoke(record_history=False)
@@ -6397,7 +6454,7 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch):
item for item in result["checks"]
if item["name"] == "Sitewide visual QA readback"
)
assert result["summary"]["total"] == 30
assert result["summary"]["total"] == 33
assert surface_check["status"] == "ok"
assert surface_check["details"]["checked_surface_count"] == 10
assert sitewide_check["status"] == "ok"

138
tests/test_pixelrag_rag_candidate_replay_service.py Normal file
View File

@@ -0,0 +1,138 @@
import json
import subprocess
import sys
from datetime import datetime, timezone
def _write_receipt(root, *, platform, manifest_id, title, url, http_status=200):
receipt_dir = root / platform / manifest_id
tiles_dir = receipt_dir / "tiles"
tiles_dir.mkdir(parents=True)
screenshot = receipt_dir / "screenshot.png"
screenshot.write_bytes(b"png")
files = [{"kind": "screenshot", "path": str(screenshot)}]
for index in range(4):
tile = tiles_dir / f"tile-{index}.png"
tile.write_bytes(b"tile")
files.append({"kind": "tile", "path": str(tile)})
receipt = {
"generated_at": datetime.now(timezone.utc).isoformat(),
"status": "captured",
"manifest_id": manifest_id,
"http_status": http_status,
"capture_target": {
"platform": platform,
"url": url,
"crawler": "PixelRAGMarketplaceSearch.visual_fallback",
},
"page_metrics": {
"title": title,
"final_url": url,
},
"tile_plan": {
"planned_tile_count": 4,
"emitted_tile_count": 4,
},
"files": files,
}
path = receipt_dir / "capture_receipt.json"
path.write_text(json.dumps(receipt, ensure_ascii=False), encoding="utf-8")
return path
def test_pixelrag_rag_candidate_replay_splits_eligible_and_blocked_receipts(tmp_path):
from services.pixelrag_rag_candidate_replay_service import (
POLICY,
build_pixelrag_rag_candidate_replay_readback,
)
_write_receipt(
tmp_path,
platform="shopee_tw",
manifest_id="shopee-ok",
title="Shopee 防曬乳",
url="https://shopee.tw/search?keyword=%E9%98%B2%E6%9B%AC%E4%B9%B3",
)
_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_rag_candidate_replay_readback(artifact_root=tmp_path)
by_platform = {item["platform"]: item for item in payload["candidates"]}
assert payload["policy"] == POLICY
assert payload["status"] == "warning"
assert payload["summary"]["receipt_count"] == 2
assert payload["summary"]["eligible_count"] == 1
assert payload["summary"]["blocked_count"] == 1
assert payload["summary"]["visual_barrier_count"] == 1
assert by_platform["shopee_tw"]["eligible_for_rag_candidate_replay"] is True
assert by_platform["shopee_tw"]["promotion_boundary"]["direct_ai_insights_write_allowed"] is False
assert "Shopee 防曬乳" in by_platform["shopee_tw"]["rag_candidate_text"]
assert by_platform["coupang_tw"]["eligible_for_rag_candidate_replay"] is False
assert by_platform["coupang_tw"]["visual_barrier_reason"] == "http_status_403"
assert by_platform["coupang_tw"]["next_machine_action"] == "run_platform_probe_or_use_structured_api"
assert payload["replay_contract"]["blocked_pages_are_not_product_data"] is True
assert payload["controlled_apply"]["db_write"] is False
def test_pixelrag_rag_candidate_replay_cli_outputs_machine_readable_json(tmp_path):
_write_receipt(
tmp_path,
platform="shopee_tw",
manifest_id="shopee-ok",
title="Shopee 防曬乳",
url="https://shopee.tw/search?keyword=sunscreen",
)
completed = subprocess.run(
[
sys.executable,
"scripts/ops/report_pixelrag_rag_candidate_replay.py",
"--artifact-root",
str(tmp_path),
"--platform",
"shopee_tw",
],
capture_output=True,
check=False,
text=True,
)
assert completed.returncode == 0
payload = json.loads(completed.stdout)
assert payload["summary"]["receipt_count"] == 1
assert payload["summary"]["eligible_count"] == 1
assert payload["candidates"][0]["platform"] == "shopee_tw"
def test_pixelrag_rag_candidate_replay_route_returns_readback(tmp_path, monkeypatch):
from flask import Flask
from routes import system_public_routes as routes
from services import pixelrag_rag_candidate_replay_service as service
_write_receipt(
tmp_path,
platform="shopee_tw",
manifest_id="shopee-ok",
title="Shopee 防曬乳",
url="https://shopee.tw/search?keyword=sunscreen",
)
monkeypatch.setattr(service, "DEFAULT_ARTIFACT_ROOT", str(tmp_path))
app = Flask(__name__)
with app.test_request_context(
"/api/ai-automation/pixelrag-rag-candidate-replay?platform=shopee_tw"
):
response = routes.ai_automation_pixelrag_rag_candidate_replay_api.__wrapped__()
payload = response.get_json()
assert payload["policy"] == "read_only_pixelrag_rag_candidate_replay_v1"
assert payload["summary"]["receipt_count"] == 1
assert payload["summary"]["eligible_count"] == 1
assert payload["replay_contract"]["writes_database"] is False