1  """Admin endpoint for testing project event webhooks.
 2  
 3  Single ``POST /projects/{id}/webhooks/test`` route — fires a synthetic
 4  ``test`` event so an admin can confirm their receiver is wired
 5  correctly without waiting for a real budget/sync/eval/routine event.
 6  """
 7  from __future__ import annotations
 8  
 9  import json
10  
11  from fastapi import APIRouter, Depends, HTTPException
12  
13  from restai.auth import get_current_username_project
14  from restai.database import DBWrapper, get_db_wrapper
15  from restai.models.models import User
16  from restai.webhooks import emit_event
17  
18  router = APIRouter()
19  
20  
21  @router.post("/projects/{projectID}/webhooks/test")
22  async def test_webhook(
23      projectID: int,
24      user: User = Depends(get_current_username_project),
25      db_wrapper: DBWrapper = Depends(get_db_wrapper),
26  ):
27      """Fire a synthetic ``test`` event to the project's webhook URL.
28      Returns ``{ok: bool, reason?: str}``. ``ok=False`` when no URL is
29      configured, the URL is unsafe (private/internal), or the event
30      isn't in the project's subscribed set."""
31      proj = db_wrapper.get_project_by_id(projectID)
32      if proj is None:
33          raise HTTPException(status_code=404, detail="project not found")
34      try:
35          opts = json.loads(proj.options) if proj.options else {}
36      except Exception:
37          opts = {}
38      if not (opts.get("webhook_url") or "").strip():
39          return {"ok": False, "reason": "no webhook_url configured"}
40      queued = emit_event(
41          projectID, proj.name, opts, "test",
42          {"message": "RESTai test webhook — receiver is reachable."},
43      )
44      if not queued:
45          return {"ok": False, "reason": "event filtered or url refused (check logs)"}
46      return {"ok": True}