Aleksandr Meshchriakov
14196a7549
Some checks failed
CI/CD Pipeline / Run Tests (pull_request) Successful in 1m13s
CI/CD Pipeline / Code Quality Checks (pull_request) Successful in 1m18s
CI/CD Pipeline / Build Docker Images (pull_request) Failing after 2m46s
CI/CD Pipeline / Push to Gitea Registry (pull_request) Has been skipped
227 lines
8.0 KiB
Python
227 lines
8.0 KiB
Python
"""Tests for core views and API endpoints."""
|
|
|
|
from unittest.mock import patch
|
|
|
|
from apps.core.models import JobStatus
|
|
from apps.core.services import BackgroundJobService
|
|
from apps.user.services import UserService
|
|
from django.urls import reverse
|
|
from rest_framework import status
|
|
from rest_framework.test import APITestCase
|
|
|
|
from tests.apps.user.factories import UserFactory
|
|
|
|
|
|
class HealthCheckViewTest(APITestCase):
|
|
"""Tests for HealthCheckView"""
|
|
|
|
def test_health_check_url_reverse(self):
|
|
"""Test reverse URL resolution for health check"""
|
|
url = reverse("core:health")
|
|
self.assertEqual(url, "/health/")
|
|
|
|
def test_health_check_success(self):
|
|
"""Test health check returns healthy status"""
|
|
url = reverse("core:health")
|
|
response = self.client.get(url)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertIn("status", response.data)
|
|
self.assertIn("version", response.data)
|
|
self.assertIn("checks", response.data)
|
|
self.assertIn("database", response.data["checks"])
|
|
|
|
def test_health_check_database_up(self):
|
|
"""Test health check reports database as up"""
|
|
url = reverse("core:health")
|
|
response = self.client.get(url)
|
|
|
|
self.assertEqual(response.data["checks"]["database"]["status"], "up")
|
|
self.assertIn("latency_ms", response.data["checks"]["database"])
|
|
|
|
|
|
class LivenessViewTest(APITestCase):
|
|
"""Tests for LivenessView"""
|
|
|
|
def test_liveness_url_reverse(self):
|
|
"""Test reverse URL resolution for liveness"""
|
|
url = reverse("core:liveness")
|
|
self.assertEqual(url, "/health/live/")
|
|
|
|
def test_liveness_returns_alive(self):
|
|
"""Test liveness probe returns alive status"""
|
|
url = reverse("core:liveness")
|
|
response = self.client.get(url)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["status"], "alive")
|
|
|
|
|
|
class ReadinessViewTest(APITestCase):
|
|
"""Tests for ReadinessView"""
|
|
|
|
def test_readiness_url_reverse(self):
|
|
"""Test reverse URL resolution for readiness"""
|
|
url = reverse("core:readiness")
|
|
self.assertEqual(url, "/health/ready/")
|
|
|
|
def test_readiness_returns_ready(self):
|
|
"""Test readiness probe returns ready when DB is available"""
|
|
url = reverse("core:readiness")
|
|
response = self.client.get(url)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["status"], "ready")
|
|
|
|
|
|
class APIVersioningURLTest(APITestCase):
|
|
"""Tests for API versioning URL structure"""
|
|
|
|
def test_api_v1_user_register_reverse(self):
|
|
"""Test reverse URL for user registration"""
|
|
url = reverse("api_v1:user:register")
|
|
self.assertEqual(url, "/api/v1/users/register/")
|
|
|
|
def test_api_v1_user_login_reverse(self):
|
|
"""Test reverse URL for user login"""
|
|
url = reverse("api_v1:user:login")
|
|
self.assertEqual(url, "/api/v1/users/login/")
|
|
|
|
def test_api_v1_user_logout_reverse(self):
|
|
"""Test reverse URL for user logout"""
|
|
url = reverse("api_v1:user:logout")
|
|
self.assertEqual(url, "/api/v1/users/logout/")
|
|
|
|
def test_api_v1_user_current_user_reverse(self):
|
|
"""Test reverse URL for current user"""
|
|
url = reverse("api_v1:user:current_user")
|
|
self.assertEqual(url, "/api/v1/users/me/")
|
|
|
|
def test_api_v1_user_token_refresh_reverse(self):
|
|
"""Test reverse URL for token refresh"""
|
|
url = reverse("api_v1:user:token_refresh")
|
|
self.assertEqual(url, "/api/v1/users/token/refresh/")
|
|
|
|
def test_api_v1_user_password_change_reverse(self):
|
|
"""Test reverse URL for password change"""
|
|
url = reverse("api_v1:user:password_change")
|
|
self.assertEqual(url, "/api/v1/users/password/change/")
|
|
|
|
|
|
class BackgroundJobApiTest(APITestCase):
|
|
"""Tests for background job API access rules."""
|
|
|
|
def setUp(self):
|
|
self.user = UserFactory.create_user()
|
|
tokens = UserService.get_tokens_for_user(self.user)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {tokens['access']}")
|
|
|
|
def test_job_status_denies_other_user_job(self):
|
|
"""Test users cannot read jobs owned by another user."""
|
|
other_user = UserFactory.create_user()
|
|
job = BackgroundJobService.create_job(
|
|
task_id="task-other-user",
|
|
task_name="test.task",
|
|
user_id=other_user.id,
|
|
)
|
|
|
|
response = self.client.get(
|
|
reverse("api_v1:jobs:job-status", kwargs={"task_id": job.task_id})
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
|
|
def test_job_status_denies_system_job_for_regular_user(self):
|
|
"""Test user_id=NULL jobs are not visible to regular users."""
|
|
job = BackgroundJobService.create_job(
|
|
task_id="task-system",
|
|
task_name="test.task",
|
|
)
|
|
|
|
response = self.client.get(
|
|
reverse("api_v1:jobs:job-status", kwargs={"task_id": job.task_id})
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
|
|
def test_job_status_allows_system_job_for_staff(self):
|
|
"""Test staff can inspect system jobs."""
|
|
staff = UserFactory.create_user(is_staff=True)
|
|
tokens = UserService.get_tokens_for_user(staff)
|
|
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {tokens['access']}")
|
|
job = BackgroundJobService.create_job(
|
|
task_id="task-system-staff",
|
|
task_name="test.task",
|
|
)
|
|
|
|
response = self.client.get(
|
|
reverse("api_v1:jobs:job-status", kwargs={"task_id": job.task_id})
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
def test_job_list_rejects_invalid_limit(self):
|
|
"""Test job list validates limit query parameter."""
|
|
response = self.client.get(
|
|
reverse("api_v1:jobs:job-list"),
|
|
{"limit": "bad"},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
@patch("apps.core.views.current_app.control.revoke")
|
|
def test_job_control_revoke_marks_user_job_revoked(self, mock_revoke):
|
|
"""Test owner can revoke a running background job."""
|
|
job = BackgroundJobService.create_job(
|
|
task_id="task-revoke",
|
|
task_name="test.task",
|
|
user_id=self.user.id,
|
|
)
|
|
job.mark_started()
|
|
|
|
response = self.client.post(
|
|
reverse("api_v1:jobs:job-control", kwargs={"task_id": job.task_id}),
|
|
{"action": "revoke"},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
mock_revoke.assert_called_once_with(job.task_id, terminate=True)
|
|
job.refresh_from_db()
|
|
self.assertEqual(job.status, JobStatus.REVOKED)
|
|
|
|
def test_job_control_denies_finished_job(self):
|
|
"""Test finished jobs cannot be revoked again."""
|
|
job = BackgroundJobService.create_job(
|
|
task_id="task-finished",
|
|
task_name="test.task",
|
|
user_id=self.user.id,
|
|
)
|
|
job.complete()
|
|
|
|
response = self.client.post(
|
|
reverse("api_v1:jobs:job-control", kwargs={"task_id": job.task_id}),
|
|
{"action": "revoke"},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_job_stream_returns_completed_sse_event(self):
|
|
"""Test dev-compatible SSE job stream endpoint."""
|
|
job = BackgroundJobService.create_job(
|
|
task_id="task-stream",
|
|
task_name="test.task",
|
|
user_id=self.user.id,
|
|
)
|
|
job.complete(result={"ok": True})
|
|
|
|
response = self.client.get(
|
|
reverse("api_v1:jobs:job-stream", kwargs={"task_id": job.task_id})
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
content = b"".join(response.streaming_content).decode("utf-8")
|
|
self.assertIn("event: completed", content)
|
|
self.assertIn('"task_id": "task-stream"', content)
|