Coverage for routers/core_dynamic.py: 67%
163 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 Body, Depends, BackgroundTasks
5from fastapi import HTTPException, status
6from starlette.responses import RedirectResponse, JSONResponse, Response
8import schemas
9from chalicelib.core import scope
10from chalicelib.core import signup
11from chalicelib.core import tenants, users, projects, license
12from chalicelib.core import webhook
13from chalicelib.core.collaborations.collaboration_slack import Slack
14from chalicelib.core.errors import errors, errors_details
15from chalicelib.utils import captcha, smtp
16from chalicelib.utils import helper
17from chalicelib.utils.TimeUTC import TimeUTC
18from or_dependencies import OR_context, OR_role
19from routers.base import get_routers
20from routers.subs import spot
22logger = logging.getLogger(__name__)
23public_app, app, app_apikey = get_routers()
25if config("LOCAL_DEV", cast=bool, default=False): 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true
26 COOKIE_PATH = "/refresh"
27else:
28 COOKIE_PATH = "/api/refresh"
31@public_app.get('/signup', tags=['signup'])
32async def get_all_signup():
33 return {"data": {"tenants": await tenants.tenants_exists(),
34 "sso": None,
35 "ssoProvider": None,
36 "enforceSSO": None,
37 "edition": license.EDITION}}
40if not tenants.tenants_exists_sync(use_pool=False): 40 ↛ 51line 40 didn't jump to line 51 because the condition on line 40 was always true
41 @public_app.post('/signup', tags=['signup'])
42 @public_app.put('/signup', tags=['signup'])
43 async def signup_handler(response: JSONResponse, data: schemas.UserSignupSchema = Body(...)):
44 content = await signup.create_tenant(data)
45 if "errors" in content:
46 return content
47 content = __process_authentication_response(response=response, data=content)
48 return content
51def __process_authentication_response(response: JSONResponse, data: dict) -> dict:
52 data["smtp"] = smtp.has_smtp()
53 refresh_token = data.pop("refreshToken")
54 refresh_token_max_age = data.pop("refreshTokenMaxAge")
55 spot_refresh_token = data.pop("spotRefreshToken")
56 spot_refresh_token_max_age = data.pop("spotRefreshTokenMaxAge")
57 data = {
58 'jwt': data.pop('jwt'),
59 "spotJwt": data.pop("spotJwt"),
60 'data': {
61 "scopeState": scope.get_scope(-1),
62 "user": data
63 }
64 }
65 response.set_cookie(key="refreshToken", value=refresh_token, path=COOKIE_PATH,
66 max_age=refresh_token_max_age, secure=True, httponly=True)
67 response.set_cookie(key="spotRefreshToken", value=spot_refresh_token, path=spot.COOKIE_PATH,
68 max_age=spot_refresh_token_max_age, secure=True, httponly=True)
69 return data
72@public_app.post('/login', tags=["authentication"])
73def login_user(response: JSONResponse, data: schemas.UserLoginSchema = Body(...)):
74 if helper.allow_captcha() and not captcha.is_valid(data.g_recaptcha_response): 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true
75 raise HTTPException(
76 status_code=status.HTTP_401_UNAUTHORIZED,
77 detail="Invalid captcha."
78 )
80 r = users.authenticate(email=data.email, password=data.password.get_secret_value())
81 if r is None:
82 raise HTTPException(
83 status_code=status.HTTP_401_UNAUTHORIZED,
84 detail="You've entered invalid Email or Password."
85 )
86 if "errors" in r: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true
87 raise HTTPException(
88 status_code=status.HTTP_401_UNAUTHORIZED,
89 detail=r["errors"][0]
90 )
91 r = __process_authentication_response(response=response, data=r)
92 return r
95@app.get('/logout', tags=["login"])
96def logout_user(response: Response, context: schemas.CurrentContext = Depends(OR_context)):
97 users.logout(user_id=context.user_id)
98 response.delete_cookie(key="refreshToken", path=COOKIE_PATH)
99 response.delete_cookie(key="spotRefreshToken", path=spot.COOKIE_PATH)
100 return {"data": "success"}
103@app.get('/refresh', tags=["login"])
104def refresh_login(response: JSONResponse, context: schemas.CurrentContext = Depends(OR_context)):
105 r = users.refresh(user_id=context.user_id)
106 content = {"jwt": r.get("jwt")}
107 response.set_cookie(key="refreshToken", value=r.get("refreshToken"), path=COOKIE_PATH,
108 max_age=r.pop("refreshTokenMaxAge"), secure=True, httponly=True)
109 return content
112@app.get('/account', tags=['accounts'])
113def get_account(context: schemas.CurrentContext = Depends(OR_context)):
114 r = users.get_user(tenant_id=context.tenant_id, user_id=context.user_id)
115 if r is None: 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true
116 return {"errors": ["current user not found"]}
117 t = tenants.get_by_tenant_id(context.tenant_id)
118 if t is not None: 118 ↛ 123line 118 didn't jump to line 123 because the condition on line 118 was always true
119 t = dict(t)
120 t["createdAt"] = TimeUTC.datetime_to_timestamp(t["createdAt"])
121 t["tenantName"] = t.pop("name")
122 else:
123 return {"errors": ["current tenant not found"]}
125 return {
126 'data': {
127 **r,
128 **t,
129 **license.get_status(context.tenant_id),
130 "smtp": smtp.has_smtp()
131 }
132 }
135@app.post('/account', tags=["account"])
136def edit_account(data: schemas.EditAccountSchema = Body(...),
137 context: schemas.CurrentContext = Depends(OR_context)):
138 return users.edit_account(tenant_id=context.tenant_id, user_id=context.user_id, changes=data)
141@app.post('/account/scope', tags=["account"])
142def change_scope(data: schemas.ScopeSchema = Body(),
143 context: schemas.CurrentContext = Depends(OR_context)):
144 data = scope.update_scope(tenant_id=-1, scope=data.scope)
145 return {'data': data}
148@app.post('/account/password', tags=["account"])
149def change_client_password(resqponse: JSONResponse, data: schemas.EditUserPasswordSchema = Body(...),
150 context: schemas.CurrentContext = Depends(OR_context)):
151 r = users.change_password(email=context.email, old_password=data.old_password.get_secret_value(),
152 new_password=data.new_password.get_secret_value(), tenant_id=context.tenant_id,
153 user_id=context.user_id)
154 if "errors" not in r:
155 r = __process_authentication_response(response=resqponse, data=r)
156 return r
159@app.post('/integrations/slack', tags=['integrations'])
160@app.put('/integrations/slack', tags=['integrations'])
161def add_slack_integration(data: schemas.AddCollaborationSchema,
162 context: schemas.CurrentContext = Depends(OR_context)):
163 n = Slack.add(tenant_id=context.tenant_id, data=data)
164 if n is None: 164 ↛ 168line 164 didn't jump to line 168 because the condition on line 164 was always true
165 return {
166 "errors": ["We couldn't send you a test message on your Slack channel. Please verify your webhook url."]
167 }
168 return {"data": n}
171@app.post('/integrations/slack/{integrationId}', tags=['integrations'])
172def edit_slack_integration(integrationId: int, data: schemas.EditCollaborationSchema = Body(...),
173 context: schemas.CurrentContext = Depends(OR_context)):
174 if len(data.url) > 0: 174 ↛ 184line 174 didn't jump to line 184 because the condition on line 174 was always true
175 old = Slack.get_integration(tenant_id=context.tenant_id, integration_id=integrationId)
176 if not old: 176 ↛ 178line 176 didn't jump to line 178 because the condition on line 176 was always true
177 return {"errors": ["Slack integration not found."]}
178 if old["endpoint"] != data.url:
179 if not Slack.say_hello(data.url):
180 return {
181 "errors": [
182 "We couldn't send you a test message on your Slack channel. Please verify your webhook url."]
183 }
184 return {"data": webhook.update(tenant_id=context.tenant_id, webhook_id=integrationId,
185 changes={"name": data.name, "endpoint": data.url.unicode_string()})}
188@app.post('/client/members', tags=["client"], dependencies=[OR_role("owner", "admin")])
189def add_member(background_tasks: BackgroundTasks, data: schemas.CreateMemberSchema = Body(...),
190 context: schemas.CurrentContext = Depends(OR_context)):
191 return users.create_member(tenant_id=context.tenant_id, user_id=context.user_id, data=data,
192 background_tasks=background_tasks)
195@public_app.get('/users/invitation', tags=['users'])
196def process_invitation_link(token: str):
197 if token is None or len(token) < 64: 197 ↛ 199line 197 didn't jump to line 199 because the condition on line 197 was always true
198 return {"errors": ["please provide a valid invitation"]}
199 user = users.get_by_invitation_token(token)
200 if user is None:
201 return {"errors": ["invitation not found"]}
202 if user["expiredInvitation"]:
203 return {"errors": ["expired invitation, please ask your admin to send a new one"]}
204 if user["expiredChange"] is not None and not user["expiredChange"] \
205 and user["changePwdToken"] is not None and user["changePwdAge"] < -5 * 60:
206 pass_token = user["changePwdToken"]
207 else:
208 pass_token = users.allow_password_change(user_id=user["userId"])
209 return RedirectResponse(url=config("SITE_URL") + config("change_password_link") % (token, pass_token))
212@public_app.post('/password/reset', tags=["users"])
213def change_password_by_invitation(response: JSONResponse, data: schemas.EditPasswordByInvitationSchema = Body(...)):
214 if data is None or len(data.invitation) < 64 or len(data.passphrase) < 8: 214 ↛ 216line 214 didn't jump to line 216 because the condition on line 214 was always true
215 return {"errors": ["please provide a valid invitation & pass"]}
216 user = users.get_by_invitation_token(token=data.invitation, pass_token=data.passphrase)
217 if user is None:
218 return {"errors": ["invitation not found"]}
219 if user["expiredChange"]:
220 return {"errors": ["expired change, please re-use the invitation link"]}
222 r = users.set_password_invitation(new_password=data.password.get_secret_value(), user_id=user["userId"])
223 r = __process_authentication_response(response=response, data=r)
224 return r
227@app.put('/client/members/{memberId}', tags=["client"], dependencies=[OR_role("owner", "admin")])
228def edit_member(memberId: int, data: schemas.EditMemberSchema,
229 context: schemas.CurrentContext = Depends(OR_context)):
230 return users.edit_member(tenant_id=context.tenant_id, editor_id=context.user_id, changes=data,
231 user_id_to_update=memberId)
234@app.get('/projects', tags=['projects'])
235def get_projects(context: schemas.CurrentContext = Depends(OR_context)):
236 return {"data": projects.get_projects(tenant_id=context.tenant_id, gdpr=True, recorded=True)}
239@app.get('/{projectId}/sessions/{sessionId}/errors/{errorId}/sourcemaps', tags=["sessions", "sourcemaps"])
240def get_error_trace(projectId: int, sessionId: int, errorId: str,
241 context: schemas.CurrentContext = Depends(OR_context)):
242 data = errors.get_trace(project_id=projectId, error_id=errorId)
243 if "errors" in data: 243 ↛ 245line 243 didn't jump to line 245 because the condition on line 243 was always true
244 return data
245 return {
246 'data': data
247 }
250@app.get('/{projectId}/errors/{errorId}', tags=['errors'])
251def errors_get_details(projectId: int, errorId: str, density24: int = 24, density30: int = 30,
252 context: schemas.CurrentContext = Depends(OR_context)):
253 data = errors_details.get_details(project_id=projectId, user_id=context.user_id, error_id=errorId,
254 **{"density24": density24, "density30": density30})
255 return data
258@app.get('/{projectId}/errors/{errorId}/sourcemaps', tags=['errors'])
259def errors_get_details_sourcemaps(projectId: int, errorId: str,
260 context: schemas.CurrentContext = Depends(OR_context)):
261 data = errors.get_trace(project_id=projectId, error_id=errorId)
262 if "errors" in data: 262 ↛ 264line 262 didn't jump to line 264 because the condition on line 262 was always true
263 return data
264 return {
265 'data': data
266 }
269# TODO: delete this if UI is not calling it
270@app.get('/delete_later/{projectId}/errors/{errorId}/sessions', tags=["errors"])
271def get_errors_sessions(projectId: int, errorId: str,
272 startDate: int = TimeUTC.now(-7), endDate: int = TimeUTC.now(),
273 context: schemas.CurrentContext = Depends(OR_context)):
274 return {
275 "data": errors.get_sessions(project_id=projectId, user_id=context.user_id, error_id=errorId,
276 start_date=startDate, end_date=endDate)}