230 lines
8.1 KiB
Python
230 lines
8.1 KiB
Python
import json
|
|
import subprocess
|
|
import sys
|
|
from types import SimpleNamespace
|
|
|
|
from tests.test_pixelrag_ocr_vlm_replay_service import _write_receipt
|
|
|
|
|
|
def test_pixelrag_vlm_replay_worker_dry_run_keeps_blocked_pages_out(tmp_path):
|
|
from services.pixelrag_vlm_replay_worker_service import (
|
|
POLICY,
|
|
run_pixelrag_ollama_vlm_replay_worker,
|
|
)
|
|
|
|
_write_receipt(
|
|
tmp_path,
|
|
platform="shopee_tw",
|
|
manifest_id="shopee-ok",
|
|
title="Shopee 防曬乳",
|
|
url="https://shopee.tw/search?keyword=sunscreen",
|
|
)
|
|
_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 = run_pixelrag_ollama_vlm_replay_worker(
|
|
artifact_root=tmp_path,
|
|
platform=("shopee_tw", "coupang_tw"),
|
|
)
|
|
|
|
assert payload["policy"] == POLICY
|
|
assert payload["status"] == "warning"
|
|
assert payload["execute"] is False
|
|
assert payload["summary"]["receipt_count"] == 2
|
|
assert payload["summary"]["ready_count"] == 1
|
|
assert payload["summary"]["dry_run_count"] == 1
|
|
assert payload["summary"]["skipped_count"] == 1
|
|
assert payload["summary"]["model_call_performed"] is False
|
|
assert payload["summary"]["artifact_write_performed"] is False
|
|
assert payload["summary"]["writes_database_count"] == 0
|
|
assert payload["controlled_apply"]["primary_human_gate_count"] == 0
|
|
assert payload["next_machine_action"] == "run_pixelrag_vlm_replay_worker_execute"
|
|
by_platform = {item["platform"]: item for item in payload["worker_items"]}
|
|
assert by_platform["shopee_tw"]["worker_status"] == "dry_run_ready"
|
|
assert by_platform["coupang_tw"]["worker_status"] == "skipped_blocked_or_not_ready"
|
|
|
|
|
|
def test_pixelrag_vlm_replay_worker_execute_writes_artifact_receipt(tmp_path, monkeypatch):
|
|
from services import pixelrag_vlm_replay_worker_service as service
|
|
|
|
_write_receipt(
|
|
tmp_path,
|
|
platform="shopee_tw",
|
|
manifest_id="shopee-ok",
|
|
title="Shopee 防曬乳",
|
|
url="https://shopee.tw/search?keyword=sunscreen",
|
|
)
|
|
|
|
class FakeOllama:
|
|
def __init__(self, model):
|
|
self.model = model
|
|
|
|
def generate(self, *args, **kwargs):
|
|
return SimpleNamespace(
|
|
success=True,
|
|
content=json.dumps({
|
|
"blocked_page_detected": False,
|
|
"fields": {
|
|
"title": {
|
|
"value": "防曬乳 SPF50",
|
|
"confidence": 0.92,
|
|
"evidence_refs": ["tile:1"],
|
|
},
|
|
"price": {
|
|
"value": "399",
|
|
"confidence": 0.88,
|
|
"evidence_refs": ["tile:2"],
|
|
},
|
|
},
|
|
"quality": {
|
|
"overall_confidence": 0.89,
|
|
"missing_required_fields": [],
|
|
"requires_identity_matcher_replay": True,
|
|
"requires_promotion_gate": True,
|
|
},
|
|
"notes": [],
|
|
}),
|
|
model="minicpm-v:latest",
|
|
error=None,
|
|
total_duration=1.5,
|
|
host="http://34.87.90.216:11434",
|
|
input_tokens=12,
|
|
output_tokens=80,
|
|
)
|
|
|
|
monkeypatch.setattr(service, "OllamaService", FakeOllama)
|
|
payload = service.run_pixelrag_ollama_vlm_replay_worker(
|
|
artifact_root=tmp_path,
|
|
output_root=tmp_path / "receipts",
|
|
platform="shopee_tw",
|
|
execute=True,
|
|
write_receipt=True,
|
|
tile_limit=1,
|
|
)
|
|
|
|
assert payload["status"] == "ok"
|
|
assert payload["summary"]["executed_count"] == 1
|
|
assert payload["summary"]["executed_ok_count"] == 1
|
|
assert payload["summary"]["receipt_written_count"] == 1
|
|
assert payload["summary"]["writes_database_count"] == 0
|
|
item = payload["worker_items"][0]
|
|
assert item["worker_status"] == "executed_ok"
|
|
assert item["provider"] == "gcp_ollama"
|
|
assert item["validation"]["valid_for_identity_matcher_replay"] is True
|
|
receipt_path = tmp_path / "receipts" / "shopee_tw" / "shopee-ok" / "vlm_replay_receipt.json"
|
|
assert receipt_path.exists()
|
|
receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
|
|
assert receipt["parsed_output"]["fields"]["title"]["value"] == "防曬乳 SPF50"
|
|
|
|
|
|
def test_pixelrag_vlm_replay_worker_writes_model_error_receipt(tmp_path, monkeypatch):
|
|
from services import pixelrag_vlm_replay_worker_service as service
|
|
|
|
_write_receipt(
|
|
tmp_path,
|
|
platform="shopee_tw",
|
|
manifest_id="shopee-ok",
|
|
title="Shopee 防曬乳",
|
|
url="https://shopee.tw/search?keyword=sunscreen",
|
|
)
|
|
|
|
class FakeOllama:
|
|
def __init__(self, model):
|
|
self.model = model
|
|
|
|
def generate(self, *args, **kwargs):
|
|
return SimpleNamespace(
|
|
success=False,
|
|
content="",
|
|
model="minicpm-v:latest",
|
|
error="all hosts failed",
|
|
total_duration=None,
|
|
host="http://34.21.145.224:11434",
|
|
input_tokens=0,
|
|
output_tokens=0,
|
|
)
|
|
|
|
monkeypatch.setattr(service, "OllamaService", FakeOllama)
|
|
payload = service.run_pixelrag_ollama_vlm_replay_worker(
|
|
artifact_root=tmp_path,
|
|
output_root=tmp_path / "receipts",
|
|
platform="shopee_tw",
|
|
execute=True,
|
|
write_receipt=True,
|
|
tile_limit=1,
|
|
)
|
|
|
|
assert payload["status"] == "critical"
|
|
assert payload["summary"]["model_error_count"] == 1
|
|
assert payload["summary"]["receipt_written_count"] == 1
|
|
assert payload["summary"]["artifact_write_performed"] is True
|
|
assert payload["controlled_apply"]["writes_database"] is False
|
|
receipt_path = tmp_path / "receipts" / "shopee_tw" / "shopee-ok" / "vlm_replay_receipt.json"
|
|
receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
|
|
assert receipt["worker_status"] == "model_error"
|
|
assert receipt["next_machine_action"] == "repair_ollama_vlm_runtime_or_model_route"
|
|
|
|
|
|
def test_pixelrag_vlm_replay_worker_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/run_pixelrag_vlm_replay_worker.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"]["dry_run_count"] == 1
|
|
assert payload["controlled_apply"]["model_call"] is False
|
|
assert payload["controlled_apply"]["writes_database"] is False
|
|
|
|
|
|
def test_pixelrag_vlm_replay_worker_route_returns_readback(tmp_path, monkeypatch):
|
|
from flask import Flask
|
|
from routes import system_public_routes as routes
|
|
from services import pixelrag_vlm_replay_worker_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-vlm-replay-worker?platform=shopee_tw"
|
|
):
|
|
response = routes.ai_automation_pixelrag_vlm_replay_worker_api.__wrapped__()
|
|
payload = response.get_json()
|
|
|
|
assert payload["policy"] == "controlled_pixelrag_ollama_vlm_replay_worker_v1"
|
|
assert payload["summary"]["receipt_count"] == 1
|
|
assert payload["summary"]["dry_run_count"] == 1
|
|
assert payload["controlled_apply"]["model_call"] is False
|