Coverage for chalicelib/core/collaborations/collaboration_msteams.py: 32%
112 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:56 +0000
1import logging
3from decouple import config
4from fastapi import HTTPException, status
6import schemas
7from chalicelib.core import webhook
8from chalicelib.core.collaborations.collaboration_base import BaseCollaboration
9from chalicelib.utils import ssrf
10from chalicelib.utils.log import sanitize
12logger = logging.getLogger(__name__)
15class MSTeams(BaseCollaboration):
16 @classmethod
17 def add(cls, tenant_id, data: schemas.AddCollaborationSchema):
18 if webhook.exists_by_name(tenant_id=tenant_id, name=data.name, exclude_id=None, 18 ↛ 20line 18 didn't jump to line 20 because the condition on line 18 was never true
19 webhook_type=schemas.WebhookType.MSTEAMS):
20 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"name already exists.")
21 try:
22 hello = cls.say_hello(data.url)
23 except ValueError as e:
24 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
25 if hello: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true
26 return webhook.add(tenant_id=tenant_id,
27 endpoint=data.url.unicode_string(),
28 webhook_type=schemas.WebhookType.MSTEAMS,
29 name=data.name)
30 return None
32 @classmethod
33 def say_hello(cls, url):
34 try:
35 r = ssrf.post_json(
36 endpoint=url,
37 json_data={
38 "@type": "MessageCard",
39 "@context": "https://schema.org/extensions",
40 "summary": "Welcome to OpenReplay",
41 "title": "Welcome to OpenReplay"
42 },
43 timeout=3)
44 if r.status_code != 200: 44 ↛ 54line 44 didn't jump to line 54 because the condition on line 44 was always true
45 logger.warning("MSTeams integration failed")
46 logger.warning(sanitize(r.text))
47 return False
48 except ValueError:
49 raise
50 except Exception as e:
51 logger.warning("!!! MSTeams integration failed")
52 logger.exception(e)
53 return False
54 return True
56 @classmethod
57 def send_raw(cls, tenant_id, webhook_id, body):
58 integration = cls.get_integration(tenant_id=tenant_id, integration_id=webhook_id)
59 if integration is None:
60 return {"errors": ["msteams integration not found"]}
61 try:
62 r = ssrf.post_json(
63 endpoint=integration["endpoint"],
64 json_data=body,
65 timeout=5)
66 if r.status_code != 200:
67 logger.warning(f"!! issue sending msteams raw; webhookId:{webhook_id} code:{r.status_code}")
68 logger.warning(sanitize(r.text))
69 return None
70 except requests.exceptions.Timeout:
71 logger.warning(f"!! Timeout sending msteams raw webhookId:{webhook_id}")
72 return None
73 except Exception as e:
74 logger.warning(f"!! Issue sending msteams raw webhookId:{webhook_id}")
75 logger.warning(e)
76 return None
77 return {"data": r.text}
79 @classmethod
80 def send_batch(cls, tenant_id, webhook_id, attachments):
81 integration = cls.get_integration(tenant_id=tenant_id, integration_id=webhook_id)
82 if integration is None:
83 return {"errors": ["msteams integration not found"]}
84 logger.debug(f"====> sending msteams batch notification: {len(attachments)}")
85 for i in range(0, len(attachments), 50):
86 part = attachments[i:i + 50]
87 for j in range(1, len(part), 2):
88 part.insert(j, {"text": "***"})
90 r = ssrf.post_json(endpoint=integration["endpoint"],
91 json_data={
92 "@type": "MessageCard",
93 "@context": "http://schema.org/extensions",
94 "summary": part[0]["activityTitle"],
95 "sections": part
96 })
97 if r.status_code != 200:
98 logger.warning("!!!! something went wrong")
99 logger.warning(sanitize(r.text))
101 @classmethod
102 def __share(cls, tenant_id, integration_id, attachement, extra=None):
103 if extra is None:
104 extra = {}
105 integration = cls.get_integration(tenant_id=tenant_id, integration_id=integration_id)
106 if integration is None:
107 return {"errors": ["Microsoft Teams integration not found"]}
108 r = ssrf.post_json(
109 endpoint=integration["endpoint"],
110 json_data={
111 "@type": "MessageCard",
112 "@context": "http://schema.org/extensions",
113 "sections": [attachement],
114 **extra
115 })
117 return r.text
119 @classmethod
120 def share_session(cls, tenant_id, project_id, session_id, user, comment, project_name=None, integration_id=None):
121 title = f"*{user}* has shared the below session!"
122 link = f"{config('SITE_URL')}/{project_id}/session/{session_id}"
123 args = {
124 "activityTitle": title,
125 "facts": [
126 {
127 "name": "Session:",
128 "value": link
129 }],
130 "markdown": True
131 }
132 if project_name and len(project_name) > 0:
133 args["activitySubtitle"] = f"On Project *{project_name}*"
134 if comment and len(comment) > 0:
135 args["facts"].append({
136 "name": "Comment:",
137 "value": comment
138 })
139 data = cls.__share(tenant_id, integration_id, attachement=args, extra={"summary": title})
140 if "errors" in data:
141 return data
142 return {"data": data}
144 @classmethod
145 def share_error(cls, tenant_id, project_id, error_id, user, comment, project_name=None, integration_id=None):
146 title = f"*{user}* has shared the below error!"
147 link = f"{config('SITE_URL')}/{project_id}/errors/{error_id}"
148 args = {
149 "activityTitle": title,
150 "facts": [
151 {
152 "name": "Session:",
153 "value": link
154 }],
155 "markdown": True
156 }
157 if project_name and len(project_name) > 0:
158 args["activitySubtitle"] = f"On Project *{project_name}*"
159 if comment and len(comment) > 0:
160 args["facts"].append({
161 "name": "Comment:",
162 "value": comment
163 })
164 data = cls.__share(tenant_id, integration_id, attachement=args, extra={"summary": title})
165 if "errors" in data:
166 return data
167 return {"data": data}
169 @classmethod
170 def get_integration(cls, tenant_id, integration_id=None):
171 if integration_id is not None:
172 return webhook.get_webhook(tenant_id=tenant_id, webhook_id=integration_id,
173 webhook_type=schemas.WebhookType.MSTEAMS)
175 integrations = webhook.get_by_type(tenant_id=tenant_id, webhook_type=schemas.WebhookType.MSTEAMS)
176 if integrations is None or len(integrations) == 0:
177 return None
178 return integrations[0]