Test connection to Haotian list_url using the provided authorization.
(
authorization: Optional[str] = Header(None),
request: HaotianTestConnectionRequest = Body(...),
)
| 61 | |
| 62 | @router.post("/test-connection") |
| 63 | async def test_haotian_connection_api( |
| 64 | authorization: Optional[str] = Header(None), |
| 65 | request: HaotianTestConnectionRequest = Body(...), |
| 66 | ) -> JSONResponse: |
| 67 | """ |
| 68 | Test connection to Haotian list_url using the provided authorization. |
| 69 | """ |
| 70 | _ = authorization |
| 71 | try: |
| 72 | ok, error_message = await test_haotian_connection_impl( |
| 73 | list_url=request.list_url, |
| 74 | external_authorization=request.authorization, |
| 75 | ) |
| 76 | if ok: |
| 77 | return JSONResponse( |
| 78 | status_code=HTTPStatus.OK, |
| 79 | content={"success": True, "message": "Connection successful"}, |
| 80 | ) |
| 81 | raise HTTPException( |
| 82 | status_code=HTTPStatus.BAD_REQUEST, |
| 83 | detail=f"Cannot connect to Haotian server: {error_message}", |
| 84 | ) |
| 85 | except HTTPException: |
| 86 | raise |
| 87 | except Exception as e: |
| 88 | logger.error(f"Error testing Haotian connection: {e}") |
| 89 | raise HTTPException( |
| 90 | status_code=HTTPStatus.INTERNAL_SERVER_ERROR, |
| 91 | detail=f"Error testing Haotian connection: {str(e)}", |
| 92 | ) |
nothing calls this directly
no test coverage detected