Aleksandr Meshchriakov
ee95628a0a
Some checks failed
CI/CD Pipeline / Run Tests (push) Failing after 37s
CI/CD Pipeline / Code Quality Checks (push) Failing after 43s
CI/CD Pipeline / Build & Push Images (push) Has been skipped
CI/CD Pipeline / Deploy (dev) (push) Has been skipped
CI/CD Pipeline / Deploy (prod) (push) Has been skipped
CI/CD Pipeline / Code Quality Checks (pull_request) Failing after 0s
CI/CD Pipeline / Run Tests (pull_request) Failing after 0s
CI/CD Pipeline / Build & Push Images (pull_request) Has been skipped
CI/CD Pipeline / Deploy (dev) (pull_request) Has been skipped
CI/CD Pipeline / Deploy (prod) (pull_request) Has been skipped
- Обновлены клиенты парсеров (checko, fns, minpromtorg, proverki, zakupki) - Добавлены новые миграции для моделей - Расширено покрытие тестами - Обновлены конфигурации и настройки проекта - Добавлены утилиты для тестирования Co-Authored-By: Warp <agent@warp.dev>
94 lines
3.3 KiB
Python
94 lines
3.3 KiB
Python
"""Tests for core admin configurations."""
|
|
|
|
from datetime import timedelta
|
|
|
|
from django.contrib.admin.sites import AdminSite
|
|
from django.contrib.messages.storage.fallback import FallbackStorage
|
|
from django.test import RequestFactory, TestCase
|
|
from django.utils import timezone
|
|
|
|
from apps.core.admin import BackgroundJobAdmin
|
|
from apps.core.models import BackgroundJob
|
|
from tests.apps.user.factories import UserFactory
|
|
from tests.utils.fixtures import fake
|
|
|
|
|
|
class CoreAdminTest(TestCase):
|
|
def setUp(self):
|
|
self.site = AdminSite()
|
|
self.admin = BackgroundJobAdmin(BackgroundJob, self.site)
|
|
self.user = UserFactory.create_superuser()
|
|
self.factory = RequestFactory()
|
|
|
|
def _request(self):
|
|
request = self.factory.get("/")
|
|
request.user = self.user
|
|
request.session = {}
|
|
request._messages = FallbackStorage(request)
|
|
return request
|
|
|
|
def test_task_name_short(self):
|
|
job = BackgroundJob.objects.create(
|
|
task_id=fake.uuid4(),
|
|
task_name="apps.parsers.tasks.parse_industrial_production",
|
|
)
|
|
self.assertEqual(self.admin.task_name_short(job), "parse_industrial_production")
|
|
job_short = BackgroundJob.objects.create(
|
|
task_id=fake.uuid4(),
|
|
task_name="short.task",
|
|
)
|
|
self.assertEqual(self.admin.task_name_short(job_short), "short.task")
|
|
|
|
def test_status_badge_and_progress(self):
|
|
job = BackgroundJob.objects.create(
|
|
task_id=fake.uuid4(),
|
|
task_name="test.task",
|
|
status="success",
|
|
progress=100,
|
|
)
|
|
badge = self.admin.status_badge(job)
|
|
bar = self.admin.progress_bar(job)
|
|
self.assertIn("span", str(badge))
|
|
self.assertIn("100%", str(bar))
|
|
|
|
def test_duration_display(self):
|
|
job = BackgroundJob.objects.create(
|
|
task_id=fake.uuid4(),
|
|
task_name="test.task",
|
|
)
|
|
self.assertEqual(self.admin.duration_display(job), "-")
|
|
job.started_at = timezone.now()
|
|
job.completed_at = job.started_at + timedelta(seconds=30)
|
|
job.save(update_fields=["started_at", "completed_at"])
|
|
self.assertIn("сек", self.admin.duration_display(job))
|
|
job.completed_at = job.started_at + timedelta(seconds=90)
|
|
job.save(update_fields=["started_at", "completed_at"])
|
|
self.assertIn("мин", self.admin.duration_display(job))
|
|
|
|
def test_permissions(self):
|
|
request = self._request()
|
|
self.assertFalse(self.admin.has_add_permission(request))
|
|
self.assertFalse(self.admin.has_change_permission(request))
|
|
|
|
def test_revoke_jobs_action_no_active(self):
|
|
BackgroundJob.objects.create(
|
|
task_id=fake.uuid4(),
|
|
task_name="test.task",
|
|
status="success",
|
|
)
|
|
request = self._request()
|
|
qs = BackgroundJob.objects.all()
|
|
self.admin.revoke_jobs(request, qs)
|
|
|
|
def test_revoke_jobs_action_pending(self):
|
|
job = BackgroundJob.objects.create(
|
|
task_id=fake.uuid4(),
|
|
task_name="test.task",
|
|
status="pending",
|
|
)
|
|
request = self._request()
|
|
qs = BackgroundJob.objects.all()
|
|
self.admin.revoke_jobs(request, qs)
|
|
job.refresh_from_db()
|
|
self.assertEqual(job.status, "revoked")
|