1223 lines
53 KiB
Python
1223 lines
53 KiB
Python
import json
|
||
|
||
|
||
def _auto_policy_guard_history_check(status="ok"):
|
||
return {
|
||
"name": "PChome auto-policy authorization guard",
|
||
"status": status,
|
||
"summary": "PChome auto-policy guard 已自動讀回 no-write/no-secret package",
|
||
"details": {
|
||
"policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_lane_guard",
|
||
"result": "WAITING_FOR_DB_APPLY_AUTHORIZATION_REQUEST_CLOSEOUT",
|
||
"guard_ready_count": 0,
|
||
"lane_guard_check_count": 12,
|
||
"lane_guard_pass_count": 11,
|
||
"lane_guard_waiting_count": 1,
|
||
"authorization_request_closeout_ready_count": 0,
|
||
"exact_request_payload_field_count": 10,
|
||
"machine_request_manifest_step_count": 6,
|
||
"lane_entry_requirement_count": 6,
|
||
"reads_secret_count": 0,
|
||
"executes_sql_count": 0,
|
||
"writes_database_count": 0,
|
||
"primary_human_gate_count": 0,
|
||
"manual_review_mode": "exception_only",
|
||
"machine_verifiable": True,
|
||
"ready_for_database_apply_now": False,
|
||
"issues_database_apply_authorization": False,
|
||
"next_machine_action": "collect_machine_fetch_evidence_for_authorization_guard",
|
||
},
|
||
}
|
||
|
||
|
||
def _auto_policy_decision_preflight_history_check(status="ok"):
|
||
return {
|
||
"name": "PChome auto-policy decision preflight",
|
||
"status": status,
|
||
"summary": "PChome auto-policy decision preflight 已用本機 machine evidence 自動通過",
|
||
"details": {
|
||
"policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_decision_preflight",
|
||
"result": "DB_APPLY_AUTHORIZATION_DECISION_PREFLIGHT_READY",
|
||
"decision_preflight_ready_count": 1,
|
||
"authorization_lane_guard_ready_count": 1,
|
||
"decision_preflight_check_count": 12,
|
||
"decision_preflight_pass_count": 12,
|
||
"decision_preflight_waiting_count": 0,
|
||
"decision_input_requirement_count": 8,
|
||
"decision_rejection_reason_count": 10,
|
||
"machine_fetch_evidence_ready_count": 1,
|
||
"reads_secret_count": 0,
|
||
"executes_sql_count": 0,
|
||
"writes_database_count": 0,
|
||
"primary_human_gate_count": 0,
|
||
"manual_review_mode": "exception_only",
|
||
"payload_source": "local_contract_fixture",
|
||
"execute_fetch": True,
|
||
"outbound_network": False,
|
||
"business_data_source": False,
|
||
"ready_for_database_apply_now": False,
|
||
"issues_database_apply_authorization": False,
|
||
"next_machine_action": "continue_to_pchome_auto_policy_decision_closeout_lane",
|
||
},
|
||
}
|
||
|
||
|
||
def _auto_policy_decision_closeout_history_check(status="ok"):
|
||
return {
|
||
"name": "PChome auto-policy decision closeout",
|
||
"status": status,
|
||
"summary": "PChome auto-policy decision closeout 已自動完成",
|
||
"details": {
|
||
"policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_decision_closeout",
|
||
"result": "DB_APPLY_AUTHORIZATION_DECISION_CLOSEOUT_READY",
|
||
"decision_closeout_ready_count": 1,
|
||
"decision_preflight_ready_count": 1,
|
||
"decision_closeout_check_count": 12,
|
||
"decision_closeout_pass_count": 12,
|
||
"decision_closeout_waiting_count": 0,
|
||
"post_apply_verifier_required_count": 1,
|
||
"same_run_truth_required_count": 1,
|
||
"reads_secret_count": 0,
|
||
"executes_sql_count": 0,
|
||
"writes_database_count": 0,
|
||
"primary_human_gate_count": 0,
|
||
"manual_review_mode": "exception_only",
|
||
"payload_source": "local_contract_fixture",
|
||
"execute_fetch": True,
|
||
"outbound_network": False,
|
||
"business_data_source": False,
|
||
"ready_for_database_apply_now": False,
|
||
"issues_database_apply_authorization": False,
|
||
"permits_future_authorization_decision_lane": True,
|
||
"next_machine_action": "continue_to_pchome_auto_policy_authorization_issuer_gate",
|
||
},
|
||
}
|
||
|
||
|
||
def test_event_router_smoke_reports_queued_deliveries(tmp_path, monkeypatch):
|
||
from services import ai_automation_metrics as metrics
|
||
from services import ai_automation_smoke_service as smoke
|
||
from services import event_router
|
||
|
||
queue_path = tmp_path / "failed_deliveries.jsonl"
|
||
queue_path.write_text('{"event_key":"a"}\n{"event_key":"b"}\n', encoding="utf-8")
|
||
monkeypatch.setattr(event_router, "_QUEUE_PATH", str(queue_path))
|
||
metrics.reset_for_tests()
|
||
metrics.record_event_router_dispatch(
|
||
tier="L1",
|
||
event_type="crawler_timeout",
|
||
delivered=False,
|
||
queued=True,
|
||
latency_ms=12,
|
||
)
|
||
|
||
result = smoke._event_router_check()
|
||
|
||
assert result["status"] == "warning"
|
||
assert result["details"]["queued_deliveries"] == 2
|
||
assert result["details"]["dispatch_metric_total"] == 1
|
||
assert result["details"]["dispatch_sync"] is True
|
||
|
||
|
||
def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
monkeypatch.setattr(smoke, "_event_router_check", lambda: smoke._check("event", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_gemini_egress_check", lambda: smoke._check("gemini", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_autoheal_check", lambda: smoke._check("autoheal", "warning", "warn"))
|
||
monkeypatch.setattr(smoke, "_nemotron_check", lambda: smoke._check("nemotron", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_embedding_queue_check", lambda: smoke._check("embedding", "critical", "boom"))
|
||
monkeypatch.setattr(smoke, "_elephant_hitl_check", lambda: smoke._check("elephant", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_controlled_apply_drift_monitor_check", lambda: smoke._check("pchome", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_auto_policy_authorization_guard_check", lambda: smoke._check("auto-policy", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_auto_policy_decision_preflight_check", lambda: smoke._check("auto-policy preflight", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_auto_policy_decision_closeout_check", lambda: smoke._check("auto-policy closeout", "ok", "ok"))
|
||
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"))
|
||
|
||
result = smoke.collect_ai_automation_smoke(record_history=False)
|
||
|
||
assert result["status"] == "critical"
|
||
assert result["summary"] == {"ok": 11, "warning": 1, "critical": 1, "total": 13}
|
||
|
||
|
||
def test_pchome_controlled_apply_drift_monitor_reports_verified_zero_drift(monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
from services import pchome_mapping_backlog_service as backlog
|
||
|
||
class FakeEngine:
|
||
disposed = False
|
||
|
||
def dispose(self):
|
||
self.disposed = True
|
||
|
||
fake_engine = FakeEngine()
|
||
monkeypatch.setattr(smoke, "_create_pchome_drift_monitor_engine", lambda _path: fake_engine)
|
||
monkeypatch.setattr(
|
||
backlog,
|
||
"build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package",
|
||
lambda **_kwargs: {
|
||
"result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED",
|
||
"summary": {
|
||
"target_selector_count": 4,
|
||
"post_apply_readback_pass_count": 4,
|
||
"executor_receipt_hash_match_count": 1,
|
||
},
|
||
},
|
||
)
|
||
monkeypatch.setattr(
|
||
backlog,
|
||
"build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package",
|
||
lambda **_kwargs: {
|
||
"result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED",
|
||
"summary": {
|
||
"target_selector_count": 4,
|
||
"post_apply_readback_pass_count": 4,
|
||
"drift_count": 0,
|
||
"drift_verified_count": 1,
|
||
"drift_verifier_artifact_materialized_count": 1,
|
||
"drift_verifier_artifact_hash_match_count": 1,
|
||
"writes_database_count": 0,
|
||
},
|
||
},
|
||
)
|
||
|
||
result = smoke._pchome_controlled_apply_drift_monitor_check()
|
||
|
||
assert result["status"] == "ok"
|
||
assert result["details"]["selector_count"] == 4
|
||
assert result["details"]["drift_count"] == 0
|
||
assert result["details"]["writes_database"] is False
|
||
assert result["details"]["materialize_artifacts"] is False
|
||
assert fake_engine.disposed is True
|
||
|
||
|
||
def test_pchome_auto_policy_authorization_guard_reports_no_write_exception_only(monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
from services import pchome_mapping_backlog_service as backlog
|
||
from services import pchome_revenue_growth_service as growth
|
||
|
||
class FakeEngine:
|
||
disposed = False
|
||
|
||
def dispose(self):
|
||
self.disposed = True
|
||
|
||
fake_engine = FakeEngine()
|
||
monkeypatch.setattr(smoke, "_create_pchome_drift_monitor_engine", lambda _path: fake_engine)
|
||
monkeypatch.setattr(growth, "build_pchome_growth_opportunities", lambda _engine, limit: {"success": True})
|
||
monkeypatch.setattr(
|
||
backlog,
|
||
"build_pchome_auto_policy_db_apply_authorization_lane_guard",
|
||
lambda *_args, **_kwargs: {
|
||
"policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_lane_guard",
|
||
"result": "WAITING_FOR_DB_APPLY_AUTHORIZATION_REQUEST_CLOSEOUT",
|
||
"source_policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_request_closeout",
|
||
"summary": {
|
||
"authorization_lane_guard_ready_count": 0,
|
||
"authorization_request_closeout_ready_count": 0,
|
||
"lane_guard_check_count": 12,
|
||
"lane_guard_pass_count": 11,
|
||
"lane_guard_waiting_count": 1,
|
||
"exact_request_payload_field_count": 10,
|
||
"machine_request_manifest_step_count": 6,
|
||
"lane_entry_requirement_count": 6,
|
||
"reads_secret_count": 0,
|
||
"executes_sql_count": 0,
|
||
"writes_database_count": 0,
|
||
"primary_human_gate_count": 0,
|
||
"manual_review_required_count": 0,
|
||
},
|
||
"future_authorization_lane_guard": {
|
||
"ready_for_database_apply_now": False,
|
||
"issues_database_apply_authorization": False,
|
||
"requires_fresh_production_truth_in_same_run": True,
|
||
},
|
||
"lane_transfer_contract": {
|
||
"machine_verifiable": True,
|
||
"writes_database": False,
|
||
"issues_database_apply_authorization": False,
|
||
},
|
||
"safety": {
|
||
"reads_secret_in_preview": False,
|
||
"executes_sql": False,
|
||
"writes_database": False,
|
||
"manual_review_mode": "exception_only",
|
||
},
|
||
},
|
||
)
|
||
|
||
result = smoke._pchome_auto_policy_authorization_guard_check()
|
||
|
||
assert result["status"] == "ok"
|
||
assert result["details"]["lane_guard_pass_count"] == 11
|
||
assert result["details"]["primary_human_gate_count"] == 0
|
||
assert result["details"]["writes_database_count"] == 0
|
||
assert result["details"]["next_machine_action"] == "collect_machine_fetch_evidence_for_authorization_guard"
|
||
assert fake_engine.disposed is True
|
||
|
||
|
||
def test_pchome_auto_policy_decision_preflight_reports_machine_fetch_no_write(monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
from services import pchome_mapping_backlog_service as backlog
|
||
captured = {}
|
||
|
||
def fake_preflight(*_args, **kwargs):
|
||
captured.update(kwargs)
|
||
return {
|
||
"policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_decision_preflight",
|
||
"result": "DB_APPLY_AUTHORIZATION_DECISION_PREFLIGHT_READY",
|
||
"source_policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_lane_guard",
|
||
"summary": {
|
||
"authorization_decision_preflight_ready_count": 1,
|
||
"authorization_lane_guard_ready_count": 1,
|
||
"decision_preflight_check_count": 12,
|
||
"decision_preflight_pass_count": 12,
|
||
"decision_preflight_waiting_count": 0,
|
||
"decision_input_requirement_count": 8,
|
||
"decision_rejection_reason_count": 10,
|
||
"lane_entry_requirement_count": 6,
|
||
"exact_request_payload_field_count": 10,
|
||
"machine_request_manifest_step_count": 6,
|
||
"reads_secret_count": 0,
|
||
"executes_sql_count": 0,
|
||
"writes_database_count": 0,
|
||
"primary_human_gate_count": 0,
|
||
"manual_review_required_count": 0,
|
||
},
|
||
"future_authorization_decision_preflight": {
|
||
"ready_for_future_authorization_decision": True,
|
||
"ready_for_database_apply_now": False,
|
||
"issues_database_apply_authorization": False,
|
||
"requires_fresh_production_truth_in_same_run": True,
|
||
},
|
||
"decision_preflight_envelope": {
|
||
"allows_authorization_decision_in_future_lane": True,
|
||
"issues_database_apply_authorization": False,
|
||
"ready_for_database_apply_now": False,
|
||
},
|
||
"safety": {
|
||
"reads_secret_in_preview": False,
|
||
"executes_sql": False,
|
||
"writes_database": False,
|
||
"manual_review_mode": "exception_only",
|
||
},
|
||
}
|
||
|
||
monkeypatch.setattr(
|
||
backlog,
|
||
"build_pchome_auto_policy_db_apply_authorization_decision_preflight",
|
||
fake_preflight,
|
||
)
|
||
|
||
result = smoke._pchome_auto_policy_decision_preflight_check()
|
||
|
||
assert result["status"] == "ok"
|
||
assert captured["execute_fetch"] is True
|
||
assert captured["http_get"] is smoke._pchome_auto_policy_machine_fetch_evidence_get
|
||
assert result["details"]["decision_preflight_pass_count"] == 12
|
||
assert result["details"]["machine_fetch_evidence_ready_count"] == 1
|
||
assert result["details"]["machine_fetch_evidence_source"] == "local_schema_fixture"
|
||
assert result["details"]["payload_source"] == "local_contract_fixture"
|
||
assert result["details"]["outbound_network"] is False
|
||
assert result["details"]["business_data_source"] is False
|
||
assert result["details"]["primary_human_gate_count"] == 0
|
||
assert result["details"]["writes_database_count"] == 0
|
||
assert result["details"]["next_machine_action"] == "continue_to_pchome_auto_policy_decision_closeout_lane"
|
||
|
||
|
||
def test_pchome_auto_policy_decision_closeout_reports_no_write_package(monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
from services import pchome_mapping_backlog_service as backlog
|
||
captured = {}
|
||
|
||
def fake_closeout(*_args, **kwargs):
|
||
captured.update(kwargs)
|
||
return {
|
||
"policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_decision_closeout",
|
||
"result": "DB_APPLY_AUTHORIZATION_DECISION_CLOSEOUT_READY",
|
||
"source_policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_decision_preflight",
|
||
"summary": {
|
||
"authorization_decision_closeout_ready_count": 1,
|
||
"authorization_decision_preflight_ready_count": 1,
|
||
"decision_closeout_check_count": 12,
|
||
"decision_closeout_pass_count": 12,
|
||
"decision_closeout_waiting_count": 0,
|
||
"decision_input_requirement_count": 8,
|
||
"decision_rejection_reason_count": 10,
|
||
"post_apply_verifier_required_count": 1,
|
||
"same_run_truth_required_count": 1,
|
||
"reads_secret_count": 0,
|
||
"executes_sql_count": 0,
|
||
"writes_database_count": 0,
|
||
"primary_human_gate_count": 0,
|
||
"manual_review_required_count": 0,
|
||
},
|
||
"future_authorization_decision_closeout": {
|
||
"ready_for_future_authorization_decision_closeout": True,
|
||
"ready_for_database_apply_now": False,
|
||
"issues_database_apply_authorization": False,
|
||
},
|
||
"future_authorization_decision_package": {
|
||
"ready_for_future_authorization_decision_package": True,
|
||
"ready_for_database_apply_now": False,
|
||
"issues_database_apply_authorization": False,
|
||
"requires_post_apply_verifier": True,
|
||
"requires_fresh_production_truth_in_same_run": True,
|
||
},
|
||
"decision_closeout_contract": {
|
||
"machine_verifiable": True,
|
||
"permits_future_authorization_decision_lane": True,
|
||
"issues_database_apply_authorization": False,
|
||
"ready_for_database_apply_now": False,
|
||
"writes_database": False,
|
||
},
|
||
"safety": {
|
||
"reads_secret_in_preview": False,
|
||
"executes_sql": False,
|
||
"writes_database": False,
|
||
"manual_review_mode": "exception_only",
|
||
},
|
||
}
|
||
|
||
monkeypatch.setattr(
|
||
backlog,
|
||
"build_pchome_auto_policy_db_apply_authorization_decision_closeout",
|
||
fake_closeout,
|
||
)
|
||
|
||
result = smoke._pchome_auto_policy_decision_closeout_check()
|
||
|
||
assert result["status"] == "ok"
|
||
assert captured["execute_fetch"] is True
|
||
assert captured["http_get"] is smoke._pchome_auto_policy_machine_fetch_evidence_get
|
||
assert result["details"]["decision_closeout_pass_count"] == 12
|
||
assert result["details"]["post_apply_verifier_required_count"] == 1
|
||
assert result["details"]["same_run_truth_required_count"] == 1
|
||
assert result["details"]["payload_source"] == "local_contract_fixture"
|
||
assert result["details"]["outbound_network"] is False
|
||
assert result["details"]["business_data_source"] is False
|
||
assert result["details"]["primary_human_gate_count"] == 0
|
||
assert result["details"]["writes_database_count"] == 0
|
||
assert result["details"]["issues_database_apply_authorization"] is False
|
||
assert result["details"]["next_machine_action"] == "continue_to_pchome_auto_policy_authorization_issuer_gate"
|
||
|
||
|
||
def test_collect_ai_automation_smoke_persists_recent_history(tmp_path, monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
history_path = tmp_path / "smoke_history.jsonl"
|
||
monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path))
|
||
monkeypatch.setattr(smoke, "_HISTORY_LIMIT", 2)
|
||
monkeypatch.setattr(smoke, "_event_router_check", lambda: smoke._check("event", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_gemini_egress_check", lambda: smoke._check("gemini", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_autoheal_check", lambda: smoke._check("autoheal", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_nemotron_check", lambda: smoke._check("nemotron", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_embedding_queue_check", lambda: smoke._check("embedding", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_elephant_hitl_check", lambda: smoke._check("elephant", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_controlled_apply_drift_monitor_check", lambda: smoke._check("pchome", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_auto_policy_authorization_guard_check", lambda: smoke._check("auto-policy", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_auto_policy_decision_preflight_check", lambda: smoke._check("auto-policy preflight", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_auto_policy_decision_closeout_check", lambda: smoke._check("auto-policy closeout", "ok", "ok"))
|
||
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"))
|
||
|
||
first = smoke.collect_ai_automation_smoke(history_limit=5)
|
||
second = smoke.collect_ai_automation_smoke(history_limit=5)
|
||
third = smoke.collect_ai_automation_smoke(history_limit=5)
|
||
|
||
assert first["status"] == "ok"
|
||
assert second["history"]["counts"]["ok"] == 2
|
||
assert third["history"]["counts"]["ok"] == 2
|
||
assert len(history_path.read_text(encoding="utf-8").strip().splitlines()) == 2
|
||
|
||
|
||
def test_smoke_history_export_and_clear(tmp_path, monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
history_path = tmp_path / "smoke_history.jsonl"
|
||
history_path.write_text(
|
||
'{"generated_at":"2026-04-29T01:00:00","status":"ok"}\n'
|
||
'{"generated_at":"2026-04-29T02:00:00","status":"warning"}\n',
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path))
|
||
|
||
export = smoke.export_smoke_history_jsonl()
|
||
cleared = smoke.clear_smoke_history()
|
||
|
||
assert export["count"] == 2
|
||
assert '"status":"warning"' in export["content"]
|
||
assert cleared["cleared"] == 2
|
||
assert not history_path.exists()
|
||
|
||
|
||
def test_smoke_history_daily_summary():
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
summary = smoke._history_summary([
|
||
{"generated_at": "2026-04-28T23:00:00", "status": "ok"},
|
||
{"generated_at": "2026-04-29T01:00:00", "status": "warning"},
|
||
{"generated_at": "2026-04-29T02:00:00", "status": "critical"},
|
||
])
|
||
|
||
assert summary["daily"] == [
|
||
{"date": "2026-04-28", "ok": 1, "warning": 0, "critical": 0, "total": 1},
|
||
{"date": "2026-04-29", "ok": 0, "warning": 1, "critical": 1, "total": 2},
|
||
]
|
||
|
||
|
||
def test_scheduled_automation_health_summary_reads_history_without_side_effects(tmp_path, monkeypatch):
|
||
from datetime import datetime
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
history_path = tmp_path / "smoke_history.jsonl"
|
||
history_path.write_text(
|
||
json.dumps({
|
||
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
||
"status": "ok",
|
||
"summary": {"ok": 13, "warning": 0, "critical": 0, "total": 13},
|
||
"checks": [
|
||
{
|
||
"name": "PChome 受控落地 drift monitor",
|
||
"status": "ok",
|
||
"summary": "PChome controlled apply drift 已驗證 4/4，目前 0 drift",
|
||
"details": {
|
||
"result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED",
|
||
"source_receipt_replay_result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED",
|
||
"selector_count": 4,
|
||
"readback_pass_count": 4,
|
||
"drift_count": 0,
|
||
},
|
||
},
|
||
_auto_policy_guard_history_check(),
|
||
_auto_policy_decision_preflight_history_check(),
|
||
_auto_policy_decision_closeout_history_check(),
|
||
{
|
||
"name": "AI surface HTML readback",
|
||
"status": "ok",
|
||
"summary": "AI surface HTML readback 通過 10/10，無產品面 raw wording",
|
||
"details": {
|
||
"policy": "read_only_ai_surface_html_readback_v1",
|
||
"checked_surface_count": 10,
|
||
"pass_count": 10,
|
||
"failed_count": 0,
|
||
"forbidden_leak_count": 0,
|
||
},
|
||
},
|
||
{
|
||
"name": "Sitewide UI/UX Agent readback",
|
||
"status": "ok",
|
||
"summary": "Sitewide UI/UX Agent ok/no-op，整站專業化 surface 無退化",
|
||
"details": {
|
||
"policy": "read_only_sitewide_ui_ux_agent_inventory_v1",
|
||
"template_count": 55,
|
||
"high_priority_template_count": 31,
|
||
"compact_guardrail_count": 29,
|
||
"issue_surface_count": 0,
|
||
"raw_engineering_issue_count": 0,
|
||
"repair_package_status": "no_op",
|
||
"repair_action_count": 0,
|
||
},
|
||
},
|
||
{
|
||
"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,
|
||
"route_count": 12,
|
||
"viewport_count": 3,
|
||
"failed_count": 0,
|
||
"overflow_issue_count": 0,
|
||
"visual_offender_count": 0,
|
||
"artifact_path": "data/ai_automation/sitewide_visual_qa_latest.json",
|
||
"artifact_generated_at": datetime.now().isoformat(timespec="seconds"),
|
||
"stale": False,
|
||
},
|
||
}
|
||
],
|
||
}, ensure_ascii=False) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path))
|
||
|
||
summary = smoke.build_scheduled_automation_health_summary(
|
||
history_limit=5,
|
||
freshness_max_age_hours=9999,
|
||
)
|
||
|
||
pchome_family = next(
|
||
item for item in summary["families"]
|
||
if item["key"] == "pchome_controlled_apply_drift_monitor"
|
||
)
|
||
assert summary["policy"] == "read_only_ai_automation_scheduled_health_summary"
|
||
assert summary["status"] == "ok"
|
||
assert summary["summary"]["total"] == 10
|
||
assert summary["summary"]["primary_human_gate_count"] == 0
|
||
assert summary["summary"]["writes_database_count"] == 0
|
||
assert pchome_family["status"] == "ok"
|
||
assert pchome_family["details"]["drift_count"] == 0
|
||
auto_policy_family = next(
|
||
item for item in summary["families"]
|
||
if item["key"] == "pchome_auto_policy_authorization_guard"
|
||
)
|
||
assert auto_policy_family["status"] == "ok"
|
||
assert auto_policy_family["details"]["lane_guard_pass_count"] == 11
|
||
assert auto_policy_family["details"]["primary_human_gate_count"] == 0
|
||
assert auto_policy_family["details"]["writes_database_count"] == 0
|
||
preflight_family = next(
|
||
item for item in summary["families"]
|
||
if item["key"] == "pchome_auto_policy_decision_preflight"
|
||
)
|
||
assert preflight_family["status"] == "ok"
|
||
assert preflight_family["details"]["decision_preflight_pass_count"] == 12
|
||
assert preflight_family["details"]["machine_fetch_evidence_ready_count"] == 1
|
||
assert preflight_family["details"]["payload_source"] == "local_contract_fixture"
|
||
assert preflight_family["details"]["outbound_network"] is False
|
||
assert preflight_family["details"]["business_data_source"] is False
|
||
assert preflight_family["details"]["primary_human_gate_count"] == 0
|
||
assert preflight_family["details"]["writes_database_count"] == 0
|
||
closeout_family = next(
|
||
item for item in summary["families"]
|
||
if item["key"] == "pchome_auto_policy_decision_closeout"
|
||
)
|
||
assert closeout_family["status"] == "ok"
|
||
assert closeout_family["details"]["decision_closeout_pass_count"] == 12
|
||
assert closeout_family["details"]["post_apply_verifier_required_count"] == 1
|
||
assert closeout_family["details"]["same_run_truth_required_count"] == 1
|
||
assert closeout_family["details"]["payload_source"] == "local_contract_fixture"
|
||
assert closeout_family["details"]["outbound_network"] is False
|
||
assert closeout_family["details"]["business_data_source"] is False
|
||
assert closeout_family["details"]["primary_human_gate_count"] == 0
|
||
assert closeout_family["details"]["writes_database_count"] == 0
|
||
surface_family = next(
|
||
item for item in summary["families"]
|
||
if item["key"] == "ai_surface_html_readback"
|
||
)
|
||
assert surface_family["status"] == "ok"
|
||
assert surface_family["details"]["checked_surface_count"] == 10
|
||
sitewide_family = next(
|
||
item for item in summary["families"]
|
||
if item["key"] == "sitewide_ui_ux_agent"
|
||
)
|
||
assert sitewide_family["status"] == "ok"
|
||
assert sitewide_family["details"]["issue_surface_count"] == 0
|
||
assert sitewide_family["details"]["repair_package_status"] == "no_op"
|
||
visual_family = next(
|
||
item for item in summary["families"]
|
||
if item["key"] == "sitewide_visual_qa"
|
||
)
|
||
assert visual_family["status"] == "ok"
|
||
assert visual_family["details"]["route_count"] == 12
|
||
assert visual_family["details"]["failed_count"] == 0
|
||
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"
|
||
assert summary["safety"]["writes_database"] is False
|
||
|
||
|
||
def test_scheduled_automation_health_summary_can_use_current_smoke_without_recording_history(tmp_path, monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
history_path = tmp_path / "smoke_history.jsonl"
|
||
monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path))
|
||
current_smoke = {
|
||
"generated_at": "2026-07-02T12:00:00",
|
||
"status": "critical",
|
||
"summary": {"ok": 12, "warning": 0, "critical": 1, "total": 13},
|
||
"checks": [
|
||
{
|
||
"name": "PChome 受控落地 drift monitor",
|
||
"status": "critical",
|
||
"summary": "PChome controlled apply 偵測到 1 筆 drift",
|
||
"details": {"drift_count": 1, "selector_count": 4, "readback_pass_count": 3},
|
||
},
|
||
_auto_policy_guard_history_check(),
|
||
_auto_policy_decision_preflight_history_check(),
|
||
_auto_policy_decision_closeout_history_check(),
|
||
{
|
||
"name": "AI surface HTML readback",
|
||
"status": "ok",
|
||
"summary": "AI surface HTML readback 通過 10/10，無產品面 raw wording",
|
||
"details": {"checked_surface_count": 10, "pass_count": 10, "failed_count": 0},
|
||
},
|
||
{
|
||
"name": "Sitewide UI/UX Agent readback",
|
||
"status": "ok",
|
||
"summary": "Sitewide UI/UX Agent ok/no-op，整站專業化 surface 無退化",
|
||
"details": {
|
||
"template_count": 55,
|
||
"compact_guardrail_count": 29,
|
||
"issue_surface_count": 0,
|
||
"repair_package_status": "no_op",
|
||
"repair_action_count": 0,
|
||
},
|
||
},
|
||
{
|
||
"name": "Sitewide visual QA readback",
|
||
"status": "ok",
|
||
"summary": "Sitewide visual QA ok，12 routes x 3 viewports 無 overflow 退化",
|
||
"details": {
|
||
"route_count": 12,
|
||
"viewport_count": 3,
|
||
"failed_count": 0,
|
||
"overflow_issue_count": 0,
|
||
"visual_offender_count": 0,
|
||
"stale": False,
|
||
},
|
||
}
|
||
],
|
||
}
|
||
|
||
summary = smoke.build_scheduled_automation_health_summary(
|
||
current_smoke=current_smoke,
|
||
freshness_max_age_hours=9999,
|
||
)
|
||
|
||
assert summary["status"] == "critical"
|
||
assert "run_pchome_drift_verifier_and_recovery_package" in summary["next_machine_actions"]
|
||
assert summary["safety"]["include_current_smoke"] is True
|
||
assert not history_path.exists()
|
||
|
||
|
||
def test_scheduled_automation_health_summary_falls_back_to_visual_qa_artifact(tmp_path, monkeypatch):
|
||
from datetime import datetime
|
||
from services import ai_automation_smoke_service as smoke
|
||
from services import ai_surface_html_readback_service as surface_service
|
||
|
||
history_path = tmp_path / "smoke_history.jsonl"
|
||
history_path.write_text(
|
||
json.dumps({
|
||
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
||
"status": "ok",
|
||
"summary": {"ok": 11, "warning": 0, "critical": 0, "total": 11},
|
||
"checks": [],
|
||
}, ensure_ascii=False) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path))
|
||
monkeypatch.setattr(
|
||
smoke,
|
||
"_pchome_auto_policy_authorization_guard_check",
|
||
lambda: _auto_policy_guard_history_check(),
|
||
)
|
||
monkeypatch.setattr(
|
||
smoke,
|
||
"_pchome_auto_policy_decision_preflight_check",
|
||
lambda: _auto_policy_decision_preflight_history_check(),
|
||
)
|
||
monkeypatch.setattr(
|
||
smoke,
|
||
"_pchome_auto_policy_decision_closeout_check",
|
||
lambda: _auto_policy_decision_closeout_history_check(),
|
||
)
|
||
monkeypatch.setattr(
|
||
surface_service,
|
||
"build_sitewide_visual_qa_readback",
|
||
lambda **_kwargs: {
|
||
"policy": "read_only_sitewide_visual_qa_readback_v1",
|
||
"status": "ok",
|
||
"artifact_path": "data/ai_automation/sitewide_visual_qa_latest.json",
|
||
"artifact_exists": True,
|
||
"artifact_generated_at": "2026-07-03T07:40:00",
|
||
"next_machine_action": "keep_sitewide_visual_qa_monitoring",
|
||
"summary": {
|
||
"result_count": 147,
|
||
"route_count": 49,
|
||
"viewport_count": 3,
|
||
"pass_count": 147,
|
||
"failed_count": 0,
|
||
"overflow_issue_count": 0,
|
||
"visual_offender_count": 0,
|
||
"stale": False,
|
||
},
|
||
},
|
||
)
|
||
|
||
summary = smoke.build_scheduled_automation_health_summary(
|
||
history_limit=5,
|
||
freshness_max_age_hours=9999,
|
||
)
|
||
|
||
visual_family = next(
|
||
item for item in summary["families"]
|
||
if item["key"] == "sitewide_visual_qa"
|
||
)
|
||
assert visual_family["status"] == "ok"
|
||
assert visual_family["details"]["route_count"] == 49
|
||
assert visual_family["details"]["failed_count"] == 0
|
||
|
||
|
||
def test_scheduled_automation_health_summary_route_returns_compact_payload(tmp_path, monkeypatch):
|
||
from datetime import datetime
|
||
from flask import Flask
|
||
from routes import system_public_routes as routes
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
history_path = tmp_path / "smoke_history.jsonl"
|
||
history_path.write_text(
|
||
json.dumps({
|
||
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
||
"status": "ok",
|
||
"summary": {"ok": 12, "warning": 0, "critical": 0, "total": 12},
|
||
"checks": [],
|
||
}, ensure_ascii=False) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path))
|
||
monkeypatch.setattr(
|
||
smoke,
|
||
"_pchome_auto_policy_authorization_guard_check",
|
||
lambda: _auto_policy_guard_history_check(),
|
||
)
|
||
monkeypatch.setattr(
|
||
smoke,
|
||
"_pchome_auto_policy_decision_preflight_check",
|
||
lambda: _auto_policy_decision_preflight_history_check(),
|
||
)
|
||
monkeypatch.setattr(
|
||
smoke,
|
||
"_pchome_auto_policy_decision_closeout_check",
|
||
lambda: _auto_policy_decision_closeout_history_check(),
|
||
)
|
||
|
||
app = Flask(__name__)
|
||
with app.test_request_context(
|
||
"/api/ai-automation/scheduled-health-summary?history_limit=5&freshness_max_age_hours=9999"
|
||
):
|
||
response = routes.ai_automation_scheduled_health_summary_api.__wrapped__()
|
||
|
||
payload = response.get_json()
|
||
assert payload["policy"] == "read_only_ai_automation_scheduled_health_summary"
|
||
assert payload["scheduled_outputs"]["health_summary_endpoint"] == (
|
||
"/api/ai-automation/scheduled-health-summary"
|
||
)
|
||
assert payload["safety"]["writes_database"] is False
|
||
|
||
|
||
def test_surface_html_readback_service_reports_compact_contracts():
|
||
from services.ai_surface_html_readback_service import build_ai_surface_html_readback
|
||
|
||
payload = build_ai_surface_html_readback()
|
||
|
||
assert payload["policy"] == "read_only_ai_surface_html_readback_v1"
|
||
assert payload["status"] == "ok"
|
||
assert payload["summary"]["checked_surface_count"] == 10
|
||
assert payload["summary"]["pass_count"] == 10
|
||
assert payload["summary"]["forbidden_leak_count"] == 0
|
||
assert payload["summary"]["primary_human_gate_count"] == 0
|
||
assert payload["summary"]["writes_database_count"] == 0
|
||
assert payload["safety"]["writes_database"] is False
|
||
|
||
|
||
def test_surface_html_readback_service_detects_regression():
|
||
from services.ai_surface_html_readback_service import build_ai_surface_html_readback
|
||
|
||
payload = build_ai_surface_html_readback(
|
||
rendered_html_by_route={
|
||
"/observability/overview": "<section>PPT_VISION_ENABLED=false</section>"
|
||
}
|
||
)
|
||
|
||
assert payload["status"] == "critical"
|
||
assert payload["summary"]["failed_count"] == 1
|
||
failed = payload["failed_surfaces"][0]
|
||
assert failed["route"] == "/observability/overview"
|
||
assert "compact-observability-workbench" in failed["missing_markers"]
|
||
assert "PPT_VISION_ENABLED=" in failed["forbidden_leaks"]
|
||
|
||
|
||
def test_surface_html_repair_package_builds_controlled_actions():
|
||
from services.ai_surface_html_readback_service import (
|
||
build_ai_surface_html_readback,
|
||
build_ai_surface_html_repair_package,
|
||
)
|
||
|
||
readback = build_ai_surface_html_readback(
|
||
rendered_html_by_route={
|
||
"/observability/overview": "<section>PPT_VISION_ENABLED=false</section>"
|
||
}
|
||
)
|
||
package = build_ai_surface_html_repair_package(source_readback=readback)
|
||
|
||
assert package["policy"] == "read_only_ai_surface_html_repair_package_v1"
|
||
assert package["status"] == "repair_ready"
|
||
assert package["summary"]["controlled_action_count"] == 2
|
||
assert package["summary"]["primary_human_gate_count"] == 0
|
||
assert package["summary"]["writes_database_count"] == 0
|
||
assert package["summary"]["executes_shell_count"] == 0
|
||
assert package["next_machine_action"] == "apply_ai_surface_html_contract_repair"
|
||
assert package["safety"]["read_only_package"] is True
|
||
actions = package["repair_items"][0]["controlled_actions"]
|
||
assert {item["action"] for item in actions} == {
|
||
"restore_density_guardrail_marker",
|
||
"remove_raw_engineering_copy_from_product_surface",
|
||
}
|
||
|
||
|
||
def test_sitewide_ui_ux_agent_inventory_scans_real_pages():
|
||
from services.ai_surface_html_readback_service import build_sitewide_ui_ux_agent_inventory
|
||
|
||
inventory = build_sitewide_ui_ux_agent_inventory()
|
||
|
||
assert inventory["policy"] == "read_only_sitewide_ui_ux_agent_inventory_v1"
|
||
assert inventory["summary"]["template_count"] >= 50
|
||
assert inventory["summary"]["high_priority_template_count"] >= 20
|
||
assert inventory["summary"]["compact_guardrail_count"] >= 29
|
||
assert inventory["summary"]["issue_surface_count"] == 0
|
||
assert inventory["summary"]["primary_human_gate_count"] == 0
|
||
assert inventory["summary"]["writes_database_count"] == 0
|
||
assert inventory["safety"]["writes_database"] is False
|
||
templates = {item["template"] for item in inventory["surfaces"]}
|
||
issue_templates = {item["template"] for item in inventory["issue_surfaces"]}
|
||
assert "templates/base.html" not in templates
|
||
assert not any("/components/" in item for item in templates)
|
||
assert not any("/." in item or item.startswith("templates/.") for item in templates)
|
||
assert "templates/ai_history.html" not in issue_templates
|
||
assert "templates/ai_recommend.html" not in issue_templates
|
||
assert "templates/admin/ppt_audit_preview.html" not in issue_templates
|
||
assert "templates/cicd_dashboard.html" not in issue_templates
|
||
assert "templates/daily_sales.html" not in issue_templates
|
||
assert "templates/dashboard.html" not in issue_templates
|
||
assert "templates/edm_dashboard.html" not in issue_templates
|
||
assert "templates/growth_analysis.html" not in issue_templates
|
||
assert "templates/pchome_crawler.html" not in issue_templates
|
||
assert "templates/price_comparison.html" not in issue_templates
|
||
assert "templates/sales_analysis.html" not in issue_templates
|
||
assert "templates/vendor_stockout_history_v2.html" not in issue_templates
|
||
assert "templates/vendor_stockout_import_v2.html" not in issue_templates
|
||
assert "templates/vendor_stockout_index_v2.html" not in issue_templates
|
||
assert "templates/vendor_stockout_list_v2.html" not in issue_templates
|
||
assert "templates/vendor_stockout_send_email_v2.html" not in issue_templates
|
||
assert "templates/vendor_stockout_vendor_management_v2.html" not in issue_templates
|
||
|
||
|
||
def test_sitewide_ui_ux_repair_package_prioritizes_controlled_template_repairs():
|
||
from services.ai_surface_html_readback_service import (
|
||
build_sitewide_ui_ux_agent_inventory,
|
||
build_sitewide_ui_ux_repair_package,
|
||
)
|
||
|
||
inventory = build_sitewide_ui_ux_agent_inventory()
|
||
package = build_sitewide_ui_ux_repair_package(source_inventory=inventory, limit=3)
|
||
|
||
assert package["policy"] == "read_only_sitewide_ui_ux_repair_package_v1"
|
||
assert package["summary"]["selected_surface_count"] <= 3
|
||
assert package["summary"]["primary_human_gate_count"] == 0
|
||
assert package["summary"]["writes_database_count"] == 0
|
||
assert package["summary"]["executes_shell_count"] == 0
|
||
assert package["safety"]["read_only_package"] is True
|
||
assert "templates/**/*.html" in package["controlled_apply_contract"]["allowed_target_globs"]
|
||
if package["repair_items"]:
|
||
assert package["next_machine_action"] == "apply_sitewide_ui_ux_controlled_repairs"
|
||
assert package["repair_items"][0]["controlled_actions"]
|
||
|
||
|
||
def test_sitewide_ui_ux_agent_reports_non_utf8_templates(tmp_path):
|
||
from services.ai_surface_html_readback_service import (
|
||
build_sitewide_ui_ux_agent_inventory,
|
||
build_sitewide_ui_ux_repair_package,
|
||
)
|
||
|
||
template_dir = tmp_path / "templates"
|
||
template_dir.mkdir()
|
||
routes_dir = tmp_path / "routes"
|
||
routes_dir.mkdir()
|
||
(template_dir / "legacy_dashboard.html").write_bytes(b"\xa3 legacy dashboard")
|
||
(template_dir / "._legacy_dashboard.html").write_bytes(b"\xa3 apple metadata")
|
||
(routes_dir / "legacy_routes.py").write_bytes(b"\xa3 legacy route source")
|
||
|
||
inventory = build_sitewide_ui_ux_agent_inventory(root=tmp_path)
|
||
issue = inventory["issue_surfaces"][0]
|
||
package = build_sitewide_ui_ux_repair_package(source_inventory=inventory)
|
||
|
||
assert inventory["status"] == "warning"
|
||
assert inventory["summary"]["template_count"] == 1
|
||
assert issue["status"] == "critical"
|
||
assert issue["findings"][0]["type"] == "template_encoding_error"
|
||
assert package["repair_items"][0]["controlled_actions"][0]["action"] == (
|
||
"normalize_template_encoding_utf8"
|
||
)
|
||
assert package["safety"]["writes_database"] is False
|
||
|
||
|
||
def test_sitewide_visual_qa_readback_consumes_artifact_payload():
|
||
from datetime import datetime
|
||
from services.ai_surface_html_readback_service import build_sitewide_visual_qa_readback
|
||
|
||
payload = build_sitewide_visual_qa_readback(
|
||
artifact_payload={
|
||
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
||
"results": [
|
||
{"route": "/ai_intelligence", "viewport": "desktop-1440", "passed": True},
|
||
{"route": "/ai_intelligence", "viewport": "mobile-390", "passed": True},
|
||
{"route": "/observability/overview", "viewport": "mobile-390", "passed": True},
|
||
],
|
||
}
|
||
)
|
||
|
||
assert payload["policy"] == "read_only_sitewide_visual_qa_readback_v1"
|
||
assert payload["status"] == "ok"
|
||
assert payload["summary"]["result_count"] == 3
|
||
assert payload["summary"]["route_count"] == 2
|
||
assert payload["summary"]["viewport_count"] == 2
|
||
assert payload["summary"]["failed_count"] == 0
|
||
assert payload["summary"]["primary_human_gate_count"] == 0
|
||
assert payload["summary"]["writes_database_count"] == 0
|
||
assert payload["safety"]["requires_browser_for_readback"] is False
|
||
assert payload["safety"]["requires_browser_for_artifact_generation"] is True
|
||
|
||
|
||
def test_sitewide_visual_qa_readback_reports_visual_failures():
|
||
from datetime import datetime
|
||
from services.ai_surface_html_readback_service import build_sitewide_visual_qa_readback
|
||
|
||
payload = build_sitewide_visual_qa_readback(
|
||
artifact_payload={
|
||
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
||
"results": [
|
||
{
|
||
"route": "/price_comparison",
|
||
"viewport": "mobile-390",
|
||
"passed": False,
|
||
"error": "body horizontal overflow 24px",
|
||
}
|
||
],
|
||
}
|
||
)
|
||
|
||
assert payload["status"] == "critical"
|
||
assert payload["summary"]["failed_count"] == 1
|
||
assert payload["summary"]["overflow_issue_count"] == 1
|
||
assert payload["next_machine_action"] == "run_sitewide_visual_qa_and_publish_artifact"
|
||
assert payload["failed_results"][0]["route"] == "/price_comparison"
|
||
|
||
|
||
def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
monkeypatch.setattr(smoke, "_event_router_check", lambda: smoke._check("event", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_gemini_egress_check", lambda: smoke._check("gemini", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_autoheal_check", lambda: smoke._check("autoheal", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_nemotron_check", lambda: smoke._check("nemotron", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_embedding_queue_check", lambda: smoke._check("embedding", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_elephant_hitl_check", lambda: smoke._check("elephant", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_controlled_apply_drift_monitor_check", lambda: smoke._check("pchome", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_auto_policy_authorization_guard_check", lambda: smoke._check("auto-policy", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_auto_policy_decision_preflight_check", lambda: smoke._check("auto-policy preflight", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_pchome_auto_policy_decision_closeout_check", lambda: smoke._check("auto-policy closeout", "ok", "ok"))
|
||
monkeypatch.setattr(smoke, "_sitewide_visual_qa_check", lambda: smoke._check(
|
||
"Sitewide visual QA readback",
|
||
"ok",
|
||
"visual ok",
|
||
{"route_count": 12, "viewport_count": 3, "failed_count": 0},
|
||
))
|
||
|
||
result = smoke.collect_ai_automation_smoke(record_history=False)
|
||
|
||
surface_check = next(
|
||
item for item in result["checks"]
|
||
if item["name"] == "AI surface HTML readback"
|
||
)
|
||
sitewide_check = next(
|
||
item for item in result["checks"]
|
||
if item["name"] == "Sitewide UI/UX Agent readback"
|
||
)
|
||
visual_check = next(
|
||
item for item in result["checks"]
|
||
if item["name"] == "Sitewide visual QA readback"
|
||
)
|
||
assert result["summary"]["total"] == 13
|
||
assert surface_check["status"] == "ok"
|
||
assert surface_check["details"]["checked_surface_count"] == 10
|
||
assert sitewide_check["status"] == "ok"
|
||
assert sitewide_check["details"]["issue_surface_count"] == 0
|
||
assert sitewide_check["details"]["repair_package_status"] == "no_op"
|
||
assert visual_check["status"] == "ok"
|
||
assert visual_check["details"]["route_count"] == 12
|
||
|
||
|
||
def test_surface_html_readback_route_returns_payload(monkeypatch):
|
||
from flask import Flask
|
||
from routes import system_public_routes as routes
|
||
|
||
app = Flask(__name__)
|
||
with app.test_request_context("/api/ai-automation/surface-html-readback"):
|
||
response = routes.ai_automation_surface_html_readback_api.__wrapped__()
|
||
|
||
payload = response.get_json()
|
||
assert payload["policy"] == "read_only_ai_surface_html_readback_v1"
|
||
assert payload["summary"]["checked_surface_count"] == 10
|
||
assert payload["safety"]["writes_database"] is False
|
||
|
||
|
||
def test_surface_and_sitewide_ui_ux_agent_routes_return_payloads():
|
||
from flask import Flask
|
||
from routes import system_public_routes as routes
|
||
|
||
app = Flask(__name__)
|
||
route_cases = [
|
||
(
|
||
"/api/ai-automation/surface-html-repair-package",
|
||
routes.ai_automation_surface_html_repair_package_api.__wrapped__,
|
||
"read_only_ai_surface_html_repair_package_v1",
|
||
),
|
||
(
|
||
"/api/ai-automation/sitewide-ui-ux-agent",
|
||
routes.ai_automation_sitewide_ui_ux_agent_api.__wrapped__,
|
||
"read_only_sitewide_ui_ux_agent_inventory_v1",
|
||
),
|
||
(
|
||
"/api/ai-automation/sitewide-ui-ux-repair-package?limit=3",
|
||
routes.ai_automation_sitewide_ui_ux_repair_package_api.__wrapped__,
|
||
"read_only_sitewide_ui_ux_repair_package_v1",
|
||
),
|
||
(
|
||
"/api/ai-automation/sitewide-visual-qa-readback?max_age_hours=9999",
|
||
routes.ai_automation_sitewide_visual_qa_readback_api.__wrapped__,
|
||
"read_only_sitewide_visual_qa_readback_v1",
|
||
),
|
||
]
|
||
|
||
for path, view_func, policy in route_cases:
|
||
with app.test_request_context(path):
|
||
response = view_func()
|
||
payload = response.get_json()
|
||
assert payload["policy"] == policy
|
||
assert payload["safety"].get("writes_database") is False
|
||
|
||
|
||
def test_gemini_egress_check_ok_when_no_calls(monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
class FakeResult:
|
||
def fetchone(self):
|
||
return {"calls": 0, "tokens": 0, "cost_usd": 0, "last_called": None}
|
||
|
||
def fetchall(self):
|
||
return []
|
||
|
||
class FakeSession:
|
||
closed = False
|
||
|
||
def execute(self, *_args, **_kwargs):
|
||
return FakeResult()
|
||
|
||
def close(self):
|
||
self.closed = True
|
||
|
||
fake_session = FakeSession()
|
||
monkeypatch.setattr(smoke, "get_session", lambda: fake_session)
|
||
monkeypatch.delenv("GEMINI_API_HARD_DISABLED", raising=False)
|
||
|
||
result = smoke._gemini_egress_check()
|
||
|
||
assert result["status"] == "ok"
|
||
assert result["details"]["calls"] == 0
|
||
assert result["details"]["hard_disabled"] is True
|
||
assert fake_session.closed is True
|
||
|
||
|
||
def test_gemini_egress_check_critical_when_hard_disabled_has_calls(monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
class FakeSummaryResult:
|
||
def fetchone(self):
|
||
return {
|
||
"calls": 2,
|
||
"tokens": 1234,
|
||
"cost_usd": 0.012345,
|
||
"last_called": "2026-05-21 07:00:00+00",
|
||
}
|
||
|
||
class FakeTopResult:
|
||
def fetchall(self):
|
||
return [{
|
||
"caller": "code_review_openclaw_gemini",
|
||
"model": "gemini-2.5-flash",
|
||
"calls": 2,
|
||
"tokens": 1234,
|
||
"cost_usd": 0.012345,
|
||
"last_called": "2026-05-21 07:00:00+00",
|
||
}]
|
||
|
||
class FakeSession:
|
||
def __init__(self):
|
||
self.calls = 0
|
||
|
||
def execute(self, *_args, **_kwargs):
|
||
self.calls += 1
|
||
return FakeSummaryResult() if self.calls == 1 else FakeTopResult()
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
monkeypatch.setattr(smoke, "get_session", lambda: FakeSession())
|
||
monkeypatch.delenv("GEMINI_API_HARD_DISABLED", raising=False)
|
||
|
||
result = smoke._gemini_egress_check()
|
||
|
||
assert result["status"] == "critical"
|
||
assert "Hard-disabled" in result["summary"]
|
||
assert result["details"]["calls"] == 2
|
||
assert result["details"]["top_callers"][0]["caller"] == "code_review_openclaw_gemini"
|
||
|
||
|
||
def test_nemotron_smoke_detects_current_dispatcher_fallback(monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
import services.nemoton_dispatcher_service as nemotron
|
||
|
||
monkeypatch.delattr(nemotron, "NemotronDispatcherService", raising=False)
|
||
monkeypatch.setattr(nemotron, "NIM_API_KEY", "")
|
||
|
||
result = smoke._nemotron_check()
|
||
|
||
assert result["status"] == "warning"
|
||
assert result["details"]["fallback_ready"] is True
|
||
assert result["details"]["dispatcher_class"] == "NemotronDispatcher"
|
||
|
||
|
||
def test_build_smoke_daily_summary_message_escapes_history(tmp_path, monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
history_path = tmp_path / "smoke_history.jsonl"
|
||
history_path.write_text(
|
||
'{"generated_at":"2026-04-29T01:00:00","status":"warning",'
|
||
'"summary":{"ok":0,"warning":1,"critical":0,"total":1},'
|
||
'"checks":[{"name":"<script>","status":"warning","summary":"bad"}]}\n',
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path))
|
||
|
||
message = smoke.build_smoke_daily_summary_message()
|
||
|
||
assert "AI 自動化 Smoke 每日摘要" in message
|
||
assert "WARNING" in message
|
||
assert "最近異常檢查" in message
|
||
assert "<script>" not in message
|
||
|
||
|
||
def test_send_smoke_daily_summary_uses_telegram_result(monkeypatch):
|
||
from services import ai_automation_smoke_service as smoke
|
||
|
||
monkeypatch.setattr(smoke, "build_smoke_daily_summary_message", lambda: "hello")
|
||
monkeypatch.setattr(
|
||
"services.telegram_templates.send_telegram_with_result",
|
||
lambda message, chat_ids=None: {
|
||
"ok": True,
|
||
"sent": 1,
|
||
"failed": 0,
|
||
"chat_ids": chat_ids or [123],
|
||
"errors": [],
|
||
},
|
||
)
|
||
|
||
result = smoke.send_smoke_daily_summary(chat_ids=[999])
|
||
|
||
assert result["status"] == "sent"
|
||
assert result["telegram"]["chat_ids"] == [999]
|