+ const candidateClass = source.has_candidate_runtime_data && !source.has_runtime_data
+ ? ' is-candidate'
+ : '';
+ return `
${escapeHtml(source.display_name || '未命名來源')}
${escapeHtml(source.status_label || '待確認')}
diff --git a/tests/test_ai_intelligence_text_density_guardrails.py b/tests/test_ai_intelligence_text_density_guardrails.py
index 98f0829..b5df76b 100644
--- a/tests/test_ai_intelligence_text_density_guardrails.py
+++ b/tests/test_ai_intelligence_text_density_guardrails.py
@@ -92,7 +92,9 @@ def test_ai_intelligence_comparison_coverage_has_explicit_scope_and_partition():
)[0]
assert "同款比價證據 · TOP 50" in template
- assert "正式資料平台" in template
+ assert "外部平台資料" in template
+ assert "正式 ·" in renderer
+ assert "候選" in renderer
assert "比價覆蓋 · TOP 50" in template
assert "具正式比價證據" in template
assert "候選 / 來源驗證" in template
diff --git a/tests/test_external_market_offer_service.py b/tests/test_external_market_offer_service.py
index 38ccf38..3274158 100644
--- a/tests/test_external_market_offer_service.py
+++ b/tests/test_external_market_offer_service.py
@@ -625,6 +625,47 @@ def test_external_source_readiness_uses_legacy_momo_reference_cache():
assert "只有 runtime 已啟用" in payload["plain_summary"]
+def test_external_source_readiness_separates_candidate_runtime_from_formal(
+ monkeypatch,
+ tmp_path,
+):
+ import json
+ from datetime import datetime, timezone
+ from services import yahoo_shopping_public_source_adapter as adapter
+ from services.external_market_offer_service import build_external_source_readiness
+
+ receipt = {
+ "policy": "controlled_yahoo_shopping_public_offer_v1",
+ "adapter_version": "yahoo_shopping_public_jsonld_v1",
+ "generated_at": datetime.now(timezone.utc).isoformat(),
+ "summary": {
+ "source_candidate_count": 4,
+ "verified_candidate_count": 0,
+ "written_count": 0,
+ "formal_source_activated": False,
+ },
+ }
+ (tmp_path / "candidate-run.json").write_text(
+ json.dumps(receipt),
+ encoding="utf-8",
+ )
+ monkeypatch.setattr(adapter, "DEFAULT_RECEIPT_ROOT", str(tmp_path))
+ from services import yahoo_shopping_public_offer_service as runtime
+
+ monkeypatch.setattr(runtime, "DEFAULT_RECEIPT_ROOT", str(tmp_path))
+
+ payload = build_external_source_readiness()
+ sources = {source["code"]: source for source in payload["sources"]}
+
+ assert payload["sources_with_data_count"] == 0
+ assert payload["candidate_runtime_source_count"] == 1
+ assert payload["sources_with_any_runtime_evidence_count"] == 1
+ assert sources["yahoo_shopping"]["runtime_stage"] == "candidate"
+ assert sources["yahoo_shopping"]["status_label"] == "候選自動化"
+ assert sources["yahoo_shopping"]["candidate_offer_count"] == 4
+ assert sources["yahoo_shopping"]["can_alert"] is False
+
+
def test_source_seeds_preserve_runtime_activation_and_readiness_uses_it():
from services import external_market_offer_service as service
@@ -649,6 +690,142 @@ def test_source_seeds_preserve_runtime_activation_and_readiness_uses_it():
assert bool(source["enabled"]) is True
+def test_targeted_marketplace_canary_activates_and_rollback_quarantines(monkeypatch):
+ from services import external_marketplace_canary_service as canary_service
+ from services.external_market_offer_service import build_external_source_readiness
+ from services.pchome_growth_same_item_reconciliation import (
+ readback_external_offer_candidates,
+ )
+
+ monkeypatch.setattr(
+ canary_service,
+ "mark_pchome_growth_cache_stale",
+ lambda: None,
+ )
+ engine = create_engine("sqlite:///:memory:")
+ _seed_external_offer_sync_tables(engine)
+ identity = {
+ "trace_id": "trace-yahoo",
+ "run_id": "run-yahoo",
+ "work_item_id": "GROWTH-P0-001-C",
+ }
+ candidates = []
+ for index in (1, 2):
+ candidates.append({
+ "product_id": f"Y-{index}",
+ "name": f"Yahoo 同款商品 {index}",
+ "price": 800 + index,
+ "currency": "TWD",
+ "product_url": f"https://tw.buy.yahoo.com/gdsale/product-{index}.html",
+ "target_pchome_product_id": f"P-{index}",
+ "target_pchome_name": f"PChome 同款商品 {index}",
+ "target_pchome_price": 900 + index,
+ "target_match_score": 1.0,
+ "target_match_reasons": [],
+ "target_comparison_mode": "exact_identity",
+ "target_match_type": "exact",
+ "target_alert_tier": "price_alert_exact",
+ "target_price_basis": "total_price",
+ "target_hard_veto": False,
+ "auto_compare_type": "total_price",
+ "source_contract_guard": {"passed": True, "reasons": []},
+ "same_item_candidate_fingerprint": f"fingerprint-{index}",
+ "same_item_reconciliation": {
+ **identity,
+ "independent_verifier_passed": True,
+ },
+ })
+
+ preview = canary_service.sync_targeted_marketplace_candidates_to_external_offers(
+ engine,
+ candidates,
+ source_code="yahoo_shopping",
+ platform_code="yahoo_shopping_tw",
+ ingestion_method="public_product_jsonld",
+ connector_key="pchome_targeted_yahoo_public_search",
+ source_label="Yahoo 購物",
+ dry_run=True,
+ )
+ with engine.connect() as conn:
+ source_count = conn.execute(
+ text("SELECT COUNT(*) FROM external_market_sources")
+ ).scalar_one()
+ offer_count = conn.execute(
+ text("SELECT COUNT(*) FROM external_offers")
+ ).scalar_one()
+ assert preview["status"] == "dry_run"
+ assert preview["selected_count"] == 2
+ assert preview["writes_database_count"] == 0
+ assert source_count == 0
+ assert offer_count == 0
+
+ sync = canary_service.sync_targeted_marketplace_candidates_to_external_offers(
+ engine,
+ candidates,
+ source_code="yahoo_shopping",
+ platform_code="yahoo_shopping_tw",
+ ingestion_method="public_product_jsonld",
+ connector_key="pchome_targeted_yahoo_public_search",
+ source_label="Yahoo 購物",
+ )
+ readback = readback_external_offer_candidates(
+ engine,
+ candidates,
+ source_code="yahoo_shopping",
+ ingestion_method="public_product_jsonld",
+ )
+ activation = canary_service.activate_external_market_source_after_canary(
+ engine,
+ source_code="yahoo_shopping",
+ candidates=candidates,
+ independent_readback=readback,
+ identity=identity,
+ minimum_verified_count=2,
+ )
+
+ assert sync["written_count"] == 2
+ assert readback["success"] is True
+ assert readback["readback_pass_count"] == 2
+ assert activation["status"] == "activated"
+ assert activation["applied"] is True
+ assert bool(activation["observed_state"]["write_enabled"]) is True
+ readiness = build_external_source_readiness(engine)
+ yahoo = next(
+ source for source in readiness["sources"] if source["code"] == "yahoo_shopping"
+ )
+ assert yahoo["runtime_stage"] == "formal"
+ assert yahoo["has_runtime_data"] is True
+ assert yahoo["write_enabled"] is True
+
+ rollback = canary_service.rollback_external_market_source_canary(
+ engine,
+ source_code="yahoo_shopping",
+ ingestion_method="public_product_jsonld",
+ candidates=candidates,
+ identity=identity,
+ activation_receipt=activation,
+ )
+
+ assert rollback["quarantined_offer_count"] == 2
+ assert rollback["source_state_restored"] is True
+ with engine.connect() as conn:
+ source = conn.execute(text("""
+ SELECT status, enabled, write_enabled
+ FROM external_market_sources
+ WHERE code = 'yahoo_shopping'
+ """)).mappings().one()
+ rows = conn.execute(text("""
+ SELECT match_status, data_quality_status
+ FROM external_offers
+ WHERE source_code = 'yahoo_shopping'
+ """)).mappings().all()
+ assert source["status"] == "paused"
+ assert bool(source["enabled"]) is False
+ assert bool(source["write_enabled"]) is False
+ assert {row["match_status"] for row in rows} == {"quarantined"}
+ assert {row["data_quality_status"] for row in rows} == {"needs_review"}
+
+
def test_source_aliases_normalize_pixelrag_platform_codes():
from services.external_market_offer_service import normalize_external_offer_payload
@@ -708,7 +885,23 @@ def test_external_offer_sync_is_registered_in_scheduler():
assert "sync_legacy_momo_reference_offers" in scheduler_source
assert "def run_pchome_growth_momo_backfill_task" in scheduler_source
assert "run_pchome_growth_momo_backfill" in scheduler_source
+ assert "def run_pchome_growth_yahoo_backfill_task" in scheduler_source
+ assert "run_pchome_growth_yahoo_backfill" in scheduler_source
assert "run_external_offer_sync_task" in run_scheduler_source
assert "run_pchome_growth_momo_backfill_task" in run_scheduler_source
+ assert "run_pchome_growth_yahoo_backfill_task" in run_scheduler_source
assert "external_offer_sync" in run_scheduler_source
assert "pchome_growth_momo_backfill" in run_scheduler_source
+ assert "pchome_growth_yahoo_backfill" in run_scheduler_source
+
+
+def test_yahoo_public_offer_readback_route_is_login_protected():
+ from pathlib import Path
+
+ route_source = Path("routes/system_public_routes.py").read_text(encoding="utf-8")
+
+ marker = "@system_public_bp.route('/api/ai-automation/yahoo-shopping-public-offer')"
+ assert marker in route_source
+ route_block = route_source[route_source.index(marker):]
+ assert "@login_required" in route_block[:200]
+ assert "build_yahoo_shopping_runtime_readback" in route_block[:600]
diff --git a/tests/test_frontend_v2_assets.py b/tests/test_frontend_v2_assets.py
index e6e2424..fae564f 100644
--- a/tests/test_frontend_v2_assets.py
+++ b/tests/test_frontend_v2_assets.py
@@ -1265,15 +1265,18 @@ def test_ai_product_pick_agent_uses_real_competitor_data_and_dashboard_action():
agent_actions_source = (ROOT / "services/agent_actions.py").read_text(encoding="utf-8")
assert "def run_pchome_match_backfill_task" in scheduler_source
assert "def run_pchome_growth_momo_backfill_task" in scheduler_source
+ assert "def run_pchome_growth_yahoo_backfill_task" in scheduler_source
assert "_save_stats('pchome_match_backfill'" in scheduler_source
assert "_save_stats('pchome_growth_momo_backfill'" in scheduler_source
assert "retryable_candidate_revalidation_total" in scheduler_source
assert "run_pchome_match_backfill_task" in run_scheduler_source
assert "run_pchome_growth_momo_backfill_task" in run_scheduler_source
+ assert "run_pchome_growth_yahoo_backfill_task" in run_scheduler_source
assert "每日 10:30：pchome_match_backfill" in run_scheduler_source
assert "每日 10:45：pchome_growth_momo_backfill" in run_scheduler_source
assert '"run_pchome_match_backfill_task"' in agent_actions_source
assert '"run_pchome_growth_momo_backfill_task"' in agent_actions_source
+ assert '"run_pchome_growth_yahoo_backfill_task"' in agent_actions_source
assert "產生今日清單" in template
assert "補齊比價資料" in template
diff --git a/tests/test_pchome_revenue_growth_service.py b/tests/test_pchome_revenue_growth_service.py
index 3ce95fd..947bfce 100644
--- a/tests/test_pchome_revenue_growth_service.py
+++ b/tests/test_pchome_revenue_growth_service.py
@@ -647,6 +647,105 @@ def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items()
assert captured["search_limit"] == 2
+def test_backfill_rolls_back_failed_write_before_source_activation(monkeypatch):
+ from services import pchome_growth_momo_backfill_service as service
+
+ engine = create_engine("sqlite:///:memory:")
+ candidate = {
+ "product_id": "Y-ROLLBACK",
+ "target_pchome_product_id": "PCH-ROLLBACK",
+ "auto_compare_type": "total_price",
+ }
+ payload = {
+ "success": True,
+ "stats": {"candidate_validation_count": 1, "unmatched_count": 0},
+ "opportunities": [
+ {
+ "pchome_product_id": "PCH-ROLLBACK",
+ "product_name": "需回滾商品",
+ "pchome_price": 900,
+ "sales_7d": 1000,
+ "external_price": None,
+ "recommended_action": {"code": "map_external_product"},
+ }
+ ],
+ }
+ rollback_calls = []
+
+ monkeypatch.setattr(
+ service,
+ "ensure_evidence_receipt_table",
+ lambda _engine: {"success": True},
+ )
+ monkeypatch.setattr(
+ service,
+ "verify_same_item_candidates",
+ lambda _targets, _candidates, *, identity: {
+ "verified_candidates": [{**candidate, "same_item_reconciliation": identity}],
+ "review_candidates": [],
+ "blocked_candidate_count": 0,
+ },
+ )
+ monkeypatch.setattr(
+ service,
+ "build_coverage_post_verifier",
+ lambda **_kwargs: {
+ "all_checks_passed": False,
+ "status": "readback_failed",
+ "readback_pass_count": 0,
+ "after": {"candidate_validation_count": 1, "unmatched_count": 0},
+ },
+ )
+ monkeypatch.setattr(
+ service,
+ "persist_reconciliation_receipts",
+ lambda *_args, **_kwargs: {"success": True, "readback_count": 1},
+ )
+
+ def rollback(_engine, **kwargs):
+ rollback_calls.append(kwargs)
+ return {
+ "success": True,
+ "status": "rolled_back_and_quarantined",
+ "writes_database_count": 1,
+ "source_state_restored": False,
+ }
+
+ result = service.run_pchome_growth_momo_backfill(
+ engine,
+ limit=1,
+ build_payload_func=lambda _engine, _limit: payload,
+ search_func=lambda _targets, _limit: (True, "found", [candidate]),
+ sync_func=lambda _engine, _candidates: {
+ "success": True,
+ "status": "synced",
+ "written_count": 1,
+ },
+ sync_review_func=lambda _engine, _candidates: {
+ "success": True,
+ "written_count": 0,
+ },
+ readback_func=lambda _engine, _candidates: {
+ "success": False,
+ "readback_pass_count": 0,
+ },
+ source_profile={"work_item_id": "GROWTH-P0-001-C"},
+ source_activation_func=lambda _engine, **_kwargs: {
+ "success": True,
+ "status": "canary_threshold_not_met",
+ "applied": False,
+ "writes_database_count": 0,
+ },
+ source_rollback_func=rollback,
+ )
+
+ assert result["success"] is False
+ assert result["data"]["terminal_status"] == "partial_post_verifier_failed"
+ assert result["data"]["source_rollback"]["status"] == "rolled_back_and_quarantined"
+ assert len(rollback_calls) == 1
+ assert rollback_calls[0]["activation_receipt"]["applied"] is False
+
+
def test_pchome_growth_momo_backfill_default_search_uses_deeper_terms(monkeypatch):
from services import pchome_growth_momo_backfill_service as service
diff --git a/tests/test_pchome_same_item_reconciliation.py b/tests/test_pchome_same_item_reconciliation.py
index 6e960ee..eca5397 100644
--- a/tests/test_pchome_same_item_reconciliation.py
+++ b/tests/test_pchome_same_item_reconciliation.py
@@ -117,6 +117,39 @@ def test_independent_verifier_promotes_exact_and_unit_candidates_but_rejects_con
assert result["blocked_candidates"][0]["reasons"]
+def test_independent_verifier_rejects_failed_source_contract_before_scoring():
+ from services.pchome_growth_same_item_reconciliation import (
+ verify_same_item_candidates,
+ )
+
+ result = verify_same_item_candidates(
+ [_target("P-SHADE", "Estee Lauder 粉持久粉底 1W2", 1690, 1)],
+ [{
+ "product_id": "Y-GENERIC",
+ "name": "Estee Lauder 粉持久粉底",
+ "price": 1590,
+ "target_pchome_product_id": "P-SHADE",
+ "source_contract_guard": {
+ "passed": False,
+ "reasons": ["target_spec_missing_from_source_title"],
+ },
+ }],
+ identity={
+ "trace_id": "trace-shade",
+ "run_id": "run-shade",
+ "work_item_id": "GROWTH-P0-001-C",
+ },
+ )
+
+ assert result["selected_candidate_count"] == 0
+ assert result["blocked_candidate_count"] == 1
+ blocked = result["blocked_candidates"][0]
+ assert blocked["reason_code"] == "source_contract_guard_rejected"
+ assert blocked["source_contract_guard_reasons"] == [
+ "target_spec_missing_from_source_title"
+ ]
+
+
def test_independent_verifier_prefers_exact_sheet_pack_over_unit_candidate():
from services.pchome_growth_same_item_reconciliation import (
verify_same_item_candidates,
diff --git a/tests/test_run_scheduler_embed_consistency.py b/tests/test_run_scheduler_embed_consistency.py
index 4d39f9e..351cc68 100644
--- a/tests/test_run_scheduler_embed_consistency.py
+++ b/tests/test_run_scheduler_embed_consistency.py
@@ -232,6 +232,7 @@ def test_roi_ai_smoke_and_daily_report_schedules_stay_staggered():
assert 'schedule.every().day.at("10:30").do(run_pchome_match_backfill_task)' in source
assert 'schedule.every(30).minutes.do(run_revenue_automation_cycle_background_task)' in source
assert 'schedule.every().day.at("10:45").do(run_pchome_growth_momo_backfill_background_task)' in source
+ assert 'schedule.every(6).hours.do(run_pchome_growth_yahoo_backfill_background_task)' in source
assert "start_revenue_automation_watchdog()" in source
assert "schedule.every(6).hours.do(run_action_plan_hygiene_task)" in source
assert "schedule.every(15).minutes.do(run_ollama_111_usage_guard_check)" in source
diff --git a/tests/test_yahoo_shopping_public_offer_service.py b/tests/test_yahoo_shopping_public_offer_service.py
new file mode 100644
index 0000000..fb330d2
--- /dev/null
+++ b/tests/test_yahoo_shopping_public_offer_service.py
@@ -0,0 +1,511 @@
+import json
+
+
+class _FakeResponse:
+ def __init__(self, url, body, status_code=200):
+ self.url = url
+ self.content = body.encode("utf-8")
+ self.status_code = status_code
+
+
+def _search_html(*, product_id="1001", title="LANCOME 蘭蔻 零粉感水凝粉餅 10g #01"):
+ payload = {
+ "search": {
+ "ecsearch": {
+ "hits": [
+ {
+ "ec_productid": product_id,
+ "ec_title": title,
+ "ec_item_url": (
+ f"https://tw.buy.yahoo.com/gdsale/product-{product_id}.html"
+ ),
+ "ec_price": "1890",
+ "ec_listprice": "2100",
+ "ec_image": "https://img.example.test/1001.jpg",
+ "ec_brand": "LANCOME",
+ "ec_storename": "Yahoo購物中心",
+ "ec_has_stock": "1",
+ "ec_quality_rating": "4.9",
+ "ec_rating_count": "18",
+ }
+ ]
+ }
+ }
+ }
+ return (
+ "
Yahoo 搜尋"
+ f""
+ ""
+ )
+
+
+def _product_html(*, product_id="1001", title="LANCOME 蘭蔻 零粉感水凝粉餅 10g #01"):
+ payload = {
+ "@context": "https://schema.org",
+ "@type": "Product",
+ "sku": product_id,
+ "name": f"{title} | Yahoo購物中心",
+ "brand": {"@type": "Brand", "name": "LANCOME"},
+ "gtin13": "3614271234567",
+ "image": ["https://img.example.test/1001.jpg"],
+ "offers": {
+ "@type": "Offer",
+ "price": "1890",
+ "priceCurrency": "TWD",
+ "url": f"https://tw.buy.yahoo.com/gdsale/gdsale.asp?gdid={product_id}",
+ "seller": {"@type": "Organization", "name": "Yahoo購物中心"},
+ },
+ "aggregateRating": {
+ "@type": "AggregateRating",
+ "ratingValue": "4.9",
+ "ratingCount": "18",
+ },
+ }
+ return (
+ "商品頁"
+ f""
+ ""
+ )
+
+
+def test_yahoo_public_parsers_extract_search_and_product_jsonld():
+ from services.yahoo_shopping_public_offer_service import (
+ parse_yahoo_product_jsonld,
+ parse_yahoo_search_html,
+ )
+
+ search_url = "https://tw.buy.yahoo.com/search/product?p=lancome"
+ product_url = "https://tw.buy.yahoo.com/gdsale/product-1001.html"
+ search = parse_yahoo_search_html(_search_html(), search_url=search_url)
+ product = parse_yahoo_product_jsonld(
+ _product_html(),
+ product_url=product_url,
+ )
+
+ assert search["success"] is True
+ assert search["candidate_count"] == 1
+ assert search["candidates"][0]["product_id"] == "1001"
+ assert search["candidates"][0]["search_price"] == 1890
+ assert search["content_sha256"]
+ assert product["success"] is True
+ assert product["product"]["product_id"] == "1001"
+ assert product["product"]["price"] == 1890
+ assert product["product"]["currency"] == "TWD"
+ assert product["product"]["gtin13"] == "3614271234567"
+ assert product["product_fingerprint_sha256"]
+
+
+def test_yahoo_source_contract_rejects_missing_pchome_spec():
+ from services.yahoo_shopping_public_offer_service import _source_contract_guard
+
+ product_url = "https://tw.buy.yahoo.com/gdsale/product-1001.html"
+ guard = _source_contract_guard(
+ target={
+ "target_public_evidence": {
+ "spec": "1W2",
+ "current_price": 1690,
+ }
+ },
+ search_candidate={"product_id": "1001"},
+ product_readback={
+ "success": True,
+ "product": {
+ "product_id": "1001",
+ "name": "Estee Lauder 粉持久完美持妝粉底",
+ "price": 1590,
+ "currency": "TWD",
+ "product_url": product_url,
+ },
+ },
+ )
+
+ assert guard["passed"] is False
+ assert "target_spec_missing_from_source_title" in guard["reasons"]
+ assert guard["direct_price_write_allowed"] is False
+
+
+def test_yahoo_source_contract_rejects_multi_variant_group_even_when_spec_appears():
+ from services.yahoo_shopping_public_offer_service import _source_contract_guard
+
+ product_url = "https://tw.buy.yahoo.com/gdsale/product-2002.html"
+ guard = _source_contract_guard(
+ target={
+ "target_public_evidence": {
+ "spec": "無香",
+ "current_price": 179,
+ }
+ },
+ search_candidate={"product_id": "2002"},
+ product_readback={
+ "success": True,
+ "product": {
+ "product_id": "2002",
+ "name": "GATSBY 魔法激凍體用噴霧170ml(海洋冰封/無香薄荷)",
+ "price": 169,
+ "currency": "TWD",
+ "product_url": product_url,
+ },
+ },
+ )
+
+ assert guard["passed"] is False
+ assert "source_multi_variant_listing" in guard["reasons"]
+
+
+def test_yahoo_source_contract_rejects_unavailable_target_or_source_stock():
+ from services.yahoo_shopping_public_offer_service import _source_contract_guard
+
+ product_url = "https://tw.buy.yahoo.com/gdsale/product-3003.html"
+ guard = _source_contract_guard(
+ target={
+ "target_public_evidence": {
+ "spec": "100ml",
+ "current_price": 975,
+ "quantity": 0,
+ }
+ },
+ search_candidate={"product_id": "3003", "stock_status": "out_of_stock"},
+ product_readback={
+ "success": True,
+ "product": {
+ "product_id": "3003",
+ "name": "施巴 抗乾敏滋潤浴露100ml",
+ "price": 299,
+ "currency": "TWD",
+ "product_url": product_url,
+ },
+ },
+ )
+
+ assert guard["passed"] is False
+ assert "pchome_stock_not_available" in guard["reasons"]
+ assert "source_stock_not_confirmed" in guard["reasons"]
+
+
+def test_yahoo_source_contract_rejects_sales_report_price_fallback():
+ from services.yahoo_shopping_public_offer_service import _source_contract_guard
+
+ product_url = "https://tw.buy.yahoo.com/gdsale/product-4004.html"
+ guard = _source_contract_guard(
+ target={
+ "target_public_evidence": {
+ "spec": "100ml",
+ "current_price": 975,
+ "current_price_field": "sales_report_fallback",
+ }
+ },
+ search_candidate={"product_id": "4004", "stock_status": "in_stock"},
+ product_readback={
+ "success": True,
+ "product": {
+ "product_id": "4004",
+ "name": "施巴 抗乾敏滋潤浴露100ml",
+ "price": 999,
+ "currency": "TWD",
+ "product_url": product_url,
+ },
+ },
+ )
+
+ assert guard["passed"] is False
+ assert "pchome_current_price_missing" in guard["reasons"]
+
+
+def test_yahoo_source_contract_accepts_product_detail_instock_evidence():
+ from services.yahoo_shopping_public_offer_service import _source_contract_guard
+
+ product_url = "https://tw.buy.yahoo.com/gdsale/product-5005.html"
+ guard = _source_contract_guard(
+ target={
+ "target_public_evidence": {
+ "spec": "100ml",
+ "current_price": 975,
+ "current_price_field": "P",
+ }
+ },
+ search_candidate={"product_id": "5005", "stock_status": "unknown"},
+ product_readback={
+ "success": True,
+ "product": {
+ "product_id": "5005",
+ "name": "施巴 抗乾敏滋潤浴露100ml",
+ "price": 999,
+ "currency": "TWD",
+ "availability": "https://schema.org/InStock",
+ "product_url": product_url,
+ },
+ },
+ )
+
+ assert guard["passed"] is True
+ assert "source_stock_not_confirmed" not in guard["reasons"]
+
+
+def test_public_source_stream_reader_stops_at_byte_limit():
+ from services.public_source_http_service import read_bounded_response_text
+
+ class StreamingResponse:
+ status_code = 200
+ headers = {}
+ closed = False
+
+ def iter_content(self, *, chunk_size):
+ assert chunk_size > 0
+ yield b"1234"
+ yield b"5678"
+
+ def close(self):
+ self.closed = True
+
+ response = StreamingResponse()
+
+ try:
+ read_bounded_response_text(response, max_bytes=6)
+ except ValueError as exc:
+ assert str(exc) == "public_source_response_too_large"
+ else:
+ raise AssertionError("oversized public response was accepted")
+ assert response.closed is True
+
+
+def test_yahoo_search_builds_guarded_candidate_from_two_public_sources(monkeypatch):
+ from services.yahoo_shopping_public_offer_service import (
+ search_yahoo_candidates_for_pchome_products,
+ )
+
+ monkeypatch.setenv("PCHOME_GROWTH_YAHOO_MAX_TERMS_PER_TARGET", "1")
+ monkeypatch.setenv("PCHOME_GROWTH_YAHOO_DETAIL_CANDIDATES_PER_TARGET", "1")
+ pchome_payload = {
+ "PCH-1": {
+ "Id": "PCH-1",
+ "Name": "LANCOME 蘭蔻 零粉感水凝粉餅 10g",
+ "Spec": "#01",
+ "Price": {"P": 1690, "M": 1890},
+ "Qty": 3,
+ }
+ }
+
+ def fake_get(url, *, timeout):
+ assert timeout > 0
+ if "ecapi.pchome.com.tw" in url:
+ body = f"try{{jsonp({json.dumps(pchome_payload)});}}catch(e){{}}"
+ return _FakeResponse(url, body)
+ if "/search/product" in url:
+ return _FakeResponse(url, _search_html())
+ if "/gdsale/" in url:
+ return _FakeResponse(url, _product_html())
+ raise AssertionError(url)
+
+ success, message, candidates = search_yahoo_candidates_for_pchome_products(
+ [
+ {
+ "product_id": "PCH-1",
+ "name": "LANCOME 蘭蔻 零粉感水凝粉餅 10g",
+ "price": 1650,
+ }
+ ],
+ 1,
+ request_get=fake_get,
+ )
+
+ assert success is True
+ assert "1 筆商品頁候選" in message
+ assert len(candidates) == 1
+ candidate = candidates[0]
+ assert candidate["target_pchome_price"] == 1690
+ assert candidate["target_public_evidence"]["spec"] == "#01"
+ assert candidate["target_public_evidence"]["current_price_field"] == "P"
+ assert candidate["target_public_evidence"]["price_fields"] == {
+ "P": 1690.0,
+ "M": 1890.0,
+ }
+ assert candidate["source_contract_guard"]["passed"] is True
+ assert candidate["source_provenance"]["content_sha256"]
+ assert candidate["source_provenance"]["product_content_sha256"]
+ assert candidate["auto_compare_type"] == "total_price"
+
+
+def test_yahoo_runtime_readback_uses_fresh_compact_receipt(tmp_path):
+ from services.yahoo_shopping_public_offer_service import (
+ _write_compact_receipt,
+ build_yahoo_shopping_runtime_readback,
+ )
+
+ identity = {
+ "trace_id": "trace-yahoo",
+ "run_id": "run-yahoo",
+ "work_item_id": "GROWTH-P0-001-C",
+ }
+ result = {
+ "success": True,
+ "data": {
+ "identity": identity,
+ "terminal_status": "verified_write_readback",
+ "next_machine_action": "continue_next_revenue_weighted_batch",
+ "target_count": 2,
+ "candidate_count": 2,
+ "auto_compare_count": 2,
+ "review_count": 0,
+ "external_offer_sync": {
+ "written_count": 2,
+ "writes_database_count": 2,
+ },
+ "source_activation": {"applied": True, "writes_database_count": 1},
+ "source_rollback": {"writes_database_count": 0},
+ "post_verifier": {"readback_pass_count": 2},
+ "receipt_persistence": {"success": True},
+ "reconciliation_receipt": {
+ "source_receipt": {"source": "public_search"},
+ "candidate_verification": {
+ "verified_candidates": [
+ {
+ "target_pchome_product_id": "PCH-1",
+ "product_id": "1001",
+ "product_url": "https://tw.buy.yahoo.com/gdsale/product-1001.html",
+ "price": 1890,
+ "currency": "TWD",
+ "target_match_score": 1.0,
+ "same_item_candidate_fingerprint": "candidate-fingerprint",
+ "source_provenance": {
+ "product_fingerprint_sha256": "product-fingerprint"
+ },
+ }
+ ]
+ },
+ },
+ },
+ }
+
+ persisted = _write_compact_receipt(result, receipt_root=tmp_path)
+ readback = build_yahoo_shopping_runtime_readback(receipt_root=tmp_path)
+
+ assert persisted["success"] is True
+ assert readback["status"] == "ready"
+ assert readback["has_candidate_runtime_data"] is True
+ assert readback["candidate_offer_count"] == 2
+ assert readback["verified_candidate_count"] == 2
+ assert readback["formal_source_activated"] is True
+ assert readback["writes_database_count"] == 0
+
+
+def test_yahoo_runtime_forces_check_mode_before_controlled_apply(monkeypatch):
+ from services import external_marketplace_canary_service as canary_service
+ from services import pchome_growth_momo_backfill_service as backfill_service
+ from services import yahoo_shopping_public_offer_service as runtime
+
+ calls = []
+
+ def fake_sync(_engine, candidates, *, dry_run, **_kwargs):
+ calls.append(dry_run)
+ return {
+ "success": True,
+ "status": "dry_run" if dry_run else "synced",
+ "candidate_count": len(candidates),
+ "selected_count": len(candidates),
+ "written_count": 0 if dry_run else len(candidates),
+ "writes_database_count": 0 if dry_run else len(candidates),
+ }
+
+ def fake_backfill(_engine, **kwargs):
+ sync = kwargs["sync_func"](_engine, [{"product_id": "Y-1"}])
+ return {
+ "success": True,
+ "data": {
+ "external_offer_sync": sync,
+ "identity": {"run_id": "run-yahoo-check-mode"},
+ "reconciliation_receipt": {},
+ },
+ }
+
+ monkeypatch.setattr(
+ canary_service,
+ "sync_targeted_marketplace_candidates_to_external_offers",
+ fake_sync,
+ )
+ monkeypatch.setattr(
+ backfill_service,
+ "run_pchome_growth_momo_backfill",
+ fake_backfill,
+ )
+ monkeypatch.setattr(
+ runtime,
+ "_write_compact_receipt",
+ lambda _result, receipt_root: {
+ "success": True,
+ "receipt_path": str(receipt_root),
+ },
+ )
+
+ result = runtime.run_pchome_growth_yahoo_backfill(
+ object(),
+ receipt_root="/tmp/yahoo-check-mode-test",
+ )
+
+ assert calls == [True, False]
+ sync = result["data"]["external_offer_sync"]
+ assert sync["written_count"] == 1
+ assert sync["check_mode_receipt"]["status"] == "dry_run"
+ assert sync["check_mode_receipt"]["writes_database_count"] == 0
+
+
+def test_yahoo_scheduler_adapter_records_terminal_receipt(monkeypatch):
+ import sqlalchemy
+ from services import pchome_growth_yahoo_scheduler_task as scheduler_task
+ from services import yahoo_shopping_public_offer_service as runtime
+
+ captured = {"stats": []}
+
+ class FakeEngine:
+ def dispose(self):
+ captured["disposed"] = True
+
+ monkeypatch.setenv("PCHOME_GROWTH_YAHOO_BACKFILL_LIMIT", "5")
+ monkeypatch.setattr(sqlalchemy, "create_engine", lambda _path: FakeEngine())
+ monkeypatch.setattr(
+ runtime,
+ "run_pchome_growth_yahoo_backfill",
+ lambda engine, limit: {
+ "success": True,
+ "message": "verified",
+ "data": {
+ "identity": {
+ "trace_id": "trace-scheduler",
+ "run_id": "run-scheduler",
+ "work_item_id": "GROWTH-P0-001-C",
+ },
+ "terminal_status": "verified_with_coverage_gain",
+ "next_machine_action": "continue_next_revenue_weighted_batch",
+ "scanned_products": limit,
+ "candidate_count": 4,
+ "auto_compare_count": 3,
+ "review_count": 1,
+ "external_offer_sync": {"written_count": 3},
+ "post_verifier": {
+ "readback_pass_count": 3,
+ "status": "verified",
+ },
+ "source_activation": {
+ "status": "activated",
+ "applied": True,
+ },
+ "artifact_receipt": {
+ "status": "persisted_and_verified",
+ "receipt_sha256": "receipt-hash",
+ },
+ },
+ },
+ )
+
+ result = scheduler_task.run_scheduled_pchome_growth_yahoo_backfill(
+ save_stats=lambda key, value: captured["stats"].append((key, value))
+ )
+
+ assert result["success"] is True
+ assert captured["disposed"] is True
+ key, stats = captured["stats"][0]
+ assert key == "pchome_growth_yahoo_backfill"
+ assert stats["status"] == "Success"
+ assert stats["work_item_id"] == "GROWTH-P0-001-C"
+ assert stats["readback_pass_count"] == 3
+ assert stats["source_activated"] is True
+ assert stats["artifact_sha256"] == "receipt-hash"