from pathlib import Path from services import pchome_mapping_backlog_service as compatibility_facade from services.pchome_mapping_backlog import ( artifacts, contracts, controlled_apply_executor, controlled_apply_receipt, controlled_apply_verifier, enrichment, evidence, policies, reporter, ) ROOT = Path(__file__).resolve().parents[1] def test_policy_registry_is_extracted_without_changing_legacy_exports(): assert len(policies.__all__) == 96 assert compatibility_facade.BACKLOG_POLICY == policies.BACKLOG_POLICY assert ( compatibility_facade.AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_PACKAGE_POLICY == policies.AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_PACKAGE_POLICY ) assert compatibility_facade.PCHOME_FETCH_ALLOWED_DOMAIN == "24h.pchome.com.tw" def test_evidence_and_contract_functions_keep_legacy_import_identity(): assert ( compatibility_facade.parse_pchome_product_page_evidence_html is evidence.parse_pchome_product_page_evidence_html ) assert compatibility_facade.parse_unit_package_basis is evidence.parse_unit_package_basis assert ( compatibility_facade._ai_exception_compatibility_fields is contracts._ai_exception_compatibility_fields ) def test_extracted_unit_parser_preserves_ai_exception_and_no_write_contract(): parsed = compatibility_facade.parse_unit_package_basis("理膚寶水 B5 40ml x2 超值組") assert parsed["estimated_total_quantity"] == 80 assert parsed["ai_exception_required"] is True assert parsed["primary_human_gate_count"] == 0 assert parsed["writes_database"] is False assert parsed["fetches_external_sites"] is False def test_controlled_apply_executor_helpers_keep_legacy_import_identity(): assert ( compatibility_facade._canonical_retry_exception_artifact_bytes is artifacts._canonical_retry_exception_artifact_bytes ) assert ( compatibility_facade._selector_to_pchome_match_params is controlled_apply_executor._selector_to_pchome_match_params ) def test_controlled_apply_executor_core_is_no_write_without_apply_request(tmp_path): preflight = { "policy": "controlled_apply_preflight_fixture", "success": True, "generated_at": "2026-07-11T12:00:00+08:00", "stats": {}, "backlog": {}, "summary": {"mutation_plan_count": 1}, "controlled_apply_preflight": { "preflight_id": "preflight-1", "run_id": "run-1", "ready_for_controlled_apply_executor": True, }, "target_selectors": [ { "selector_id": "selector-1", "momo_product_id": "MOMO-1", "momo_product_name": "Example 40ml", "target_pchome_product_id": "PCH-1", "target_pchome_product_name": "Example 40ml", "target_match_score": 0.91, } ], } result = controlled_apply_executor.build_controlled_apply_executor_from_preflight( preflight, artifact_root=tmp_path, ) assert result["result"] == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_READY" assert result["summary"]["target_selector_count"] == 1 assert result["summary"]["applied_record_count"] == 0 assert result["summary"]["rollback_step_count"] == 1 assert result["safety"]["execute_apply"] is False assert result["safety"]["writes_database_count"] == 0 assert result["safety"]["writes_artifact_count"] == 0 def test_controlled_apply_drift_verifier_core_is_no_write(tmp_path): replay = { "result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED", "summary": { "target_selector_count": 1, "post_apply_readback_count": 1, "post_apply_readback_pass_count": 1, "executor_receipt_hash_match_count": 1, "missing_artifact_count": 0, }, "post_apply_readbacks": [ { "selector_id": "selector-1", "momo_icode": "MOMO-1", "expected_pchome_id": "PCH-1", "actual_pchome_id": "PCH-1", "passed": True, "writes_database": False, } ], "receipt_replay": {"run_id": "run-1"}, "missing_artifacts": [], } result = controlled_apply_verifier.build_controlled_apply_drift_verifier_from_replay( replay, root=tmp_path, source_receipt_replay=replay, ) assert result["result"] == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED" assert result["success"] is True assert result["summary"]["drift_count"] == 0 assert result["summary"]["drift_verified_count"] == 1 assert result["all_checks_passed"] is True assert result["safety"]["writes_database_count"] == 0 assert result["safety"]["writes_artifact_count"] == 0 def test_controlled_apply_receipt_replay_keeps_identity_and_no_write_default(tmp_path): assert ( compatibility_facade.build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package is controlled_apply_receipt.build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package ) assert ( compatibility_facade._find_retry_exception_artifact_file is controlled_apply_receipt._find_retry_exception_artifact_file ) result = controlled_apply_receipt.build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package( artifact_root=tmp_path, run_id="missing-run", ) assert result["result"] == "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_REPLAY_ARTIFACTS" assert result["summary"]["missing_artifact_count"] == 3 assert result["summary"]["writes_database_count"] == 0 assert result["materialized_executor_artifacts"] == [] assert result["safety"]["reads_database"] is False assert result["safety"]["writes_database"] is False assert result["safety"]["writes_artifact_count"] == 0 assert len(result["executor_receipt_artifact"]["payload_sha256"]) == 64 def test_reporter_keeps_legacy_identity_and_read_only_preview(): assert ( compatibility_facade.summarize_pchome_mapping_backlog is reporter.summarize_pchome_mapping_backlog ) assert ( compatibility_facade.build_pchome_mapping_operator_preview is reporter.build_pchome_mapping_operator_preview ) assert compatibility_facade.compact_mapping_item is reporter.compact_mapping_item assert ( compatibility_facade._evidence_completeness is reporter._evidence_completeness ) assert ( compatibility_facade._build_external_benchmark_alignment is reporter._build_external_benchmark_alignment ) assert ( compatibility_facade._build_ai_automation_plan is reporter._build_ai_automation_plan ) payload = { "success": True, "generated_at": "2026-07-11T14:00:00+08:00", "stats": {}, "opportunities": [], } summary = reporter.summarize_pchome_mapping_backlog(payload) preview = reporter.build_pchome_mapping_operator_preview(payload) assert summary["result"] == "PASS" assert preview["result"] == "NO_DIRECT_MAPPING_TARGETS" assert preview["safety"]["read_only_preview"] is True assert preview["safety"]["executes_search"] is False assert preview["safety"]["writes_database"] is False def test_enrichment_keeps_legacy_identity_and_no_write_preview(): assert ( compatibility_facade.build_pchome_evidence_enrichment_preview is enrichment.build_pchome_evidence_enrichment_preview ) assert ( compatibility_facade.build_pchome_evidence_source_preview is enrichment.build_pchome_evidence_source_preview ) assert ( compatibility_facade._build_evidence_task is enrichment._build_evidence_task ) assert ( compatibility_facade._build_fetch_gate_candidates is enrichment._build_fetch_gate_candidates ) assert ( compatibility_facade._field_enrichment_sources is enrichment._field_enrichment_sources ) assert ( compatibility_facade._source_plan_for_field is enrichment._source_plan_for_field ) payload = { "success": True, "generated_at": "2026-07-11T15:30:00+08:00", "stats": {}, "opportunities": [], } preview = enrichment.build_pchome_evidence_enrichment_preview(payload) source_preview = enrichment.build_pchome_evidence_source_preview(payload) assert preview["result"] == "NO_TARGETS" assert source_preview["result"] == "NO_TARGETS" assert preview["safety"]["fetches_external_sites"] is False assert preview["safety"]["writes_database"] is False assert source_preview["safety"]["fetches_external_sites"] is False assert source_preview["safety"]["writes_database"] is False def test_legacy_facade_is_smaller_after_sixth_p0_extraction(): facade = ROOT / "services/pchome_mapping_backlog_service.py" assert sum(1 for _ in facade.open(encoding="utf-8")) < 42_900