Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 6 additions & 9 deletions app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,9 @@

@app.get("/version")
async def _version():
"""
Returns the current version of the FastAPI application as a JSON object.
"""
return {"version": app.version}


Expand Down Expand Up @@ -97,15 +100,9 @@ async def ko_fi_webhook(data: str = Form(...)):
@app.websocket("/ws/{verification_token}")
async def websocket_endpoint(websocket: WebSocket, verification_token: str):
"""
Establishes a WebSocket connection with the client and forwards incoming
Ko-fi webhooks to the corresponding connection.

The endpoint expects a verification token as a path parameter, which is used
to identify the connection. The endpoint will keep the connection alive by
sending a "pong" response to the "ping" message sent by the client.

If the connection is closed, the endpoint will remove the connection from the
active connections dictionary.
Handles a WebSocket connection for receiving Ko-fi webhook notifications.

Establishes a WebSocket connection associated with the provided verification token, maintains connection health via ping/pong messages, and ensures proper cleanup when the connection is closed.
"""
await websocket.accept()
active_connections[verification_token].add(websocket)
Expand Down
13 changes: 11 additions & 2 deletions tests/test_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,11 @@ async def mock_send_json(*args, **kwargs):

@pytest.mark.asyncio
async def test_webhook_websocket_disconnect(client, sample_webhook_data):
"""Test webhook WebSocket disconnect."""
"""
Tests that when a WebSocket disconnect occurs during webhook processing, the connection is removed from active connections.

Opens a WebSocket connection, mocks the send_json method to raise a disconnect, posts webhook data, and verifies the connection is cleaned up.
"""
async def mock_send_json(*args, **kwargs):
raise WebSocketDisconnect()

Expand All @@ -167,7 +171,12 @@ async def mock_send_json(*args, **kwargs):


def test_version_endpoint(client):
"""Test the version endpoint."""
"""
Tests that the /version endpoint returns the correct application version.

Sends a GET request to /version and asserts the response status is 200 and the JSON
payload contains the expected version string.
"""
response = client.get("/version")
assert response.status_code == 200
assert response.json() == {"version": app.version}
Expand Down