Coverage for routers/subs/mcp.py: 62%

1from fastapi import Depends, Body, BackgroundTasks 

2from fastapi import HTTPException 

3 

4import schemas 

5from chalicelib.core.mcp import authorizers, tracer 

6from or_dependencies import OR_context 

7from routers.base import get_routers 

8 

9public_app, app, app_apikey = get_routers() 

10 

11 

12@app.post('/v1/mcp/authorize', tags=["mcp"]) 

13def authorize_mcp_app(background_tasks: BackgroundTasks, 

14 data: schemas.MCP.AuthorizeSchema = Body(...), 

15 context: schemas.CurrentContext = Depends(OR_context)): 

16 if not (context.email.endswith("asayer.io") or context.email.endswith("openreplay.com")): 16 ↛ 18line 16 didn't jump to line 18 because the condition on line 16 was always true

17 raise HTTPException(status_code=401, detail="Unauthorized") 

18 authorizers.store_token_request(data=data, cotext=context) 

19 background_tasks.add_task(tracer.store_client_id, 

20 user_id=context.user_id, 

21 client_id=data.client_id) 

22 return {"data": {"success": True}} 

23 

24 

25@public_app.get("/v1/mcp/auth-status", tags=["mcp"]) 

26def get_mcp_jwt(client_id: str, state: str): 

27 jwt = authorizers.get_token_by_state(client_id=client_id, state=state) 

28 if jwt is None: 

29 raise HTTPException(status_code=404, detail="Unauthorized") 

30 return {"jwt": jwt}