64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from src.services.awooop_truth_chain_service import _clean_row, _truth_status
|
|
|
|
|
|
def test_clean_row_parses_json_text_fields_for_gateway_visibility() -> None:
|
|
row = {
|
|
"gate_result": '{"schema_version":"legacy_mcp_bridge_v1","policy_enforced":false}',
|
|
"source_envelope": '{"adapter":"legacy_telegram_gateway"}',
|
|
"plain_text": '{"not":"parsed"}',
|
|
}
|
|
|
|
cleaned = _clean_row(row)
|
|
|
|
assert cleaned["gate_result"]["schema_version"] == "legacy_mcp_bridge_v1"
|
|
assert cleaned["gate_result"]["policy_enforced"] is False
|
|
assert cleaned["source_envelope"]["adapter"] == "legacy_telegram_gateway"
|
|
assert cleaned["plain_text"] == '{"not":"parsed"}'
|
|
|
|
|
|
def test_truth_status_marks_no_action_approval_as_manual_required() -> None:
|
|
status = _truth_status(
|
|
incident={"incident_id": "INC-1", "status": "INVESTIGATING"},
|
|
approvals=[{"status": "APPROVED", "action": "未知操作 | NO_ACTION"}],
|
|
evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 0}],
|
|
automation_ops=[],
|
|
drift=None,
|
|
drift_repeat_count=0,
|
|
gateway_mcp_total=0,
|
|
legacy_mcp_total=8,
|
|
outbound_visible_total=0,
|
|
)
|
|
|
|
assert status["current_stage"] == "manual_required"
|
|
assert status["stage_status"] == "blocked"
|
|
assert status["needs_human"] is True
|
|
assert "approval_resolved_no_action_without_execution" in status["blockers"]
|
|
assert "all_evidence_sensors_failed" in status["blockers"]
|
|
assert "awooop_mcp_gateway_audit_empty" in status["blockers"]
|
|
|
|
|
|
def test_truth_status_marks_repeated_pending_drift_as_human_needed() -> None:
|
|
status = _truth_status(
|
|
incident=None,
|
|
approvals=[],
|
|
evidence_rows=[],
|
|
automation_ops=[],
|
|
drift={
|
|
"report_id": "7f858956",
|
|
"status": "pending",
|
|
"interpretation": {"confidence": 0.0},
|
|
},
|
|
drift_repeat_count=12,
|
|
gateway_mcp_total=0,
|
|
legacy_mcp_total=0,
|
|
outbound_visible_total=0,
|
|
)
|
|
|
|
assert status["current_stage"] == "dedup_or_repeat_updated"
|
|
assert status["stage_status"] == "pending"
|
|
assert status["needs_human"] is True
|
|
assert "drift_report_pending_without_resolution" in status["blockers"]
|
|
assert "drift_ai_confidence_zero" in status["blockers"]
|