Daniil
248 lines
8.0 KiB
Python
248 lines
8.0 KiB
Python
"""
|
|
Tests for webhooks endpoints.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from cpv3.modules.webhooks.models import Webhook
|
|
from cpv3.modules.users.models import User
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_webhook(test_db_session: AsyncSession, test_user: User) -> Webhook:
|
|
"""Create a test webhook owned by test_user."""
|
|
webhook = Webhook(
|
|
id=uuid.uuid4(),
|
|
user_id=test_user.id,
|
|
event="job.completed",
|
|
url="https://example.com/webhook",
|
|
secret="test-secret-123",
|
|
is_active=True,
|
|
)
|
|
test_db_session.add(webhook)
|
|
await test_db_session.commit()
|
|
await test_db_session.refresh(webhook)
|
|
return webhook
|
|
|
|
|
|
@pytest.fixture
|
|
async def other_webhook(test_db_session: AsyncSession, other_user: User) -> Webhook:
|
|
"""Create a webhook owned by another user."""
|
|
webhook = Webhook(
|
|
id=uuid.uuid4(),
|
|
user_id=other_user.id,
|
|
event="job.failed",
|
|
url="https://other.com/webhook",
|
|
secret="other-secret-456",
|
|
is_active=True,
|
|
)
|
|
test_db_session.add(webhook)
|
|
await test_db_session.commit()
|
|
await test_db_session.refresh(webhook)
|
|
return webhook
|
|
|
|
|
|
class TestListWebhooksEndpoint:
|
|
"""Tests for GET /api/webhooks/."""
|
|
|
|
async def test_list_webhooks(self, auth_client: AsyncClient, test_webhook: Webhook):
|
|
"""Test listing webhooks."""
|
|
response = await auth_client.get("/api/webhooks/")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
async def test_list_webhooks_unauthenticated(self, async_client: AsyncClient):
|
|
"""Test listing webhooks without auth returns 401."""
|
|
response = await async_client.get("/api/webhooks/")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestCreateWebhookEndpoint:
|
|
"""Tests for POST /api/webhooks/."""
|
|
|
|
async def test_create_webhook_success(self, auth_client: AsyncClient):
|
|
"""Test creating a webhook."""
|
|
response = await auth_client.post(
|
|
"/api/webhooks/",
|
|
json={
|
|
"event": "transcription.completed",
|
|
"url": "https://myapp.com/webhook",
|
|
"secret": "my-webhook-secret",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["event"] == "transcription.completed"
|
|
assert data["url"] == "https://myapp.com/webhook"
|
|
|
|
async def test_create_webhook_minimal(self, auth_client: AsyncClient):
|
|
"""Test creating a webhook with minimal fields."""
|
|
response = await auth_client.post(
|
|
"/api/webhooks/",
|
|
json={"url": "https://minimal.com/hook"},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["url"] == "https://minimal.com/hook"
|
|
|
|
async def test_create_webhook_unauthenticated(self, async_client: AsyncClient):
|
|
"""Test creating webhook without auth returns 401."""
|
|
response = await async_client.post(
|
|
"/api/webhooks/",
|
|
json={"url": "https://test.com/hook"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestRetrieveWebhookEndpoint:
|
|
"""Tests for GET /api/webhooks/{webhook_id}/."""
|
|
|
|
async def test_retrieve_own_webhook(
|
|
self, auth_client: AsyncClient, test_webhook: Webhook
|
|
):
|
|
"""Test retrieving own webhook."""
|
|
response = await auth_client.get(f"/api/webhooks/{test_webhook.id}/")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == str(test_webhook.id)
|
|
assert data["url"] == test_webhook.url
|
|
|
|
async def test_retrieve_other_webhook_as_staff(
|
|
self, staff_client: AsyncClient, test_webhook: Webhook
|
|
):
|
|
"""Test staff can retrieve any webhook."""
|
|
response = await staff_client.get(f"/api/webhooks/{test_webhook.id}/")
|
|
|
|
assert response.status_code == 200
|
|
|
|
async def test_retrieve_nonexistent_webhook(self, auth_client: AsyncClient):
|
|
"""Test retrieving nonexistent webhook returns 404."""
|
|
fake_id = uuid.uuid4()
|
|
response = await auth_client.get(f"/api/webhooks/{fake_id}/")
|
|
|
|
assert response.status_code == 404
|
|
assert response.json()["detail"] == "Not found"
|
|
|
|
async def test_retrieve_other_webhook_forbidden(
|
|
self, auth_client: AsyncClient, other_webhook: Webhook
|
|
):
|
|
"""Test regular user cannot retrieve other user's webhook."""
|
|
response = await auth_client.get(f"/api/webhooks/{other_webhook.id}/")
|
|
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"] == "Forbidden"
|
|
|
|
|
|
class TestPatchWebhookEndpoint:
|
|
"""Tests for PATCH /api/webhooks/{webhook_id}/."""
|
|
|
|
async def test_patch_own_webhook(
|
|
self, auth_client: AsyncClient, test_webhook: Webhook
|
|
):
|
|
"""Test updating own webhook."""
|
|
response = await auth_client.patch(
|
|
f"/api/webhooks/{test_webhook.id}/",
|
|
json={
|
|
"url": "https://updated.com/webhook",
|
|
"event": "job.started",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["url"] == "https://updated.com/webhook"
|
|
assert data["event"] == "job.started"
|
|
|
|
async def test_patch_webhook_deactivate(
|
|
self, auth_client: AsyncClient, test_webhook: Webhook
|
|
):
|
|
"""Test deactivating a webhook."""
|
|
response = await auth_client.patch(
|
|
f"/api/webhooks/{test_webhook.id}/",
|
|
json={"is_active": False},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["is_active"] is False
|
|
|
|
async def test_patch_other_webhook_as_staff(
|
|
self, staff_client: AsyncClient, test_webhook: Webhook
|
|
):
|
|
"""Test staff can update any webhook."""
|
|
response = await staff_client.patch(
|
|
f"/api/webhooks/{test_webhook.id}/",
|
|
json={"event": "staff.updated"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
async def test_patch_nonexistent_webhook(self, auth_client: AsyncClient):
|
|
"""Test patching nonexistent webhook returns 404."""
|
|
fake_id = uuid.uuid4()
|
|
response = await auth_client.patch(
|
|
f"/api/webhooks/{fake_id}/",
|
|
json={"url": "https://test.com"},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
async def test_patch_other_webhook_forbidden(
|
|
self, auth_client: AsyncClient, other_webhook: Webhook
|
|
):
|
|
"""Test regular user cannot update other user's webhook."""
|
|
response = await auth_client.patch(
|
|
f"/api/webhooks/{other_webhook.id}/",
|
|
json={"url": "https://hacked.com"},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
class TestDeleteWebhookEndpoint:
|
|
"""Tests for DELETE /api/webhooks/{webhook_id}/."""
|
|
|
|
async def test_delete_own_webhook(
|
|
self, auth_client: AsyncClient, test_webhook: Webhook
|
|
):
|
|
"""Test deleting own webhook."""
|
|
response = await auth_client.delete(f"/api/webhooks/{test_webhook.id}/")
|
|
|
|
assert response.status_code == 204
|
|
|
|
async def test_delete_other_webhook_as_staff(
|
|
self, staff_client: AsyncClient, test_webhook: Webhook
|
|
):
|
|
"""Test staff can delete any webhook."""
|
|
response = await staff_client.delete(f"/api/webhooks/{test_webhook.id}/")
|
|
|
|
assert response.status_code == 204
|
|
|
|
async def test_delete_nonexistent_webhook(self, auth_client: AsyncClient):
|
|
"""Test deleting nonexistent webhook returns 404."""
|
|
fake_id = uuid.uuid4()
|
|
response = await auth_client.delete(f"/api/webhooks/{fake_id}/")
|
|
|
|
assert response.status_code == 404
|
|
|
|
async def test_delete_other_webhook_forbidden(
|
|
self, auth_client: AsyncClient, other_webhook: Webhook
|
|
):
|
|
"""Test regular user cannot delete other user's webhook."""
|
|
response = await auth_client.delete(f"/api/webhooks/{other_webhook.id}/")
|
|
|
|
assert response.status_code == 403
|