Coverage for open_webui/routers/notifications.py: 59%

1from __future__ import annotations 

2 

3from typing import Any 

4 

5from fastapi import APIRouter, Depends, HTTPException, Request, status 

6from pydantic import BaseModel 

7 

8from open_webui.constants import ERROR_MESSAGES 

9from open_webui.models.config import Config 

10from open_webui.utils.access_control import has_permission 

11from open_webui.utils.auth import get_verified_user 

12from open_webui.utils.notifications import ( 

13 create_target, 

14 delete_target, 

15 get_notification_event_catalog, 

16 list_targets, 

17 set_default_target, 

18 test_target, 

19 update_target, 

20) 

21 

22router = APIRouter() 

23 

24 

25class NotificationTargetForm(BaseModel): 

26 id: str | None = None 

27 type: str | None = None 

28 enabled: bool | None = None 

29 events: list[str] | None = None 

30 delivery: str | None = None 

31 config: dict[str, Any] | None = None 

32 

33 

34async def _check_notifications_access(user) -> None: 

35 if not await Config.get('ui.enable_user_webhooks'): 35 ↛ 38line 35 didn't jump to line 38 because the condition on line 35 was always true

36 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND) 

37 

38 if user.role != 'admin' and not await has_permission( 

39 user.id, 'features.webhooks', await Config.get('user.permissions') 

40 ): 

41 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACCESS_PROHIBITED) 

42 

43 

44@router.get('/events') 

45async def get_notification_events(user=Depends(get_verified_user)): 

46 await _check_notifications_access(user) 

47 return {'events': get_notification_event_catalog()} 

48 

49 

50@router.get('/targets') 

51async def get_notification_targets(user=Depends(get_verified_user)): 

52 await _check_notifications_access(user) 

53 return await list_targets(user.id) 

54 

55 

56@router.post('/targets') 

57async def create_notification_target(form_data: NotificationTargetForm, user=Depends(get_verified_user)): 

58 await _check_notifications_access(user) 

59 try: 

60 return await create_target(user.id, form_data.model_dump(exclude_unset=True)) 

61 except ValueError as e: 

62 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) 

63 

64 

65@router.put('/targets/{target_id}') 

66async def update_notification_target( 

67 target_id: str, form_data: NotificationTargetForm, user=Depends(get_verified_user) 

68): 

69 await _check_notifications_access(user) 

70 try: 

71 return await update_target(user.id, target_id, form_data.model_dump(exclude_unset=True)) 

72 except ValueError as e: 

73 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) 

74 

75 

76@router.delete('/targets/{target_id}') 

77async def delete_notification_target(target_id: str, user=Depends(get_verified_user)): 

78 await _check_notifications_access(user) 

79 if not await delete_target(user.id, target_id): 

80 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND) 

81 return {'ok': True} 

82 

83 

84@router.put('/targets/{target_id}/default') 

85async def set_default_notification_target(target_id: str, user=Depends(get_verified_user)): 

86 await _check_notifications_access(user) 

87 try: 

88 return await set_default_target(user.id, target_id) 

89 except ValueError as e: 

90 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) 

91 

92 

93@router.post('/targets/{target_id}/test') 

94async def test_notification_target(request: Request, target_id: str, user=Depends(get_verified_user)): 

95 await _check_notifications_access(user) 

96 try: 

97 # LICENSE covers this Open WebUI notification identifier. 

98 # Do not alter, remove, obscure, or replace it except as LICENSE permits: 

99 # https://docs.openwebui.com/license. 

100 app_name = getattr(request.app.state, 'WEBUI_NAME', 'Open WebUI') 

101 return await test_target(user.id, target_id, app_name) 

102 except ValueError as e: 

103 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))