204 lines
8.4 KiB
Python
204 lines
8.4 KiB
Python
from flask import Flask
|
|
|
|
from services.pchome_mapping_backlog_service import (
|
|
build_pchome_direct_mapping_candidate_decision_lane_closeout_package,
|
|
)
|
|
|
|
|
|
def _payload():
|
|
return {
|
|
"success": True,
|
|
"system_name": "MOMO Pro",
|
|
"generated_at": "2026-06-28T01:16:02",
|
|
"cache_state": "fresh",
|
|
"stats": {
|
|
"candidate_count": 4,
|
|
"mapped_count": 1,
|
|
"mapping_rate": 25.0,
|
|
"needs_mapping_count": 3,
|
|
"review_candidate_count": 1,
|
|
"overall_latest_sales_date": "2026-06-24",
|
|
"overall_sales_7d": 2020234.0,
|
|
"action_counts": {
|
|
"先補商品對應": 2,
|
|
"確認候選": 1,
|
|
"放大價格優勢": 1,
|
|
},
|
|
"action_code_counts": {
|
|
"map_external_product": 2,
|
|
"review_external_candidate": 1,
|
|
"amplify_price_advantage": 1,
|
|
},
|
|
},
|
|
"opportunities": [
|
|
{
|
|
"pchome_product_id": "PCH-1",
|
|
"product_name": "Mapped product",
|
|
"sales_7d": 0,
|
|
"external_price": {
|
|
"momo_sku": "M-1",
|
|
"price_basis": "unit_price",
|
|
"gap_pct": 12.5,
|
|
},
|
|
"recommended_action": {
|
|
"code": "amplify_price_advantage",
|
|
"label": "放大價格優勢",
|
|
},
|
|
"priority_score": 75.0,
|
|
},
|
|
{
|
|
"pchome_product_id": "PCH-2",
|
|
"product_name": "Direct mapping product 40ml x2",
|
|
"sales_7d": 9800,
|
|
"pchome_price": 1200,
|
|
"external_price": None,
|
|
"recommended_action": {"code": "map_external_product", "label": "先補商品對應"},
|
|
"priority_score": 88.0,
|
|
"reason_lines": ["需要補商品對應"],
|
|
},
|
|
{
|
|
"pchome_product_id": "PCH-3",
|
|
"product_name": "Review candidate product",
|
|
"sales_7d": 1200,
|
|
"external_price": None,
|
|
"review_candidate": {
|
|
"id": 725,
|
|
"momo_sku": "5868343",
|
|
"momo_name": "MOMO candidate",
|
|
"quality_score": 94.8,
|
|
},
|
|
"recommended_action": {"code": "review_external_candidate", "label": "確認候選"},
|
|
"priority_score": 64.0,
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
def test_candidate_decision_lane_closeout_outputs_receipt_replay_drift_and_readiness(tmp_path):
|
|
def fake_search(targets, limit_per_product, max_products, max_terms_per_product, min_score):
|
|
return True, "found", [
|
|
{
|
|
"product_id": "MOMO-1",
|
|
"name": "Direct mapping product 40ml x2",
|
|
"price": 999,
|
|
"target_pchome_product_id": "PCH-2",
|
|
"target_pchome_name": "Direct mapping product 40ml x2",
|
|
"target_match_score": 0.92,
|
|
"auto_compare_type": "total_price",
|
|
"target_hard_veto": False,
|
|
"target_price_basis": "total_price",
|
|
"target_gap_pct": 16.8,
|
|
"target_search_term": "direct mapping product 40ml x2",
|
|
},
|
|
{
|
|
"product_id": "MOMO-2",
|
|
"name": "Variant candidate",
|
|
"price": 899,
|
|
"target_pchome_product_id": "PCH-2",
|
|
"target_pchome_name": "Direct mapping product 40ml x2",
|
|
"target_match_score": 0.51,
|
|
"auto_compare_type": "manual_review",
|
|
"target_hard_veto": False,
|
|
},
|
|
]
|
|
|
|
package = build_pchome_direct_mapping_candidate_decision_lane_closeout_package(
|
|
_payload(),
|
|
batch_size=1,
|
|
execute_search=True,
|
|
limit_per_product=3,
|
|
search_func=fake_search,
|
|
artifact_root=tmp_path,
|
|
)
|
|
|
|
assert package["policy"] == "read_only_pchome_growth_direct_mapping_candidate_decision_lane_closeout"
|
|
assert package["result"] == "DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CLOSEOUT_READY"
|
|
assert package["summary"]["candidate_decision_count"] == 2
|
|
assert package["summary"]["auto_compare_decision_count"] == 1
|
|
assert package["summary"]["machine_review_decision_count"] == 1
|
|
assert package["summary"]["receipt_payload_hash_ready_count"] == 1
|
|
assert package["summary"]["replay_verified_count"] == 1
|
|
assert package["summary"]["drift_count"] == 0
|
|
assert package["summary"]["primary_human_gate_count"] == 0
|
|
assert package["lane_receipt"]["receipt_status"] == "candidate_decision_lane_receipt_ready"
|
|
assert package["receipt_replay"]["passed"] is True
|
|
assert package["drift_verifier"]["status"] == "verified"
|
|
assert package["product_readiness"]["status"] == "ready_for_next_automation"
|
|
assert package["product_readiness"]["automation_policy"]["primary_flow"] == "ai_controlled"
|
|
assert package["product_readiness"]["primary_human_gate_count"] == 0
|
|
assert package["safety"]["writes_database"] is False
|
|
assert package["safety"]["writes_artifact_count"] == 0
|
|
assert package["all_checks_passed"] is True
|
|
|
|
|
|
def test_candidate_decision_lane_closeout_route_defaults_to_cached_no_db_no_search(monkeypatch):
|
|
from routes import ai_routes as routes
|
|
|
|
monkeypatch.setattr(routes, "_get_cached_pchome_growth_payload", lambda: _payload())
|
|
|
|
def fail_engine(database_path):
|
|
raise AssertionError("cached candidate decision lane closeout should not open a DB engine")
|
|
|
|
monkeypatch.setattr(routes, "_create_icaim_dashboard_engine", fail_engine)
|
|
|
|
app = Flask(__name__)
|
|
with app.test_request_context(
|
|
"/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-decision-lane-closeout-package?batch_size=1"
|
|
):
|
|
response = (
|
|
routes
|
|
.api_pchome_growth_direct_mapping_candidate_decision_lane_closeout_package
|
|
.__wrapped__()
|
|
)
|
|
|
|
payload = response.get_json()
|
|
assert payload["success"] is True
|
|
assert payload["policy"] == "read_only_pchome_growth_direct_mapping_candidate_decision_lane_closeout"
|
|
assert payload["source_endpoint"] == (
|
|
"/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-decision-package"
|
|
)
|
|
assert payload["result"] == "WAITING_FOR_DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CANDIDATES"
|
|
assert payload["summary"]["candidate_decision_count"] == 0
|
|
assert payload["summary"]["replay_verified_count"] == 1
|
|
assert payload["summary"]["drift_count"] == 0
|
|
assert payload["product_readiness"]["status"] == "waiting_for_read_only_search"
|
|
assert payload["product_readiness"]["primary_human_gate_count"] == 0
|
|
assert payload["safety"]["executes_search"] is False
|
|
assert payload["safety"]["writes_database"] is False
|
|
|
|
|
|
def test_ai_automation_surface_summary_route_defaults_to_cached_no_db(monkeypatch):
|
|
from routes import ai_routes as routes
|
|
|
|
monkeypatch.setattr(routes, "_get_cached_pchome_growth_payload", lambda: _payload())
|
|
|
|
def fail_engine(database_path):
|
|
raise AssertionError("cached AI automation surface summary should not open a DB engine")
|
|
|
|
monkeypatch.setattr(routes, "_create_icaim_dashboard_engine", fail_engine)
|
|
|
|
app = Flask(__name__)
|
|
with app.test_request_context(
|
|
"/api/ai/pchome-growth/ai-automation-surface-summary"
|
|
"?batch_size=1&include_receipt_replay=false&include_drift_verifier=false"
|
|
):
|
|
response = routes.api_pchome_growth_ai_automation_surface_summary.__wrapped__()
|
|
|
|
payload = response.get_json()
|
|
signals = {signal["key"]: signal for signal in payload["golden_signals"]}
|
|
assert payload["success"] is True
|
|
assert payload["policy"] == "read_only_pchome_growth_ai_automation_surface_summary"
|
|
assert payload["source_endpoint"] == "/api/ai/pchome-growth/ai-automation-readiness"
|
|
assert payload["summary"]["safe_lane_count"] >= 4
|
|
assert payload["summary"]["primary_human_gate_count"] == 0
|
|
assert payload["summary"]["writes_database_count"] == 0
|
|
assert set(signals) == {
|
|
"automated-landing",
|
|
"verified",
|
|
"change-state",
|
|
"next-machine-action",
|
|
}
|
|
assert signals["next-machine-action"]["next_machine_action"]
|
|
assert payload["safety"]["writes_database"] is False
|
|
assert payload["safety"]["llm_calls_in_preview"] is False
|