78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
from datetime import datetime, timedelta
|
|
|
|
|
|
def test_action_plan_hygiene_preview_closes_only_stale_advisory_sources():
|
|
from services.action_plan_hygiene import build_action_plan_hygiene_preview
|
|
|
|
now = datetime(2026, 5, 19, 12, 0, 0)
|
|
rows = [
|
|
{
|
|
"id": 1,
|
|
"status": "pending",
|
|
"priority": 1,
|
|
"created_at": now - timedelta(hours=100),
|
|
"action_type": "openclaw_recommendation",
|
|
"description": "old advisory",
|
|
},
|
|
{
|
|
"id": 2,
|
|
"status": "auto_pending",
|
|
"priority": 1,
|
|
"created_at": now - timedelta(hours=90),
|
|
"action_type": "code_review_fix",
|
|
"description": "old code review",
|
|
},
|
|
{
|
|
"id": 3,
|
|
"status": "pending",
|
|
"priority": 1,
|
|
"created_at": now - timedelta(hours=90),
|
|
"created_by": "nemotron",
|
|
"description": "must stay",
|
|
},
|
|
{
|
|
"id": 4,
|
|
"status": "pending",
|
|
"priority": 1,
|
|
"created_at": now - timedelta(hours=2),
|
|
"action_type": "openclaw_recommendation",
|
|
"description": "fresh advisory",
|
|
},
|
|
{
|
|
"id": 5,
|
|
"status": "pending",
|
|
"priority": 3,
|
|
"created_at": now - timedelta(hours=110),
|
|
"created_by": "nemotron",
|
|
"payload": {
|
|
"dispatch_to": "direct_response",
|
|
"action_plan": [{"action": "reply_simple"}],
|
|
},
|
|
"description": "old chat response",
|
|
},
|
|
{
|
|
"id": 6,
|
|
"status": "pending",
|
|
"priority": 3,
|
|
"created_at": now - timedelta(hours=110),
|
|
"created_by": "nemotron",
|
|
"payload": {
|
|
"dispatch_to": "human_review",
|
|
"action_plan": [{"action": "flag_for_human_review"}],
|
|
},
|
|
"description": "must stay",
|
|
},
|
|
]
|
|
|
|
preview = build_action_plan_hygiene_preview(rows, now=now, stale_hours=72)
|
|
|
|
assert preview["candidate_count"] == 3
|
|
assert preview["by_source"] == {
|
|
"openclaw_recommendation": 1,
|
|
"code_review_fix": 1,
|
|
"nemotron_direct_response": 1,
|
|
}
|
|
assert {item["id"] for item in preview["candidates"]} == {1, 2, 5}
|
|
target_by_id = {item["id"]: item["to_status"] for item in preview["candidates"]}
|
|
assert target_by_id == {1: "rejected", 2: "auto_disabled", 5: "auto_disabled"}
|